单人 Agent 写报告常有两个毛病:要么事实没核对就开写,要么写完没人挑刺。多智能体协作的解法很简单——让一个 Agent 专门「生产」,另一个专门「审校」,两人像在群里对话一样来回改,直到审校点头。微软的 AG2(AutoGen 的新包名)把这套「群聊编排」做得非常顺手:定义好几个 Agent,交给一个 Team 按顺序或按模型选择来调度即可。
先搞懂:单 Agent vs 群聊 Team
| 方式 | 适合 | AG2 对应 |
|---|---|---|
| 单个 AssistantAgent | 一问一答、简单任务 | AssistantAgent |
| 固定顺序轮流 | 「生产→审校」这类明确流水线 | RoundRobinGroupChat |
| 模型自选发言者 | 角色多、谁该说不确定 | SelectorGroupChat |
AG2 不是 pyautogen。旧版包名叫 pyautogen、旧 API 是 autogen.GroupChat;新版统一为 ag2 包 + autogen_agentchat 命名空间。本文用新版。若你机器上装的是老版,import 会完全对不上——务必按下面 Step 1 装 ag2。
Step 1:安装与新版 import
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
export OPENAI_API_KEY=sk-...。AG2 也支持其它模型后端(通过对应的 ChatCompletionClient),但本文示例用 OpenAI 兼容接口。Step 2:定义两个角色 Agent
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 让它们轮流发言
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 才停。
Step 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 让模型自己选下一个发言者
当角色超过两个、且「谁该接着说」不确定时,让一个模型来判断更自然:
from autogen_agentchat.teams import SelectorGroupChat
selector_team = SelectorGroupChat(
[researcher, reviewer],
model_client=model, # 用一个小模型做「调度决策」
termination_condition=term,
selector_prompt="根据当前对话,选择下一个最该发言的 Agent(返回其名字)。",
)
RoundRobinGroupChat 更简单更便宜;角色多、流程灵活时再上 SelectorGroupChat。Step 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 没划清职责;明确「只输出内容 / 只审不写」 |