进阶 📋 7 个步骤 第 454 / 455 篇

用 AutoGen(AG2)搭多智能体对话框架:研究员 + 审校协同写报告

微软 AG2(AutoGen)把多智能体协作抽象成「可对话的 Agent + 群聊编排」。本文用 AgentChat API 搭一个研究员 + 审校的双 Agent 群聊,自动产出带引用的调研报告,覆盖安装、Agent 定义、RoundRobin / Selector 编排与终止条件。

2026.09.19· 23 分钟阅读· 约 1133 字· 🤝 AutoGen / 🤖 多智能体

单人 Agent 写报告常有两个毛病:要么事实没核对就开写,要么写完没人挑刺。多智能体协作的解法很简单——让一个 Agent 专门「生产」,另一个专门「审校」,两人像在群里对话一样来回改,直到审校点头。微软的 AG2(AutoGen 的新包名)把这套「群聊编排」做得非常顺手:定义好几个 Agent,交给一个 Team 按顺序或按模型选择来调度即可。

🤝 学完你能做到:① 用 AG2 的 AgentChat API 定义多个角色 Agent;② 用 RoundRobinGroupChat 让它们轮流发言;③ 用终止条件控制何时停;④ 用 SelectorGroupChat 让模型自己选下一个发言者;⑤ 把最终结论落地成文件。

先搞懂:单 Agent vs 群聊 Team

方式适合AG2 对应
单个 AssistantAgent一问一答、简单任务AssistantAgent
固定顺序轮流「生产→审校」这类明确流水线RoundRobinGroupChat
模型自选发言者角色多、谁该说不确定SelectorGroupChat

AG2 不是 pyautogen。旧版包名叫 pyautogen、旧 API 是 autogen.GroupChat;新版统一为 ag2 包 + autogen_agentchat 命名空间。本文用新版。若你机器上装的是老版,import 会完全对不上——务必按下面 Step 1 装 ag2

Step 1:安装与新版 import

1 装 ag2 并确认入口
pip install ag2

python -c "import autogen_agentchat; print('ok')"

本文用到的几个入口:

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat, SelectorGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
💡 需要 OpenAI Key:export OPENAI_API_KEY=sk-...。AG2 也支持其它模型后端(通过对应的 ChatCompletionClient),但本文示例用 OpenAI 兼容接口。

Step 2:定义两个角色 Agent

2 研究员负责写,审校负责挑
import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient

model = OpenAIChatCompletionClient(model="gpt-4o-mini")

researcher = AssistantAgent(
    name="researcher",
    model_client=model,
    system_message="你是研究员。针对给定主题,给出 3 个关键事实,并尽量标注可查证来源。只输出内容,不要自我评价。",
)

reviewer = AssistantAgent(
    name="reviewer",
    model_client=model,
    system_message="你是审校。检查研究员内容的事实与逻辑。有问题就指出并给修改建议;没问题就只回复 APPROVE。",
)

Step 3:用 RoundRobinGroupChat 让它们轮流发言

3 建 Team 并跑起来(流式)
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination

team = RoundRobinGroupChat(
    [researcher, reviewer],
    termination_condition=TextMentionTermination("APPROVE"),
)

async def main():
    stream = team.run_stream(task="写一段关于「Agent 记忆」的 200 字简介")
    async for event in stream:
        print(event)

asyncio.run(main())

运行后你会看到:研究员先发一版 → 审校指出问题或回 APPROVE → 若没通过,轮流再来一轮,直到审校说 APPROVE 才停。

🔑 这就是多智能体协作的精髓:终止条件决定「何时收工」。没有它,两个 Agent 会无限互聊。

Step 4:设置终止条件(防死循环)

4 关键词终止 + 最大轮数兜底
from autogen_agentchat.conditions import MaxMessageTermination

# 两个条件「或」:要么审校说 APPROVE,要么最多 6 条消息后强制停
term = TextMentionTermination("APPROVE") | MaxMessageTermination(6)
team = RoundRobinGroupChat([researcher, reviewer], termination_condition=term)

务必给最大轮数兜底。万一审校永远不写 APPROVE(比如研究员反复答非所问),只有关键词终止会让群聊卡到模型超时/烧钱。MaxMessageTermination 是保命开关。

Step 5:用 SelectorGroupChat 让模型自己选下一个发言者

5 角色多时用「选择器」更省

当角色超过两个、且「谁该接着说」不确定时,让一个模型来判断更自然:

from autogen_agentchat.teams import SelectorGroupChat

selector_team = SelectorGroupChat(
    [researcher, reviewer],
    model_client=model,                 # 用一个小模型做「调度决策」
    termination_condition=term,
    selector_prompt="根据当前对话,选择下一个最该发言的 Agent(返回其名字)。",
)
💡 Selector 的调度决策本身也要花 token。角色少、流程固定时,RoundRobinGroupChat 更简单更便宜;角色多、流程灵活时再上 SelectorGroupChat

Step 6:把结果落地成文件

6 取最后一条消息写盘
async def main():
    result = await team.run(task="写一段关于 Agent 记忆的 200 字简介")
    # result.messages 是全部发言;取最后一条作为定稿
    final = result.messages[-1].content
    with open("report.md", "w", encoding="utf-8") as f:
        f.write(final)
    print("已写入 report.md")

run 还是 run_stream?run() 等全部结束返回 TaskResult(方便取 .messages 落盘);run_stream() 边跑边吐事件(适合打印进度)。落盘用 run() 更方便。

常见问题与避坑

现象原因 & 解决
import autogen_agentchat 失败装的是旧 pyautogen;卸载后 pip install ag2
群聊停不下来没设 / 没生效终止条件;加 MaxMessageTermination 兜底
报模型 Key 错误未设 OPENAI_API_KEY;或换后端时 ChatCompletionClient 用错
Agent 互相复读system_message 没划清职责;明确「只输出内容 / 只审不写」
← 返回教程中心