18. 多代理系统 —— 监督者模式

ORDERS = {
    "12345": {"item": "无线耳机", "amount": 89,
              "status": "in_transit", "status_text": "运输中(明天送达)"},
    "67890": {"item": "机械键盘", "amount": 129,
              "status": "delivered", "status_text": "已送达"},
    "24680": {"item": "降噪耳机", "amount": 249,
              "status": "delivery_failed", "status_text": "配送失败(已退回 —— 未找到收件人)"},
}

from langchain.tools import tool
from langchain.agents import create_agent
 
@tool
def get_order_status(order_id: str) -> str:
    """根据订单号查询商品、支付金额和发货状态。"""
    order = ORDERS.get(order_id)
    if order is None:
        return f"未找到订单 {order_id}。"
 
    return (f"订单 {order_id}: {order['item']}, "
            f"${order['amount']:,}, 状态: {order['status_text']}")
 
 
order_agent = create_agent(
    name="order_expert",
    model="openai:gpt-5.4-mini",
    tools=[get_order_status],
    system_prompt=(
        "你是订单查询专家。查询订单状态并回答。\n"
        "在你的最终答案中包含订单号、商品、支付金额和发货状态。\n"
        "不要判断是否应当退款,也不要以任何方式提及退款。你的职责仅限于订单查询和状态报告。"
    ),
)

@tool
def check_refund_eligibility(order_id: str) -> str:
    """判断订单是否符合退款资格。只有配送失败的订单才符合条件。"""
    order = ORDERS.get(order_id)
    if order is None:
        return f"未找到订单 {order_id}。"
 
    if order["status"] == "delivery_failed":
        return f"订单 {order_id} 符合退款资格(原因: 配送失败)。"
 
    return f"订单 {order_id} 不符合退款资格(当前状态: {order['status_text']})。"
 
 
@tool
def issue_refund(order_id: str) -> str:
    """处理退款。调用此工具前请务必先用 check_refund_eligibility 确认资格。"""
    order = ORDERS.get(order_id)
    if order is None:
        return f"未找到订单 {order_id}。"
 
    return (f"退款完成: 订单 {order_id} 的 ${order['amount']:,} "
            f"将在 3–5 个工作日内退回。(审批编号: RF-{order_id})")
 
 
refund_agent = create_agent(
    name="refund_expert",
    model="openai:gpt-5.4-mini",
    tools=[check_refund_eligibility, issue_refund],
    system_prompt=(
        "你是退款处理专家。\n"
        "务必先用 check_refund_eligibility 确认资格,只有"
        "在那之后才调用 issue_refund。\n"
        "如果你处理了退款,在最终答案中包含金额和审批编号。\n"
        "如果订单不符合资格,不要处理它 —— 而是报告原因。"
    ),
)

from typing import Literal
from langgraph.graph import MessagesState
from langgraph.types import Command
 
def some_node(state: MessagesState) -> Command[Literal["refund_expert_proxy"]]:
    return Command(
        goto="refund_expert_proxy",             # 去哪里: 下一个要运行的节点
        update={"messages": [...]},             # 传递什么: 工作者的汇报(加入 State 的载荷)
    )

from typing import Literal
from pydantic import BaseModel, Field
 
class Route(BaseModel):
    reason: str = Field(description="做出此决定的原因。")
    next: Literal["order_expert_proxy", "refund_expert_proxy", "FINISH"] = Field(
        description="下一个要运行的工作者节点。如果请求已完全处理则为 FINISH。"
    )

from typing import Literal
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.types import Command
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
 
supervisor_llm = ChatOpenAI(model="gpt-5.4")
 
SUPERVISOR_PROMPT = (
    "你是一个客服团队的监督者。你管理两个工作者:\n"
    "- order_expert_proxy: 查询订单状态。\n"
    "- refund_expert_proxy: 检查退款资格并处理退款。\n"
    "要判断是否需要退款,你必须先检查订单状态。\n"
    "一次只指派一个工作者,一旦请求完全处理完毕就回复 FINISH。"
)
 
 
def supervisor(
    state: MessagesState,
) -> Command[Literal["order_expert_proxy", "refund_expert_proxy", "__end__"]]:
    messages = [{"role": "system", "content": SUPERVISOR_PROMPT}, *state["messages"]]
    decision = supervisor_llm.with_structured_output(Route).invoke(messages)
    print(f"[supervisor] → {decision.next} ({decision.reason})")
 
    if decision.next == "FINISH":
        final = supervisor_llm.invoke(
            [{"role": "system", "content": "根据迄今为止的对话,写一份给客户的回复。"},
             *state["messages"]]
        )
        return Command(goto=END, update={"messages": [final]})   # 交接给 END
 
    return Command(goto=decision.next)                            # 交接给工作者代理

def order_expert_proxy(state: MessagesState) -> Command[Literal["supervisor"]]:
    result = order_agent.invoke(state)
    last = result["messages"][-1]
    return Command(
        goto="supervisor",
        update={"messages": [HumanMessage(content=last.content, name="order_expert")]},
    )
 
 
def refund_expert_proxy(state: MessagesState) -> Command[Literal["supervisor"]]:
    result = refund_agent.invoke(state)
    last = result["messages"][-1]
    return Command(
        goto="supervisor",
        update={"messages": [HumanMessage(content=last.content, name="refund_expert")]},
    )

builder = StateGraph(MessagesState)
builder.add_node("supervisor", supervisor)
builder.add_node("order_expert_proxy", order_expert_proxy)
builder.add_node("refund_expert_proxy", refund_expert_proxy)
builder.add_edge(START, "supervisor")
 
team = builder.compile()
 
result = team.invoke(
    {"messages": [HumanMessage(
        content="订单 24680 一直没到。如果有问题,请退款。"
    )]},
    config={"recursion_limit": 15},
)

[supervisor] → order_expert_proxy (在决定退款之前需要先检查订单状态。)
[supervisor] → refund_expert_proxy (已确认配送失败,所以检查退款资格并处理它。)
[supervisor] → FINISH (订单检查和退款处理都已完成。)

from langchain.tools import tool
 
# order_agent 和 refund_agent 是 18.2 中的工作者
 
@tool
def lookup_order(request: str) -> str:
    """查询订单的商品、支付金额和发货状态。当你需要知道订单状态时使用此工具。
 
    输入: 一个自然语言查询请求(例如,'告诉我订单 24680 的发货状态')。
    """
    print("[tool call] lookup_order")
    print(f"    request: {request}")
 
    result = order_agent.invoke({"messages": [{"role": "user", "content": request}]})
    return result["messages"][-1].content
 
 
@tool
def handle_refund(request: str) -> str:
    """检查退款资格并处理退款。当客户想要退款且你已经确认订单状态时使用此工具。
 
    输入: 一个自然语言退款请求。包含订单号以及查询所确认的发货状态。
    (例如,'订单 24680 处于配送失败状态。如果符合资格请处理退款。')
    """
    print("[tool call] handle_refund")
    print(f"    request: {request}")
 
    result = refund_agent.invoke({"messages": [{"role": "user", "content": request}]})
    return result["messages"][-1].content

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
 
TOOL_SUPERVISOR_PROMPT = (
    "你是一个客服团队的监督者。\n"
    "要判断是否需要退款,你必须先检查订单状态。\n"
    "不要自己做实际工作 —— 委派给工作者。\n"
    "工作者看不到这段对话。当你委派时,把他们需要的一切都放进请求里。\n"
    "当所有工作完成后,把工作者的结果综合成一份给客户的回复。"
)
 
supervisor_agent = create_agent(
    model="openai:gpt-5.4",
    tools=[lookup_order, handle_refund],
    system_prompt=TOOL_SUPERVISOR_PROMPT,
)
 
result = supervisor_agent.invoke(
    {"messages": [HumanMessage(
        content="订单 24680 一直没到。如果有问题,请退款。"
    )]}
)
 
print("\n\n[final response]")
print(result["messages"][-1].content)

[tool call] lookup_order
    request: 客户说订单 24680 还没到。为了判断是否需要退款,
             请给我订单 24680 的商品、支付金额和当前发货状态。
[tool call] handle_refund
    request: 订单 24680 是一副降噪耳机,$249,其发货状态
             已确认为'配送失败(已退回 —— 未找到收件人)'。客户正在
             请求退款,所以检查资格,如果符合就处理它。
 
 
[final response]
我查了一下,订单 24680 处于配送失败状态(已退回 —— 未找到收件人)。
 
它符合退款资格,我已经完成了退款。
- 商品: 降噪耳机
- 退款金额: $249
- 退款审批编号: RF-24680
 
根据您的支付方式,退款通常需要几个工作日才能到账。

from langgraph.checkpoint.memory import InMemorySaver
 
supervisor_agent = create_agent(
    model="openai:gpt-5.4",
    tools=[lookup_order, handle_refund],
    system_prompt=TOOL_SUPERVISOR_PROMPT,
    checkpointer=InMemorySaver(),      # 检查点器只放在顶层代理上
)
 
config = {"configurable": {"thread_id": "cs-1"}}
 
supervisor_agent.invoke(
    {"messages": [HumanMessage(content="我的订单 12345 在哪里?")]},
    config,
)
follow_up = supervisor_agent.invoke(
    {"messages": [HumanMessage(content="那个多少钱?")]},
    config,
)
print(follow_up["messages"][-1].content)

您的订单 12345,无线耳机,是 $89。

from langgraph_supervisor import create_supervisor
from langchain_openai import ChatOpenAI
 
workflow = create_supervisor(
    agents=[order_agent, refund_agent],   # 每个代理都必须设置了 name
    model=ChatOpenAI(model="gpt-5.4"),
    prompt="把订单检查指派给 order_expert,把退款指派给 refund_expert。",
)
app = workflow.compile()