主工作流:LangGraph 编排分类、检索、执行、起草、自检

今天要解决的问题

前四天的零件都齐了:知识库能检索,工具能查账单、能建退款申请,数据在库里。今天把它们串成一条能处理真实工单的流水线。

今天结束时你应该做到:

1. 项目里有一个统一的 LLMClient:记账(花了多少钱)、结构化输出(校验失败自动重试)、流式输出(边生成边推给前端)。
2. 一张 LangGraph 图,9 个节点,4 个条件分支,每个节点做什么、为什么这么分,能讲清楚。
3. 退款申请创建后,图停在审批节点等人;人点了通过或驳回,图从停下的地方接着跑。
4. 不连数据库、不调大模型,用假 LLM 跑通 6 个单测:退款审批、转人工时作废待审批申请、纯政策问题、转人工、质检两次不过、违规承诺被拦。
5. 终端里跑 10 张演示工单,看着每一步在干什么。

预计用时 5–7 小时,是 12 天里最核心的一天。前置:Day 9(手写 Agent 循环)、Day 13(五种 workflow 模式)、Day 24–25(LangGraph 基础和人工审批)、Day 45–46。

LLMClient:从 W1 搬过来,加三样东西

W1 的 Day 4 你封装过一个 LLMClient。现在搬进项目,补上产品级需要的三样东西:

# supportpilot/src/supportpilot/llm.py
"""W1 的 LLMClient 搬过来,加了三样东西:成本账本、结构化输出重试、流式回调。"""

import json
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any

from openai import OpenAI
from pydantic import BaseModel, ValidationError

from .config import Settings, get_settings


@dataclass
class Usage:
    prompt_tokens: int = 0
    completion_tokens: int = 0
    calls: int = 0

    def add(self, other: "Usage") -> None:
        self.prompt_tokens += other.prompt_tokens
        self.completion_tokens += other.completion_tokens
        self.calls += other.calls

    def cost_cny(self, s: Settings) -> float:
        return (self.prompt_tokens * s.price_in_per_m + self.completion_tokens * s.price_out_per_m) / 1_000_000


@dataclass
class ChatResult:
    content: str
    tool_calls: list[dict] = field(default_factory=list)  # [{"id","name","arguments"(str)}]
    usage: Usage = field(default_factory=Usage)
    raw_message: dict = field(default_factory=dict)  # 原样回传给下一轮 messages 用


class LLMClient:
    def __init__(self, settings: Settings | None = None, model: str | None = None):
        self.s = settings or get_settings()
        self.model = model or self.s.llm_model
        # max_retries 交给 SDK:429 和 5xx 会自动指数退避重试
        self.client = OpenAI(base_url=self.s.llm_base_url, api_key=self.s.llm_api_key, max_retries=3, timeout=60)
        self.ledger = Usage()  # 这个实例累计花了多少 token

    def _usage(self, u: Any) -> Usage:
        got = Usage(prompt_tokens=getattr(u, "prompt_tokens", 0) or 0,
                    completion_tokens=getattr(u, "completion_tokens", 0) or 0, calls=1)
        self.ledger.add(got)
        return got

    def chat(self, messages: list[dict], *, tools: list[dict] | None = None, temperature: float = 0.2,
             json_mode: bool = False, model: str | None = None) -> ChatResult:
        kwargs: dict[str, Any] = {"model": model or self.model, "messages": messages, "temperature": temperature}
        if tools:
            kwargs["tools"] = tools
        if json_mode:
            kwargs["response_format"] = {"type": "json_object"}
        if self.s.llm_extra_body:  # 比如关掉思考:{"thinking": {"type": "disabled"}}
            kwargs["extra_body"] = self.s.llm_extra_body
        resp = self.client.chat.completions.create(**kwargs)
        msg = resp.choices[0].message
        calls = [{"id": c.id, "name": c.function.name, "arguments": c.function.arguments or "{}"}
                 for c in (msg.tool_calls or [])]
        raw: dict[str, Any] = {"role": "assistant", "content": msg.content or ""}
        if calls:
            raw["tool_calls"] = [{"id": c["id"], "type": "function",
                                  "function": {"name": c["name"], "arguments": c["arguments"]}} for c in calls]
        return ChatResult(content=msg.content or "", tool_calls=calls, usage=self._usage(resp.usage), raw_message=raw)

    def stream(self, messages: list[dict], *, on_token: Callable[[str], None] | None = None,
               temperature: float = 0.3, model: str | None = None) -> ChatResult:
        """流式生成。每来一段文字就调 on_token,最后返回拼好的完整结果。"""
        resp = self.client.chat.completions.create(
            model=model or self.model, messages=messages, temperature=temperature,
            stream=True, stream_options={"include_usage": True},
            extra_body=self.s.llm_extra_body or None,
        )
        parts: list[str] = []
        last_usage = None
        for chunk in resp:
            if chunk.usage:  # 有的厂商只在最后一块带 usage,有的每块都带累计值,所以只记最后一次
                last_usage = chunk.usage
            if chunk.choices and chunk.choices[0].delta.content:
                piece = chunk.choices[0].delta.content
                parts.append(piece)
                if on_token:
                    on_token(piece)
        text = "".join(parts)
        usage = self._usage(last_usage)  # 循环结束后只入账一次
        return ChatResult(content=text, usage=usage, raw_message={"role": "assistant", "content": text})

    def structured[T: BaseModel](self, messages: list[dict], schema: type[T], *, retries: int = 2,
                                 model: str | None = None) -> T:
        """JSON mode + pydantic 校验,校验失败把错误贴回去让模型改,最多重试 retries 次。"""
        hint = ("只输出一个 JSON 对象,不要任何解释。JSON Schema 如下:\n"
                + json.dumps(schema.model_json_schema(), ensure_ascii=False))
        msgs = [*messages, {"role": "system", "content": hint}]
        last_err = ""
        for _ in range(retries + 1):
            res = self.chat(msgs, json_mode=True, temperature=0, model=model)
            try:
                return schema.model_validate_json(res.content)
            except ValidationError as e:
                last_err = str(e)[:800]
                msgs = [*msgs, res.raw_message,
                        {"role": "user", "content": f"上面的 JSON 没通过校验,错误是:\n{last_err}\n请改正后只输出 JSON。"}]
        raise ValueError(f"结构化输出重试 {retries} 次仍失败:{last_err}")

成本账本 ledger。 每次调用都把 usage 累加到实例的账本上,cost_cny() 按配置的单价换算成人民币。每个节点在开始前记一下账本读数,结束后算差值,就知道这一步花了多少。整张工单的成本在状态里累加,评测报告里就能出现”平均每张工单 ¥0.03”这种数字。

结构化输出 structured()。 用 JSON mode 加 pydantic 校验。JSON mode 只保证输出是合法 JSON,不保证字段对,所以还要校验。校验失败时把错误信息原样贴回去让模型改,最多重试两次。DeepSeek 的文档特别提醒:开 JSON mode 必须在提示词里说”输出 JSON”,否则模型可能一直输出空白直到 token 用完,所以 hint 里明确写了”只输出一个 JSON 对象”并附上 schema。

def structured[T: BaseModel](...) 是 Python 3.12 的泛型写法,意思是”传进来什么 schema 类型,就返回什么类型的实例”,编辑器能正确提示返回值的字段。

流式输出 stream()。 stream_options={"include_usage": True} 让流式响应也带上 token 用量。DeepSeek 我实测过,只有最后一块带 usage,前面的块都是空的;但不是所有厂商都这样,有的每一块都带一个累计值,那种情况下每块都入账,成本会被算成好几十倍。所以循环里只记住最后一次的 usage,循环结束后入账一次,两种行为都不会算错。

关掉思考 extra_body。 全课的惯例是用 .env 里的 LLM_EXTRA_BODY 关掉 deepseek-flash 默认开着的思考(Day 4 的 llm.py 就是这么做的)。Day 44 的 Settings 里还没有这一项,今天在 config.py 的 Settings 类里补一行:

    # Day 47:透传给大模型接口的额外参数(JSON)。deepseek-flash 默认开思考,填 {"thinking": {"type": "disabled"}} 关掉
    llm_extra_body: dict = {}

.env 里写 LLM_EXTRA_BODY={"thinking": {"type": "disabled"}},pydantic-settings 会把这段 JSON 解析成字典。chat() 和 stream() 都把它原样透传给接口。不关的代价很实在:我实测问一句「CloudLedger 是什么」,开着思考输出 256 个 token,关掉后 28 个;10 张演示工单全跑一遍,开着思考约 0.26 元,关掉后约 0.17 元。客服回复不需要模型先长篇思考,这钱省得心安理得。

max_retries=3 交给 openai SDK,它会对 429 和 5xx 自动做指数退避重试,这是 Day 12 你手写过的逻辑,SDK 已经帮你做好了。

先用假的 OpenAI 客户端测一下这三样东西:

# supportpilot/tests/test_llm.py
"""LLMClient 的单测:用假的 OpenAI 客户端,验证成本记账和结构化输出的重试。"""

from types import SimpleNamespace as NS

import pytest
from pydantic import BaseModel

from supportpilot.config import Settings
from supportpilot.llm import LLMClient


class Answer(BaseModel):
    category: str
    confidence: float


def fake_client(replies: list[str]):
    calls = []

    def create(**kw):
        calls.append(kw)
        text = replies.pop(0)
        msg = NS(content=text, tool_calls=None)
        return NS(choices=[NS(message=msg)], usage=NS(prompt_tokens=1000, completion_tokens=100))

    return NS(chat=NS(completions=NS(create=create))), calls


def make(replies):
    llm = LLMClient(Settings(llm_api_key="test", price_in_per_m=2.0, price_out_per_m=8.0))
    llm.client, calls = fake_client(replies)
    return llm, calls


def test_structured_retries_with_validation_error():
    llm, calls = make(['{"category": "refund"}', '{"category": "refund", "confidence": 0.9}'])
    out = llm.structured([{"role": "user", "content": "x"}], Answer)
    assert out.category == "refund" and len(calls) == 2
    assert "没通过校验" in calls[1]["messages"][-1]["content"]      # 第二次带上了错误信息
    assert calls[0]["response_format"] == {"type": "json_object"}


def test_structured_gives_up():
    llm, _ = make(["not json"] * 3)
    with pytest.raises(ValueError, match="重试 2 次"):
        llm.structured([{"role": "user", "content": "x"}], Answer)


def test_ledger_cost():
    llm, _ = make(['{"category": "a", "confidence": 1}'])
    llm.structured([{"role": "user", "content": "x"}], Answer)
    # 1000 输入 × 2 元/百万 + 100 输出 × 8 元/百万 = 0.002 + 0.0008
    assert llm.ledger.cost_cny(llm.s) == pytest.approx(0.0028)

状态:工作流里流动的是什么

# supportpilot/src/supportpilot/agent/state.py
import operator
from typing import Annotated, Literal, TypedDict

from pydantic import BaseModel, Field

Category = Literal["billing", "refund", "invoice", "account", "technical", "plan_change", "api", "other"]


class Ticket(TypedDict):
    id: str
    customer_email: str
    subject: str
    body: str


class SupportState(TypedDict, total=False):
    ticket: Ticket
    customer: dict | None
    triage: dict
    hits: list[dict]
    tool_log: Annotated[list[dict], operator.add]   # 每次工具调用追加一条,不覆盖
    findings: str
    refund_request: dict | None
    draft: str
    review: dict
    revisions: int
    approval: dict | None
    followup: str
    outcome: str            # resolved / waiting_approval / escalated
    handoff_reason: str
    reply: str              # 最终发给客户的文本(引用已渲染成链接)
    cost_cny: Annotated[float, operator.add]


class Triage(BaseModel):
    category: Category = Field(description="工单分类,定义见系统提示")
    needs_account_data: bool = Field(description="是否必须查这个客户的订阅/账单/历史才能回答")
    needs_human: bool = Field(description="是否必须直接转人工:投诉升级、法律威胁、安全事件、明确要求人工")
    urgency: Literal["low", "normal", "high"]
    summary: str = Field(description="一句话复述客户诉求,不超过 40 字")
    search_query: str = Field(description="用来检索帮助中心的查询:保留报错码、产品名词,去掉寒暄和情绪")


class Review(BaseModel):
    passed: bool
    problems: list[str] = Field(default_factory=list, description="不通过时列出具体问题,每条一句话")

LangGraph 的状态是一个字典,每个节点读它、返回一个”要更新哪些字段”的小字典。大部分字段是覆盖:节点返回 {"draft": "..."},状态里的 draft 就被替换。有两个字段用了 Annotated[..., operator.add],意思是累加:

• tool_log:act 节点每次返回的工具调用记录追加到列表末尾。
• cost_cny:每个节点返回自己花的钱,累加成整张工单的总成本。

为什么 hits 存的是字典列表而不是 Hit 对象?因为状态会被 checkpointer 序列化存进数据库(Day 48),纯字典和列表最稳妥。节点里用的时候 Hit(**h) 还原回来。

Triage 和 Review 是两个结构化输出的 schema。Triage 里我特意加了 search_query 字段:让分诊模型顺手把客户的话改写成适合检索的查询,去掉”你好""麻烦了""急急急”这些对检索没用的词。一次调用干两件事,省一次 API。

Category 用 Literal 限定了 8 个取值,模型输出其他值时 pydantic 校验会失败,触发重试。

提示词:集中放,带版本号

# supportpilot/src/supportpilot/agent/prompts.py
"""所有提示词集中放这里,带版本号。评测报告里会记录 PROMPT_VERSION,改提示词就改版本号。"""

PROMPT_VERSION = "v1"

CLASSIFY = """你是 CloudLedger(云簿,在线记账 SaaS)的工单分诊员。读客户工单,输出分类结果。

分类定义:
- billing:扣费、扣款失败、重复扣款、支付方式、账单查询(不涉及退钱的诉求)
- refund:客户要求退钱,或者问能不能退、退多少
- invoice:开发票、发票错误、重开发票
- account:登录、验证码、密码、账号锁定、绑定手机邮箱、成员权限、注销
- technical:导入导出、银行流水同步、结账、报表、App 同步等产品使用问题和报错
- plan_change:升级、降级、换年付月付、取消订阅、试用
- api:开放 API、API Key、限流、Webhook
- other:以上都不是

needs_account_data:回答必须依赖这个客户自己的订阅、账单或历史记录时为 true(比如"我被扣了两次""我能退多少");
纯政策或操作问题(比如"专票怎么开")为 false。
needs_human:客户威胁投诉到监管/媒体/律师、涉及账号被盗等安全事件、或者明确要求人工客服时为 true。"""

ACT = """你是 CloudLedger 客服后台的调查员。你的任务是用工具查清回答这张工单需要的客户数据,不直接回复客户。

规则:
1. 客户 ID 已经给你了,不需要再查 get_customer。
2. 涉及退款:先用 list_invoices 找到相关账单,再对它调用 check_refund_eligibility。退款金额只以工具返回为准,不要自己算。
3. 只有当客户明确要求退款、并且 check_refund_eligibility 返回 eligible=true 时,才调用 create_refund_request。
   客户只是询问"能不能退"时不要创建申请。
4. 只查和这张工单有关的数据,查到够用就停。
5. 结束时不再调用工具,输出「调查结论」:用几条短句列出查到的关键事实(账单号、金额、状态、规则结论),不要写给客户的话。"""

DRAFT = """你是 CloudLedger 的客服,给客户写工单回复。

写作要求:
- 只使用【帮助中心资料】和【调查结论】里的事实。资料里没有的,不要猜,直接说会转给同事跟进。
- 用到帮助中心资料的句子,在句末标注来源编号,格式严格为 [KB-xxx-000],只能用资料里出现过的编号。
- 如果有待审批的退款申请:只能说"已提交退款申请,审核通过后会原路退回",不能说已经退款或承诺到账时间以外的东西。
- 如果客户不符合退款条件:说明依据的规则,给出可行的替代办法(比如关闭自动续费)。
- 如果没查到客户账户:请客户提供注册邮箱。
- 语气礼貌、直接,先回答问题再补充说明。不超过 250 字。称呼用"您好"开头,不要落款。"""

CHECK = """你是客服质检员。检查这封回复草稿能不能发给客户。

逐条核对:
1. 草稿里每一个具体事实(金额、天数、规则、操作路径)是否都能在【帮助中心资料】或【调查结论】里找到依据。
2. 是否答非所问,或者漏掉了客户的主要诉求。
3. 是否承诺了没有依据的事(比如"已经为您退款""马上到账")。

全部没问题 passed=true。有任何一条问题 passed=false,并在 problems 里逐条写清楚哪句话有什么问题。"""

所有提示词放一个文件,开头有 PROMPT_VERSION。Day 50 的评测报告会记录用的是哪个版本,Day 51 改提示词时就改版本号。提示词就是代码,要能 diff、能回滚、能对应到分数。

几个写法上的考虑:

分诊提示词把每个分类的边界写清楚了,尤其是 needs_account_data 举了正反例。模型分类出错,十有八九是因为分类定义模糊,而不是模型笨。

act 提示词第 2、3 条是退款的操作规程:先查账单、再判资格、符合条件且客户明确要求才建申请。这些规则在工具描述里也写了一遍。重复是故意的,工具描述是”这个工具怎么用”,系统提示是”这件事的流程是什么”,两处都有,模型遵守得更稳。

起草提示词明确禁止”已经退款""承诺到账时间”这类话。但我不完全信任提示词,质检节点还会用正则再查一遍。

依赖注入:让节点可以被测试

# supportpilot/src/supportpilot/agent/deps.py
"""节点需要的外部依赖都装在 Deps 里。生产环境用 from_conn() 组装;测试时换成假的,不连数据库不花钱。"""

from collections.abc import Callable
from dataclasses import dataclass
from functools import partial
from typing import Any

from ..config import Settings, get_settings
from ..kb.retriever import HybridRetriever
from ..llm import LLMClient
from ..tools import business as biz
from ..tools.registry import call_tool


@dataclass
class Deps:
    llm: LLMClient                 # 起草用的主力模型
    small_llm: LLMClient           # 分类、质检用的便宜模型(Day 52 才真的分开,现在可以是同一个)
    retriever: Any                 # 有 search(query, k) -> list[Hit] 就行
    call_tool: Callable[..., tuple[str, bool]]   # (name, arguments, ticket_id=...) -> (text, is_error)
    escalate: Callable[[str, str], Any]
    decide_refund: Callable[[str, bool, str, str], dict]
    finish_ticket: Callable[[str, str, str | None, str], None]
    settings: Settings

    @classmethod
    def from_conn(cls, conn, settings: Settings | None = None) -> "Deps":
        s = settings or get_settings()
        llm = LLMClient(s)
        small = LLMClient(s, model=s.llm_small_model) if s.llm_small_model else llm
        return cls(llm=llm, small_llm=small, retriever=HybridRetriever(conn, settings=s),
                   call_tool=partial(call_tool, conn),
                   escalate=partial(biz.escalate_to_human, conn),
                   decide_refund=partial(biz.decide_refund, conn),
                   finish_ticket=partial(biz.finish_ticket, conn), settings=s)

    def spent(self) -> float:
        """两个客户端累计花了多少钱;同一个对象只算一次。"""
        total = self.llm.ledger.cost_cny(self.settings)
        if self.small_llm is not self.llm:
            total += self.small_llm.ledger.cost_cny(self.settings)
        return total

节点需要的外部东西(大模型、检索器、工具、数据库操作)全装进一个 Deps。生产环境用 from_conn() 组装真的;测试时塞假的进去,节点代码一行都不用改。

call_tool、escalate 这些用 functools.partial 把数据库连接提前绑定好,节点调用时只需要传业务参数。节点代码里看不到任何数据库连接,它也不需要知道。

small_llm 现在和 llm 是同一个对象(配置里 LLM_SMALL_MODEL 留空),Day 52 做模型路由时才分开。spent() 里判断了是不是同一个对象,避免重复计费。

九个节点

# supportpilot/src/supportpilot/agent/nodes.py
import json
import re
from dataclasses import asdict

from langgraph.config import get_stream_writer
from langgraph.types import interrupt

from ..kb.cite import build_context, check_citations, render_sources
from ..kb.retriever import Hit
from ..tools.registry import Level, openai_tools
from . import prompts
from .deps import Deps
from .state import Review, SupportState, Triage

PROMISE_RE = re.compile(r"已(经)?(为您|给您)?(办理|完成)?退款(?!申请)|退款已(到账|完成|办理|退回)|马上到账")


def _ticket_text(state: SupportState) -> str:
    t = state["ticket"]
    return f"发件邮箱:{t['customer_email']}\n主题:{t['subject']}\n正文:{t['body']}"


def _emit(event: dict) -> None:
    """往 custom 流里推一条事件(工具调用、token 等)。非流式运行时什么都不做。"""
    try:
        get_stream_writer()(event)
    except RuntimeError:  # 不在 LangGraph 运行上下文里(比如单测直接调节点函数)
        pass


def make_nodes(deps: Deps) -> dict:
    """返回 {节点名: 函数}。用闭包把 deps 带进去,图本身不关心依赖从哪来。"""

    def load_customer(state: SupportState) -> dict:
        text, err = deps.call_tool("get_customer", {"email": state["ticket"]["customer_email"]},
                                   ticket_id=state["ticket"]["id"])
        customer = None if err else json.loads(text)
        _emit({"type": "step", "node": "load_customer", "found": customer is not None})
        return {"customer": customer, "revisions": 0, "refund_request": None, "approval": None}

    def classify(state: SupportState) -> dict:
        before = deps.spent()
        triage = deps.small_llm.structured(
            [{"role": "system", "content": prompts.CLASSIFY}, {"role": "user", "content": _ticket_text(state)}],
            Triage)
        _emit({"type": "step", "node": "classify", "category": triage.category})
        return {"triage": triage.model_dump(), "cost_cny": deps.spent() - before}

    def retrieve(state: SupportState) -> dict:
        tr = state["triage"]
        query = tr.get("search_query") or f"{state['ticket']['subject']} {state['ticket']['body']}"
        hits = deps.retriever.search(query, k=deps.settings.retrieve_k)
        _emit({"type": "step", "node": "retrieve", "docs": [h.doc_id for h in hits]})
        return {"hits": [asdict(h) for h in hits]}

    def act(state: SupportState) -> dict:
        before = deps.spent()
        tid = state["ticket"]["id"]
        cust = state["customer"]
        msgs: list[dict] = [
            {"role": "system", "content": prompts.ACT},
            {"role": "user", "content": f"客户档案:{json.dumps(cust, ensure_ascii=False)}\n\n{_ticket_text(state)}\n\n"
                                        f"分诊结论:{state['triage']['summary']}"},
        ]
        log: list[dict] = []
        refund_request = None
        findings = ""
        tools = openai_tools({Level.READ, Level.PROPOSE})
        for step in range(deps.settings.max_act_steps):
            res = deps.llm.chat(msgs, tools=tools, temperature=0)
            if not res.tool_calls:
                findings = res.content
                break
            msgs.append(res.raw_message)
            for call in res.tool_calls:
                text, err = deps.call_tool(call["name"], call["arguments"], ticket_id=tid, customer_id=cust["id"])
                log.append({"step": step, "name": call["name"], "args": call["arguments"], "error": err,
                            "result": text[:500]})
                _emit({"type": "tool", "name": call["name"], "args": call["arguments"], "error": err})
                if call["name"] == "create_refund_request" and not err:
                    refund_request = json.loads(text)
                msgs.append({"role": "tool", "tool_call_id": call["id"], "content": text})
        else:
            findings = "(调查达到步数上限,以下为已查到的工具结果)\n" + "\n".join(
                f"- {x['name']}: {x['result'][:200]}" for x in log)
        return {"tool_log": log, "findings": findings, "refund_request": refund_request,
                "cost_cny": deps.spent() - before}

    def draft(state: SupportState) -> dict:
        before = deps.spent()
        hits = [Hit(**h) for h in state.get("hits", [])]
        parts = [f"【客户工单】\n{_ticket_text(state)}",
                 f"【帮助中心资料】\n{build_context(hits)}",
                 f"【调查结论】\n{state.get('findings') or '(没有查询客户数据)'}"]
        if state.get("customer") is None and state["triage"].get("needs_account_data"):
            parts.append("【注意】没有找到这个邮箱对应的客户账户。")
        if state.get("refund_request"):
            rr = state["refund_request"]
            parts.append(f"【待审批退款申请】{rr['id']},金额 {rr['amount_cny']:.2f} 元,规则 {rr['rule']}")
        if state.get("review") and not state["review"].get("passed"):
            parts.append("【上一版草稿被质检退回,问题如下,请改正】\n" + "\n".join(state["review"]["problems"]))
        res = deps.llm.stream(
            [{"role": "system", "content": prompts.DRAFT}, {"role": "user", "content": "\n\n".join(parts)}],
            on_token=lambda t: _emit({"type": "token", "text": t}))
        return {"draft": res.content, "cost_cny": deps.spent() - before}

    def check(state: SupportState) -> dict:
        before = deps.spent()
        hits = [Hit(**h) for h in state.get("hits", [])]
        problems: list[str] = []
        _, fake = check_citations(state["draft"], hits)
        if fake:
            problems.append(f"引用了不存在于资料中的编号:{', '.join(fake)}")
        if PROMISE_RE.search(state["draft"]):
            problems.append("草稿承诺了退款已完成/马上到账,退款必须经过人工审核")
        if not problems:  # 规则检查过了再花钱让模型检查
            review = deps.small_llm.structured([
                {"role": "system", "content": prompts.CHECK},
                {"role": "user", "content": f"【客户工单】\n{_ticket_text(state)}\n\n【帮助中心资料】\n{build_context(hits)}"
                                            f"\n\n【调查结论】\n{state.get('findings') or '无'}\n\n【回复草稿】\n{state['draft']}"},
            ], Review)
            problems.extend(review.problems if not review.passed else [])
        passed = not problems
        _emit({"type": "step", "node": "check", "passed": passed, "problems": problems})
        return {"review": {"passed": passed, "problems": problems},
                "revisions": state.get("revisions", 0) + (0 if passed else 1),
                "cost_cny": deps.spent() - before}

    def approval(state: SupportState) -> dict:
        rr = state["refund_request"]
        # interrupt 之前的代码在恢复时会重跑一遍,所以这里不能有副作用(不写库、不发消息)
        decision = interrupt({"type": "refund_approval", "ticket_id": state["ticket"]["id"], "request": rr,
                              "draft": state["draft"],
                              "needs_finance_review": rr["amount_cny"] > deps.settings.refund_review_threshold})
        approved = bool(decision.get("approved"))
        result = deps.decide_refund(rr["id"], approved, decision.get("approver", "unknown"), decision.get("note", ""))
        if approved:
            followup = f"补充:您的退款申请已审核通过,{rr['amount_cny']:.2f} 元将原路退回,到账时间以支付渠道为准。"
        else:
            followup = ("补充:很抱歉,您的退款申请经审核未通过" + (f",原因:{decision['note']}" if decision.get("note") else "")
                        + "。如有疑问可以直接回复这个工单。")
        return {"approval": {**decision, "result_status": result.get("status")}, "followup": followup}

    def handoff(state: SupportState) -> dict:
        reason = state.get("handoff_reason") or (
            "客户要求人工或涉及投诉/安全" if state.get("triage", {}).get("needs_human")
            else "草稿多次未通过质检:" + ";".join(state.get("review", {}).get("problems", [])))
        rr = state.get("refund_request")
        if rr and rr.get("status") == "pending":
            # act 已经建了申请、但草稿没过质检就转人工:这条申请不会再走到 approval 节点,作废它,
            # 免得它挂在审批列表里,有人点通过却没有对应的图可以恢复
            deps.decide_refund(rr["id"], False, "system:handoff", "工单已转人工,这条申请作废,由人工客服重新判断")
        deps.escalate(state["ticket"]["id"], reason)
        reply = "您好,您的问题已经转给人工客服同事,会在 1 个工作日内通过这个工单回复您。"
        return {"outcome": "escalated", "handoff_reason": reason, "reply": reply}

    def finalize(state: SupportState) -> dict:
        hits = [Hit(**h) for h in state.get("hits", [])]
        reply = render_sources(state["draft"], hits)
        if state.get("followup"):
            reply += "\n\n" + state["followup"]
        deps.finish_ticket(state["ticket"]["id"], "resolved", state["triage"]["category"], state["triage"]["summary"])
        return {"outcome": "resolved", "reply": reply}

    return {"load_customer": load_customer, "classify": classify, "retrieve": retrieve, "act": act,
            "draft": draft, "check": check, "approval": approval, "handoff": handoff, "finalize": finalize}

逐个说。

load_customer:用发件邮箱查客户。查不到不报错,customer 为 None,后面的节点会据此让客户提供注册邮箱。这一步不调大模型,零成本。

classify:分诊。用小模型、温度 0、结构化输出。输出分类、是否需要查账户、是否需要转人工、紧急程度、一句话摘要、检索查询。

retrieve:用分诊给的 search_query 去检索。如果分诊没给就用原文。

act:全图唯一的 agent 循环,和 Day 9 手写的那个结构一样:调模型 → 有工具调用就执行 → 结果喂回 → 直到模型不再调工具。区别在于加了这些:

• 步数上限 max_act_steps=5,用 for ... else 写:循环正常跑完(没 break)说明到了上限,这时把已经查到的工具结果拼成调查结论,不浪费已经花的钱。
• 每次工具调用都记进 tool_log,并通过 _emit 推一个事件给前端。
• 工具错误(err=True)照常喂回模型,让它自己改参数重试。
• create_refund_request 成功时,把返回的申请记进 refund_request,后面的路由要用。
• 客户 ID 已经放进了用户消息,提示词也说”不需要再查 get_customer”,省一轮。

draft:起草。把工单、检索资料(带编号)、调查结论、待审批申请、上一版被退回的原因全拼进去。用流式生成,每段文字通过 _emit 推出去,Day 49 的界面上就能看到打字机效果。

check:质检,先规则后模型。规则一:引用了检索结果里没有的编号。规则二:正则匹配到”已为您退款""马上到账”这类违规承诺。规则查出问题就不再调模型,省钱;规则过了才让小模型核对事实依据。这就是 Day 13 的 evaluator-optimizer:不通过就带着问题清单回到 draft 重写。

approval:退款审批。调用 interrupt(),图在这里停下,把申请详情交给外面。人的决定通过 Command(resume=...) 传回来,成为 interrupt() 的返回值。然后调 decide_refund 执行或驳回,写一句补充说明给客户。

这里有个 LangGraph 的规则一定要记住:**节点恢复时会从头重跑,interrupt() 之前的代码会执行两次。**所以 interrupt 之前只放没有副作用的代码(构造 payload),写库、发通知都放在 interrupt 之后或者别的节点里。Day 46 把审批通知放在工具层就是这个原因。

handoff:转人工。调 escalate 把工单标为 escalated,给客户一句固定的话。原因写进 handoff_reason,人工接手时能看到。还有一种情况要收尾:act 已经建了退款申请,草稿却两次没过质检转了人工。这条申请永远走不到 approval 节点,如果留着 pending,它会出现在 Day 49 的审批列表里,有人点通过却没有对应的图可以恢复(Day 49 实测返回 409)。所以转人工前把它作废(标成 rejected 并写明原因),要不要退由接手的人重新判断。

finalize:定稿。把草稿里的 [KB-xxx] 渲染成 [1] [2] 和文末链接,有审批结果的话加上补充说明,更新工单状态。

_emit 包了一层 try。get_stream_writer() 只在 LangGraph 运行时有效,单测里直接调节点函数时没有运行上下文,包一层让节点在任何地方都能调。

图和路由

# supportpilot/src/supportpilot/agent/graph.py
from langgraph.graph import END, START, StateGraph

from .deps import Deps
from .nodes import make_nodes
from .state import SupportState


def after_classify(state: SupportState) -> str:
    return "handoff" if state["triage"]["needs_human"] else "retrieve"


def after_retrieve(state: SupportState) -> str:
    if state["triage"]["needs_account_data"] and state.get("customer"):
        return "act"
    return "draft"


def after_check(deps: Deps):
    def route(state: SupportState) -> str:
        if state["review"]["passed"]:
            rr = state.get("refund_request")
            return "approval" if rr and rr.get("status") == "pending" else "finalize"
        if state.get("revisions", 0) <= deps.settings.max_revisions:
            return "draft"
        return "handoff"
    return route


def build_graph(deps: Deps, checkpointer=None):
    n = make_nodes(deps)
    g = StateGraph(SupportState)
    for name, fn in n.items():
        g.add_node(name, fn)
    g.add_edge(START, "load_customer")
    g.add_edge("load_customer", "classify")
    g.add_conditional_edges("classify", after_classify, ["retrieve", "handoff"])
    g.add_conditional_edges("retrieve", after_retrieve, ["act", "draft"])
    g.add_edge("act", "draft")
    g.add_edge("draft", "check")
    g.add_conditional_edges("check", after_check(deps), ["approval", "finalize", "draft", "handoff"])
    g.add_edge("approval", "finalize")
    g.add_edge("finalize", END)
    g.add_edge("handoff", END)
    return g.compile(checkpointer=checkpointer)

四个路由函数都是纯函数,只看状态里的字段做判断,不调模型。分支逻辑一目了然,出了问题看一眼状态就知道为什么走了这条路。

after_check 里的 revisions <= max_revisions:第一次质检不过,revisions 变成 1,1 ≤ 1,回去重写;第二次还不过,revisions 变成 2,2 > 1,转人工。也就是说最多重写一次。为什么不多给几次?因为同一个模型、同样的资料,第三次写出来通常和第二次差不多,多重试只是多花钱。两次都过不了,说明资料不够或者问题超出了能力范围,该交给人。

add_conditional_edges 的第三个参数列出了所有可能去的节点。不写也能跑,但写上之后 LangGraph 能在编译时检查,画图时也能画出所有分支。

为什么不用现成的预置 agent

LangGraph 有现成的预置 ReAct agent,几行代码就能得到一个”模型 + 工具”的循环,Day 24 你用过。为什么 act 节点要自己写?

一是要控制。我需要在工具调用之间插入自己的逻辑:记录日志、推送事件、识别退款申请、步数用完时把已查到的结果整理成结论。预置 agent 也能通过钩子做到一部分,但每加一样就要去翻一次文档看它支持不支持。自己写的循环就二十几行,想加什么直接加。

二是要看得懂。面试时被问”你的 agent 循环是怎么实现的”,指着自己写的二十行讲清楚,比说”我调了某个函数”有说服力得多。Day 9 手写过,这里只是把它放进了产品代码。

如果你的场景就是标准的”模型加几个工具”,没有这些定制需求,用预置的完全没问题。框架是工具,不是信仰。

单测:不花一分钱验证整条流程

工作流的逻辑(路由对不对、审批停没停、质检打回后有没有重写)和大模型的能力是两回事。前者用假 LLM 测,确定、免费、几秒钟;后者 Day 50 用评测测。

# supportpilot/tests/fakes.py
"""假的 LLM / 检索 / 工具,让工作流的单测不连数据库、不花钱、结果确定。"""

import json

from supportpilot.agent import prompts
from supportpilot.agent.deps import Deps
from supportpilot.config import Settings
from supportpilot.kb.retriever import Hit
from supportpilot.llm import ChatResult, Usage


class FakeLLM:
    def __init__(self, triage: dict, act_script: list[list[tuple[str, dict]]], draft: str, review_ok: bool = True):
        self.triage, self.act_script, self.draft_text, self.review_ok = triage, list(act_script), draft, review_ok
        self.ledger = Usage()

    def structured(self, messages, schema, **_):
        self.ledger.add(Usage(100, 20, 1))
        if messages[0]["content"] == prompts.CLASSIFY:
            return schema(**self.triage)
        return schema(passed=self.review_ok, problems=[] if self.review_ok else ["金额没有依据"])

    def chat(self, messages, tools=None, **_):
        self.ledger.add(Usage(300, 50, 1))
        if not self.act_script:
            return ChatResult(content="调查结论:- 已查清")
        calls = [{"id": f"c{i}", "name": n, "arguments": json.dumps(a)} for i, (n, a) in
                 enumerate(self.act_script.pop(0))]
        raw = {"role": "assistant", "content": "", "tool_calls": [
            {"id": c["id"], "type": "function", "function": {"name": c["name"], "arguments": c["arguments"]}}
            for c in calls]}
        return ChatResult(content="", tool_calls=calls, raw_message=raw)

    def stream(self, messages, on_token=None, **_):
        self.ledger.add(Usage(800, 150, 1))
        for piece in (self.draft_text[:10], self.draft_text[10:]):
            if on_token:
                on_token(piece)
        return ChatResult(content=self.draft_text)


class FakeRetriever:
    def search(self, query, k=5):
        return [Hit(1, "KB-refund-002", "年付 14 天无理由退款", "条件", "首次购买的年付订阅 14 天内全额退款。",
                    "/help/refund/002", "refund", 0.03, ["vector", "bm25"])]


class FakeTools:
    def __init__(self, customer: dict | None = None):
        self.customer = customer
        self.calls: list[str] = []
        self.decided: list[tuple] = []
        self.escalated: list[str] = []
        self.finished: list[tuple] = []

    def __call__(self, name, arguments, *, ticket_id, customer_id=None):
        self.calls.append(name)
        args = json.loads(arguments) if isinstance(arguments, str) else arguments
        if name == "get_customer":
            return (json.dumps(self.customer), False) if self.customer else ("没有找到客户", True)
        if name == "list_invoices":
            return json.dumps({"total": 1, "returned": 1, "invoices": [{"id": "INV-202609-0001"}]}), False
        if name == "check_refund_eligibility":
            return json.dumps({"invoice_id": args["invoice_id"], "eligible": True, "refund_amount_cny": 950.0,
                               "rule": "annual_first_14_days"}), False
        if name == "create_refund_request":
            return json.dumps({"id": f"RF-{ticket_id}-{args['invoice_id']}", "status": "pending",
                               "amount_cny": 950.0, "rule": "annual_first_14_days"}), False
        return "未知工具", True


def make_deps(llm: FakeLLM, tools: FakeTools) -> Deps:
    return Deps(llm=llm, small_llm=llm, retriever=FakeRetriever(), call_tool=tools,
                escalate=lambda tid, reason: tools.escalated.append(tid),
                decide_refund=lambda rid, ok, who, note: (tools.decided.append((rid, ok)) or
                                                          {"status": "executed" if ok else "rejected"}),
                finish_ticket=lambda *a: tools.finished.append(a), settings=Settings())

FakeLLM 按系统提示判断自己被哪个节点调用:分诊时返回预设的分诊结果,act 时按剧本依次返回工具调用,起草时返回预设草稿。FakeTools 模拟工具返回,同时记录被调用了哪些工具、审批了什么、转人工了什么,测试里可以断言。

# supportpilot/tests/test_graph_smoke.py
import pytest
from fakes import FakeLLM, FakeTools, make_deps
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

from supportpilot.agent.graph import build_graph

TICKET = {"id": "T000900", "customer_email": "lina@example.com", "subject": "想退款",
          "body": "上周买的专业版年付,用不上,请帮我退款。"}
REFUND_TRIAGE = {"category": "refund", "needs_account_data": True, "needs_human": False, "urgency": "normal",
                 "summary": "年付首购想退款", "search_query": "年付 退款"}
ACT = [[("list_invoices", {"customer_id": "C0002"})],
       [("check_refund_eligibility", {"invoice_id": "INV-202609-0001"})],
       [("create_refund_request", {"invoice_id": "INV-202609-0001", "reason": "首购 7 天不需要"})]]
DRAFT = "您好,您的年付订阅在 14 天内,符合全额退款条件 [KB-refund-002]。已提交退款申请,审核通过后会原路退回。"


def test_refund_goes_through_approval_then_resumes():
    tools = FakeTools(customer={"id": "C0002", "email": "lina@example.com"})
    graph = build_graph(make_deps(FakeLLM(REFUND_TRIAGE, ACT, DRAFT), tools), checkpointer=InMemorySaver())
    cfg = {"configurable": {"thread_id": TICKET["id"]}}

    out = graph.invoke({"ticket": TICKET}, cfg)
    assert "__interrupt__" in out                          # 停在审批节点
    payload = out["__interrupt__"][0].value
    assert payload["request"]["status"] == "pending"
    assert tools.calls == ["get_customer", "list_invoices", "check_refund_eligibility", "create_refund_request"]
    assert tools.decided == []                              # 人还没点,钱一分没动

    out = graph.invoke(Command(resume={"approved": True, "approver": "finance-amy"}), cfg)
    assert out["outcome"] == "resolved"
    assert tools.decided == [("RF-T000900-INV-202609-0001", True)]
    assert "[1]" in out["reply"] and "参考文章" in out["reply"]
    assert out["cost_cny"] > 0


def test_handoff_withdraws_pending_refund():
    tools = FakeTools(customer={"id": "C0002", "email": "lina@example.com"})
    deps = make_deps(FakeLLM(REFUND_TRIAGE, ACT, DRAFT, review_ok=False), tools)
    out = build_graph(deps).invoke({"ticket": TICKET})
    assert out["outcome"] == "escalated"
    assert tools.decided == [("RF-T000900-INV-202609-0001", False)]   # 申请被作废,不会留在审批列表里


def test_policy_question_skips_tools():
    triage = {**REFUND_TRIAGE, "category": "invoice", "needs_account_data": False, "summary": "专票怎么开"}
    tools = FakeTools(customer={"id": "C0001"})
    graph = build_graph(make_deps(FakeLLM(triage, [], DRAFT), tools))
    out = graph.invoke({"ticket": {**TICKET, "body": "专票怎么开"}})
    assert tools.calls == ["get_customer"]                 # 没进 act
    assert out["outcome"] == "resolved"


def test_angry_customer_goes_to_human():
    triage = {**REFUND_TRIAGE, "needs_human": True}
    tools = FakeTools(customer={"id": "C0002"})
    out = build_graph(make_deps(FakeLLM(triage, [], DRAFT), tools)).invoke({"ticket": TICKET})
    assert out["outcome"] == "escalated" and tools.escalated == ["T000900"]


def test_failed_review_twice_escalates():
    tools = FakeTools(customer={"id": "C0001"})
    triage = {**REFUND_TRIAGE, "needs_account_data": False}
    out = build_graph(make_deps(FakeLLM(triage, [], DRAFT, review_ok=False), tools)).invoke({"ticket": TICKET})
    assert out["outcome"] == "escalated"
    assert out["revisions"] == 2


def test_promise_rule_blocks_draft_without_llm():
    tools = FakeTools(customer={"id": "C0001"})
    triage = {**REFUND_TRIAGE, "needs_account_data": False}
    bad = "您好,已为您退款,马上到账 [KB-refund-002]。"
    out = build_graph(make_deps(FakeLLM(triage, [], bad), tools)).invoke({"ticket": TICKET})
    assert out["outcome"] == "escalated"
    assert any("承诺" in p for p in out["review"]["problems"])


@pytest.mark.parametrize("text,bad", [
    ("已为您退款,马上到账", True), ("退款已到账", True), ("已为您完成退款", True),
    ("已为您提交退款申请,审核通过后原路退回", False), ("已为您办理退款申请", False),
])
def test_promise_regex(text, bad):
    from supportpilot.agent.nodes import PROMISE_RE
    assert bool(PROMISE_RE.search(text)) is bad

前六个测试各守一条底线:

• 退款工单一定会停在审批节点,人没点之前 decided 是空的,一分钱没动;点了通过之后才执行,并且最终回复里引用被渲染成了 [1] 和链接。
• 退款申请已经建了、草稿却两次没过质检转人工时,申请被作废,不会留在审批列表里。
• 政策问题不进 act,工具只调了 get_customer。
• needs_human 直接转人工。
• 质检两次不过就转人工,revisions 是 2。
• 草稿里有”已为您退款,马上到账”,规则检查直接拦下,而且根本没走到模型质检。

最后一个参数化测试是给质检正则本身的。我第一版正则把”已为您办理退款申请”也当成了违规承诺,这是一句完全合规的话,误判会让好草稿被打回、白花一次重写的钱。判断规则也是代码,也会写错,也要测。

这些测试我在本机都跑过,全部通过。InMemorySaver 是内存版的 checkpointer,审批需要它:没有 checkpointer,图停下后状态就丢了,没法恢复。

端到端跑一遍

{"customer_email": "zhangwei@example.com", "subject": "怎么开专票", "body": "我们公司要增值税专用发票,需要准备什么?"}
{"customer_email": "zhangwei@example.com", "subject": "导入失败", "body": "导入凭证一直提示 E2001,文件是从老系统导出的 xls。"}
{"customer_email": "lina@example.com", "subject": "想退款", "body": "上周买的专业版年付,发现用不上,请帮我退款。"}
{"customer_email": "lina@example.com", "subject": "能退吗", "body": "如果我现在不想用了,年付能退多少?先问问。"}
{"customer_email": "wangfang@example.com", "subject": "申请退款", "body": "5 月份买的专业版年付,现在用不上了,请退款。"}
{"customer_email": "liuyang@example.com", "subject": "扣了两次", "body": "9 月 12 号被扣了两次 39 元,请把多的退给我。"}
{"customer_email": "chenjing@example.com", "subject": "月费退款", "body": "这个月的 39 元请退给我,我不用了。"}
{"customer_email": "yanglei@example.com", "subject": "账套变只读了", "body": "为什么突然不能新增凭证了?"}
{"customer_email": "nobody@example.com", "subject": "退款", "body": "请把我的年费退了。"}
{"customer_email": "chenjing@example.com", "subject": "投诉", "body": "你们乱扣费,再不处理我就去 12315 投诉。"}

10 张工单覆盖了主要路径:政策问答、报错码、退款要求、退款咨询、按比例退款、重复扣款、月付退款被拒、扣款失败、找不到客户、投诉转人工。

# supportpilot/scripts/run_tickets.py
"""端到端跑 data/seed/demo_tickets.jsonl 里的 10 张工单,终端里看每一步。遇到退款审批会停下来问你。
uv run python scripts/run_tickets.py            # 全部
uv run python scripts/run_tickets.py 3          # 只跑第 3 张
"""

import json
import sys
from pathlib import Path

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

from supportpilot.agent.deps import Deps
from supportpilot.agent.graph import build_graph
from supportpilot.db import connect
from supportpilot.tools import business as biz

DEMO = Path(__file__).resolve().parent.parent / "data" / "seed" / "demo_tickets.jsonl"


def show(graph, payload, cfg) -> None:
    for mode, chunk in graph.stream(payload, cfg, stream_mode=["updates", "custom"]):
        if mode == "custom":
            if chunk["type"] == "token":
                print(chunk["text"], end="", flush=True)
            elif chunk["type"] == "tool":
                print(f"\n  🔧 {chunk['name']}({chunk['args']}){' ✗' if chunk['error'] else ''}")
            elif chunk["type"] == "step":
                info = {k: v for k, v in chunk.items() if k not in ("type", "node")}
                print(f"\n  ▸ {chunk['node']} {info}")
        elif "__interrupt__" in chunk:
            req = chunk["__interrupt__"][0].value["request"]
            print(f"\n  ⏸ 退款审批:{req['id']} ¥{req['amount_cny']:.2f} 规则={req['rule']}")


def main() -> None:
    tickets = [json.loads(x) for x in DEMO.read_text(encoding="utf-8").splitlines() if x.strip()]
    if len(sys.argv) > 1:
        tickets = [tickets[int(sys.argv[1]) - 1]]
    conn = connect()
    graph = build_graph(Deps.from_conn(conn), checkpointer=InMemorySaver())
    for t in tickets:
        tid = biz.create_ticket(conn, t["customer_email"], t["subject"], t["body"])
        cfg = {"configurable": {"thread_id": tid}}
        print(f"\n===== {tid} · {t['customer_email']} · {t['subject']} =====")
        show(graph, {"ticket": {"id": tid, **t}}, cfg)
        while graph.get_state(cfg).next:  # 停在审批节点
            ok = input("\n  通过这笔退款吗?[y/N] ").strip().lower() == "y"
            show(graph, Command(resume={"approved": ok, "approver": "cli", "note": "" if ok else "演示驳回"}), cfg)
        final = graph.get_state(cfg).values
        print(f"\n----- 最终回复({final.get('outcome')},¥{final.get('cost_cny', 0):.4f})-----\n{final.get('reply')}")


if __name__ == "__main__":
    main()

uv run python scripts/run_tickets.py 3      # 只跑第 3 张:李娜要退款

输出大致长这样(工单号、检索到的文章、模型的措辞每次会有差异):

===== T000153 · lina@example.com · 想退款 =====

  ▸ load_customer {'found': True}

  ▸ classify {'category': 'refund'}

  ▸ retrieve {'docs': ['KB-refund-002', 'KB-refund-001', ...]}

  🔧 list_invoices({"customer_id": "C0002"})

  🔧 check_refund_eligibility({"invoice_id": "INV-202609-0002"})

  🔧 create_refund_request({"invoice_id": "INV-202609-0002", "reason": "..."})
[notify:dry-run]
【退款待审批】RF-T000153-INV-202609-0002
...
您好,您的专业版年付订阅是 9 月 24 日首次购买的,目前在 14 天内,符合全额退款条件 [KB-refund-002]。我们已为您提交退款申请……
  ▸ check {'passed': True, 'problems': []}

  ⏸ 退款审批:RF-T000153-INV-202609-0002 ¥950.00 规则=annual_first_14_days

  通过这笔退款吗?[y/N] y

----- 最终回复(resolved,¥0.03 左右)-----
您好,……符合全额退款条件 [1]。……

参考文章:
[1] 年付 14 天无理由退款 https://help.cloudledger.example/help/refund/002

补充:您的退款申请已审核通过,950.00 元将原路退回,到账时间以支付渠道为准。

跑全部 10 张,每张都对照预期看一遍:第 4 张(只是问问)不应该建申请,第 7 张(月付)应该解释不退并建议关闭自动续费,第 9 张(找不到客户)应该请客户提供注册邮箱,第 10 张(投诉)应该直接转人工。

你会发现有些工单处理得不完美。比如”只是问问”那张,模型可能还是建了申请;或者某个回复漏了关键数字。别急着改,把问题记下来。Day 50 会把这些变成评测用例,Day 51 系统地修。今天的目标是”整条流程通了”,不是”每张都完美”。

动手练习

步骤 1:放文件。 config.py 的 Settings 加上 llm_extra_body,.env 里填好 LLM_EXTRA_BODY;src/supportpilot/llm.py,agent/ 下 state.py、prompts.py、deps.py、nodes.py、graph.py,scripts/run_tickets.py,data/seed/demo_tickets.jsonl,tests/ 下 fakes.py、test_graph_smoke.py、test_llm.py。

步骤 2:跑单测。 make test,确认新加的测试全过。这一步不需要 API key。

步骤 3:画出你的图。 LangGraph 能把编译好的图画成 Mermaid:

from supportpilot.agent.graph import build_graph
from fakes import FakeLLM, FakeTools, make_deps   # 在 tests 目录下运行
g = build_graph(make_deps(FakeLLM({}, [], ""), FakeTools()))
print(g.get_graph().draw_mermaid())

把输出贴到 mermaid.live 里看,和上面的图对照。

步骤 4:端到端。 确认 .env 里填了 LLM_API_KEY 和 EMB_API_KEY,数据库起着、数据导好了,然后 uv run python scripts/run_tickets.py,10 张全跑一遍。审批时第 3 张选 y,第 5 张选 n,看两种补充说明。

步骤 5:记一份问题清单。 新建 docs/day47-observations.md,每张工单一行:工单、期望、实际、哪里不对。至少记下 3 个问题,这是 Day 51 错误分析的原材料。

验收标准: make test 全绿(不需要 API);10 张演示工单都能跑完不崩;退款工单会停下等审批,批完能看到补充说明;投诉工单直接转人工;每张工单打印出了成本,大约在几分钱以内。

名词解释

名词	意思
成本账本	LLMClient 上累计 token 用量的计数器,用来算每一步花了多少钱
JSON mode	让模型保证输出合法 JSON 的接口选项,不保证字段正确
reducer	LangGraph 状态字段的合并方式,默认覆盖,operator.add 表示追加/累加
依赖注入	把外部依赖从参数传进来,而不是在函数里自己创建,方便替换成假的
interrupt	LangGraph 中让图暂停并把数据交给外部的函数
Command(resume)	恢复暂停的图,并把值作为 interrupt() 的返回值
checkpointer	保存图状态的组件,暂停和恢复都依赖它
evaluator-optimizer	起草和质检分开,质检不过就带着意见重写
for … else	Python 语法,循环没有被 break 时执行 else 块

常见误区

整个工单丢给一个大 agent。 能跑,但分诊错了、检索漏了、承诺乱说,你分不清是哪一步的问题,也没法单独优化。拆成节点,每个节点有明确的输入输出。

interrupt 之前写副作用。 恢复时会再执行一次。写库、发通知、调外部接口,都放在 interrupt 之后或者别的节点。

质检全交给模型。 能用规则查的(编号是否存在、是否有违禁承诺)先用规则,确定、免费、不会漏。模型质检只做规则做不了的判断。

单测里调真的大模型。 慢、花钱、结果不确定,测试会时好时坏。工作流逻辑用假 LLM 测,模型能力用评测测,两件事分开。

流式 usage 每块都入账。 成本会被夸大很多倍,而且很难察觉,直到月底看账单。

一看到不完美的回复就改提示词。 今天改一句、明天改一句,改了什么、变好还是变坏都说不清。先记下来,等有了评测集再系统地改。

学习资料

文档与文章

• LangGraph 概览 和 Graph API:State、reducer、条件边的官方说明,1 小时。
• LangGraph Interrupts:重点读”节点恢复时从头重跑”和”interrupt 前的副作用要幂等”两段,20 分钟。
• LangGraph Streaming:stream_mode 的几种模式,特别是 custom 模式和 get_stream_writer,20 分钟。
• DeepSeek JSON Output 和 Function Calling:你用的厂商的结构化输出和工具调用说明,换厂商就看对应文档,20 分钟。
• Anthropic · Building effective agents:重读 evaluator-optimizer 那一节,10 分钟。

B 站视频

• 上面嵌的敲代码的码农《挑战 19 分钟搞定,LangGraph 快速入门与原理剖析》。
• 尚硅谷《LangGraph 教程》(BV1z3NY66EY1,132 集合集):系统性最强的中文 LangGraph 课,今天用到的条件边、checkpointer、人工介入都有专门的集,按目录挑着看,别从头刷。

其他经典视频

• 上面嵌的 LangChain《LangGraph interrupt》。
• LangChain《Build a Customer Support Bot | LangGraph》(YouTube,b3XsvoFWp4c):航空客服机器人,对比它和 SupportPilot 的设计差异:它的敏感操作是怎么确认的、节点是怎么分的。
• LangChain Academy · Introduction to LangGraph:官方免费课,Module 3(人机交互)和 Module 4(子图、并行)和今天最相关。

书籍

• Chip Huyen《AI Engineering》第 6 章 Agents 部分的”Planning”和”Agent Failure Modes and Evaluation”:规划、工具选择错误、目标偏离等失败模式,对照我们为什么用固定流程,约 1 小时。
• Jay Alammar、Maarten Grootendorst《Hands-On Large Language Models》第 7 章 Advanced Text Generation Techniques and Tools:后半部分讲 Agent 和 ReAct,图很清楚,40 分钟。

开源项目与代码

• langchain-ai/langgraph:libs/langgraph/langgraph/types.py 里看 interrupt 和 Command 的实现,理解为什么恢复时会重跑节点。
• sierra-research/tau-bench:看它的 agent 是怎么调工具的,和我们的 act 节点对比。

自测

1. tool_log 和 cost_cny 为什么要用 Annotated[..., operator.add]?去掉会怎样?

答案

这两个字段需要累加而不是覆盖。tool_log 记录整张工单所有的工具调用,cost_cny 是各节点花费之和。去掉的话,每个节点返回的值会覆盖之前的,最后只剩最后一个节点的花费,工具记录也只剩最后一批。

2. 质检第一次不通过、第二次也不通过,工作流会怎么走?revisions 分别是多少?

答案

第一次不通过,revisions 从 0 变成 1,1 <= max_revisions(1),回到 draft 重写;第二次不通过,revisions 变成 2,2 > 1,走 handoff 转人工。

3. 审批节点里,如果把”写一条日志到数据库”放在 interrupt() 之前,会发生什么?

答案

会写两次。图第一次执行到审批节点时写一次,然后在 interrupt 处暂停;人做出决定后图恢复,审批节点从头重跑,又写一次。interrupt 之前的代码必须没有副作用或者幂等。

4. 为什么 check 节点先跑规则,规则不过就不调模型了?

答案

规则检查确定、免费、不会漏(编号是否存在、是否有违禁承诺);规则发现问题时草稿反正要重写,再花钱让模型检查没有意义。规则过了,才需要模型做规则做不了的判断(事实是否都有依据)。

5. 「面试题」你的工作流怎么测试?大模型的输出不确定,怎么写单测?

答案

分两层。工作流逻辑(路由、审批暂停恢复、重写次数、转人工条件)用假 LLM 测:通过依赖注入把假的大模型、检索、工具传进节点,按剧本返回固定结果,测试确定、免费、几秒钟跑完。模型能力(分类准不准、回复好不好)不在单测里测,用评测集跑真实模型,看统计指标。两者分开,单测保证”流程对”,评测衡量”效果好”。

6. 「面试题」LangGraph 的 interrupt 是怎么实现暂停和恢复的?需要什么前提?

答案

interrupt() 内部抛一个特殊异常让图停下,当前状态由 checkpointer 按 thread_id 保存;外部用同一个 thread_id 调 invoke(Command(resume=值)) 时,LangGraph 读回状态,从被中断的节点开头重新执行,这次 interrupt() 直接返回 resume 的值。前提是编译时配了 checkpointer,恢复时用同一个 thread_id;因为节点会重跑,interrupt 之前的代码要没有副作用,也不能用 try/except 把 interrupt 包住吞掉异常。

明天预告

Day 48 · 多轮与记忆:会话状态、客户画像、上下文压缩。今天的工作流只能处理”一问一答”,客户追问一句它就忘了前面说过什么。明天加上三种记忆:同一张工单内的多轮对话(状态存进 Postgres,进程重启也不丢)、跨工单的客户画像(吴刚第四次来问银行同步,Agent 应该知道他之前遇到过三次)、对话太长时的自动压缩。