Data Agents-Part 2 ：揭秘 AI 量产关键：多智能体编排

class State(MessagesState):
    user_query: Optional[str]               # 用户原始查询
    enabled_agents: Optional[List[str]]     # 启用的智能体列表
    plan: Optional[Dict[str, Dict[str, Any]]]  # 任务规划
    current_step: int                       # 当前执行步骤
    agent_query: Optional[str]              # 发给专项智能体的子查询
    last_reason: Optional[str]             # 上一步的执行理由
    replan_flag: Optional[bool]             # 重规划标记
    replan_attempts: Optional[Dict[int, int]]  # 各步骤的重规划次数

reasoning_llm = ChatOpenAI(
    model="o3",
    model_kwargs={"response_format": {"type": "json_object"}},
)
def planner_node(state: State) -> Command["executor"]:
    llm_reply = reasoning_llm.invoke([plan_prompt(state)])
    parsed_plan = json.loads(llm_reply.content)
    return Command(
        update={
            "plan": parsed_plan,
            "messages": [HumanMessage(content=llm_reply.content)],
            "user_query": state["user_query"],
            "current_step": 1,
            "replan_flag": state.get("replan_flag", False),
        },
        goto="executor",
    )

def executor_node(state: State) -> Command:
    plan = state.get("plan", {})
    step = state.get("current_step", 1)
    llm_reply = reasoning_llm.invoke([executor_prompt(state)])
    parsed = json.loads(llm_reply.content)  
    replan = parsed["replan"]  # 是否重规划
    goto   = parsed["goto"]    # 下一步要调用的节点
    query  = parsed["query"]   # 发给专项智能体的子查询

if replan:
    if step_replans < MAX_REPLANS:
        updates["replan_flag"] = True
        return Command(update=updates, goto="planner")

updates["current_step"] = step + 1
return Command(update=updates, goto=goto)

tavily_tool = TavilySearch(max_results=5)
web_search_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o"),
    tools=[tavily_tool],
    prompt=agent_system_prompt("你是一名专业的网页研究员，负责精准获取公开网络上的信息…")
)
def web_research_node(state: State) -> Command["executor"]:
    result = web_search_agent.invoke({"messages": state["agent_query"]})
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="web_researcher"
    )
    return Command(update={"messages": result["messages"]}, goto="executor")

chart_agent = create_react_agent(
    llm,
    [python_repl_tool],
    prompt=agent_system_prompt("""
        你只会干一件事：根据输入的数据，生成对应的可视化图表…
    """),
)
def chart_node(state: State) -> Command["chart_summarizer"]:
    result = chart_agent.invoke(state)
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="chart_generator"
    )
    return Command(update={"messages": result["messages"]}, goto="chart_summarizer")

chart_summary_agent = create_react_agent(
    llm,
    tools=[],
    prompt=agent_system_prompt("你是一名专业的图表解读员，只会给输入的图表生成清晰易懂的文字说明…"),
)
def chart_summary_node(state: State) -> Command[END]:
    result = chart_summary_agent.invoke(state)
    return Command(
        update={
            "messages": result["messages"],
            "final_answer": result["messages"][-1].content,
        },
        goto=END,
    )

def synthesizer_node(state: State) -> Command[END]:
    relevant_msgs = [
        m.content for m in state.get("messages", [])
        if getattr(m, "name", None) in ("web_researcher", "chart_generator", "chart_summarizer")
    ]
    llm_reply = llm.invoke([
        HumanMessage(content=f"用户问题：{state['user_query']}\n\n参考资料：\n" + "\n---\n".join(relevant_msgs))
    ])
    return Command(
        update={"final_answer": llm_reply.content},
        goto=END,
    )

workflow = StateGraph(State)
# 注册所有节点
workflow.add_node("planner", planner_node)
workflow.add_node("executor", executor_node)
workflow.add_node("web_researcher", web_research_node)
workflow.add_node("chart_generator", chart_node)
workflow.add_node("chart_summarizer", chart_summary_node)
workflow.add_node("synthesizer", synthesizer_node)
# 定义起始节点
workflow.add_edge(START, "planner")
# 编译生成可执行的图
graph = workflow.compile()

state = {
    "messages": [HumanMessage(content="Chart the market cap of top 5 US banks")],
    "user_query": "Chart the market cap of top 5 US banks",
    "enabled_agents": ["web_researcher", "chart_generator", "chart_summarizer", "synthesizer"],
}
graph.invoke(state)

state = {
    "messages": [HumanMessage(content="Identify current US financial regulations")],
    "user_query": "Identify current US financial regulations",
    "enabled_agents": ["web_researcher", "synthesizer"],
}
graph.invoke(state)