AutoGen 多智能体框架：从对话到协作

┌─────────────────────────────────┐
│     Application Layer           │
│   (User Interface / API)        │
├─────────────────────────────────┤
│     Orchestration Layer         │
│   ┌───────────┐ ┌────────────┐ │
│   │ Group Chat│ │ Workflow   │ │
│   │ Manager   │ │ Engine     │ │
│   └───────────┘ └────────────┘ │
├─────────────────────────────────┤
│     Agent Layer                 │
│   AssistantAgent│UserProxyAgent│
│   MultimodalAgent│CustomAgent  │
├─────────────────────────────────┤
│     Communication Layer         │
│   Message Queue │ Event Bus     │
├─────────────────────────────────┤
│     Model Layer                 │
│   OpenAI │ Azure │ Local Models │
└─────────────────────────────────┘

# 安装 AutoGen
# pip install autogen-agentchat autogen-ext

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 配置模型
model_client = OpenAIChatCompletionClient(
    model="gpt-5",
    api_key="your-api-key",
    # Azure OpenAI 配置
    # base_url="https://your-resource.openai.azure.com",
    # api_version="2026-03-01-preview",
)

# 创建专业化的 Agent 团队
product_manager = AssistantAgent(
    name="ProductManager",
    model_client=model_client,
    system_message="""
    你是一个产品经理。你的职责：
    1. 理解用户需求
    2. 将需求拆解为具体的功能点
    3. 定义验收标准
    
    不要写代码，专注于需求分析。
    当需求分析完成时，说"需求已明确，请开发人员开始实现"。
    """,
)

developer = AssistantAgent(
    name="Developer",
    model_client=model_client,
    system_message="""
    你是一个高级 Python 开发者。你的职责：
    1. 根据产品经理的需求编写代码
    2. 确保代码质量（类型标注、错误处理）
    3. 编写单元测试
    
    写完代码后，说"代码已完成，请测试人员审查"。
    """,
    # 可以执行代码
    reflect_on_tool_history=True,
)

tester = AssistantAgent(
    name="Tester",
    model_client=model_client,
    system_message="""
    你是一个 QA 测试工程师。你的职责：
    1. 审查代码的逻辑正确性
    2. 检查边界条件和错误处理
    3. 提出改进建议
    
    审查完成后，说"APPROVED"表示通过，或指出需要修改的问题。
    """,
)

# 创建团队（轮询模式）
team = RoundRobinGroupChat(
    participants=[product_manager, developer, tester],
    termination_condition=TextMentionTermination("APPROVED"),
    max_turns=20,
)

# 执行任务
async def main():
    result = await team.run(
        task="实现一个函数，接收日期字符串，返回该日期是星期几，"
        "支持多种日期格式输入"
    )
    print(result)

asyncio.run(main())

from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

# 使用智能选择器（Selector）模式
# 由一个"管理者"Agent 决定下一个发言者
team = SelectorGroupChat(
    participants=[product_manager, developer, tester],
    model_client=model_client,  # 用于选择发言者
    selector_prompt="""
    你是一个团队协调者。根据当前对话状态，
    选择下一个应该发言的团队成员。
    
    可选成员：{participants}
    对话历史：{history}
    
    返回下一个发言者的名字。
    """,
    termination_condition=MaxMessageTermination(30),
    allow_repeated_speaker=False,  # 同一成员不能连续发言
)

# 带工具的工作流
async def run_with_tools():
    # 给 Developer 添加代码执行工具
    from autogen_ext.tools.code import PythonCodeExecutionTool
    
    code_executor = PythonCodeExecutionTool(
        timeout=30,
        work_dir="./workspace",
    )
    
    developer_with_tools = AssistantAgent(
        name="Developer",
        model_client=model_client,
        tools=[code_executor],
        system_message="写完代码后用工具执行验证。",
    )
    
    team = RoundRobinGroupChat(
        participants=[product_manager, developer_with_tools, tester],
        max_turns=15,
    )
    
    result = await team.run(task="实现并验证一个快速排序算法")
    return result

from autogen_agentchat.teams import GraphWorkflow
from autogen_agentchat.graph import Graph, Node, Edge

# 定义有向无环图工作流
async def build_research_workflow():
    # 创建专业化 Agent
    researcher = AssistantAgent(
        name="Researcher",
        model_client=model_client,
        system_message="你是一个研究员，负责搜索和整理信息。",
    )
    
    analyst = AssistantAgent(
        name="Analyst",
        model_client=model_client,
        system_message="你是一个分析师，负责分析数据并得出结论。",
    )
    
    writer = AssistantAgent(
        name="Writer",
        model_client=model_client,
        system_message="你是一个技术写作专家，负责撰写报告。",
    )
    
    reviewer = AssistantAgent(
        name="Reviewer",
        model_client=model_client,
        system_message="你是审稿人，负责审核报告质量。",
    )
    
    # 构建工作流图
    graph = Graph()
    
    # 添加节点
    graph.add_node(Node("research", researcher))
    graph.add_node(Node("analyze", analyst))
    graph.add_node(Node("write", writer))
    graph.add_node(Node("review", reviewer))
    
    # 添加边（定义执行顺序）
    graph.add_edge(Edge("research", "analyze"))
    graph.add_edge(Edge("analyze", "write"))
    graph.add_edge(Edge("write", "review"))
    # 审核不通过可以回退
    graph.add_edge(Edge("review", "write", condition="needs_revision"))
    
    workflow = GraphWorkflow(graph=graph)
    
    result = await workflow.run(
        task="研究 2026 年 AI 编程助手市场格局并撰写分析报告"
    )
    return result

# AutoGen 的消息结构
from autogen_agentchat.messages import TextMessage, MultiModalMessage

# 文本消息
text_msg = TextMessage(
    source="Developer",
    content="我已经实现了用户认证模块，请审查。",
    metadata={"timestamp": "2026-06-25T10:00:00Z"}
)

# 多模态消息（包含图片）
multimodal_msg = MultiModalMessage(
    source="Tester",
    content=[
        "测试结果截图如下：",
        Image.from_file("test-results.png"),
    ]
)

from autogen_core import MessageContext, RoutedAgent, message_handler

class TaskMessage:
    def __init__(self, content: str, priority: int = 0):
        self.content = content
        self.priority = priority

class ResultMessage:
    def __init__(self, content: str, status: str):
        self.content = content
        self.status = status

class CustomWorkerAgent(RoutedAgent):
    def __init__(self, model_client):
        super().__init__("Worker")
        self.model_client = model_client
        self._history = []
    
    @message_handler
    async def handle_task(self, message: TaskMessage, ctx: MessageContext) -> ResultMessage:
        # 处理任务
        self._history.append(message.content)
        
        response = await self.model_client.create(
            messages=[{"role": "user", "content": message.content}]
        )
        
        return ResultMessage(
            content=response.content,
            status="completed"
        )

from autogen_core import CancellationToken
import json
import os

class CheckpointManager:
    """Agent 会话状态持久化"""
    
    def __init__(self, checkpoint_dir: str = "./checkpoints"):
        self.checkpoint_dir = checkpoint_dir
        os.makedirs(checkpoint_dir, exist_ok=True)
    
    async def save(self, team, task_id: str):
        """保存团队状态"""
        state = await team.save_state()
        path = os.path.join(self.checkpoint_dir, f"{task_id}.json")
        with open(path, "w") as f:
            json.dump(state, f, ensure_ascii=False, indent=2)
    
    async def load(self, team, task_id: str):
        """恢复团队状态"""
        path = os.path.join(self.checkpoint_dir, f"{task_id}.json")
        if os.path.exists(path):
            with open(path) as f:
                state = json.load(f)
            await team.load_state(state)
            return True
        return False

# 使用示例
checkpoint = CheckpointManager()

# 保存中断的会话
await checkpoint.save(team, "task-001")

# 恢复会话
await checkpoint.load(team, "task-001")
result = await team.run(task="继续之前的任务")

from autogen_agentchat.teams import ConcurrentGroupChat

# 并行执行模式：多个 Agent 同时工作
concurrent_team = ConcurrentGroupChat(
    participants=[researcher, data_analyst, market_analyst],
    model_client=model_client,
    aggregation_strategy="merge",  # merge | select | vote
    max_turns=10,
)

# 三个 Agent 并行研究不同方面，然后合并结果
result = await concurrent_team.run(
    task="分析竞品：研究员查技术栈，数据分析师查性能指标，市场分析师查市场份额"
)

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 启动 Agent 服务
CMD ["python", "-m", "autogen_server", "--host", "0.0.0.0", "--port", "8080"]

# requirements.txt
autogen-agentchat>=0.5.0
autogen-ext[openai,azure]>=0.5.0
fastapi>=0.115.0
uvicorn>=0.30.0

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI(title="AutoGen Agent Service")

class TaskRequest(BaseModel):
    task: str
    team_config: dict = {}
    max_turns: int = 20

class TaskResponse(BaseModel):
    task_id: str
    status: str
    result: str | None = None

@app.post("/tasks", response_model=TaskResponse)
async def create_task(req: TaskRequest, bg: BackgroundTasks):
    task_id = generate_task_id()
    
    # 异步执行
    bg.add_task(run_agent_team, task_id, req.task, req.team_config)
    
    return TaskResponse(task_id=task_id, status="running")

@app.get("/tasks/{task_id}", response_model=TaskResponse)
async def get_task(task_id: str):
    status, result = await get_task_status(task_id)
    return TaskResponse(task_id=task_id, status=status, result=result)