AutoGen 2026多智能体：协作AI的新范式

模式一：对话式协作
Agent A ←→ Agent B
（两个Agent通过对话解决问题）

模式二：层级式协作
Manager Agent
├── Worker Agent 1
├── Worker Agent 2
└── Worker Agent 3
（管理者分配任务给工作者）

模式三：流水线协作
Agent A → Agent B → Agent C
（每个Agent处理一个阶段）

模式四：竞争式协作
Agent A ↘
Agent B → Judge Agent
Agent C ↗
（多个Agent竞争，裁判选择最佳）

from autogen import AssistantAgent, UserProxyAgent

# 创建用户代理
user_proxy = UserProxyAgent(
    name="user",
    human_input_mode="TERMINATE",
    max_consecutive_auto_reply=10
)

# 创建助手
assistant = AssistantAgent(
    name="assistant",
    system_prompt="你是一个Python编程助手。",
    llm_config={"model": "gpt-5"}
)

# 开始对话
user_proxy.initiate_chat(
    assistant,
    message="帮我写一个快速排序算法"
)

from autogen import AssistantAgent, GroupChat, GroupChatManager

# 创建多个专家Agent
coder = AssistantAgent(
    name="coder",
    system_prompt="你是一个Python程序员，负责写代码。",
    llm_config={"model": "gpt-5"}
)

reviewer = AssistantAgent(
    name="reviewer", 
    system_prompt="你是一个代码审查专家，负责检查代码质量。",
    llm_config={"model": "claude-4-opus"}
)

tester = AssistantAgent(
    name="tester",
    system_prompt="你是一个测试工程师，负责编写测试用例。",
    llm_config={"model": "gpt-5"}
)

# 创建群聊
group_chat = GroupChat(
    agents=[coder, reviewer, tester],
    messages=[],
    max_round=20
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"model": "gpt-5"}
)

# 开始协作
user_proxy.initiate_chat(
    manager,
    message="实现一个LRU缓存，包括代码、审查和测试"
)

from autogen import Workflow

# 定义工作流
workflow = Workflow()

# 添加节点
workflow.add_node("researcher", research_agent)
workflow.add_node("writer", writing_agent)
workflow.add_node("editor", editing_agent)

# 定义流程
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "editor")
workflow.add_edge("editor", "writer", condition="needs_revision")

# 执行
result = workflow.run("写一篇关于AI的科普文章")

from autogen import register_function

# 注册工具
@register_function("search")
def search_web(query: str) -> str:
    """搜索网络"""
    return web_search(query)

@register_function("code_exec")
def execute_code(code: str) -> str:
    """执行Python代码"""
    return exec_python(code)

# Agent可以使用这些工具
agent = AssistantAgent(
    name="tool_agent",
    tools=["search", "code_exec"],
    llm_config={"model": "gpt-5"}
)

from autogen import trace

# 追踪Agent交互
with trace("my_conversation"):
    user_proxy.initiate_chat(assistant, message="...")

# 查看追踪
trace.visualize()  # 生成交互图

from autogen import save_state, load_state

# 保存对话状态
save_state(assistant, "agent_state.pkl")

# 加载状态继续对话
assistant = load_state("agent_state.pkl")
user_proxy.initiate_chat(assistant, message="继续之前的对话")

# 多Agent协作开发软件
product_manager = AssistantAgent(
    name="PM",
    system_prompt="你是产品经理，负责需求分析和项目规划。"
)

architect = AssistantAgent(
    name="Architect",
    system_prompt="你是架构师，负责技术设计。"
)

developer = AssistantAgent(
    name="Developer",
    system_prompt="你是开发者，负责编码实现。"
)

qa = AssistantAgent(
    name="QA",
    system_prompt="你是测试工程师，负责质量保证。"
)

team = GroupChat(
    agents=[product_manager, architect, developer, qa],
    max_round=50
)

# 多Agent协作写研究报告
researcher = AssistantAgent(
    name="Researcher",
    system_prompt="你是研究员，负责收集和分析资料。"
)

analyst = AssistantAgent(
    name="Analyst", 
    system_prompt="你是分析师，负责数据分析和可视化。"
)

writer = AssistantAgent(
    name="Writer",
    system_prompt="你是技术写作专家，负责撰写报告。"
)

editor = AssistantAgent(
    name="Editor",
    system_prompt="你是编辑，负责审核和修改。"
)

# 分层Agent客服
triage_agent = AssistantAgent(
    name="Triage",
    system_prompt="你是客服分流Agent，判断问题类型并路由。"
)

tech_agent = AssistantAgent(
    name="Tech",
    system_prompt="你是技术支持Agent。"
)

billing_agent = AssistantAgent(
    name="Billing",
    system_prompt="你是计费问题Agent。"
)

# 多Agent并行工作
import asyncio

async def parallel_agents():
    tasks = [
        agent1.ainvoke("任务1"),
        agent2.ainvoke("任务2"),
        agent3.ainvoke("任务3")
    ]
    results = await asyncio.gather(*tasks)
    return results

# 根据任务复杂度选择模型
def select_model(task_complexity):
    if task_complexity == "simple":
        return "gpt-5o-mini"
    elif task_complexity == "medium":
        return "gpt-5o"
    else:
        return "gpt-5"