用 LangChain 1.x 的 create_agent 做一个客服退款助手

用 LangChain 1.x 的 create_agent 做一个客服退款助手：

• get_order 只读，自动执行；
• refund 会修改数据，由 HumanInTheLoopMiddleware 拦下，人工批准或拒绝之后才继续；
• refund 内部还会再做一次业务校验（金额、状态），批准了也不能超额退款；
• ToolCallLimitMiddleware 限制单次运行的工具调用次数。

先看清成品（box 上真实运行，订单数据是虚构的）

• qwen3 0.6B：A1001 退 199 元时暂停等待审批，批准后执行，审计记录 [('A1001', 199.0, '耳机杂音')]；拒绝时模型收到“User rejected...”，审计记录为空；A1002（运输中）由模型查询后直接告知不能退款。
• qwen2.5 1.5B：在这个任务上失败了，两种失败情况都在调试示例里如实记录。
• pytest 5 项通过（用脚本化的测试替身，不需要 Ollama）。

合规提醒：本系列只用本地 Ollama 模型，数据不出本机。如果换成云端模型，请先确认数据合规和服务商条款，API Key 放在环境变量里，不要共享账号。模型的许可证以它在 Ollama 或 Hugging Face 页面上写的为准。示例数据都是我自己编的。本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

安装：pip install -r requirements.txt

关键 API（来自 langchain 1.4.6 源码和文档）

文件结构

60-langchain-C-src/
├── agent.py
├── requirements.txt
└── test_agent.py

第 1 步：智能体

"""客服退款助手：LangChain create_agent + 人工审批。
- get_order 只读，自动执行；refund 会改数据，由 HumanInTheLoopMiddleware 拦下，必须人工批准、修改或拒绝
- refund 内部还会再校验一次（金额不超过实付、状态允许），不把安全寄托在模型或审批人身上
- ToolCallLimitMiddleware 限制单次运行的工具调用次数，防止小模型陷入循环
订单数据是教程虚构的。"""
import os
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware, ToolCallLimitMiddleware
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

MODEL = os.environ.get("OLLAMA_MODEL", "qwen3-local:0.6b")   # 实测 qwen2.5 1.5B 在本例里不调用 refund，见教程
ORDERS = {
    "A1001": {"item": "蓝牙耳机", "paid": 199.0, "status": "已签收", "refunded": 0.0},
    "A1002": {"item": "机械键盘", "paid": 459.0, "status": "运输中", "refunded": 0.0},
}
AUDIT = []

@tool
def get_order(order_id: str) -> dict:
    """查询订单：商品、实付金额、状态、已退金额。order_id 形如 A1001。"""
    o = ORDERS.get(order_id.strip().upper())
    return {"order_id": order_id, **o} if o else {"error": f"订单 {order_id} 不存在"}

@tool
def refund(order_id: str, amount: float, reason: str) -> str:
    """给订单退款。只有已签收的订单可以退，累计退款不能超过实付金额。"""
    o = ORDERS.get(order_id.strip().upper())
    if not o:
        return f"失败：订单 {order_id} 不存在"
    if o["status"] != "已签收":
        return f"失败：订单状态为 {o['status']}，不能退款"
    if amount <= 0 or o["refunded"] + amount > o["paid"]:
        return f"失败：可退金额为 {o['paid'] - o['refunded']:.2f}"
    o["refunded"] += amount
    AUDIT.append((order_id, amount, reason))
    return f"成功：订单 {order_id} 已退款 {amount:.2f} 元"

SYSTEM = ("你是网店客服。先用 get_order 查订单，再决定是否调用 refund。"
          "“已签收”的订单可以退款，“运输中”的不能退。退款金额不能超过实付金额。回答用中文，简短。")

def build(model=None):
    if model is None:
        from langchain_ollama import ChatOllama
        model = ChatOllama(model=MODEL, temperature=0)
    return create_agent(model, tools=[get_order, refund], system_prompt=SYSTEM,
                        middleware=[HumanInTheLoopMiddleware(interrupt_on={"refund": True, "get_order": False}),
                                    ToolCallLimitMiddleware(run_limit=6)],
                        checkpointer=InMemorySaver())

def pending(agent, config):
    """返回待审批的工具调用列表（没有则为空）。"""
    st = agent.get_state(config)
    return [a for it in st.interrupts for a in it.value["action_requests"]]

def run_cli(text, thread="demo", decide=input):
    agent, cfg = build(), {"configurable": {"thread_id": thread}}
    agent.invoke({"messages": [{"role": "user", "content": text}]}, cfg)
    while reqs := pending(agent, cfg):
        decisions = []
        for r in reqs:
            ans = decide(f"批准 {r['name']}({r['args']})？[y=批准 / n=拒绝] ").strip().lower()
            decisions.append({"type": "approve"} if ans == "y" else {"type": "reject", "message": "审批人拒绝了这次退款"})
        agent.invoke(Command(resume={"decisions": decisions}), cfg)
    msgs = agent.get_state(cfg).values["messages"]
    for m in msgs:
        calls = getattr(m, "tool_calls", None)
        print(f"[{m.type}]", (f"调用 {[(c['name'], c['args']) for c in calls]}" if calls else str(m.content)[:200]))
    return msgs

if __name__ == "__main__":
    import sys
    run_cli(" ".join(sys.argv[1:]) or "订单 A1001 的耳机有杂音，我要退款 199 元")
    print("审计记录：", AUDIT)

第 2 步：测试

import pytest
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
from langgraph.types import Command
import agent as A

class ScriptedModel(GenericFakeChatModel):
    """按顺序吐出预设消息（可带 tool_calls）；bind_tools 原样返回自己。"""
    def bind_tools(self, tools, **kw):
        return self

def call(name, args, i):
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": f"c{i}", "type": "tool_call"}])

@pytest.fixture(autouse=True)
def reset():
    A.ORDERS["A1001"]["refunded"] = 0.0; A.AUDIT.clear()

ORIG_BUILD = A.build        # 测试里会 monkeypatch A.build，先保存原函数

def make(script):
    return ORIG_BUILD(ScriptedModel(messages=iter(script)))

CFG = {"configurable": {"thread_id": "t"}}
REFUND_FLOW = lambda amt: [call("get_order", {"order_id": "A1001"}, 1),
                           call("refund", {"order_id": "A1001", "amount": amt, "reason": "杂音"}, 2),
                           AIMessage(content="已处理")]

def test_refund_pauses_for_approval_then_runs():
    ag = make(REFUND_FLOW(199))
    ag.invoke({"messages": [{"role": "user", "content": "退款"}]}, CFG)
    reqs = A.pending(ag, CFG)
    assert [r["name"] for r in reqs] == ["refund"] and A.AUDIT == []      # get_order 已自动执行，refund 被拦下
    ag.invoke(Command(resume={"decisions": [{"type": "approve"}]}), CFG)
    assert A.AUDIT == [("A1001", 199, "杂音")] and A.pending(ag, CFG) == []

def test_reject_does_not_refund():
    ag = make(REFUND_FLOW(199))
    ag.invoke({"messages": [{"role": "user", "content": "退款"}]}, CFG)
    ag.invoke(Command(resume={"decisions": [{"type": "reject", "message": "不批"}]}), CFG)
    msgs = ag.get_state(CFG).values["messages"]
    assert A.AUDIT == [] and any("不批" in str(m.content) for m in msgs if m.type == "tool")

def test_tool_guard_even_if_approved():
    ag = make(REFUND_FLOW(500))
    ag.invoke({"messages": [{"role": "user", "content": "退款"}]}, CFG)
    ag.invoke(Command(resume={"decisions": [{"type": "approve"}]}), CFG)
    tool_out = [m.content for m in ag.get_state(CFG).values["messages"] if m.type == "tool"][-1]
    assert "可退金额为 199.00" in tool_out and A.AUDIT == []

def test_in_transit_blocked():
    assert "运输中" in A.refund.invoke({"order_id": "A1002", "amount": 1, "reason": "x"})

def test_cli_with_scripted_decisions(monkeypatch, capsys):
    monkeypatch.setattr(A, "build", lambda: make(REFUND_FLOW(50)))
    A.run_cli("退款", decide=lambda prompt: "y")
    assert A.AUDIT == [("A1001", 50, "杂音")] and "已处理" in capsys.readouterr().out

第 3 步：运行

$ ollama pull qwen3:0.6b && export OLLAMA_MODEL=qwen3:0.6b     # box 上是 qwen3-local:0.6b
$ python agent.py                     # 输入 y
批准 refund({'order_id': 'A1001', 'amount': 199, 'reason': '耳机杂音'})？[y=批准 / n=拒绝] y
[human] 订单 A1001 的耳机有杂音，我要退款 199 元
[ai] 调用 [('refund', {'order_id': 'A1001', 'amount': 199, 'reason': '耳机杂音'})]
[tool] 成功：订单 A1001 已退款 199.00 元
[ai] 成功：订单 A1001 已退款 199.00 元。需要其他帮助吗？
审计记录： [('A1001', 199.0, '耳机杂音')]

$ python agent.py                     # 输入 n
[tool] User rejected the tool call for `refund` with reason: 审批人拒绝了这次退款
[ai] 订单A1001的退款已拒绝，原因可能是审批人未通过。...
审计记录： []

$ python agent.py 订单 A1002 的键盘我不要了，退 459 元
[ai] 调用 [('get_order', {'order_id': 'A1002'})]
[tool] {"order_id": "A1002", "item": "机械键盘", "paid": 459.0, "status": "运输中", "refunded": 0.0}
[ai] 订单A1002的运输中状态，无法申请退款。...
$ python -m pytest -q
5 passed

注意第一段：qwen3 0.6B 跳过了 get_order，直接调用了 refund。提示词要求先查询，但模型没有照做。这正是 refund 内部必须自己校验的原因。

调试示例

1. qwen2.5 1.5B 把规则理解反了。 用最初的提示词时，模型查到 A1001 的状态是“已签收”，却回答“目前订单已签收，无法退款”，根本没有调用 refund。我在提示词里明确写出“‘已签收’的订单可以退款”之后，它仍然回答“订单 A1001 的耳机已签收，无法退款。”

2. 提示词写长之后，小模型把工具调用写成了普通文本。 我试过把规则写成 1/2/3 步骤的长提示词，结果 qwen2.5 1.5B 连续 4 次都输出了 {{"name": "get_order", "arguments": ...}} 这样的文本，而不是真正的 tool_calls，所以没有任何工具被执行，审计记录也为空（从安全角度看，这算是“安全地失败”）。最后我把默认模型改成在这个任务上能正确调用工具的 qwen3 0.6B，并把这两种失败都写进了教程。

3. 测试里的 monkeypatch 套娃。 test_cli_with_scripted_decisions 把 A.build 替换成 lambda: make(...)，而 make 内部又调用 A.build(...)，于是报错 <lambda>() takes 0 positional arguments but 1 was given。在模块加载时先保存 ORIG_BUILD = A.build 就解决了。

常见问题

模板规则

• 会修改数据的工具一律经过人工审批。
• 工具内部做最终校验。
• 记录审计日志。
• 换模型前，先用真实任务跑一遍批准、拒绝、拦截三条路径。

项目资源

智能体可以决定“想做什么”，但真正改数据之前，要经过人工审批和工具内部的校验。