多智能体系统搭建实战：Orchestrator-Worker 模式从设计到部署

pip install crewai
pip install crewai[tools]  # 包含内置工具集

from crewai import Agent, Task, Crew, Process

# 数据分析师 Worker
analyst = Agent(
    role="数据分析师",
    goal="分析销售数据，识别核心趋势和模式",
    backstory="你是一个资深数据分析师，擅长从数字中发现故事",
    verbose=True,
    allow_delegation=False,
)

# 风险分析师 Worker
risk_analyst = Agent(
    role="风险分析师",
    goal="识别数据中的异常和风险点",
    backstory="你专注于发现隐藏的风险，对异常值极其敏感",
    verbose=True,
    allow_delegation=False,
)

# 可视化顾问 Worker
viz_consultant = Agent(
    role="可视化顾问",
    goal="根据分析结果推荐最合适的可视化方式",
    backstory="你擅长用图表讲故事，知道什么数据该用什么图",
    verbose=True,
    allow_delegation=False,
)

# 报告撰写者 — 充当编排器角色
report_writer = Agent(
    role="报告主编",
    goal="整合所有分析结果，撰写一份完整的分析报告",
    backstory="你是总编，负责把各专家的分析组装成一篇高质量的最终报告",
    verbose=True,
    allow_delegation=True,  # 允许编排
)

# 并行执行的分析任务
analyze_task = Task(
    description="分析{period}的销售数据：{sales_data}，找出销售趋势",
    agent=analyst,
    expected_output="结构化的趋势分析报告，包含同比增长率、环比变化等",
)

risk_task = Task(
    description="分析{period}的销售数据：{sales_data}，识别异常和风险",
    agent=risk_analyst,
    expected_output="风险清单，每个风险包含影响程度和概率评估",
)

viz_task = Task(
    description="根据分析结果推荐可视化方案",
    agent=viz_consultant,
    expected_output="推荐图表类型及其说明",
)

# 汇总任务（依赖前三个任务的结果）
report_task = Task(
    description="整合趋势分析、风险评估和可视化建议，撰写完整报告",
    agent=report_writer,
    expected_output="一份完整的销售分析报告",
)

# 组装 Crew
crew = Crew(
    agents=[analyst, risk_analyst, viz_consultant, report_writer],
    tasks=[analyze_task, risk_task, viz_task, report_task],
    process=Process.sequential,  # 按顺序执行（简单模式）
    verbose=True,
)

# 在实际的生产场景中，可以使用 Process.hierarchical 来让
# report_writer 自动编排其他三个 Agent 并行工作

result = crew.kickoff(inputs={
    "period": "2026 年第一季度",
    "sales_data": "...（从数据库获取的实际数据）",
})
print(result)

pip install langgraph langchain langchain-openai

from typing import TypedDict, List, Any, Optional
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# 定义系统状态（LangGraph 的核心概念）
class AnalysisState(TypedDict):
    period: str
    raw_data: str
    trend_analysis: Optional[str]
    risk_analysis: Optional[str]
    viz_recommendation: Optional[str]
    final_report: Optional[str]
    errors: List[str]
    retry_count: int

model = ChatOpenAI(model="gpt-4.5", temperature=0.2)

# 节点 1：趋势分析
def analyze_trends(state: AnalysisState):
    try:
        response = model.invoke([
            SystemMessage(content="你是一个数据分析专家，分析销售趋势。"),
            HumanMessage(content=f"分析 {state['period']} 的销售数据：{state['raw_data']}"),
        ])
        return {"trend_analysis": response.content, "retry_count": 0}
    except Exception as e:
        return {"errors": [f"趋势分析失败: {str(e)}"], "retry_count": state["retry_count"] + 1}

# 节点 2：风险分析
def analyze_risks(state: AnalysisState):
    response = model.invoke([
        SystemMessage(content="你是一个风险分析师，识别数据中的异常。"),
        HumanMessage(content=f"分析这些数据中的风险：{state['raw_data']}"),
    ])
    return {"risk_analysis": response.content}

# 节点 3：可视化推荐
def recommend_viz(state: AnalysisState):
    response = model.invoke([
        SystemMessage(content="你是一个数据可视化专家。"),
        HumanMessage(content=f"趋势：{state['trend_analysis']}\n风险：{state['risk_analysis']}\n推荐图表"),
    ])
    return {"viz_recommendation": response.content}

# 节点 4：报告生成（编排器聚合）
def generate_report(state: AnalysisState):
    response = model.invoke([
        SystemMessage(content="你是一个报告总编。"),
        HumanMessage(content=f"""整合以下分析结果为一份完整报告：
趋势：{state['trend_analysis']}
风险：{state['risk_analysis']}
可视化建议：{state['viz_recommendation']}"""),
    ])
    return {"final_report": response.content}

# 重试决策函数
def should_retry(state: AnalysisState):
    if len(state["errors"]) > 0 and state["retry_count"] < 3:
        return "analyze_trends"  # 重试失败的节点
    return "analyze_risks"  # 继续执行

# 构建图
workflow = StateGraph(AnalysisState)

# 添加节点
workflow.add_node("analyze_trends", analyze_trends)
workflow.add_node("analyze_risks", analyze_risks)
workflow.add_node("recommend_viz", recommend_viz)
workflow.add_node("generate_report", generate_report)

# 设置执行边
workflow.set_entry_point("analyze_trends")

# 条件边：趋势分析失败则重试
workflow.add_conditional_edges(
    "analyze_trends",
    should_retry,
    {"analyze_trends": "analyze_trends", "analyze_risks": "analyze_risks"}
)

# 并行执行两条分析路径
workflow.add_edge("analyze_risks", "recommend_viz")
workflow.add_edge("recommend_viz", "generate_report")
workflow.add_edge("generate_report", END)

# 编译和运行
app = workflow.compile()
result = app.invoke({
    "period": "2026 Q1",
    "raw_data": "...（销售数据）",
    "errors": [],
    "retry_count": 0
})
print(result["final_report"])