LangGraph 客户支持 Email Agent：从设计思维到可运行代码

上篇你学会了七种 WorkFlow 模式，能点名、能画图。但模式只是菜谱——真给你一封客服邮件，你能画出它该走哪条路吗？这篇用一个完整的客户支持 Email Agent 项目，把 Routing、Prompt Chaining、HITL 串成一张可运行的图。

小明把七模式钉墙上，仍不会开一张业务图

小明刚学完 LangGraph 七种 WorkFlow，觉得信心满满。产品甩过来一个需求：

自动读收件箱 → 分类 → 能答就查知识库回信 → Bug 开工单 → 账单/疑难直接人工 → 必要时跟进。

小明本能反应：create_agent + 一堆工具，让模型自己决定调哪个。

老张拦住他：”你想想，重复扣款的邮件和密码重置的邮件，能走同一条路吗？”

“呃……好像不行？扣款得直接人工，密码重置可以自动回复。”

“对。工具解决的是『会不会干』；这张单子的灵魂是『不同类型必须走不同路』。 七种模式告诉你有哪些拓扑零件；Think in LangGraph 告诉你怎么用这些零件拼一张业务图。”

老张在白板上写了五个字：先想图，再写码。

五阶段思维：从需求到拓扑

老张画了一张流程：

① 拆 Node    → 把任务切成离散步骤
② 明确节点职责  → 每个节点只干一件事；标清谁做决策
③ 设计 State    → 节点之间流转什么数据
④ 写节点 + 异常   → 函数实现、失败怎么收口
⑤ 连边成 Graph  → 串起来；决策可在节点内或条件边

“阶段 ①～③ 先落纸面，④～⑤ 对照仓库 day6-customer-support/。别一上来就接真实 SMTP——先用 mock 数据把图跑通。”

小明问：”第一步拆 Node，怎么拆？”

“看需求里有哪些动作。读邮件是一个动作，分类是一个，搜知识库是一个……每个动作就是一个节点。”

七个节点，各司其职

          ┌──► search_docs ──► draft_reply ──► finalize_auto ──► END
          │     │
          │     └──（feature）──► tag_feature ──► …
START → read_email → classify_intent
          │
          ├──► create_bug_ticket ──► human_review ──► END
          │
          ├──► escalate_human ─────────────────────► END

小明注意到：”classify_intent 是唯一的路由决策点，human_review 是人工决策点——其他节点都是干活的。”

“对。决策点必须显式存在，不能藏在 prompt 里让模型自己猜。这是图和链的核心区别。”

设计 State：节点的合同

“节点拆完了，下一步设计 State——节点之间流转什么数据。”

老张打开 state.py：

# state.py

from typing import TypedDict

class SupportState(TypedDict, total=False):
  # 输入 / 读信
  email_id: str
  from_addr: str
  subject: str
  body: str
  # 分类
  intent: str      # password_reset | bug | billing | feature | tech_complex
  urgency: str     # low | medium | high | critical
  next_action: str   # search_docs | create_bug_ticket | escalate_human
  # 检索 / 工单 / 草稿
  doc_hits: list[str]
  ticket_id: str
  feature_tag: str
  draft: str
  reply_status: str  # auto_sent | queued | skipped | escalated
  # HITL
  human_decision: str  # approve | reject
  followup_at: str
  # 可观测
  trace: list[str]
  error: str

小明：”total=False 是什么意思？”

“意思是所有字段都是可选的。LangGraph 的节点返回部分更新（partial update）——read_email 只写 from_addr/subject/body，classify_intent 只写 intent/urgency/next_action。不需要每个节点都填满整个 State。”

“所以 State 像一份合同：分类节点负责写 intent，检索节点负责写 doc_hits，各写各的，互不干涉。”

“对。别让每个节点都去解析原始 MIME 邮件——那是 read_email 的活。下游只读结构化字段。”

五个场景，五条路

“图设计好了，但怎么验收？老张给了五封代表邮件：”

# mailbox_data.py（节选）

SAMPLE_EMAILS = {
  "e_password": {
    "from": "user@example.com",
    "subject": "无法登录，需要重置密码",
    "body": "你好，我忘记密码了，请问如何重置密码？",
  },
  "e_bug": {
    "from": "pm@example.com",
    "subject": "导出 PDF 时客户端崩溃",
    "body": "点击导出 PDF 后应用闪退，可稳定复现。请帮忙建 Bug 跟踪。",
  },
  "e_billing": {
    "from": "cfo@example.com",
    "subject": "重复扣款紧急处理",
    "body": "本月订阅被重复扣款两次，请立刻人工处理退款。",
  },
  "e_feature": {
    "from": "designer@example.com",
    "subject": "希望增加暗黑模式",
    "body": "产品有暗黑模式吗？如果没有，希望作为功能请求跟进。",
  },
  "e_tech": {
    "from": "dev@example.com",
    "subject": "API 间歇性返回 504",
    "body": "调用 /v1/report 时偶发 504，复杂排查请转技术支持人工。",
  },
}

五封邮件，五种意图，五条路径：

小明复述：”同样是『读了邮件』，后面分叉。账单和疑难禁止走自动温馨提示链。”

“对。如果硬做成一条链『读信→检索→回信→结束』，重复扣款也会被自动礼貌回复糊弄过去——合规与体验双爆。这就是需要图而不是链的原因。“

代码实战：从节点到图

1. 分类节点：规则路由，不靠模型

老张打开 nodes.py，指向 classify_intent：

# nodes.py（节选）

def classify_intent(state: SupportState) -> dict:
  """规则分类（课堂稳定、无需 LLM；生产可换成结构化输出）。"""
  if state.get("error"):
    return {"next_action": "escalate_human",
        "trace": _trace(state, "classify:skip_error")}

  text = f"{state.get('subject', '')} {state.get('body', '')}".lower()
  intent = "tech_complex"
  urgency = "medium"
  next_action = "escalate_human"  # 默认兜底：拿不准就升级人工

  if any(k in text for k in ("密码", "重置", "忘记密码", "登录")):
    intent, urgency, next_action = "password_reset", "low", "search_docs"
  elif any(k in text for k in ("崩溃", "bug", "闪退", "导出 pdf", "pdf")):
    intent, urgency, next_action = "bug", "high", "create_bug_ticket"
  elif any(k in text for k in ("扣款", "退款", "账单", "重复扣")):
    intent, urgency, next_action = "billing", "critical", "escalate_human"
  elif any(k in text for k in ("暗黑", "dark", "功能请求", "希望增加")):
    intent, urgency, next_action = "feature", "low", "search_docs"
  elif any(k in text for k in ("504", "api", "间歇")):
    intent, urgency, next_action = "tech_complex", "high", "escalate_human"

  return {
    "intent": intent,
    "urgency": urgency,
    "next_action": next_action,
    "trace": _trace(state, f"classify:{intent}/{urgency}→{next_action}"),
  }

小明问：”为什么不用 LLM 分类？”

“两个原因：第一，课堂无 Key / 离线也能跑通五条路径；第二，规则分类稳定可复现——同样的输入永远走同一条路，方便调试。”

“生产环境呢？”

“换成 LLM 结构化输出，图拓扑不用改。classify_intent 内部从规则换成模型，对外接口不变——返回 intent/urgency/next_action。这就是节点解耦的好处。”

老张补充了一个细节：”注意默认值是 escalate_human——拿不准就升级人工。这是安全兜底，别让模型在不确定时硬答。”

2. 条件边：分类之后去哪

分类节点写了 next_action，但谁来读这个字段决定下一跳？

def route_after_classify(state: SupportState) -> str:
  return state.get("next_action") or "escalate_human"

就一行。读 next_action，返回节点名。None 时兜底到人工。

def route_after_search(state: SupportState) -> str:
  if state.get("intent") == "feature":
    return "tag_feature"
  return "draft_reply"

搜完知识库也有分叉：功能请求走 tag_feature 打标签，其他走 draft_reply 直接撰稿。

3. HITL：interrupt 暂停，Command 恢复

Bug 路径最特殊——建完工单后要暂停等人审：

def human_review(state: SupportState) -> dict:
  """Bug 工单人工审核：interrupt 暂停，resume 后继续。"""
  payload = interrupt(
    {
      "kind": "bug_ticket_review",
      "ticket_id": state.get("ticket_id"),
      "subject": state.get("subject"),
      "from_addr": state.get("from_addr"),
      "prompt": "approve 或 reject 该 Bug 工单",
    }
  )
  if isinstance(payload, dict):
    decision = str(payload.get("type") or payload.get("decision") or "approve")
  else:
    decision = str(payload or "approve")
  decision = decision.lower()
  if decision not in ("approve", "reject"):
    decision = "approve"
  return {
    "human_decision": decision,
    "reply_status": "queued" if decision == "approve" else "skipped",
    "trace": _trace(state, f"human_review:{decision}"),
  }

小明：”interrupt() 是什么？”

“LangGraph 的暂停指令。执行到这里，图会冻结当前状态，把 payload 抛给调用方。人在外面看完信息，用 Command(resume=...) 把决策塞回去，图接着跑。”

“这和之前学的 create_agent + Middleware 有什么区别？”

“create_agent 的 Middleware 是发信前拦截——零件级；这里的 interrupt 是图节点级暂停——总装级。零件和总装对照着学。”

4. 组装图：add_node + add_edge

graph.py 把所有零件拼起来：

# graph.py

def build_graph(*, checkpointer: InMemorySaver | None = None):
  g = StateGraph(SupportState)
  g.add_node("read_email", read_email)
  g.add_node("classify_intent", classify_intent)
  g.add_node("search_docs", search_docs)
  g.add_node("tag_feature", tag_feature)
  g.add_node("draft_reply", draft_reply)
  g.add_node("finalize_auto", finalize_auto)
  g.add_node("create_bug_ticket", create_bug_ticket)
  g.add_node("human_review", human_review)
  g.add_node("escalate_human", escalate_human)

  g.add_edge(START, "read_email")
  g.add_edge("read_email", "classify_intent")
  g.add_conditional_edges(
    "classify_intent",
    route_after_classify,
    {
      "search_docs": "search_docs",
      "create_bug_ticket": "create_bug_ticket",
      "escalate_human": "escalate_human",
    },
  )
  g.add_conditional_edges(
    "search_docs",
    route_after_search,
    {"tag_feature": "tag_feature", "draft_reply": "draft_reply"},
  )
  g.add_edge("tag_feature", "draft_reply")
  g.add_edge("draft_reply", "finalize_auto")
  g.add_edge("finalize_auto", END)
  g.add_edge("create_bug_ticket", "human_review")
  g.add_edge("human_review", END)
  g.add_edge("escalate_human", END)

  saver = checkpointer if checkpointer is not None else InMemorySaver()
  return g.compile(checkpointer=saver)

小明逐行读：”先 add_node 注册九个节点，再 add_edge 串边。直边是固定路径，add_conditional_edges 是分叉路径。”

“对。注意 human_review 用了 interrupt，所以必须配 checkpointer——否则暂停后状态没地方存，恢复时找不到。这里用 InMemorySaver，生产换 PostgresSaver。”

5. 交互式 CLI：让 HITL 可触摸

图组装好了，怎么跑？main.py 不是一键脚本，而是一个交互式命令行：

# main.py（核心循环）

def main() -> None:
  print("=== Day6 客户支持 Email Agent ===")
  app = build_graph()
  pending_cfg: dict | None = None   # 暂停中的图配置
  seq = 0

  while True:
    prompt = "审核> " if pending_cfg is not None else "> "
    line = input(prompt).strip()

    # ... 命令解析（list / run / help / exit）...

    # HITL：暂停中只接受审批
    if pending_cfg is not None:
      decision = _normalize_decision(cmd)  # a→approve, r→reject
      out = app.invoke(Command(resume={"type": decision}), pending_cfg)
      _print_case(f"人工审核后继续（{decision}）", out)
      pending_cfg = None
      continue

    # 正常执行
    cfg = {"configurable": {"thread_id": f"cli-{seq}-{email_id}"}}
    result = app.invoke(_empty_state(email_id), cfg)
    if _has_interrupt(result):
      _print_interrupt(_interrupt_payload(result))
      pending_cfg = cfg  # 记住配置，等下一步恢复
      continue

    _print_case(mail["subject"], result)

小明：”这个设计很巧妙——pending_cfg 为 None 时是正常输入，不为 None 时切换到 审核> 提示。”

“对。Bug 路径走到 human_review 时，图会 interrupt 暂停。main.py 检测到暂停后，把 cfg 存进 pending_cfg，下一轮输入提示变成 审核>，只接受 a/approve 或 r/reject。”

“恢复时呢？”

“用 Command(resume={"type": decision}) 再 invoke 一次，传同一个 thread_id。图从冻结点继续执行，把决策塞回 interrupt() 的返回值。”

老张补充：”注意 _normalize_decision 做了容错——a、approve、y、yes 都映射到 approve；输入乱七八糟的返回 None，提示重新输入。人机交互的鲁棒性和图逻辑一样重要。“

完整运行：交互式五场景验收

“启动 CLI，逐封处理邮件：”

cd course-langgraph/day6-customer-support
python main.py

=== Day6 客户支持 Email Agent ===
命令:
  list        列出样例邮件
  run <email_id>  处理一封邮件（如 run e_bug）
  <email_id>    同上简写（如 e_password）
  help        显示帮助
  /exit | quit    退出

HITL 暂停时（Bug 工单）:
  a | approve     审批通过并恢复执行
  r | reject    拒绝并恢复执行

样例邮件: e_password, e_bug, e_billing, e_feature, e_tech

> list
  e_password: 无法登录，需要重置密码  (from user@example.com)
  e_bug: 导出 PDF 时客户端崩溃  (from pm@example.com)
  e_billing: 重复扣款紧急处理  (from cfo@example.com)
  e_feature: 希望增加暗黑模式  (from designer@example.com)
  e_tech: API 间歇性返回 504  (from dev@example.com)

> e_password
--- 无法登录，需要重置密码 ---
  intent=password_reset urgency=low next=search_docs
  reply_status=auto_sent
  draft: 您好，关于「无法登录，需要重置密码」：密码重置：登录页点击「忘记密码」，验证邮箱后 15 分钟内有效。如仍无法解决请回复本邮件。
  trace: read_email:e_password → classify:password_reset/low→search_docs → search_docs:hits=1 → draft_reply:auto_sent → finalize:auto_sent

> run e_bug
--- HITL 暂停：等待人工审核 ---
  kind: bug_ticket_review
  ticket_id: BUG-1001
  subject: 导出 PDF 时客户端崩溃
  from_addr: pm@example.com
  prompt: approve 或 reject 该 Bug 工单
  输入 a/approve 通过，或 r/reject 拒绝。

审核> a
  resume: approve
--- 人工审核后继续（approve） ---
  intent=bug urgency=high next=create_bug_ticket
  reply_status=queued ticket=BUG-1001
  human_decision=approve
  trace: read_email:e_bug → classify:bug/high→create_bug_ticket → create_bug_ticket:BUG-1001 → human_review:approve

> e_billing
--- 重复扣款紧急处理 ---
  intent=billing urgency=critical next=escalate_human
  reply_status=escalated
  draft: [已升级人工] intent=billing urgency=critical from=cfo@example.com
  trace: read_email:e_billing → classify:billing/critical→escalate_human → escalate_human:billing

> e_feature
--- 希望增加暗黑模式 ---
  intent=feature urgency=low next=search_docs
  reply_status=auto_sent feature_tag=feature:dark-mode
  draft: 您好，已登记功能请求（feature:dark-mode）。参考：dark: 外观设置…感谢反馈。
  trace: read_email:e_feature → classify:feature/low→search_docs → search_docs:hits=1 → tag_feature:feature:dark-mode → draft_reply:auto_sent → finalize:auto_sent

> e_tech
--- API 间歇性返回 504 ---
  intent=tech_complex urgency=high next=escalate_human
  reply_status=escalated followup_at=T+1 技术支持回访
  trace: read_email:e_tech → classify:tech_complex/high→escalate_human → escalate_human:tech_complex

> /exit

小明逐条对照：”五条路径全通——密码重置自动回复，Bug 建单后人审，扣款和 504 直接升级人工，功能请求打标签再回复。”

“注意 Bug 路径——run e_bug 后提示变成 审核>。你看 ticket_id: BUG-1001、subject、from_addr 都显示出来了，人看完再决定 approve 还是 reject。输入 a 后，Command(resume={"type": "approve"}) 把决策塞回去，图继续跑。”

“如果 reject 呢？”

“reply_status 变成 skipped，工单不发出。但 trace 里仍然记录了 human_review:reject——审计可追溯。”

为何不是一条 Prompt Chain

链适合「步骤永远一样」；客服邮件的步骤故意不一样——所以用图。

三个常见坑

坑 1：用一条 Chain「先回信再分类」

症状：紧急账单已自动发出安抚邮件，财务被动。

原因：把 Chaining 当默认——读信 → 检索 → 回信，一条直线。

修法：classify 必须在自动回复之前；critical 意图直接路由到 escalate_human，跳过自动回信链。

坑 2：所有决策只写在 Prompt 里

症状：图上一条直线，分叉全靠模型自觉。换个模型行为就变了。

原因：Think 阶段偷懒，没标决策节点。把”如果是 Bug 就建工单”写在 prompt 里，模型偶尔不听。

修法：至少有一个显式 classify_intent 节点；路由结果写进 State（next_action），用条件边控制下一跳。决策是拓扑，不是 prompt。

坑 3：interrupt 不配 checkpointer

症状：human_review 里调了 interrupt()，但恢复时报错”找不到状态”。

原因：interrupt 依赖 checkpointer 持久化中间状态。没配 saver，暂停后状态丢失。

修法：g.compile(checkpointer=InMemorySaver())，并且每次 invoke 传同一个 thread_id。生产换 PostgresSaver。

总结

老张说：”LangGraph 专题四篇走完：为何 Graph → 三要素 → 七模式 → 怎么想一张业务图。”

“三个核心理解：

1. 五阶段思维 —— 拆 Node → 定职责 → 设计 State → 写节点异常 → 连 Graph。前三步落纸面，后两步落代码。
2. 客服要用图 —— 五场景五条路；链无法表达『有的信禁止自动回』。决策点必须显式存在，不能藏在 prompt 里。
3. 设计先于代码，代码已可验收 —— 拓扑 + State + 路径表三件套；交互式 CLI 让 HITL 可触摸——审核> 提示让人审不再是代码里的概念，而是真实操作。HITL 零件见 day12，总装见 day6-customer-support/。”

LangGraph 专题进度：

01 介绍（可靠度 × 自主性）
    ↓
02 三要素（State / Node / Edge）
    ↓
03 七种 WorkFlow（模式菜单）
    ↓
04 Think in LG：客服 Email Agent（本篇收束）
    ↓
落地：day6-customer-support（交互式 CLI + 五场景 + Bug HITL）
选修：Postgres / Studio / 多 Agent handoff

小明把五场景路径表拍进纪要：”设计说明按这张交；CLI 我今晚逐封跑绿。”

先画对路，再写会跑的节点。Think in LangGraph，想的是业务拓扑，不是又一个 while True。