第 10 章 从单图到团队：多 Agent 架构

Agent = 子图（通常是 ReAct 循环）+ 专属 prompt + 专属工具集
多 Agent 系统 = 若干 Agent 子图 + 协作拓扑（谁调谁、怎么传话）

from typing import Annotated, TypedDict
import operator
from langchain_core.messages import AnyMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.types import Command

class TeamState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]   # 团队共享消息黑板
    resume_text: str
    parsed: dict                # parser 产物
    tech_report: str            # tech 产物
    final_report: str           # report 产物

# ---- 三个专职 Agent：演示版各是一个简单节点 ----
# 真实实现里，每个都是第 8 章那样的 ReAct 子图
def parser_agent(state: TeamState) -> Command:
    lines = [l.strip() for l in state["resume_text"].splitlines() if l.strip()]
    parsed = {
        "name": lines[0].split("：")[0],
        "skills": next((l.replace("- 技能：", "").split("、") for l in lines if l.startswith("- 技能")), []),
    }
    # 干完活，控制权交回主管
    return Command(update={"parsed": parsed}, goto="supervisor")

def tech_agent(state: TeamState) -> Command:
    score = min(100, 30 + len(state["parsed"]["skills"]) * 15)
    report = f"技术评估：{state['parsed']['name']} 得分 {score}"
    return Command(update={"tech_report": report}, goto="supervisor")

def report_agent(state: TeamState) -> Command:
    final = f"{state['tech_report']}\n解析字段数：{len(state['parsed'])}"
    return Command(update={"final_report": final}, goto=END)   # 报告完成即收工

# ---- 主管：读状态，决定下一个干活的 Agent ----
def supervisor(state: TeamState) -> Command:
    if not state.get("parsed"):
        return Command(goto="parser_agent")            # 还没解析 → 先解析
    if not state.get("tech_report"):
        return Command(goto="tech_agent")              # 没评估 → 去评估
    return Command(goto="report_agent")                # 齐了 → 出报告

builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor)
builder.add_node("parser_agent", parser_agent)
builder.add_node("tech_agent", tech_agent)
builder.add_node("report_agent", report_agent)
builder.add_edge(START, "supervisor")
graph = builder.compile()

result = graph.invoke({
    "resume_text": "李思远：Python 后端\n- 技能：Python、FastAPI、PostgreSQL",
    "messages": [],
})
print(result["final_report"])
# 技术评估：李思远 得分 75
# 解析字段数：2

def parser_agent(state: TeamState) -> Command:
    lines = [l.strip() for l in state["resume_text"].splitlines() if l.strip()]
    parsed = {"name": lines[0].split("：")[0]}
    # 不回主管，直接点名下一个接手者
    return Command(update={"parsed": parsed}, goto="tech_agent")

def tech_agent(state: TeamState) -> Command:
    report = f"技术评估完成：{state['parsed']['name']}"
    return Command(update={"tech_report": report}, goto="report_agent")

def report_agent(state: TeamState) -> Command:
    return Command(update={"final_report": state["tech_report"] + "（终版）"}, goto=END)

builder = StateGraph(TeamState)
builder.add_node("parser_agent", parser_agent)
builder.add_node("tech_agent", tech_agent)
builder.add_node("report_agent", report_agent)
builder.add_edge(START, "parser_agent")
graph = builder.compile()

result = graph.invoke({
    "resume_text": "李思远：Python 后端",
    "messages": [],
})
print(result["final_report"])   # 技术评估完成：李思远（终版）

# 每个 Agent = ReAct 子图（独立 schema）+ 专属 prompt + 专属工具
tech_agent_graph = build_react_graph(
    system_prompt=TECH_PROMPT,
    tools=[lookup_company, get_job_requirements],
).compile()

# 在 supervisor 图里注册（模式二：包装翻译）
def tech_agent(state: TeamState) -> Command:
    sub_result = tech_agent_graph.ainvoke(...)   # 或同步 invoke
    return Command(update={...}, goto="supervisor")

builder.add_node("tech_agent", tech_agent)