生命不息，折腾不止。前四课攒下的三件套（工具手、落盘、审批闸门），今天一把焊成一个真的能干活的工单 Agent。

LangGraph 实战这个系列写到今天，总算可以收尾了。前面四课我们分别攒了三样硬货：第 2 课学会用 interrupt() 给 Agent 装人工审批闸门，第 3 课用 SqliteSaver 让 Agent 能落盘「倒带重来」，第 4 课用 @tool + ToolNode 给 Agent 装了会干活的「手」。

但光有零件不算本事，把它们串起来才叫 Agent。今天我们就手搓一个客服工单 Agent：用户报个订单号，它自动查单、算优惠、起草发货单，然后——重点来了——卡在发货前等人点头，人回一句「发货」才真正往下走。全程断了电、关了机，隔天开回来还能接着上次的进度继续跑。这就是把三节课焊成一个完整闭环。

一、先盘一遍手里的三件套

动手之前，把三个零件各归各位，写代码的时候心里才有数：

  • 工具手（第 4 课）：@tool 装饰器把普通 Python 函数变成模型能调的工具，bind_tools 把工具「塞给」模型，ToolNode 负责真的去执行工具并回传结果。
  • 落盘倒带（第 3 课）：SqliteSaver 把每一步的状态存进 SQLite 数据库，thread_id 是「这一单」的唯一标识，随时能停、随时能续。
  • 人工闸门（第 2 课）：interrupt() 在节点里暂停整张图，把要审批的内容甩给人；人回个结果，用 Command(resume=...) 让它接着往下跑。

串起来的流程张这样：大脑节点（调工具）→ 工具节点（执行）→ 大脑节点（继续想）→ … → 走路到审批节点（卡住等人）→ 人身点头 → 收工。前四课的活儿，全在这一条链子里。

二、先把工单 Agent 的骨架想清楚

写代码前，先想清楚这 Agent 得管哪些事、状态长啥样。我们做的是「自动查单 + 算优惠 + 人工确认发货」三步。状态用一个 TypedDict 装下整单信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import sqlite3
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# 状态：一张表装下整单信息
class OrderState(TypedDict):
    messages: Annotated[list, add_messages]   # 对话历史，自动追加、按 ID 去重
    order_id: str                             # 当前处理的订单号
    approval: str                             # 人工审批结果：approve / reject

messages 用 add_messages 这个 reducer 是 LangGraph 的规矩：模型每轮说的话往上追加，而不是覆盖。这是第 4 课就讲过的基础，今天照用。

接着定义两个工具。教程里查单、算优惠用模拟数据顶一下，生产环境你把 fake_db 换成真库查询就行，工具的口子是完全一样的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@tool
def query_order(order_id: str) -> str:
    """根据订单号查订单。入参是订单号字符串。"""
    fake_db = {
        "1001": {"货品": "机械键盘 ×1", "原价": 399, "状态": "待发货"},
        "1002": {"货品": "显示器 ×1", "原价": 1299, "状态": "待发货"},
    }
    order = fake_db.get(order_id)
    return f"查无此单" if not order else str(order)

@tool
def calc_discount(amount: int, coupon: str) -> str:
    """按优惠码计算折后价。入参：原价金额、优惠码。"""
    discount_map = {"NEW10": 0.9, "VIP20": 0.8}   # 打 9 折 / 8 折
    ratio = discount_map.get(coupon)
    if not ratio:
        return f"优惠码 {coupon} 无效"
    return f"折后实付 {int(amount * ratio)} 元"

tools = [query_order, calc_discount]

工具函数的 docstring 别偷懒——模型就是靠它来理解「这工具是干嘛的、参数传什么」。写得越清楚，模型调得越准。

三、把整张图拼起来

零件齐了，上模型。这种中文客服场景，DeepSeek 便宜又够用，是本篇首选；我直接走中转站 ai.aklibk.com（一个 key 通吃 DeepSeek / Claude / GPT 多家模型，按量付费，比每家单独开账号省心）：

1
2
3
4
5
llm = ChatOpenAI(
    model="deepseek-chat",
    base_url="https://ai.aklibk.com/v1",   # 也可直接填 api.deepseek.com
    api_key="sk-你的key",
).bind_tools(tools)

然后定义三个节点——大脑、手、闸门：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def agent(state: OrderState):
    """大脑节点：决定查单/算优惠，还是给出结论。"""
    resp = llm.invoke(state["messages"])
    return {"messages": [resp]}

def approve_ship(state: OrderState):
    """人工闸门：发货前必须让人点头。"""
    decision = interrupt({
        "提示": "是否确认发货？",
        "货品": "机械键盘 ×1",
        "折后价": 359,
        "下单账号": "峰哥",
    })
    return {"approval": decision}

def should_continue(state: OrderState) -> Literal["tools", "approve_ship"]:
    """条件边：模型还要调工具就继续，否则走向人工审批。"""
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):
        return "tools"
    return "approve_ship"

最后拼图 + 落盘，compile(checkpointer=...) 这一步是关键——不挂 checkpointer，interrupt() 直接报错：

1
2
3
4
5
6
7
8
9
10
11
12
13
graph = StateGraph(OrderState)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))          # 第 4 课的手
graph.add_node("approve_ship", approve_ship)      # 第 2 课的闸门

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")                  # 工具跑完回大脑继续想
graph.add_edge("approve_ship", END)

conn = sqlite3.connect("orders.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)                  # 第 3 课的落盘
app = graph.compile(checkpointer=checkpointer)

整条链你看一眼就懂：大脑想 → 有工具要调就去「手」那儿执行 → 回来继续想 → 直到想完出结论，走进审批闸门 → 等人点头 → 结束。一个标准的 ReAct「想—做—再想」循环，末尾接了一道人肉闸门。

四、跑一遍完整流程

开工。第一段调用，Agent 会自动查单、算优惠，走到发货前主动停下等人：

1
2
3
4
5
6
7
8
config = {"configurable": {"thread_id": "order-1001"}}

result = app.invoke(
    {"messages": [("user", "帮我处理订单 1001，用优惠码 NEW10，准备发货")]},
    config=config,
)
print(result["__interrupt__"])   # 卡在 approve_ship，等人拍板
# 输出类似：[Interrupt(value={'提示': '是否确认发货？', ...})]

看到没，它自己把订单查了、折扣算了，最后把「是否发货」这件事交还给人，而不是闷头就发货。这就是人工闸门的价值——不可逆的动作，永远别让它自己拍板。

人回一句「approve」，接着跑完：

1
2
final = app.invoke(Command(resume="approve"), config=config)
print(final["approval"])   # approve

流程跑通了。再加一层好处：因为挂了 SqliteSaver 落盘，你可以现在就关电脑，明天用同一个 thread_id 再 invoke 一次，它照样记得这单卡在哪儿。断点续跑，免费白送。

五、几个最容易踩的坑

收尾前把血泪坑再钉一遍，省得你回头翻：

  1. interrupt() 必须配 checkpointer + thread_id。少了任何一个，不是报错就是新建了个空线程，根本恢复不了。
  2. Command(resume=...) 是唯一该当 invoke 输入的命令。别手滑传成 Command(update=...) 或普通 dict，否则图看着像「卡死」了。
  3. 节点恢复时是从头重跑。所以发邮件、扣款这类副作用，一定要写在 interrupt() 之后，别写前面，否则每次恢复都会重复执行。
  4. thread_id 用业务 ID（订单号、工单号），别用随机 uuid。出了事你要靠它在数据库里扒这条线程的 checkpoint，业务 ID 一眼定位。

本篇把查单、算优惠用模拟数据顶上了，是个能跑通的闭环骨架；你接上真库、真支付，它就是一个小型生产级客服工单 Agent。LangGraph 实战系列到这儿收尾，前五课连起来看，你已经能把「工具调用 + 状态持久化 + 人工审批」这三块硬骨头啃下来了。

生命不息，折腾不止。LangGraph 系列收官，下一篇开个新坑——DeepSeek Harness 进阶玩法系列第 1 篇：手写一个 dsh 插件，搞懂「一切皆插件」的 Cordis 架构，教你给这个国产 Agent 框架装上一个自定义 provider，接进任何 OpenAI 兼容接口。