LangChain / LangGraph源码学习路线：第 17 篇 Multi-Agent、Subagents 与 Handoffs源码解剖#

1
builder.add_conditional_edges("supervisor", route)

1
from typing import Annotated, TypedDict
2
from langgraph.graph import StateGraph, START, END
3
from langgraph.graph.message import add_messages
4

5

6
class MultiAgentState(TypedDict):
7
    messages: Annotated[list[BaseMessage], add_messages]
8
    active_agent: str | None
9
    allowed_agents: list[str]
10
    handoff_reason: str | None
11
    handoff_history: list[dict]
12
    research_notes: list[dict]
13
    plan_draft: str | None
14
    final_answer: str | None
15

16

17
builder = StateGraph(MultiAgentState)
18
builder.add_node("supervisor", supervisor_node)
19
builder.add_node("travel_planner", travel_planner_node)
20
builder.add_node("research_agent", research_agent_node)
21
builder.add_node("booking_guard", booking_guard_node)
22
builder.add_node("finalize", finalize_node)
23

24
builder.add_edge(START, "supervisor")
25
builder.add_edge("finalize", END)
26

27
multi_agent_graph = builder.compile()

1
Command(goto="research_agent", update={...})

1
def make_handoff(
2
    *,
3
    source: str,
4
    target: str,
5
    reason: str,
6
    state: MultiAgentState,
7
) -> Command:
8
    if target not in state["allowed_agents"]:
9
        return Command(
10
            goto="booking_guard",
11
            update={"handoff_error": f"disallowed target: {target}"},
12
        )
13

14
    return Command(
15
        goto=target,
16
        update={
17
            "active_agent": target,
18
            "handoff_reason": reason,
19
            "handoff_history": [
20
                *state.get("handoff_history", []),
21
                {"from": source, "to": target, "reason": reason},
22
            ],
23
        },
24
    )

1
tool = subagent.as_tool(...)

1
research_tool = research_agent.as_tool(
2
    name="research_travel_facts",
3
    description=(
4
        "Use for bounded fact lookup. Input must include a focused question "
5
        "and a scope. Returns summarized notes with sources."
6
    ),
7
)
8

9

10
def planner_node(state: MultiAgentState) -> dict:
11
    result = research_tool.invoke({
12
        "question": "大阪城到道顿堀晚间交通方式",
13
        "scope": "transport",
14
    })
15
    return {"research_notes": parse_research_tool_result(result)}

1
class ResearchState(TypedDict):
2
    messages: Annotated[list[BaseMessage], add_messages]
3
    task: str
4
    notes: list[dict]
5

6

7
research_builder = StateGraph(ResearchState)
8
research_builder.add_node("search", search_node)
9
research_builder.add_node("verify", verify_node)
10
research_builder.add_node("return_to_parent", return_to_parent_node)
11
research_builder.add_edge(START, "search")
12
research_builder.add_edge("search", "verify")
13
research_builder.add_edge("verify", "return_to_parent")
14

15
research_subgraph = research_builder.compile()
16
parent_builder.add_node("research_agent", research_subgraph)

1
def supervisor_node(state: MultiAgentState, config: RunnableConfig) -> Command:
2
    routing_input = {
3
        "conversation": trim_messages(state["messages"]),
4
        "active_agent": state.get("active_agent"),
5
        "research_notes": summarize_notes(state.get("research_notes", [])),
6
        "plan_draft": state.get("plan_draft"),
7
    }
8

9
    route = supervisor_router.invoke(routing_input, config=config)
10

11
    if route.next == "research_agent":
12
        return make_handoff(
13
            source="supervisor",
14
            target="research_agent",
15
            reason=route.reason,
16
            state=state,
17
        )
18

19
    if route.next == "travel_planner":
20
        return make_handoff(
21
            source="supervisor",
22
            target="travel_planner",
23
            reason=route.reason,
24
            state=state,
25
        )
26

27
    return Command(goto="finalize", update={"handoff_reason": route.reason})

1
def select_messages_for_worker(messages: list[BaseMessage], worker: str) -> list[BaseMessage]:
2
    if worker == "research_agent":
3
        return [
4
            msg for msg in messages
5
            if msg.type in {"human", "ai"} and not contains_private_booking_data(msg)
6
        ][-6:]
7

8
    if worker == "booking_guard":
9
        return [msg for msg in messages if mentions_booking_or_payment(msg)]
10

11
    return messages[-10:]
12

13

14
def worker_input_from_parent(state: MultiAgentState, worker: str) -> dict:
15
    return {
16
        "messages": select_messages_for_worker(state["messages"], worker),
17
        "task": state["handoff_reason"],
18
    }

1
def travel_planner_node(state: MultiAgentState) -> Command[Literal["supervisor"]]:
2
    planner_input = {
3
        "messages": select_messages_for_worker(state["messages"], "travel_planner"),
4
        "research_notes": state.get("research_notes", []),
5
    }
6

7
    plan = planner_chain.invoke(planner_input)
8

9
    return Command(
10
        goto="supervisor",
11
        graph=Command.PARENT,
12
        update={
13
            "messages": [AIMessage(content=plan.summary, name="travel_planner")],
14
            "plan_draft": plan.markdown,
15
            "active_agent": "supervisor",
16
        },
17
    )

1
def call_research_as_tool(state: MultiAgentState) -> dict:
2
    result = research_tool.invoke({
3
        "question": latest_user_question(state["messages"]),
4
        "scope": "facts_only",
5
    })
6

7
    return {
8
        "messages": [AIMessage(content="已完成事实核查。", name="supervisor")],
9
        "research_notes": result["notes"],
10
    }

1
def validate_handoff_target(target: str, state: MultiAgentState) -> Command | None:
2
    if target not in REGISTERED_NODES:
3
        return Command(
4
            goto="supervisor",
5
            update={"handoff_error": f"unknown node: {target}"},
6
        )
7

8
    if target not in state["allowed_agents"]:
9
        return Command(
10
            goto="booking_guard",
11
            update={"handoff_error": f"disallowed handoff: {target}"},
12
        )
13

14
    return None

1
def route_with_handoff_limit(command: Command, state: MultiAgentState) -> Command:
2
    history = state.get("handoff_history", [])
3

4
    if len(history) >= 8:
5
        return Command(
6
            goto="finalize",
7
            update={
8
                "handoff_error": "handoff limit reached",
9
                "final_answer": "任务在多 Agent 协作中未能稳定收敛，需要人工检查。",
10
            },
11
        )
12

13
    last_three = [(item["from"], item["to"]) for item in history[-3:]]
14
    if len(set(last_three)) == 1:
15
        return Command(
16
            goto="supervisor",
17
            update={"handoff_error": "repeated handoff loop detected"},
18
        )
19

20
    return command

1
def safe_worker_node(state: MultiAgentState) -> Command:
2
    try:
3
        return research_agent_node(state)
4
    except RecoverableToolError as exc:
5
        return Command(
6
            goto="supervisor",
7
            graph=Command.PARENT,
8
            update={
9
                "messages": [AIMessage(content=f"研究 Agent 暂时失败：{exc}", name="research_agent")],
10
                "handoff_error": str(exc),
11
                "active_agent": "supervisor",
12
            },
13
        )
14
    except Exception:
15
        raise

1
def should_handoff(state: MultiAgentState, decision: RouteDecision) -> Command:
2
    if decision.confidence < 0.55:
3
        return Command(
4
            goto="supervisor",
5
            update={"handoff_error": "low routing confidence"},
6
        )
7

8
    if decision.next_agent == "booking_guard" and not state.get("user_confirmed_booking"):
9
        return Command(
10
            goto="supervisor",
11
            update={"handoff_error": "booking handoff requires user confirmation"},
12
        )
13

14
    return make_handoff(
15
        source="supervisor",
16
        target=decision.next_agent,
17
        reason=decision.reason,
18
        state=state,
19
    )

1
def load_worker_memory(user_id: str, worker: str, store: BaseStore) -> dict:
2
    global_preferences = store.search(("user", user_id, "preferences"))
3
    worker_memory = store.search(("user", user_id, "agents", worker))
4

5
    return {
6
        "global_preferences": redact_sensitive(global_preferences),
7
        "worker_memory": worker_memory,
8
    }

1
def tag_handoff_config(config: RunnableConfig, command: Command, state: MultiAgentState) -> RunnableConfig:
2
    return {
3
        **config,
4
        "tags": [*config.get("tags", []), "multi-agent", "handoff"],
5
        "metadata": {
6
            **config.get("metadata", {}),
7
            "from_agent": state.get("active_agent") or "supervisor",
8
            "to_agent": command.goto,
9
            "handoff_reason": state.get("handoff_reason"),
10
            "handoff_count": len(state.get("handoff_history", [])),
11
        },
12
    }

1
def research_agent_node(state: MultiAgentState) -> Command:
2
    rag_result = rag_graph.invoke({
3
        "question": state["handoff_reason"],
4
        "documents": [],
5
        "answer": None,
6
    })
7

8
    return Command(
9
        goto="supervisor",
10
        graph=Command.PARENT,
11
        update={
12
            "research_notes": extract_notes(rag_result),
13
            "messages": [AIMessage(content=rag_result["answer"], name="research_agent")],
14
        },
15
    )