Multi-Agent Orchestration 深度解析：从 CrewAI 到 LangGraph，多智能体协作架构的四种模式与实战

用户输入 → 分析 Agent → 搜索 Agent → 综合 Agent → 审查 Agent → 最终输出

用户提出问题 → 所有 Agent 同时收到问题
→ Agent A 发言 → Agent B 基于 A 的发言补充或反驳
→ Agent C 综合 A 和 B 的观点 → Manager Agent 判断是否达成共识
→ 达成共识 → 输出最终答案；未达成共识 → 继续讨论

用户提交复杂任务 → Manager Agent 分解为子任务
→ 分配给 Worker A（子任务 1）和 Worker B（子任务 2）
→ Worker A 和 Worker B 并行执行
→ Manager 汇总结果并整合 → 输出

[START] → [Router: 判断任务类型]
  → 代码任务 → [Code Agent] → [Review Agent] → [END]
  → 写作任务 → [Writer Agent] → [Editor Agent] → [END]
  → 研究任务 → [Search Agent] → [Synthesis Agent] → [END]

pip install crewai crewai-tools
export OPENAI_API_KEY="your-api-key"

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

# 创建工具
search_tool = SerperDevTool()
web_tool = ScrapeWebsiteTool()

# 定义三个专业角色
researcher = Agent(
    role="高级研究员",
    goal="深入搜索和分析指定主题的最新研究进展",
    backstory="""你是一位拥有 15 年经验的资深研究员，擅长从海量信息中
    筛选出关键洞察。你精通学术论文阅读、技术趋势分析和跨领域知识整合。
    你的分析报告以逻辑严密、论据充分著称。""",
    tools=[search_tool, web_tool],
    verbose=True,
    allow_delegation=False,
)

analyst = Agent(
    role="数据分析师",
    goal="对研究成果进行数据驱动的分析和量化评估",
    backstory="""你是一位数据驱动的分析专家，擅长将定性的研究结论
    转化为可量化的指标。你精通统计分析、可视化设计和趋势预测，
    能够从数据中发现隐藏的模式和关联。""",
    tools=[search_tool],
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role="技术撰稿人",
    goal="将研究和分析结果转化为高质量的技术报告",
    backstory="""你是一位获奖的科技撰稿人，擅长将复杂的技术概念
    转化为通俗易懂的文章。你的写作风格既专业准确又生动有趣，
    能够让不同背景的读者都能理解并产生兴趣。""",
    verbose=True,
    allow_delegation=False,
)

# 定义任务 - 每个任务绑定到特定 Agent
research_task = Task(
    description="研究 {topic} 领域的最新进展，重点关注 2026 年的突破性成果。"
                "搜索至少 5 篇相关论文或项目，总结核心创新点和实际影响。",
    expected_output="一份包含 5 个关键发现的研究摘要，每个发现包含标题、"
                    "核心创新、技术路线和潜在影响。",
    agent=researcher,
)

analysis_task = Task(
    description="基于研究结果，对 {topic} 领域的发展趋势进行量化分析。"
                "评估各项目的技术成熟度、市场影响力和创新独特性。",
    expected_output="一份分析报告，包含趋势图表、项目对比矩阵和未来 6-12"
                    "个月的预测。",
    agent=analyst,
)

writing_task = Task(
    description="将研究和分析结果整合为一篇面向技术决策者的深度报告。"
                "报告需要包含执行摘要、技术分析和战略建议。",
    expected_output="一篇 3000 字以上的技术报告，结构清晰、论据充分、"
                    "结论可操作。",
    agent=writer,
)

# 组装 Crew
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    verbose=True,
    # process=Process.sequential  # 流水线模式（默认）
)

# 执行
result = crew.kickoff(inputs={"topic": "多智能体编排框架"})
print(result)

# 添加 Manager Agent
manager = Agent(
    role="研究总监",
    goal="统筹管理研究团队，确保研究质量与时效性",
    backstory="你是一位经验丰富的研究总监，擅长团队管理和质量控制。",
    allow_delegation=True,  # 关键：允许委派任务给 Worker
)

hierarchical_crew = Crew(
    agents=[manager, researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    verbose=True,
    process=Process.hierarchical,  # 切换到层次化模式
    manager_agent=manager,
)

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator

# 定义全局状态
class CodeReviewState(TypedDict):
    code: str           # 待审查代码
    review_comments: Annotated[list, operator.add]  # 审查意见（自动追加）
    bugs_found: int     # 发现的 Bug 数量
    needs_rework: bool  # 是否需要返工
    final_verdict: str  # 最终结论

# 定义节点函数
def static_analysis(state: CodeReviewState) -> dict:
    """静态分析：检查代码风格和常见错误"""
    code = state["code"]
    comments = []
    bugs = 0

    # 模拟静态分析逻辑
    if "TODO" in code:
        comments.append("发现 TODO 标记，需要处理")
        bugs += 1
    if len(code.split("\n")) > 100:
        comments.append("函数过长，建议拆分")
        bugs += 1

    return {
        "review_comments": comments,
        "bugs_found": bugs,
    }

def security_review(state: CodeReviewState) -> dict:
    """安全审查：检查潜在安全风险"""
    code = state["code"]
    comments = []
    bugs = 0

    if "eval(" in code or "exec(" in code:
        comments.append("发现危险的 eval/exec 调用")
        bugs += 2  # 安全问题权重更高
    if "password" in code.lower():
        comments.append("代码中包含密码相关逻辑，需要额外审查")
        bugs += 1

    return {
        "review_comments": comments,
        "bugs_found": bugs,
    }

def decide_verdict(state: CodeReviewState) -> dict:
    """根据审查结果做出最终判定"""
    total_bugs = state["bugs_found"]
    if total_bugs == 0:
        return {"final_verdict": "✅ 通过，代码质量优秀", "needs_rework": False}
    elif total_bugs <= 2:
        return {"final_verdict": "⚠️ 有条件通过，请修复发现的问题", "needs_rework": True}
    else:
        return {"final_verdict": "❌ 不通过，需要重大修改后重新审查", "needs_rework": True}

# 构建图
graph = StateGraph(CodeReviewState)

# 添加节点
graph.add_node("static_analysis", static_analysis)
graph.add_node("security_review", security_review)
graph.add_node("verdict", decide_verdict)

# 设置入口
graph.set_entry_point("static_analysis")

# 添加边
graph.add_edge("static_analysis", "security_review")
graph.add_edge("security_review", "verdict")
graph.add_edge("verdict", END)

# 编译
app = graph.compile()

# 执行
result = app.invoke({
    "code": """
def process_data(data):
    # TODO: add error handling
    result = eval(data['expression'])
    password = data.get('password', '')
    return result
""",
    "review_comments": [],
    "bugs_found": 0,
    "needs_rework": False,
    "final_verdict": "",
})

print(f"审查意见: {result['review_comments']}")
print(f"Bug 数量: {result['bugs_found']}")
print(f"最终判定: {result['final_verdict']}")

def should_loop_back(state: CodeReviewState) -> str:
    """决定是否需要重新审查"""
    if state["needs_rework"] and state["bugs_found"] > 3:
        return "rework"  # 重大问题，需要返工
    return "done"

# 添加条件边
graph.add_conditional_edges(
    "verdict",
    should_loop_back,
    {
        "rework": "static_analysis",  # 回到静态分析重新开始
        "done": END,
    }
)

# 给每个 Agent 添加独立的日志记录
import logging

logger = logging.getLogger("multi-agent-debug")

class TracedAgent(Agent):
    def execute(self, task):
        logger.info(f"[{self.role}] 开始执行: {task[:50]}...")
        result = super().execute(task)
        logger.info(f"[{self.role}] 完成，产出长度: {len(str(result))} 字符")
        return result

# 在 CrewAI 中检查任务间传递的内容
for task in crew.tasks:
    print(f"任务: {task.description}")
    print(f"输入来源: {task.context}")  # 查看这个任务依赖哪些前置任务
    print("---")

# CrewAI 支持并行任务
parallel_tasks = [
    Task(description="搜索 {topic} 的技术论文", agent=researcher),
    Task(description="搜索 {topic} 的商业报告", agent=researcher),
]

# 这两个任务可以并行执行
crew = Crew(
    agents=[researcher],
    tasks=parallel_tasks,
    # 默认就是并行的
)

import hashlib
import json
from pathlib import Path

cache_dir = Path(".agent_cache")
cache_dir.mkdir(exist_ok=True)

def cached_search(query: str) -> str:
    """缓存搜索结果，避免重复搜索"""
    cache_key = hashlib.md5(query.encode()).hexdigest()
    cache_file = cache_dir / f"{cache_key}.json"

    if cache_file.exists():
        with open(cache_file) as f:
            return json.load(f)["result"]

    # 执行搜索
    result = search_tool.run(query)

    with open(cache_file, "w") as f:
        json.dump({"query": query, "result": result}, f)

    return result