AutoGen 通过对话驱动实现更自然的 Agent 合作

你在上一章用 LangGraph 搭建了一个精确的状态机，Agent 按照你预设的节点流转、决策、执行。这种确定性带来了可预测性，但当你面对需要两个 Agent 自由碰撞、互相纠错的任务时——比如让它们讨论技术方案、挑战彼此的假设——状态图反而成了束缚。你会想：“如果它们能像人类团队那样，在对话中自然协调，不需要我预先编排就好了。”

这就是 AutoGen 的价值所在。2023 年 8 月，微软研究院发布了 AutoGen 论文，提出“对话即协议”的理念：多个 Agent 不需要显式的管道和路由，它们通过消息传递、轮流发言、自主决策，就能在对话过程中涌现出合作行为。截至 2025 年底，AutoGen 已经演进到稳定的 API 版本（0.7+），支持异步对话、工具调用、代码执行等核心功能，成为探索性多 Agent 任务的利器。

本文会带你用 Python 代码实操，理解 AutoGen 的对话驱动哲学，并与 LangGraph 的确定性控制形成对比。你将看到：何时需要精确的状态机，何时可以让 Agent 自由对话。

你需要什么

总时间投入：约 60 分钟（包括理解概念 + 运行代码 + 调试踩坑）。

最终成果

你将构建一个双 Agent 对话系统：

• UserProxyAgent 扮演“人类代表”，负责提出需求、执行代码、反馈结果。
• AssistantAgent 扮演“技术顾问”，负责生成方案、编写代码、解释结果。

两个 Agent 通过多轮对话，完成一个数据分析任务：读取 CSV 数据、计算统计指标、生成可视化图表。整个过程由对话驱动，没有预先编排的流程图。

为什么做这个：这是探索性任务的典型场景——你有一个模糊的需求（“帮我分析这份数据”），需要 Agent 自主决定分析思路、编写代码、查看结果、调整策略。对话式协作天然适合这种“边做边想”的模式。

步骤 1：安装依赖并配置环境

首先安装所需包：

pip install autogen-agentchat autogen-ext[openai]

预期结果：安装成功，pip list 中可以看到 autogen-agentchat 和 autogen-ext。

接下来设置 API Key。在终端中设置环境变量：

export OPENAI_API_KEY="sk-..."

或者在 Python 脚本中直接传入（不推荐在生产环境中硬编码 Key）。

⚠️ 注意踩坑：截至 2026 年 6 月，AutoGen 已经进入维护模式，不再添加新功能。微软官方建议新项目使用 Microsoft Agent Framework (MAF)，但 AutoGen 作为学习工具和现有项目框架仍然可用。如果你准备开始一个全新生产项目，请评估 MAF 或继续使用 AutoGen 的稳定版本。本文代码基于 autogen-agentchat 0.7+，与早期 0.2 版本不兼容，不要混用旧版 API。

步骤 2：创建第一个 AssistantAgent 并测试对话

我们先创建单个 Agent，理解 AutoGen 的异步对话模式。打开 Python 文件 step2_hello.py：

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main():
    # 初始化模型客户端，指定使用 gpt-4o
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    # 创建 AssistantAgent，它本质是一个可对话的 LLM 封装
    agent = AssistantAgent(
        name="assistant",  # Agent 的唯一标识
        model_client=model_client,
        system_message="你是一个友好的助手，用中文简短回答。"
    )

    # 调用 run() 发起一次对话式任务
    result = await agent.run(task="什么是'对话即协议'？请用一两句话解释。")
    print(result)

# 运行异步主函数
asyncio.run(main())

运行这段代码，你会看到类似输出：

对话即协议是一种多Agent协作范式，Agent之间通过消息对话来协调任务分工和执行，
而不是依赖显式的、预先定义的控制流。

预期结果：Agent 成功响应，返回了与你提问相关的中文回答。

发生了什么：

1. OpenAIChatCompletionClient 封装了模型调用，AssistantAgent 通过它与 GPT-4o 交互。
2. agent.run(task=...) 将任务字符串转换为一次对话消息，发送给 LLM，等待回复。
3. 整个过程是异步的，await 确保你在收到完整结果前不会继续执行。

这是 AutoGen 对话驱动模式的最小单元：一个 Agent 就是一个“对话参与者”。后面我们会看到，当多个 Agent 互相发送消息时，合作就自然产生了。

步骤 3：定义 UserProxyAgent 并启动双 Agent 对话

单 Agent 只是回答问题，真正的多 Agent 合作需要至少两个角色。我们引入 UserProxyAgent，它代表“人类”一方，负责发起任务、执行代码、反馈结果。AssistantAgent 则负责思考方案、生成代码。

创建文件 step3_two_agents.py：

import asyncio
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main():
    # 创建模型客户端
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    # 1. 定义 AssistantAgent：负责生成方案和代码
    assistant = AssistantAgent(
        name="assistant",
        model_client=model_client,
        system_message="""你是一个数据分析专家。当收到用户请求时：
1. 先描述你的分析思路
2. 编写 Python 代码
3. 如果用户反馈了执行结果，解读结果并给出下一步建议。
始终保持对话式、协作的语气。"""
    )

    # 2. 定义 UserProxyAgent：代表人类，负责执行代码
    user_proxy = UserProxyAgent(
        name="user_proxy",
        description="人类代表，执行代码并反馈结果",
        code_execution_config={
            "work_dir": "coding",       # 代码执行的工作目录
            "use_docker": False        # 不使用 Docker，直接本地执行
        }
    )

    # 3. 创建对话团队：使用轮询模式，让两个 Agent 轮流发言
    team = RoundRobinGroupChat(
        [assistant, user_proxy],
        max_turns=10  # 最多 10 轮对话，防止无限循环
    )

    # 4. 发起对话：给出一个需要编程的任务
    task = """
    请完成以下数据分析任务：
    1. 生成一个包含 100 个随机数的 numpy 数组
    2. 计算均值、标准差、最大值、最小值
    3. 打印这些统计结果
    请编写并执行 Python 代码。
    """

    # 运行对话流，并打印每一轮消息
    stream = team.run_stream(task=task)
    async for message in stream:
        # 消息对象包含 source（谁发的）、content（内容）
        print(f"\n[{message.source}]: {message.content}")
        print("-" * 60)

asyncio.run(main())

预期结果：你会看到一系列消息流转：

1. [user_proxy] 先发出任务（框架自动将 task 作为 user_proxy 的首条消息）。
2. [assistant] 响应，描述分析思路，并生成 Python 代码。
3. [user_proxy] 自动执行代码（因为 assistant 的消息中包含代码块），并反馈执行结果。
4. [assistant] 解读结果，可能给出进一步建议。
5. 对话持续到任务完成或达到 max_turns。

典型的输出片段：

[user_proxy]: 请完成以下数据分析任务：
1. 生成一个包含 100 个随机数的 numpy 数组
2. 计算均值、标准差、最大值、最小值
3. 打印这些统计结果
请编写并执行 Python 代码。
------------------------------------------------------------

[assistant]: 好的，我们先生成 100 个随机数并计算统计指标。
我先编写代码：
```python
import numpy as np

data = np.random.randn(100)
mean = np.mean(data)
std = np.std(data)
max_val = np.max(data)
min_val = np.min(data)

print(f"均值: {mean:.4f}")
print(f"标准差: {std:.4f}")
print(f"最大值: {max_val:.4f}")
print(f"最小值: {min_val:.4f}")

均值: 0.0342
标准差: 1.0211
最大值: 2.8912
最小值: -2.4511

[assistant]: 从结果看，这 100 个随机数近似标准正态分布...

**发生了什么**：
- `RoundRobinGroupChat` 是 AutoGen 内置的对话协调器。它按照列表顺序轮流给每个 Agent 发言权，直到某个 Agent 选择不发言或达到最大轮数。
- `UserProxyAgent` 默认会检测消息中的代码块（\`\`\`python），并自动执行。执行结果作为新消息追加到对话中。
- `assistant` 收到执行结果后，可以基于结果生成新的分析或建议——这就是“对话驱动协作”的核心：没有预设流程图，Agent 根据对话历史自主决定下一步。

> **⚠️ 注意踩坑**：
> - 如果你在 Windows 上运行，确保 Python 环境中有 NumPy，否则代码执行会失败。UserProxyAgent 会捕捉错误并反馈给助理 Agent，触发纠错对话。
> - `use_docker=False` 表示在本地执行代码。如果你的环境支持 Docker，可以设为 True 以提高安全性。但在本地测试时，False 更方便。
> - `max_turns` 是重要的安全阀。如果没有限制，两个 Agent 可能陷入“讨论循环”——assistant 反复建议更多分析，user_proxy 反复执行。在实际任务中，建议根据复杂度设置 5-20 轮。

## 步骤 4：集成工具箱——让 Agent 在对话中调用 Python 函数

对话不只是聊天，真正的合作需要 Agent 能“动手”。AutoGen 允许将自定义 Python 函数注册为工具，Agent 可以在对话中自动调用它们。

创建文件 `step4_tools.py`，我们给 AssistantAgent 添加一个能计算简单统计函数的工具：

```python
import asyncio
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.tools import FunctionTool

# 定义两个可被调用的工具函数
def calculate_summary(numbers: list[float]) -> dict:
    """计算一组数字的统计摘要"""
    import numpy as np
    arr = np.array(numbers)
    return {
        "mean": float(np.mean(arr)),
        "std": float(np.std(arr)),
        "min": float(np.min(arr)),
        "max": float(np.max(arr)),
        "count": len(numbers)
    }

def generate_sample_data(size: int = 50) -> list[float]:
    """生成指定大小的随机样本数据"""
    import numpy as np
    return list(np.random.randn(size))

async def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    # 将函数包装为 AutoGen 工具
    summary_tool = FunctionTool(
        func=calculate_summary,
        description="接收一个数字列表，返回均值、标准差、最小值、最大值和计数"
    )
    data_tool = FunctionTool(
        func=generate_sample_data,
        description="生成指定大小的随机样本数据，返回浮点数列表"
    )

    # 创建助理 Agent，并注册工具
    assistant = AssistantAgent(
        name="assistant",
        model_client=model_client,
        tools=[summary_tool, data_tool],  # 工具列表
        system_message="你是一个数据分析助理。当用户需要数据时，使用 generate_sample_data 生成；"
                       "当用户需要统计时，使用 calculate_summary 计算。用中文回复。"
    )

    # UserProxyAgent 不需要代码执行能力，只负责发出指令
    user_proxy = UserProxyAgent(
        name="user_proxy",
        description="人类代表，发起任务请求"
    )

    team = RoundRobinGroupChat([assistant, user_proxy], max_turns=6)

    task = "请生成 30 个随机样本，然后计算它们的统计摘要，并解释结果。"

    stream = team.run_stream(task=task)
    async for message in stream:
        print(f"\n[{message.source}]: {message.content}")
        print("-" * 60)

asyncio.run(main())

预期结果：

1. user_proxy 发出任务。
2. assistant 的回复中会包含工具调用（在内部消息中体现，最终输出可能显示为“调用 generate_sample_data(size=30)”）。
3. 工具执行后，结果追加到对话历史中。
4. assistant 基于工具返回的数据，继续调用 calculate_summary。
5. 最终给出中文解读。

你会看到类似：

[user_proxy]: 请生成 30 个随机样本，然后计算它们的统计摘要，并解释结果。
------------------------------------------------------------

[assistant]: 好的，我先帮你生成 30 个随机样本。
[工具调用: generate_sample_data(size=30)]
------------------------------------------------------------

[user_proxy]: (工具返回: [0.12, -1.34, 0.89, ...])
------------------------------------------------------------

[assistant]: 数据已生成。现在计算统计摘要。
[工具调用: calculate_summary(numbers=[0.12, -1.34, ...])]
------------------------------------------------------------

[user_proxy]: (工具返回: {"mean": 0.034, "std": 0.98, ...})
------------------------------------------------------------

[assistant]: 这 30 个样本的均值为 0.034，接近 0，说明数据近似以 0 为中心...

发生了什么：

• FunctionTool 将 Python 函数包装成 LLM 可理解的工具描述。当 assistant 认为需要调用工具时，会生成工具调用请求。
• AutoGen 框架拦截这些请求，实际执行 Python 函数，并将结果作为消息返回给对话。
• assistant 根据工具返回的真实数据，进行下一步推理。这和代码执行是两种不同的“动手能力”：工具更适合结构化函数调用，代码执行适合开放的脚本编写。

⚠️ 注意踩坑：

• 工具函数的参数类型提示（type hints）非常重要。LLM 会根据类型提示生成符合格式的参数。如果没有类型提示，可能导致参数格式错误。
• 工具的 description 字段是 LLM 决定“何时调用工具”的关键信息。过于模糊的描述会让 LLM 不知道该用哪个工具；过于冗长会占用上下文窗口。

步骤 5：与外部 Skills 桥接——将已有的函数能力暴露为 AutoGen 工具

你可能有现成的“技能”（Skills）——比如一个已经封装好的数据分析管道、一个第三方 API 调用函数、或者从其他 Agent 框架（如 Claude Skills）迁移而来的能力包。AutoGen 的 FunctionTool 提供了自然的桥接方式。

假设你有一个现成的“异常检测”函数，原本是 Claude Skill 的一部分：

def detect_anomalies(data: list[float], threshold: float = 2.0) -> list[int]:
    """检测异常值：返回 Z-score 超过阈值的索引"""
    import numpy as np
    arr = np.array(data)
    z_scores = np.abs((arr - np.mean(arr)) / np.std(arr))
    return list(np.where(z_scores > threshold)[0])

把它注册为 AutoGen 工具只需一行：

from autogen_core.tools import FunctionTool

anomaly_tool = FunctionTool(
    func=detect_anomalies,
    description="检测数字列表中的异常值。threshold 指定 Z-score 阈值，默认 2.0。"
)

然后将它添加到 AssistantAgent 的 tools 列表中即可。在对话中，助理 Agent 可以基于用户请求自动调用它：

用户请求 → assistant 判断需要异常检测 → 调用 detect_anomalies → 返回异常索引 → assistant 解释结果。

这种桥接的设计哲学是：Skills 是封闭的能力单元，对话是连接的纽带。你不需要修改原有的 Skill 代码，只需暴露接口，让 Agent 在对话中按需调用。

回顾

你在本章中完成了：

总耗时：约 60 分钟（包括阅读 + 编码 + 调试）。

核心领悟：AutoGen 的“对话即协议”不是一句口号。当你在 RoundRobinGroupChat 中观察两个 Agent 的消息流转，你会发现：没有哪一行代码规定了“第一步做什么、第二步做什么”。Agent 根据对话历史、工具反馈、执行结果，自主决定下一步。这种模式的强项在于探索性任务——当你不完全清楚任务应该怎么分解、步骤应该怎么编排时，让 Agent 在对话中自主协商，往往比硬编码状态机更灵活。

当然，这种灵活性也有代价：Agent 可能跑偏、陷入循环、或无意义地“客气”回复。这就是为什么 LangGraph 的确定性控制在生产环境中仍然重要。如果你的任务步骤明确、分支清晰，用 LangGraph；如果你的任务需要 Agent 自行探索、互相纠错，用 AutoGen。

下一步行动清单

1. 调整 max_turns：试着设置不同值（3、10、20），观察对话的长度和完成度变化。
2. 修改 system_message：给 AssistantAgent 一个“批判性思维”角色，看它如何挑战 UserProxyAgent 的假设。
3. 添加第三个 Agent：引入一个“代码审查者”，在 UserProxy 执行代码前先审查代码质量。
4. 注册更多工具：将你项目中已有的函数暴露为工具，观察 Agent 如何在对话中自主组合它们。
5. 尝试不同的对话模式：AutoGen 还支持 SelectorGroupChat（基于选择器的发言顺序），对比它与 RoundRobinGroupChat 的区别。

如果你已经习惯了 AutoGen 的自由对话风格，你可能会想：“有没有一种方式，既能保留角色化分工的清晰结构，又不需要手动编排每一步？”下一章，我们将进入 CrewAI 的世界，看它如何用“角色-目标-任务”三层抽象，让你快速搭建出结构化的 Agent 团队——既有明确分工，又有对话的灵活性。