AutoGen(五) Human-in-the-Loop（人类在环）实战与进阶：多智能体协作与Web交互全流程（附代码）

《AutoGen Human-in-the-Loop 实战指南：多智能体协作与Web交互》摘要：本文深入探讨AutoGen框架中Human-in-the-Loop（HITL）机制的应用。HITL允许人类在多智能体协作过程中实时介入，解决需要人工审核、决策的复杂场景（如代码评审、企业审批等）。文章详细介绍了两种核心交互模式：运行时同步反馈（通过UserProxyAgent实现）和运行后异步反馈（通

qq_27492797

614人浏览 · 2025-06-27 22:36:23

qq_27492797 · 2025-06-27 22:36:23 发布

AutoGen Human-in-the-Loop（人类在环）实战与进阶：多智能体协作与Web交互全流程（附代码）

引言：AI自动化的极限与人类参与的价值

在大模型（LLM）驱动的AI应用开发中，完全自动化虽然高效，但在代码评审、内容审核、企业审批等复杂场景下，AI往往难以独立胜任。人类在环（Human-in-the-Loop, HITL）机制，让AI与人类协作，兼顾效率与安全，成为智能体系统落地的关键能力。

1. Human-in-the-Loop核心概念

定义：在多智能体团队运行过程中，允许人类实时或阶段性介入、反馈和控制。
适用场景：需要人工审核、确认、补充信息或决策的复杂任务，如代码评审、内容审核、智能问答、企业审批流等。
关键角色：
- UserProxyAgent（用户代理）：代表人类参与团队对话
- AssistantAgent（AI助手）：负责自动化任务

在这里插入图片描述

2. 典型交互方式

2.1 运行中交互（同步反馈）

通过 UserProxyAgent，团队运行时主动请求用户输入。
适合审批、确认、补充信息等即时反馈场景。

代码示例：控制台人类在环

from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
assistant = AssistantAgent("assistant", model_client=model_client)
user_proxy = UserProxyAgent("user_proxy", input_func=input)  # 控制台输入

termination = TextMentionTermination("APPROVE")
team = RoundRobinGroupChat([assistant, user_proxy], termination_condition=termination)

await Console(team.run_stream(task="Write a 4-line poem about the ocean."))
await model_client.close()

运行时，团队会在需要用户反馈时暂停，等待用户输入（如“APPROVE”）。

2.2 运行后交互（异步反馈）

团队运行结束后，用户/应用基于结果反馈，再次运行团队。
适合多轮对话、任务分批处理、持久化会话等场景。

代码示例：max_turns 控制多轮反馈

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
assistant = AssistantAgent("assistant", model_client=model_client)
team = RoundRobinGroupChat([assistant], max_turns=1)

task = "Write a 4-line poem about the ocean."
while True:
    await Console(team.run_stream(task=task))
    task = input("Enter your feedback (type 'exit' to leave): ")
    if task.lower().strip() == "exit":
        break
await model_client.close()

3. 进阶用法：WebSocket/前端集成

在实际业务中，用户输入往往来自Web前端、App等。AutoGen的 UserProxyAgent支持自定义 input_func，可对接WebSocket、API、UI等多端输入。

代码片段：FastAPI+WebSocket集成

@app.websocket("/ws/chat")
async def chat(websocket: WebSocket):
    await websocket.accept()

    async def _user_input(prompt: str, cancellation_token: CancellationToken | None) -> str:
        data = await websocket.receive_json() # 等待前端消息
        message = TextMessage.model_validate(data)
        return message.content

    user_proxy = UserProxyAgent("user_proxy", input_func=_user_input)
    # 创建团队并运行
    # ...