👥 多 Agent 与工作流

单 Agent 的局限

• 认知过载：单个 Agent 处理复杂任务时容易"迷失"，上下文窗口有限
• 专业化不足：一个 System Prompt 很难同时胜任多个领域
• 容错性差：单点故障会导致整个任务失败
• 无法并行：多个独立子任务只能串行处理

多 Agent 的优势

五种核心协作模式

Supervisor 架构图

CrewAI 概述

CrewAI 是专为多 Agent 协作设计的框架，核心概念是用"团队"隐喻来组织 Agent：Agent（角色）、Task（任务）、Crew（团队）。

代码示例：研究团队

from crewai import Agent, Task, Crew, Process

# 定义 Agent（角色）
researcher = Agent(
    role="资深研究员",
    goal="深入调研给定主题，收集权威信息",
    backstory="你拥有 20 年研究经验，擅长从海量信息中提取关键洞见。",
    tools=[search_tool, wiki_tool],
    verbose=True
)

writer = Agent(
    role="技术写作专家",
    goal="将研究成果转化为清晰易懂的技术文章",
    backstory="你曾为 Google、Meta 撰写技术博客，文风简洁有力。",
    verbose=True
)

editor = Agent(
    role="主编",
    goal="确保文章质量、准确性和可读性",
    backstory="你是顶级科技媒体的主编，对内容质量有极高标准。",
    verbose=True
)

# 定义 Task（任务）
research_task = Task(
    description="调研 {topic}，收集关键信息、数据和最新进展",
    agent=researcher,
    expected_output="包含关键发现的详细研究报告"
)

write_task = Task(
    description="基于研究报告撰写一篇技术文章",
    agent=writer,
    expected_output="2000 字的技术文章",
    context=[research_task]  # 依赖研究任务的输出
)

edit_task = Task(
    description="审查和润色文章",
    agent=editor,
    expected_output="润色后的最终版文章",
    context=[write_task]
)

# 组建 Crew（团队）
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.sequential,  # 顺序执行
    verbose=True
)

# 启动执行
result = crew.kickoff(inputs={"topic": "2026年AI Agent发展趋势"})

AutoGen 概述

AutoGen 是微软研究院推出的多 Agent 对话框架，核心理念是基于对话的 Agent 协作——Agent 之间通过消息传递来协调工作。

AutoGen 0.4 (AG2) 新架构

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination

# 创建 Agent
coder = AssistantAgent(
    name="Coder",
    model_client=model_client,
    system_message="你是 Python 专家，负责编写代码。"
)

reviewer = AssistantAgent(
    name="Reviewer",
    model_client=model_client,
    system_message="你是代码审查专家，检查代码质量和安全性。当代码合格时回复 APPROVE。"
)

# 组建团队
termination = TextMentionTermination("APPROVE")
team = RoundRobinGroupChat(
    [coder, reviewer],
    termination_condition=termination
)

# 运行
result = await team.run(
    task="编写一个 Python 快速排序函数"
)

CrewAI vs AutoGen 对比

LangGraph 的多 Agent 优势

LangGraph 是构建多 Agent 系统最灵活的选择——它不预设协作模式，而是让你用图结构自由定义 Agent 之间的通信和控制流。

• 完全可控：每条边、每个条件都可自定义
• 状态共享：通过 TypedDict 共享状态，Agent 间信息传递清晰
• 人在回路：可在任意节点插入人工审批
• 持久化：支持 Checkpoint，长时间任务可暂停恢复

Supervisor 模式实现

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
from typing import TypedDict, Literal

class TeamState(TypedDict):
    messages: list
    next_agent: str
    results: dict

# 创建专业 Agent
search_agent = create_react_agent(llm, [tavily_tool])
code_agent = create_react_agent(llm, [python_repl])
write_agent = create_react_agent(llm, [])

# Supervisor 节点
async def supervisor(state: TeamState):
    response = await llm.ainvoke(
        f"""你是项目主管。根据对话决定下一步：
选项: search_agent, code_agent, write_agent, FINISH
当前消息: {state['messages'][-1]}
返回 JSON: {{"next": "..."}}"""
    )
    return {"next_agent": parse_json(response)["next"]}

# 路由函数
def route(state: TeamState) -> Literal[
    "search_agent", "code_agent", "write_agent", "__end__"
]:
    if state["next_agent"] == "FINISH":
        return "__end__"
    return state["next_agent"]

# 构建图
graph = StateGraph(TeamState)
graph.add_node("supervisor", supervisor)
graph.add_node("search_agent", search_node)
graph.add_node("code_agent", code_node)
graph.add_node("write_agent", write_node)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route)

# 所有 Agent 完成后都回到 Supervisor
for agent in ["search_agent", "code_agent", "write_agent"]:
    graph.add_edge(agent, "supervisor")

app = graph.compile()

Agent vs Workflow

Agent 和 Workflow 是构建 LLM 应用的两种互补方式：

常见工作流模式

多 Agent 系统的挑战

• 调试困难：多个 Agent 的交互路径难以追踪和复现
• 成本控制：Agent 间通信消耗大量 Token
• 死循环：Agent 可能陷入无限对话循环
• 一致性：多个 Agent 的输出可能相互矛盾
• 延迟：多跳通信导致总延迟增高

解决方案清单

项目目标

构建一个自动化内容生产系统：输入一个主题，输出一篇高质量的技术博客文章。