05 Handoffs 交接：靠状态切换行为，像客服转接电话

1
from typing import Callable
2
from langchain.agents import AgentState, create_agent
3
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
4
from langchain.tools import tool, ToolRuntime
5
from langchain.messages import ToolMessage
6
from langgraph.checkpoint.memory import InMemorySaver
7
from langgraph.types import Command
8

9
# 1) 定义带 current_step 的状态
10
class SupportState(AgentState):
11
    """记录当前处于哪一步。"""
12
    current_step: str = "triage"
13
    warranty_status: str | None = None
14

15
# 2) 工具通过 Command 更新 current_step
16
@tool
17
def record_warranty_status(status: str, runtime: ToolRuntime[None, SupportState]) -> Command:
18
    """Record warranty status and transition to next step."""
19
    return Command(update={
20
        "messages": [
21
            ToolMessage(
22
                content=f"Warranty status recorded: {status}",
23
                tool_call_id=runtime.tool_call_id,
24
            )
25
        ],
26
        "warranty_status": status,
27
        "current_step": "specialist",   # ← 迁移到下一步
28
    })
29

30
# 3) 中间件：按 current_step 套用不同配置（提示词 + 工具集）
31
@wrap_model_call
32
def apply_step_config(
33
    request: ModelRequest,
34
    handler: Callable[[ModelRequest], ModelResponse],
35
) -> ModelResponse:
36
    """根据 current_step 配置 Agent 行为。"""
37
    step = request.state.get("current_step", "triage")
38

39
    configs = {
40
        "triage": {
41
            "prompt": "Collect warranty information...",
42
            "tools": [record_warranty_status],
43
        },
44
        "specialist": {
45
            "prompt": "Provide solutions based on warranty: {warranty_status}",
46
            "tools": [provide_solution, escalate],
47
        },
48
    }
49

50
    config = configs[step]
51
    request = request.override(                      # 改写这次模型调用的配置
52
        system_prompt=config["prompt"].format(**request.state),
53
        tools=config["tools"],
54
    )
55
    return handler(request)
56

57
# 4) 创建带中间件的 Agent
58
agent = create_agent(
59
    model,
60
    tools=[record_warranty_status, provide_solution, escalate],
61
    state_schema=SupportState,
62
    middleware=[apply_step_config],
63
    checkpointer=InMemorySaver(),   # 关键：状态要跨回合持久化
64
)

1
from typing import Literal
2
from langchain.agents import AgentState, create_agent
3
from langchain.messages import AIMessage, ToolMessage
4
from langchain.tools import tool, ToolRuntime
5
from langgraph.graph import StateGraph, START, END
6
from langgraph.types import Command
7
from typing_extensions import NotRequired
8

9
# 1) 状态里记录"当前活跃的是哪个 Agent"
10
class MultiAgentState(AgentState):
11
    active_agent: NotRequired[str]
12

13
# 2) 两个方向的交接工具（transfer_to_sales / transfer_to_support）
14
#    实现同上：取 AIMessage + 造 ToolMessage + goto + Command.PARENT
15

16
# 3) 各自带"换人"工具的 Agent
17
sales_agent = create_agent(
18
    model="google_genai:gemini-3.6-flash",
19
    tools=[transfer_to_support],
20
    system_prompt="You are a sales agent... If asked about technical issues, transfer to the support agent.",
21
)
22
support_agent = create_agent(
23
    model="google_genai:gemini-3.6-flash",
24
    tools=[transfer_to_sales],
25
    system_prompt="You are a support agent... If asked about pricing, transfer to the sales agent.",
26
)
27

28
# 4) 每个 Agent 一个节点
29
def call_sales_agent(state: MultiAgentState) -> Command:
30
    return sales_agent.invoke(state)
31

32
def call_support_agent(state: MultiAgentState) -> Command:
33
    return support_agent.invoke(state)
34

35
# 5) 路由：最后一条 AIMessage 没有工具调用 → 结束；否则回到 active_agent
36
def route_after_agent(state: MultiAgentState) -> Literal["sales_agent", "support_agent", "__end__"]:
37
    messages = state.get("messages", [])
38
    if messages:
39
        last_msg = messages[-1]
40
        if isinstance(last_msg, AIMessage) and not last_msg.tool_calls:
41
            return "__end__"        # 没有待办工具调用 = 这一轮说完了
42
    active = state.get("active_agent", "sales_agent")
43
    return active if active else "sales_agent"
44

45
def route_initial(state: MultiAgentState) -> Literal["sales_agent", "support_agent"]:
46
    return state.get("active_agent") or "sales_agent"
47

48
# 6) 组图
49
builder = StateGraph(MultiAgentState)
50
builder.add_node("sales_agent", call_sales_agent)
51
builder.add_node("support_agent", call_support_agent)
52
builder.add_conditional_edges(START, route_initial, ["sales_agent", "support_agent"])
53
builder.add_conditional_edges("sales_agent", route_after_agent, ["sales_agent", "support_agent", END])
54
builder.add_conditional_edges("support_agent", route_after_agent, ["sales_agent", "support_agent", END])
55
graph = builder.compile()