Autogen GroupChat 实战：搞定多智能体协作

2025-04-03 22:54:25

搞定 Autogen 多智能体协作：GroupChat 实战指南

用 Autogen 的时候，你是不是也遇到了这样的情况：明明定义了好几个智能体（Agent），比如一个写代码的 Coder，一个专门挑刺的 Critic，还有一个负责用户交互的 UserProxy，结果跑起来却发现只有 UserProxy 和 Coder 在聊，那个 Critic 就跟隐身了一样，完全不参与对话？

就像下面这个 Stack Overflow 上的问题一样：

问题： How can I get multiple autogen agents to work together? [closed]

**** ... 我正在尝试让多个 autogen 智能体一起沟通，包括 user_proxy, coder 和 critic。然而，只有 user_proxy 和 coder 在沟通...

(问题附带了定义 user_proxy, coder, critic 以及使用 user_proxy.initiate_chat(coder, critic, message=task) 的代码)

代码看起来没毛病啊？Agent 都定义了，initiate_chat 也把 Critic 加进去了，为什么它就是不说话呢？

别急，这个问题挺常见的。这篇就来帮你把这事儿弄明白，并给出能直接跑起来的解决方案。

一、为什么 Critic “不见了”？

问题主要出在 initiate_chat 这个方法的使用上。

当你像这样调用：

user_proxy.initiate_chat(
    coder,
    critic, # <--- 你以为加上它就能参与对话
    message=task
)

autogen 默认的行为，尤其是对于 UserProxyAgent 来说，initiate_chat 主要是用来启动 与另一个智能体 的对话。在这里，它首先看到的是 coder，于是就启动了 user_proxy 和 coder 之间的对话流程。

虽然你把 critic 作为参数传了进去，但在这个基础的 initiate_chat 调用里，它并没有一个内置的机制去理解 “coder 说完话，应该轮到 critic 说，然后 critic 说完，再看情况轮到谁” 这种多方轮流对话的逻辑。critic 就这样被“晾”在一边了，压根没机会插话。

简单说，initiate_chat(agent_A, agent_B, ...) 并不直接意味着一个包含所有列出智能体的群聊会自动开始并按某种规则轮流发言。你需要一个更明确的方式来组织这种多智能体协作。

二、解决方案：引入 GroupChat 和 GroupChatManager

要让多个 Agent 有条不紊地协作，autogen 提供了专门的“群聊”机制：GroupChat 和 GroupChatManager。

GroupChat : 想象成一个聊天室。它定义了这个聊天室里有哪些成员（Agents）以及管理聊天记录。
GroupChatManager : 这就是聊天室的管理员或主持人。它负责根据一定的规则（比如轮流发言、自动判断下一个谁发言等）来引导对话的进行，决定接下来应该由哪个 Agent 发言。

用这个组合，就能轻松实现 Coder 写代码 -> Critic 评价 -> Coder 修改 -> ... 这样的工作流。

怎么用 GroupChat 和 GroupChatManager？

下面我们来改造一下提问者代码，用上群聊机制：

保持 Agent 定义不变 : user_proxy, coder, critic 的定义可以基本保持原样。它们的角色和系统提示（System Message）决定了它们各自的功能。
创建 Agent 列表 : 把需要参与群聊的 Agent 都放进一个 Python 列表里。
实例化 GroupChat : 用这个 Agent 列表和一些配置（比如最大对话轮数 max_round）来创建一个 GroupChat 对象。
实例化 GroupChatManager : 用创建好的 GroupChat 对象和 llm_config 来初始化一个 GroupChatManager。这个 Manager 将使用大模型来理解对话内容并决定发言顺序（如果设置为自动模式）。
用 initiate_chat 启动群聊 : 这次不是 user_proxy 对某个特定 Agent 启动对话，而是 user_proxy （作为群聊发起者和参与者）对 GroupChatManager （群聊主持人）发起对话。

代码示例

我们来整合一下，完整的代码大概是这样：

import autogen
import os
# 建议使用环境变量来管理 API Key，更安全
# from dotenv import load_dotenv
# load_dotenv()
# api_key = os.getenv("OPENAI_API_KEY")

# 如果你没有设置环境变量，可以直接用下面的方式，但不推荐用于生产环境
# 请确保你的 Key 安全
api_key = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" # 替换成你自己的 Key

# 检查 API Key 是否设置
if not api_key:
    raise ValueError("请设置 OPENAI_API_KEY 环境变量或直接在代码中提供 API Key")

config_list = [
    {
        'model': 'gpt-4-0613', # 或者 gpt-3.5-turbo 等
        'api_key': api_key,
    }
]

llm_config={
    "request_timeout": 600,
    "seed": 42, # 用于缓存和可复现性
    "config_list": config_list,
    "temperature": 0, # 温度设为 0，让输出更稳定
}

# --- Agent 定义 (基本保持不变) ---

# Coder Agent: 负责编写代码
coder = autogen.AssistantAgent(
    name="Coder",
    llm_config=llm_config,
    system_message="""Coder, you are a helpful assistant. Write Python code to fulfill the user's request.
    Ensure the code is runnable and addresses all aspects of the task.
    Add comments to explain key parts of the code.
    Generate the full code implementation.
    """,
)

# Critic Agent: 负责评估代码质量
critic = autogen.AssistantAgent(
    name="Critic",
    system_message="""Critic. You are a helpful assistant highly skilled in evaluating the quality of Python code, especially for games or visualizations.
    Evaluate the given code based on the following criteria by providing a score from 1 to 10 for each:
    - Bugs (bugs): Are there bugs, logic errors, syntax errors or typos? Any reason it might fail? How to fix it?
    - Readability (readability): Is the code easy to understand? Is it well-commented?
    - Functionality (functionality): Does the code implement the requested features correctly?
    - Efficiency (efficiency): Is the code reasonably efficient? Any obvious performance bottlenecks?
    - Goal Compliance (compliance): How well does the code meet ALL the specified goals in the original request?

    You MUST provide a score dictionary for these dimensions. Example: {bugs: 8, readability: 7, functionality: 9, efficiency: 8, compliance: 9}
    Do not suggest code examples yourself. Just point out issues and areas for improvement.
    Finally, based on the critique above, suggest a concrete LIST of actionable improvements for the Coder. If the code is perfect, say "No improvements needed."
    """,
    llm_config=llm_config,
)

# User Proxy Agent: 代表用户，执行代码，控制流程
user_proxy = autogen.UserProxyAgent(
    name="User_Proxy",
    human_input_mode="TERMINATE", # 当需要人介入或任务完成时，输入 'TERMINATE' 结束
    max_consecutive_auto_reply=10, # 增加自动回复次数，允许更长的 Agent 间对话
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={
        "work_dir": "coding_sandbox", # 指定代码执行的工作目录
        "use_docker": False,  # 设置为 False 则在本地执行，True 则需要 Docker 环境
    },
    llm_config=llm_config, # 虽然主要执行代码，有时也需要 LLM 理解对话
    system_message="""Reply TERMINATE if the task has been solved to your full satisfaction (e.g., code runs correctly and meets all requirements).
    Otherwise, reply CONTINUE, or briefly state why the task is not solved yet (e.g., 'Code has bugs', 'Feature missing').
    You are the primary initiator and executor."""
)

# --- 创建 GroupChat ---

# 定义群聊成员列表
agents = [user_proxy, coder, critic]

# 创建 GroupChat 对象
groupchat = autogen.GroupChat(
    agents=agents,
    messages=[], # 初始消息列表为空
    max_round=15 # 设置群聊最大轮数，防止无限循环
)

# 创建 GroupChatManager (群聊管理员)
manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config,
    # system_message 可以不设置，默认会根据 groupchat 的 agents 自动生成
    # 如果需要更精细的控制，可以设置 system_message 指导 Manager 如何选择下一个发言者
    # 例如："Manage the conversation between Coder, Critic, and User_Proxy. Prioritize Coder to implement, then Critic to review. User_Proxy executes code and provides final feedback."
)

# --- 发起群聊 ---

# 初始任务
task = """
Build a simple snake game using Python's Pygame library.
The game should have:
1. A snake controlled by arrow keys.
2. Food that appears randomly. When the snake eats food, it grows longer.
3. A score counter displayed at the top-left corner. Initialize score to 0.
4. Eating regular food increases the score by 1 point.
5. Every 10 points (score 10, 20, 30, etc.), a special flashing 'bonus' food item should appear at a random location for 5 seconds.
6. If the snake eats the bonus food within 5 seconds, the score increases by 5 points. The bonus food then disappears. If not eaten in 5 seconds, it disappears.
7. The game ends if the snake hits the window boundaries or itself. Display 'Game Over' and the final score.
Provide the complete, runnable Python code.
"""

# 由 User Proxy 发起对 GroupChatManager 的聊天
user_proxy.initiate_chat(
    manager, # 注意！这里是 manager，不是 coder 或 critic
    message=task
)

运行这段代码会发生什么？

user_proxy 向 manager 发起对话，并给出初始任务 task。
manager (利用 llm_config 里的大模型能力) 分析任务和当前的对话状态（刚开始）。它会判断，这个任务是写代码，最适合交给 Coder。
manager 将任务分配给 coder。coder 收到任务后，生成代码。
coder 发言（输出代码）。
manager 看到 coder 提交了代码。根据它的理解（或者预设规则），接下来应该让 Critic 评价代码。
manager 将 coder 的代码和原始任务信息传递给 critic。critic 进行评估，输出评分和改进建议。
critic 发言。
manager 看到 critic 的评价。可能会判断需要 coder 根据评价修改代码，于是把 critic 的反馈转给 coder。或者，如果 critic 的评价很好，且代码可以执行，它可能会让 user_proxy 尝试执行代码。
user_proxy 收到代码（通常是从 coder 或 manager 那里），尝试在 code_execution_config 指定的环境中执行。
user_proxy 根据执行结果和任务要求，判断任务是否完成。如果没完成（比如代码有错、功能不全），它会回复 CONTINUE 或具体原因。如果完成了，回复 TERMINATE。
这个 “执行 -> 评价 -> 修改 -> 再执行” 的循环会持续下去，直到 user_proxy 认为任务满意并回复 TERMINATE，或者达到 groupchat 设置的 max_round 上限。

这样，coder 和 critic 就能按照我们期望的流程，在 GroupChatManager 的协调下参与到对话中了。

安全建议

API Key 安全 ：千万不要把你的 OpenAI API Key 直接硬编码在代码里，尤其如果代码需要共享或上传到 Git 仓库。推荐使用环境变量（如 .env 文件配合 python-dotenv 库）或者专门的密钥管理服务来存储和加载 API Key。
```
# Example using python-dotenv
import os
from dotenv import load_dotenv

load_dotenv() # Loads variables from .env file into environment variables
api_key = os.getenv("OPENAI_API_KEY")

config_list = [{'model': 'gpt-4-0613', 'api_key': api_key}]
```

进阶使用技巧

发言人选择模式 (speaker_selection_method) : GroupChatManager 有一个重要参数 speaker_selection_method，可以控制如何选择下一个发言人：
- "auto" (默认): Manager 使用 LLM 来根据对话历史和 Agent 的决定下一个发言者。最智能但也可能最慢、最贵。
- "manual": 需要人工（通过 human_input_mode 为 ALWAYS 的 UserProxyAgent）明确指定下一个发言者。
- "random": 随机选择一个 Agent。
- "round_robin": 按 Agent 在列表中的顺序轮流发言。简单可预测，适用于流程固定的场景。
你可以在创建 GroupChatManager 时指定：
```
manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config,
    speaker_selection_method="round_robin" # 或者 "auto", "manual", "random"
)
```
对于 Coder -> Critic -> UserProxy -> Coder... 这样的流程，"round_robin" 或许是个简单有效的选择。但如果需要更复杂的判断（比如 Critic 说没问题了就跳过 Coder 直接给 UserProxy 执行），"auto" 可能更合适。
定制 GroupChatManager 的决策逻辑 : 如果 "auto" 模式不够精确，你还可以通过设置 GroupChatManager 的 system_message 来更明确地指导它如何选择发言人，甚至可以继承 GroupChatManager 类并重写 _select_speaker 方法来实现完全自定义的逻辑。
调整 Agent 的 system_message : system_message 对 Agent 的行为至关重要。精心设计每个 Agent 的 system_message，明确它们的职责、能力、输出格式要求，能显著提高多智能体协作的效率和效果。比如，在 Critic 的提示里强调“必须提供评分字典”，在 Coder 的提示里要求“提供完整的可运行代码”。
代码执行配置 (code_execution_config) : 注意 UserProxyAgent 的 code_execution_config。
- work_dir: 指定代码文件保存和执行的目录。确保这个目录存在且有写入权限。使用独立目录（如 coding_sandbox）是个好习惯，避免污染项目主目录。
- use_docker: 如果设为 True，autogen 会尝试在 Docker 容器中执行代码。这提供了更好的隔离性和环境一致性，但需要本地安装并运行 Docker。如果为 False，代码将在本地直接执行，需要确保本地环境有所需的库（比如例子中的 pygame）。初次尝试或简单任务设为 False 通常更方便。