AutoGen 多 Agent 协作系统开发实战：从架构设计到生产部署

发布时间：2026-07-08 | 阅读时间：约 25 分钟 | 分类：AI Agent

摘要：本文深入讲解如何使用 Microsoft AutoGen 构建生产级多 Agent 协作系统，涵盖 GroupChat、UserProxyAgent、ConversableAgent 核心概念，提供完整的智能客服系统代码示例，包含 Docker 部署与监控方案。

为什么需要多 Agent 协作？

单个大模型 Agent 在处理复杂任务时往往力不从心。想象一个电商售后场景：用户咨询"我上周买的手机电池不耐用，想退货但订单找不到了"。这个请求涉及：

1. 身份验证 - 确认用户身份
2. 订单查询 - 检索历史订单
3. 政策判断 - 判断是否符合退货条件
4. 物流协调 - 生成退货单并通知仓库
5. 客户沟通 - 用友好语气回复用户

单 Agent 需要同时掌握所有这些能力，容易混乱且难以维护。多 Agent 架构让每个 Agent 专注于一个领域，通过协作完成复杂任务。

AutoGen 核心架构解析

Agent 类型一览

GroupChat 协作模式

AutoGen 提供三种群聊策略：

1. round_robin - 轮流发言，适合平等协作
2. random - 随机选择，用于探索性讨论
3. auto - 智能路由，Manager 根据上下文选择最合适的 Agent

生产环境推荐使用 auto 模式，配合自定义的 speaker_selection_method：

from autogen import GroupChat, GroupChatManager

groupchat = GroupChat(
    agents=[user_proxy, order_agent, policy_agent, logistics_agent],
    messages=[],
    max_round=20,
    speaker_selection_method="auto",
    allow_repeat_speaker=False,
)

实战：构建智能售后客服系统

Step 1：环境准备

# 创建虚拟环境
python -m venv autenv
source autenv/bin/activate

# 安装依赖
pip install pyautogen==0.4.0 openai python-dotenv

# 目录结构
mkdir autogen-customer-service && cd autogen-customer-service
touch .env main.py agents.py tools.py

.env 文件：

OPENAI_API_KEY=sk-your-api-key
OPENAI_BASE_URL=https://api.openai.com/v1

Step 2：定义工具函数

tools.py：

import json
from typing import List, Dict, Optional

def verify_customer(phone: str) -> Dict:
    """验证客户身份"""
    # 实际生产连接 CRM 数据库
    mock_db = {
        "13800138000": {"name": "张三", "level": "VIP", "orders": ["ORD-2026-001", "ORD-2026-015"]},
        "13900139000": {"name": "李四", "level": "普通", "orders": ["ORD-2026-008"]},
    }
    return mock_db.get(phone, {"error": "用户未找到"})

def get_order_detail(order_id: str) -> Dict:
    """查询订单详情"""
    mock_orders = {
        "ORD-2026-001": {
            "product": "iPhone 16 Pro",
            "date": "2026-06-25",
            "price": 8999,
            "status": "已完成",
            "warranty_days": 15
        },
        "ORD-2026-015": {
            "product": "小米15 Ultra",
            "date": "2026-06-28",
            "price": 6499,
            "status": "已完成",
            "warranty_days": 7
        }
    }
    return mock_orders.get(order_id, {"error": "订单不存在"})

def check_return_policy(order_date: str, product: str) -> Dict:
    """检查退货政策"""
    from datetime import datetime, timedelta
    order_dt = datetime.strptime(order_date, "%Y-%m-%d")
    days_passed = (datetime.now() - order_dt).days

    if days_passed <= 7:
        return {"eligible": True, "reason": "7天无理由退货", "deadline": (order_dt + timedelta(days=7)).strftime("%Y-%m-%d")}
    elif days_passed <= 30 and "质量问题" in product:
        return {"eligible": True, "reason": "30天质量问题包退", "deadline": (order_dt + timedelta(days=30)).strftime("%Y-%m-%d")}
    else:
        return {"eligible": False, "reason": f"已超期{days_passed}天", "suggestion": "建议联系厂商售后"}

def create_return_order(order_id: str, reason: str) -> Dict:
    """创建退货单"""
    return_id = f"RET-{order_id.replace('ORD-', '')}"
    return {
        "return_id": return_id,
        "status": "已创建",
        "pickup_date": (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d"),
        "label_url": f"https://logistics.example.com/label/{return_id}"
    }

def send_notification(user_phone: str, message: str) -> str:
    """发送通知（模拟）"""
    return f"短信已发送至 {user_phone}：{message[:50]}..."

Step 3：配置 Agent

agents.py：

import os
from dotenv import load_dotenv
from autogen import ConversableAgent, UserProxyAgent
from autogen.agentchat.contrib.capabilities import teachability

load_dotenv()

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": os.getenv("OPENAI_API_KEY"), "base_url": os.getenv("OPENAI_BASE_URL")}],
    "temperature": 0.2,
    "timeout": 120,
}

# 系统提示词设计原则：明确职责、约束行为、指定输出格式

user_proxy = UserProxyAgent(
    name="UserProxy",
    system_message="你是一个用户接入代理。你的职责是：1) 接收用户原始输入；2) 提取关键信息（手机号、订单号、诉求）；3) 将结构化信息传递给其他 Agent；4) 当需要人工确认时，向用户请求确认。不要自行做决策。",
    code_execution_config=False,
    human_input_mode="NEVER",
)

order_agent = ConversableAgent(
    name="OrderAgent",
    system_message="""你是订单查询专家。职责：
1. 使用 verify_customer 验证客户身份
2. 使用 get_order_detail 查询订单详情
3. 返回格式化的订单信息
4. 如果发现异常订单，标记并转给 PolicyAgent
不要回答与订单无关的问题。""",
    llm_config=llm_config,
)

policy_agent = ConversableAgent(
    name="PolicyAgent",
    system_message="""你是售后政策专家。职责：
1. 根据订单信息和使用 check_return_policy 判断是否符合退货/换货条件
2. 明确告知用户政策依据
3. 如符合条件，通知 LogisticsAgent 创建退货单
4. 如不符合，提供替代方案（维修、折价回收等）
必须基于具体政策条款做判断，不能随意承诺。""",
    llm_config=llm_config,
)

logistics_agent = ConversableAgent(
    name="LogisticsAgent",
    system_message="""你是物流协调专家。职责：
1. 使用 create_return_order 创建退货单
2. 安排上门取件时间
3. 生成物流标签
4. 使用 send_notification 通知用户取件信息
确保物流信息准确，时间合理。""",
    llm_config=llm_config,
)

coordinator_agent = ConversableAgent(
    name="Coordinator",
    system_message="""你是客服协调员。职责：
1. 汇总所有 Agent 的处理结果
2. 用友好、专业的语气向用户回复
3. 确保回复包含：处理结果、下一步操作、预计时间
4. 如果问题未解决，提供转人工客服的选项
回复要简洁明了，避免技术术语。""",
    llm_config=llm_config,
)

Step 4：主流程编排

main.py：

import os
from dotenv import load_dotenv
from autogen import GroupChat, GroupChatManager
from agents import user_proxy, order_agent, policy_agent, logistics_agent, coordinator_agent
from tools import verify_customer, get_order_detail, check_return_policy, create_return_order, send_notification

load_dotenv()

# 注册工具到对应 Agent
for agent in [order_agent, policy_agent, logistics_agent]:
    agent.register_function(
        function_map={
            "verify_customer": verify_customer,
            "get_order_detail": get_order_detail,
            "check_return_policy": check_return_policy,
            "create_return_order": create_return_order,
            "send_notification": send_notification,
        }
    )

# 创建群聊
groupchat = GroupChat(
    agents=[user_proxy, order_agent, policy_agent, logistics_agent, coordinator_agent],
    messages=[],
    max_round=15,
    speaker_selection_method="auto",
    allow_repeat_speaker=False,
)

manager = GroupChatManager(groupchat=groupchat, llm_config={
    "config_list": [{"model": "gpt-4o", "api_key": os.getenv("OPENAI_API_KEY"), "base_url": os.getenv("OPENAI_BASE_URL")}],
    "temperature": 0.1,
})

# 启动对话
def handle_customer_request(user_input: str):
    """处理客户请求入口"""
    user_proxy.initiate_chat(
        manager,
        message=f"用户输入：{user_input}\n\n请按流程处理：UserProxy 提取信息 → OrderAgent 查订单 → PolicyAgent 判断政策 → LogisticsAgent 处理物流 → Coordinator 汇总回复。"
    )
    return groupchat.messages

if __name__ == "__main__":
    # 模拟用户请求
    user_msg = "我上周买的iPhone电池有问题，想退货。手机号13800138000"
    messages = handle_customer_request(user_msg)

    # 打印最终回复
    for msg in reversed(messages):
        if msg["name"] == "Coordinator":
            print("\n=== 客服回复 ===")
            print(msg["content"])
            break

Step 5：运行与验证

python main.py

预期输出流程：

1. UserProxy 提取手机号 13800138000 和诉求"退货"
2. OrderAgent 调用 verify_customer 和 get_order_detail 获取订单信息
3. PolicyAgent 调用 check_return_policy 判断符合 7 天无理由退货
4. LogisticsAgent 调用 create_return_order 和 send_notification
5. Coordinator 汇总并生成友好回复

生产部署方案

Docker 容器化

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 8000

CMD ["python", "main.py"]

docker-compose.yml：

version: '3.8'
services:
  autogen-service:
    build: .
    env_file: .env
    ports:
      - "8000:8000"
    restart: unless-stopped
    logging:
      driver: json-file
      options:
        max-size: "10m"
        max-file: "3"

监控与日志

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(f'logs/autogen-{datetime.now().strftime("%Y%m%d")}.log'),
        logging.StreamHandler()
    ]
)

# 记录每个 Agent 的决策
class AgentLogger:
    @staticmethod
    def log_decision(agent_name: str, decision: str, context: dict):
        logging.info(f"[AGENT:{agent_name}] Decision: {decision} | Context: {context}")

常见问题 FAQ

Q1：AutoGen 和 LangGraph 有什么区别？

A：AutoGen 更侧重多 Agent 对话协作，强调 Agent 之间的自然语言交互；LangGraph 更侧重状态机和图结构控制流，适合需要严格流程控制的场景。两者可以结合使用。

Q2：如何处理 Agent 之间的循环对话？

A：设置 max_round 限制最大回合数；使用 allow_repeat_speaker=False 防止同一 Agent 连续发言；在 GroupChatManager 中自定义 speaker_selection_method 实现更智能的路由。

Q3：生产环境如何管理 API 成本？

A：1) 使用便宜模型做初步分类（如 GPT-3.5），复杂任务再调用 GPT-4o；2) 设置 max_tokens 限制输出长度；3) 启用缓存避免重复调用；4) 监控每个 Agent 的 token 消耗。

Q4：如何实现人工审核介入？

A：将 UserProxyAgent 的 human_input_mode 设为 "ALWAYS" 或 "TERMINATE"，关键节点（如退款审批）设置 is_termination_msg 触发人工确认。

Q5：Agent 调用工具失败怎么办？

A：在工具函数中添加异常处理；设置重试机制（最多 3 次）；失败时返回结构化错误信息，让 Coordinator Agent 向用户解释并转人工。

Q6：支持国产大模型吗？

A：支持。通过修改 llm_config 的 base_url 和 model 即可接入 DeepSeek、文心一言、通义千问等兼容 OpenAI API 格式的模型。

Q7：对话历史如何持久化？

A：groupchat.messages 是标准 Python 列表，可以序列化为 JSON 存入 Redis 或数据库。恢复对话时重新赋值即可。

Q8：如何实现 Agent 知识库增强？

A：使用 RAG 技术，在 Agent 的 system_message 中注入检索到的相关知识；或将知识库查询封装为工具函数供 Agent 调用。

进阶优化方向

1. 并行执行：使用 async 模式让多个 Agent 同时处理独立子任务
2. 记忆增强：集成向量数据库实现长期记忆
3. 安全沙箱：代码执行 Agent 放入 Docker 沙箱
4. A/B 测试：对比不同提示词策略的解决率

结语

多 Agent 协作是大模型应用走向生产的关键架构模式。通过 AutoGen，你可以快速构建可扩展、可维护的智能系统。本文的智能客服系统只是起点，同样的架构可以应用于招聘筛选、医疗分诊、法律咨询等多种场景。

本文首发于 1630.top，转载请注明出处。