Q52 · 如何设计 LangChain 到 LangGraph 的迁移路径？

某个电商客服应用已经用 LangChain 的 Agent 接入订单查询和政策查询。用户问“订单 A123 能退款吗”，它会让模型选择工具、读取结果并回答。现在业务新增两项要求：证据不全时必须补查；涉及真正的退款动作时，先等人工批准，审批可能跨天，服务重启后仍能继续。负责人问你：“要不要把系统迁到 LangGraph？怎么迁才不会让线上客服突然乱答、重复退款？”

这道题考的不是把 AgentExecutor 改名为 StateGraph。要先判断现有系统到底缺什么控制能力，再把业务行为、数据和副作用逐步搬进明确的流程，并用旧系统作为对照验证。尤其要纠正一个常见误解：今天的 LangChain Agent 本身就建立在 LangGraph 之上。如果目前只需要模型与工具的常规循环、再加少量中间件，继续使用 langchain.agents.create_agent 可能最省事；当你需要自己定义复杂分支、独立状态、可恢复的人工关口和步骤级错误处理，才有理由构建定制 LangGraph。迁移是改变编排与控制权，未必是“从一个完全不用 LangGraph 的库换到另一个库”。LangChain 概览、LangGraph 概览

先解释迁移时会碰到的词

词	小白可以怎样理解	本题里的对应物
Agent	能根据当前信息决定是否调用工具、调用哪个工具，再继续回答的运行流程	已上线的售后助手
LangChain create_agent	当前 LangChain 的高层 Agent 创建入口，默认提供模型与工具的循环	原本能查订单和政策的 Agent
LangGraph	用节点、状态和边组织可控制、可恢复流程的运行框架	把补查、审批、执行拆清楚的客服流程
State（状态）	当前这件事已知的事实和处理进度，不是提示词全文	订单号、查询结果、政策版本、是否已批准
Node（节点）	执行一个职责并返回状态更新的步骤	查订单、查政策、核对证据、等待审核
Edge（边）	指定下一步去哪儿的连接	证据齐全去生成答复；缺政策去补查
Checkpointer	保存某条图流程的状态快照，让它之后能继续	审批跨天后找回待处理草稿
thread_id	用来定位同一条图执行的标识，应用必须自己设计其归属	某次售后申请的执行 ID，不直接等于用户 ID
人工中断	流程到关口先停下，等人给出批准、修改或拒绝	真退款前等客服主管审核
幂等	同一业务请求重试多次，最终也只产生一次有效动作	同一审批单不会退两次钱
影子流量	新流程处理真实输入并记录结果，但不替旧流程向用户输出或执行动作	对比两套判断，线上仍由旧系统答复
灰度发布	先让少量请求走新流程，观察后再扩大比例	从内部账号开始，再逐步扩大
回滚	新流程出问题时让新请求回到旧入口，并妥善处理已在新流程中的任务	停止新请求切入，保留未完审批记录

文中的 A123 是虚构订单号，v3 是虚构的现行政策版本。它们只是教学示例，不能当成真实商家的退款规定。代码里的 state 是当前一条申请的状态字典；route 是决定下一节点的函数名；approved 表示审核结果。先知道这些名字的意思，再看代码，才不会误以为变量本身有某种“框架魔法”。

第一步：判断要迁移哪一部分

图左的模型与工具已经能正常循环；右边的图把“查订单后根据条件分流”以及“要人工审核的路径”显式画了出来。桥上的三个路标不是要求某天一次性重写，而是提醒你：记录旧行为、比较新旧行为、小范围切流应当是三个可分别验收的阶段。图示是示意，生产系统里的审批记录、权限、退款接口还需要独立设计。

盘点时先把现有应用分成四层：

1. 入口与用户契约：用户说什么，返回什么格式；是否必须给政策来源；不知道时该说“无法核实”还是转人工。
2. 模型决策部分：当前提示词、模型、工具定义、消息历史、最多调用几轮、解析失败如何处理。
3. 业务事实与动作：订单查询、政策查询、退款执行分别由哪个服务完成；身份和权限在哪里核验；写操作是否有幂等键。
4. 运行与观测：超时、重试、日志、追踪、成本、延迟、失败率、人工队列、旧版系统的回滚开关。

这样盘点有一个实际目的：先把已有正确行为写成可检查的契约。例如：“A123 已拆封，当前政策要求已拆封转人工，Agent 不得直接说可自动退款，更不得调用退款 API。”如果只比较最终文字的语气，新旧两版说法稍有不同就会误判；如果只比较“有没有报错”，却可能漏掉退款条件发生变化。更可靠的对照要记录订单事实、政策版本、工具调用、审批是否出现、最终结论和是否产生外部动作。

然后做技术分流：

当前实际需求	倾向路径	原因
只有常规的模型选工具、工具结果返回模型，并希望补一个提示词或工具错误处理中间件	保留或升级到当前 create_agent	不必为了出现“迁移”二字而手写每条边
固定的“模板→模型→解析”顺序调用	保留 Runnable/LCEL 组合	固定流水线并不需要 Agent 图流程
必须明确“缺证据补查、达到上限停止、敏感动作审批、审批跨天恢复”等业务路径	定制 LangGraph	这些路径要成为代码中可测试的状态与路由

如果现有代码用的是旧 AgentExecutor、LLMChain 或旧版 create_react_agent，还要先识别版本。LangChain v1 的官方迁移指南把旧链等放进 langchain-classic，并推荐把旧预构建 create_react_agent 迁到 langchain.agents.create_agent；它不等于要求所有人手写 StateGraph。迁移计划应注明依赖版本、旧 import 与目标接口，避免照着早年的教程写出在当前环境不存在的 API。LangChain v1 迁移指南

第二步：把业务流程先画出来，再决定节点

以 A123 为例，先不用代码写业务路径。前半段按顺序执行：查订单，取得下单人、签收日和拆封状态；查当前政策，取得版本、生效日期、退款条件和例外；然后核验证据。

核验之后才选择后续路径：

• 缺字段或政策版本过期：在限定次数和时间内补查。
• 补查后仍无法核实：说明原因，停止资格判断。
• 用户只需要解释规则：生成带政策来源的答复。
• 用户要求执行退款：先取得人工批准，批准后再次核验订单，最后调用退款服务。

为什么“再次核验订单”要放在审批之后？因为审批可能隔天才结束。原先的订单状态、额度或政策可能已经变了。Checkpointer 能保存之前的进度，不能保证外部世界一直不变；在真正写入前要读最新事实。也不要把“生成答复”与“执行退款”合成同一节点：前者是文本生成，后者是有资金副作用的业务动作，失败和重试要求完全不同。

接着定义状态。只保存后续步骤确实需要、又不能轻易重建的事实：order_id、核实过的订单资料、政策版本和来源、已补查次数、审核请求 ID 与结果。不要把整份冗长的提示词或 240 条工具结果硬塞入状态。可在状态里放文档 ID 与必要片段，节点需要时再从受控数据源取；提示词在生成节点中按当前事实构造。LangGraph 官方建议按业务步骤划分节点，并让状态保存原始数据而非预先格式化的提示文本。Thinking in LangGraph

节点粒度要和失败边界一致：订单查询服务可以单独超时或重试；政策检索有自己的版本校验；人工审核需要中断恢复；退款写入必须有独立的权限和幂等保护。这样某个政策服务超时，流程可以明确进入“稍后重试或无法核实”，而不是让模型在不完整资料上编一个肯定答案。

一个没有模型密钥的小例子：把关键分支写成图

下面代码只演示迁移中最关键的“状态、节点、分支”骨架，故意用固定的虚构订单和政策数据代替真实 API、模型及人工审批。读者可以只看执行顺序；真正上线时，load_order、load_policy 必须接有权限控制的服务，并把“审核通过后执行退款”放进后续独立流程。安装当前兼容版本的 langgraph 后可运行这个示意例子。LangGraph Graph API Quickstart

from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END


class RefundState(TypedDict, total=False):
    order_id: str             # 输入的订单号
    opened: bool              # 订单系统核实的“是否已拆封”
    policy_version: str       # 当前政策版本
    policy_found: bool        # 是否查到可核实的现行政策
    answer: str               # 给用户看的说明


def load_order(state: RefundState) -> dict:
    if state["order_id"] != "A123":
        raise ValueError("演示数据只有 A123")
    return {"opened": True}  # 假装订单服务确认：已拆封


def load_policy(state: RefundState) -> dict:
    return {"policy_version": "v3", "policy_found": True}


def route(state: RefundState) -> Literal["explain", "need_human"]:
    if not state["policy_found"]:
        return "need_human"  # 没证据，不允许肯定回答
    if state["opened"]:
        return "need_human"  # 教学假设：已拆封转人工
    return "explain"


def explain(state: RefundState) -> dict:
    return {"answer": f"请依据现行政策 {state['policy_version']} 核实退款条件。"}


def need_human(state: RefundState) -> dict:
    return {"answer": "当前不能自动判定退款，请人工核验。"}


builder = StateGraph(RefundState)
builder.add_node("load_order", load_order)
builder.add_node("load_policy", load_policy)
builder.add_node("explain", explain)
builder.add_node("need_human", need_human)
builder.add_edge(START, "load_order")
builder.add_edge("load_order", "load_policy")
builder.add_conditional_edges("load_policy", route)
builder.add_edge("explain", END)
builder.add_edge("need_human", END)
graph = builder.compile()

result = graph.invoke({"order_id": "A123"})
print(result["answer"])
# 当前不能自动判定退款，请人工核验。

从上到下读：RefundState 规定同一件售后申请可携带哪些字段；load_order 和 load_policy 各自返回一个状态更新字典；图把这些更新合入当前状态；route 查看已核实的 opened 与 policy_found，返回下一节点的名字。A123 在虚构数据里已拆封，因此走 need_human，最后只输出“转人工”说明。代码里的 need_human 没有真正暂停等待人，也没有执行退款；这样写是为了先把可核查的分支与状态机制讲清楚，不能误当成完整生产方案。

正式添加“跨天审批”时，应在审核节点使用官方 interrupt() 与 checkpointer 保存流程，恢复时带同一 thread_id 和审核结果继续；所选保存后端必须能跨进程持久化，内存保存器不能承担服务重启后的恢复承诺。interrupt() 恢复时节点会从开头重新执行，所以中断前不能放不可重试的退款写入。实际退款动作还要有后端权限、审批单核验、业务幂等键和最新订单状态检查；不能把“图状态显示 approved=true”当作支付系统的授权。LangGraph Interrupts、LangGraph Persistence

第三步：用旧系统的真实行为检验新系统

迁移前先固定一组代表性的历史与人工构造案例，再让旧版和新版在同一组输入、同一份订单与政策快照上运行。要比较的是业务事实与安全约束，而非逐字比较模型生成的语句：

场景	期望结果	需要抓到的错误
A123 已拆封，政策 v3 要求转人工	不自动执行退款，说明依据	旧版本政策或“7 天”单条件被误用
政策服务超时	有界重试后说暂无法核实	把“未查到”说成“不符合”或“符合”
政策查到但缺例外条款	补查或停在人工核验	只凭前半段文字做决定
审核员拒绝	无退款动作，有拒绝记录	图在审批前已调用写入工具
审核通过后服务重启	从正确申请恢复，再核验最新状态	丢失审批、错误复用别人的 thread_id
退款 API 返回超时后重试	查询动作结果、用同一幂等键处理	重复退款

对于模型参与的步骤，要保存提示词版本、模型版本、工具输入输出和最终判断；可通过固定工具快照让测试更稳定。评测不能只看回答“像不像”：还要看错误退款率、无依据肯定回答率、人工审批覆盖率、证据引用正确率、平均与尾部延迟、调用成本、恢复成功率。用户看不到的影子流量仍可能触发真实工具，所以影子模式必须禁用真实退款、发信等外部写入，或者用明确的沙箱服务；否则“只是比较一下”也会造成资金副作用。

这一步还要检查状态兼容性。原系统历史消息怎样映射到新 State？旧会话 ID 与图的 thread_id 怎样区分？新版本上线后，已暂停的旧图实例是让旧代码处理到结束，还是有经过验证的状态迁移脚本？不能只切换代码版本，却假设持久化记录自动理解新状态结构。状态字段变更要有版本和迁移策略，尤其是跨天审批中的记录。

第四步：灰度切换与可执行的回滚

先在内部账号或极少量低风险请求中开启新流程，观察到足够覆盖分支与失败路径后逐步扩大。切流开关最好在请求入口按稳定业务键决定，让同一售后申请的后续请求继续走同一版本；不要这轮由新图保存审批，下轮突然被路由到只懂旧会话格式的 Agent。逐步放量时，观察人工待办堆积、图中断数量、重试次数、答复质量和外部写入差异，而不是只盯服务是否存活。

如果出现错误，停止把新请求切入新图通常很容易；真正难的是已有的新图任务。回滚方案应提前写清：待审核申请由哪个版本继续、保存记录是否保留、已执行退款怎样查证、是否需要人工接管。对于已经发生的外部动作，回滚程序代码不会自动撤销钱款；要依据支付或业务系统的正式补偿流程处理。新旧两版并存一段时间、保留旧版只读或处理旧任务的能力，比简单删除新状态数据安全得多。

面试时可以这样回答

我会先确认迁移的目的：当前 LangChain create_agent 已建立在 LangGraph 上，常规模型与工具循环未必需要手写图；如果业务要求明确的补查分支、人工审批、跨天恢复和不同步骤的重试边界，就考虑定制 LangGraph。先盘点旧系统的输入输出、工具、提示词、历史状态、副作用与指标，并把关键案例固定成回归集。然后从业务流程设计状态、节点和边，把订单查询、政策核实、审核与退款执行分开；持久化用可跨进程的 checkpointer，外部写入靠业务幂等和重新核验。先在相同数据快照下比较新旧结果，再跑不产生真实写入的影子流量，随后小比例灰度。切流时按申请稳定路由，保留旧版处理在途任务；回滚既要能停止新流量，也要安排已暂停和已执行的任务。核心是迁移后业务判断更可控、可恢复，且不会因为重试或版本切换产生重复退款。

如果面试官追问“create_agent 已经基于 LangGraph，还需要迁吗”，答：不一定。只要高层 Agent 已满足需求，继续使用它和中间件更简单；只有要显式控制节点、状态、分支、暂停与恢复粒度时才定制图。如果追问“迁完是否天然保证恰好一次退款”，答：不能。图的 checkpoint 管的是流程状态，退款接口另有事务边界；需业务幂等键、当前状态核验和对账。

资料来源

• LangChain 官方文档：Overview
• LangChain 官方文档：LangChain v1 migration guide
• LangGraph 官方文档：Overview
• LangGraph 官方文档：Thinking in LangGraph
• LangGraph 官方文档：Quickstart
• LangGraph 官方文档：Interrupts
• LangGraph 官方文档：Persistence