1
先运行 Supervisor + Subagent
2
-> 对比 Subagent / Router / Handoff
3
-> 先看 Handoff 完整代码和 invoke 调用链
4
-> 再拆 ToolRuntime、Command、middleware 等部件
5
-> 判断什么时候不该拆
6
-> 主动练习
7
-> 最后集中排错

1
def route_question(state: dict) -> str:
2
    question = state["question"]
3
    return "research" if "知识库" in question else "answer"

1
from collections.abc import Callable  # Python 标准库类型：描述 handler 的输入和输出。
2
from typing import Any, cast
3

4
from typing_extensions import NotRequired
5

6
from langchain.agents import AgentState, create_agent  # AgentState 提供 messages；create_agent 组装 Agent 循环。
7
from langchain.agents.middleware import (
8
    ModelRequest,  # 本次模型调用的框架请求，带着 State、Prompt 和 tools。
9
    ModelResponse,  # 模型调用完成后，handler 返回给 middleware 的响应对象。
10
    wrap_model_call,  # 将函数注册为每次模型调用前都会经过的 middleware 钩子。
11
)
12
from langchain.messages import SystemMessage, ToolMessage
13
from langchain_core.runnables import RunnableConfig
14
from langchain_core.tools import BaseTool
15
from langchain.tools import ToolRuntime, tool  # ToolRuntime 由框架在调用工具时自动注入。
16
from langgraph.checkpoint.memory import InMemorySaver  # 按 thread_id 保存和恢复 State 快照。
17
from langgraph.types import Command  # 工具返回它，让框架一次合并多个 State 字段。
18

19
from app.multi_agent import build_llm
20

21

22
class HandoffState(AgentState):
23
    # 继承来的 messages 保存对话和工具消息；这个字段额外记录当前角色模式。
24
    active_agent: NotRequired[str]
25

26

27
@tool
28
def transfer_to_support(
29
    runtime: ToolRuntime[None, HandoffState],
30
) -> Command:
31
    """发现技术问题时，将当前会话转入支持角色模式。"""
32
    # 模型决定调用本工具后，框架把本次工具调用的运行时信息放进 runtime。
33
    # 返回 Command 后，框架把 update 合并回共享 State，而不是由工具手动修改 State。
34
    return Command(
35
        update={
36
            "messages": [
37
                ToolMessage(
38
                    content="已转交给支持角色。",
39
                    tool_call_id=runtime.tool_call_id,
40
                )
41
            ],
42
            "active_agent": "support_agent",
43
        }
44
    )
45

46

47
@tool
48
def check_technical_issue(question: str) -> str:
49
    """支持角色使用的演示工具：记录并检查技术问题。"""
50
    return f"技术支持已收到问题：{question}"
51

52

53
def select_prompt(state: HandoffState) -> str:
54
    """根据共享 State 选择下一次模型调用要用的角色 Prompt。"""
55
    if state.get("active_agent") == "support_agent":
56
        return "你是支持角色。先调用 check_technical_issue，再给出简短解决建议。"
57
    return "你是分诊角色。遇到技术问题必须调用 transfer_to_support。"
58

59

60
@wrap_model_call
61
def apply_role_config(
62
    request: ModelRequest,  # 框架注入本轮即将发送给模型的请求。
63
    handler: Callable[[ModelRequest], ModelResponse],  # 交给它才会继续真正的模型调用。
64
) -> ModelResponse:
65
    """每次模型调用前，按 active_agent 覆盖本次的 Prompt 与 tools。"""
66
    # 每一次 Agent 准备调用模型都会先到这里；同一 thread_id 的旧 State 已由 checkpointer 恢复。
67
    state = cast(HandoffState, request.state)
68
    prompt = select_prompt(state)
69

70
    available_tools: list[BaseTool | dict[str, Any]]
71
    if state.get("active_agent") == "support_agent":
72
        available_tools = [check_technical_issue]
73
    else:
74
        available_tools = [transfer_to_support]
75

76
    updated_request = request.override(
77
        # 用 State 推导出的 Prompt 和工具子集覆盖本次请求，不改共享 State。
78
        system_message=SystemMessage(content=prompt),
79
        tools=available_tools,
80
    )
81
    # handler 接收临时请求后，框架才真正调用模型；模型可能回答，也可能再发起工具调用。
82
    return handler(updated_request)

1
state = cast(HandoffState, request.state)   # 告诉类型检查器具体 State 类型
2
prompt = select_prompt(state)               # 得到 Prompt 字符串
3
updated_request = request.override(         # 覆盖本次模型请求
4
    system_message=SystemMessage(content=prompt),
5
    tools=available_tools,
6
)

1
middleware=[apply_role_config]

1
Agent 循环
2
  -> apply_role_config（middleware 钩子）
3
  -> select_prompt（返回 Prompt 字符串）
4
  -> request.override（覆盖本次请求）
5
  -> handler(updated_request)
6
  -> 模型

1
handoff_agent = create_agent(
2
    model=build_llm(),
3
    # 先注册全部候选工具；middleware 会在每次模型调用前按角色选出可见子集。
4
    tools=[transfer_to_support, check_technical_issue],
5
    state_schema=HandoffState,
6
    middleware=[apply_role_config],
7
    # 每轮结束保存 State；下一次使用同一个 thread_id 的 invoke 才能恢复 active_agent。
8
    checkpointer=InMemorySaver(),
9
)
10

11
thread_config: RunnableConfig = {
12
    "configurable": {"thread_id": "handoff-demo-user-1"},
13
}
14

15
first_result = handoff_agent.invoke(
16
    # 首次输入进入 messages；分诊 Prompt 会引导模型调用 transfer_to_support。
17
    {"messages": [{"role": "user", "content": "我的登录一直报错"}]},
18
    config=thread_config,
19
)
20

21
# 相同 thread_id 取回 support_agent 状态；middleware 因而改用支持 Prompt 和支持工具处理追问。
22
second_result = handoff_agent.invoke(
23
    {"messages": [{"role": "user", "content": "错误码是 401"}]},
24
    config=thread_config,
25
)
26

27
print(second_result["messages"][-1].content)

1
1. LangChain 准备第一次调用模型。
2
2. 因为 middleware=[apply_role_config]，框架先调用 apply_role_config(request, handler)。
3
3. apply_role_config 调用 select_prompt(state)。
4
4. active_agent 不存在，所以 select_prompt 返回“分诊角色”的 Prompt；available_tools 只有 transfer_to_support。
5
5. handler(updated_request) 继续真正的模型调用。模型看到“技术问题必须转交”，于是发起 transfer_to_support tool call。
6
6. 框架执行 transfer_to_support，拿到 Command(update=...)，把 ToolMessage 和 active_agent="support_agent" 写回 State。
7
7. Agent 工具循环准备下一次模型调用；middleware 再次运行。
8
8. 这次 select_prompt 读到 active_agent="support_agent"，返回“支持角色”的 Prompt；available_tools 改为 check_technical_issue。
9
9. 模型以支持角色配置继续调用工具或给出答案。

1
first_result = handoff_agent.invoke(
2
    {"messages": [{"role": "user", "content": "我的登录一直报错"}]},
3
    config=thread_config,
4
)

1
second_result = handoff_agent.invoke(
2
    {"messages": [{"role": "user", "content": "错误码是 401"}]},
3
    config=thread_config,
4
)

1
相同 thread_id
2
-> InMemorySaver 取回第一次的 State
3
-> 追加“错误码是 401”
4
-> middleware 读到 active_agent="support_agent"
5
-> 直接使用支持 Prompt 和 check_technical_issue
6
-> 模型继续回答
7
-> 保存更新后的 State

1
State = Agent 当前共享的事实，例如 active_agent="support_agent"
2
ModelRequest = middleware 根据 State 整理出的本轮模型请求

1
State
2
  -> request.state
3
  -> select_prompt(state)                 # 选择 Prompt
4
  -> 根据 active_agent 选择 tools 子集
5
  -> request.override(...)                # 生成本轮的新 ModelRequest
6
  -> handler(updated_request)             # 继续真实模型调用
7
  -> 模型读取 messages、system Prompt、tools

1
state["active_agent"] == "support_agent"

1
print("before:", request.state)
2
print("selected prompt:", prompt)
3
print("selected tools:", [tool.name for tool in available_tools])
4

5
updated_request = request.override(
6
    system_message=SystemMessage(content=prompt),
7
    tools=available_tools,
8
)
9

10
print("after override:", updated_request.system_message)
11
return handler(updated_request)

1
@tool
2
def check_technical_issue(question: str) -> str:
3
    """支持角色使用的演示工具：记录并检查技术问题。"""
4
    return f"技术支持已收到问题：{question}"

1
{
2
  "name": "check_technical_issue",
3
  "arguments": {"question": "我的登录一直报错"}
4
}

1
@tool
2
def transfer_to_support(
3
    runtime: ToolRuntime[None, HandoffState],
4
) -> Command:
5
    return Command(
6
        update={
7
            "messages": [
8
                ToolMessage(
9
                    content="已转交给支持角色。",
10
                    tool_call_id=runtime.tool_call_id,
11
                )
12
            ],
13
            "active_agent": "support_agent",
14
        }
15
    )

1
ToolRuntime[None, HandoffState]
2
#            ^     ^
3
#            |     runtime.state 的类型
4
#            没有额外自定义 context 类型

1
ToolRuntime[ContextT, StateT]

1
None          -> ContextT，本次没有自定义运行时 context
2
HandoffState  -> StateT，runtime.state 的类型

1
Command(update={"active_agent": "support_agent"})

1
runtime.tool_call_id

1
AIMessage(
2
    tool_calls=[
3
        {
4
            "name": "transfer_to_support",
5
            "args": {},
6
            "id": "call_123",
7
        }
8
    ]
9
)

1
1. 根据 name 找到 transfer_to_support Tool。
2
2. 创建本次执行用的 ToolRuntime，并把 id="call_123" 放入 runtime.tool_call_id。
3
3. 调用 transfer_to_support(runtime)。

1
runtime.tool_call_id  # "call_123"

1
ToolMessage(
2
    content="已转交给支持角色。",
3
    tool_call_id=runtime.tool_call_id,
4
)

1
runtime.state              # HandoffState：一组共享 State 字段
2
runtime.tool_call_id       # "call_123"：当前工具调用的关联 ID
3
runtime.context            # None：本例没有自定义 context

1
@tool
2
def check_technical_issue(question: str) -> str:
3
    """检查技术问题。"""
4
    return f"技术支持已收到问题：{question}"

1
工具返回字符串
2
    -> Agent / ToolNode 自动创建 ToolMessage
3
    -> 自动把 ToolMessage 追加到 messages
4
    -> 模型读取工具结果并继续回答

1
return Command(
2
    update={
3
        "messages": [ToolMessage(...)],
4
        "active_agent": "support_agent",
5
    }
6
)

1
工具返回 Command
2
    -> 框架读取 Command.update
3
    -> 写入 ToolMessage
4
    -> 写入 active_agent="support_agent"
5
    -> 下一次 middleware 根据 active_agent 切换 Prompt 和 tools

1
以前：工具只返回数据，框架自动把数据变成消息并写入 messages
2
现在：工具返回控制指令，框架按 update 修改多个 State 字段

1
class HandoffState(AgentState):
2
    active_agent: NotRequired[str]
3
    retry_count: NotRequired[int]
4
    audit_log: NotRequired[list[str]]

1
return Command(
2
    update={
3
        "messages": [ToolMessage(...)],
4
        "active_agent": "support_agent",
5
        "retry_count": 1,
6
        "audit_log": ["已转交支持角色"],
7
    }
8
)

1
messages       对话和工具消息
2
active_agent   当前角色模式
3
retry_count    当前流程的重试次数
4
audit_log      当前流程的审计记录

1
thread_id      由 invoke(config=...) 指定，用来定位会话线程
2
model 参数     由 Agent 配置或 middleware 控制
3
context        通常在 invoke(context=...) 时传入的调用上下文

1
Command(update=...)  -> 更新 State 数据
2
config               -> 配置本次运行和 thread_id
3
middleware           -> 动态修改本次模型请求
4
context              -> 提供本次调用的外部依赖或用户信息

1
Command(goto="support_agent", graph=Command.PARENT)
2
builder.add_node("support_agent", call_support_agent)

1
list[BaseTool | dict[str, Any]]

1
available_tools: list[BaseTool | dict[str, Any]]

1
from typing import cast
2

3
state = cast(HandoffState, request.state)
4
prompt = select_prompt(state)

1
from langchain_core.runnables import RunnableConfig
2

3
thread_config: RunnableConfig = {
4
    "configurable": {"thread_id": "handoff-demo-user-1"},
5
}