前面几篇介绍的 Persistence、Interrupt、Durable Execution、Subgraph 等能力，最终都服务于同一个目标：让 Agent 在真实环境中长期、稳定地运行。

本文从生产环境出发，把这些能力放回一个完整的执行过程，看看一个 LangGraph Agent Runtime 应该怎样组织。

一、Demo 与生产环境

Demo 通常验证一条业务路径能否成立。

例如做一个简单的售后 Agent：

代码片段Text
用户提出退款
   ↓
LLM 判断意图
   ↓
查询订单
   ↓
生成处理意见
   ↓
返回用户

只要模型能够正确调用订单查询 Tool，并根据结果给出回答，这个 Demo 就已经完成了验证。

真正接入退款系统后，流程会复杂很多：

代码片段Text
退款请求
   ↓
读取订单
   ↓
判断退款条件
   ↓
生成退款方案
   ↓
人工审批
   ↓
调用支付平台
   ↓
保存退款结果
   ↓
通知用户

这里开始出现很多模型之外的问题。

支付平台暂时不可用怎么办？审批人员第二天才处理怎么办？服务重启之后还能否找到原来的退款任务？支付接口已经成功，但程序在写数据库之前崩溃，又怎样避免再次退款？

生产 Runtime 应提供处理这些问题的能力。

LangGraph 在这一层提供了比较完整的机制。Graph 描述 State、Node 和执行路径；Checkpointer 保存运行状态；Retry 处理暂时性故障；Interrupt 负责暂停和恢复；Streaming 将运行过程展示给调用端。

因此，一个生产级 Agent 至少要关注以下三个问题：

代码片段Text
当前执行到了哪里？
执行中断以后，从哪里继续？
外部系统怎样观察和控制这次执行？

后面的设计基本都围绕这三个问题展开。

二、Runtime 的职责

这里需要区分两个层面的 Runtime。

本文所说的 Agent Runtime，是指承载 Agent 执行的一整套运行机制，包括状态、持久化、恢复、流式输出以及部署环境。

LangGraph 代码中还有一个具体的 Runtime 类。Node 可以通过它访问本次执行的 Context、Store 等运行期数据。两者范围不同，不应混在一起。

一个完整的运行结构大致如下：

代码片段Text
Application
    │
    ▼
Graph
State / Node / Edge / Subgraph
    │
    ▼
Execution
Retry / Interrupt / Streaming
    │
    ▼
Persistence
Thread / Checkpoint / Store
    │
    ▼
Infrastructure
API / Queue / Worker / Database
    │
    ▼
Observability
Trace / Run / Metrics

以退款 Agent 为例，Graph 描述的是“查询订单 → 判断条件 → 审批 → 执行退款”这样的业务关系。

Runtime 负责让这条流程可靠执行。某个 Node 失败后是否重试，审批期间保存什么状态，同一个退款任务恢复时如何找到之前的执行现场，都属于这一层。

这种分层很重要。项目早期如果把所有内容都写进 Node，后期很容易把数据库连接、用户信息、重试代码、日志代码和业务逻辑混在一起，Graph 本身反而越来越复杂。

三、状态与运行上下文

生产环境中最容易暴露设计问题的地方，通常是 State。

State 保存的是 Graph 后续执行仍然需要，并且恢复后应该继续存在的数据。

例如退款流程可以定义：

代码片段Python
from typing import TypedDict
 
 
class RefundState(TypedDict):
    order_id: str
    reason: str
    amount: float
    approved: bool | None
    refund_id: str | None
    status: str

TypedDict 用来描述 State 包含哪些字段以及字段类型。它主要服务于类型检查；Graph 中实际流转的仍然是普通字典。

这里保存的都是业务信息：

代码片段Text
order_id    哪个订单
reason      为什么退款
amount      退款金额
approved    是否通过审批
refund_id   支付平台返回的退款编号
status      当前处理状态

这些数据在服务重启以后依然有意义，因此应由 Checkpoint 维护。

相比之下，HTTP Client、数据库连接、模型实例以及临时格式化出来的 Prompt，通常没有必要存进 State。

例如退款判断的 Prompt 可以在 Node 中根据 reason、订单数据和退款规则动态生成。这样以后修改 Prompt 模板时，不需要改变 State Schema，历史 Checkpoint 也不会保存大量已经失去意义的文本。

Runtime Context

还有一些数据每次执行都要使用，但并不属于退款任务本身。

例如：

代码片段Text
当前用户 ID
租户 ID
数据库连接
支付服务 Client
环境配置

这些信息更适合放在运行上下文中。

代码片段Python
from dataclasses import dataclass
 
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
 
 
@dataclass
class Context:
    user_id: str
 
 
builder = StateGraph(
    RefundState,
    context_schema=Context,
)
 
 
def load_order(
    state: RefundState,
    runtime: Runtime[Context],
):
    user_id = runtime.context.user_id
 
    # 实际项目中根据 user_id 和 order_id 查询订单
    return {
        "status": f"order_loaded:{user_id}"
    }

注意这里的 context_schema 和 Runtime[Context]：

context_schema=Context 声明这张 Graph 可以接收一个 Context 对象。

Runtime[Context] 使用泛型标明 runtime.context 的类型，因此 IDE 和类型检查工具可以知道其中存在 user_id 字段。

调用 Graph 时通过 context 传入：

代码片段Python
graph.invoke(
    input_state,
    context={"user_id": "user-1001"},
)

这样，退款流程自身的状态和运行环境保持了清晰边界。

还需要区分 Store。

Checkpointer 保存的是某个 Thread 内的 Graph 状态；Store 更适合保存跨 Thread 使用的数据，例如用户长期偏好、账户画像或者其他长期记忆。

四、节点边界与恢复

Node 在代码里表现为函数，但进入生产环境后，它还决定了执行和恢复的粒度。

假设把整个退款流程写进一个 Node：

代码片段Text
查询订单
校验退款条件
调用退款接口
记录退款结果
发送通知

如果最后一步通知发送失败，这个 Node 可能重新执行。

此时问题就出现了：前面的退款接口是否也会被再次调用？

因此，更稳妥的结构通常会把具有独立执行意义的步骤分开：

代码片段Text
load_order
    ↓
check_policy
    ↓
approval
    ↓
refund
    ↓
save_result
    ↓
notify

这样 notify 失败时，可以只处理通知相关逻辑；支付平台已经完成的退款没有必要跟着重新执行。

Node 应该多细，没有统一答案。

实际项目中可以用一个较为简单的标准判断：

某一步如果需要单独重试、单独观察，或者希望失败后从这里重新执行，就适合成为独立 Node。

这样的结构也更容易排查线上问题。

Trace 中可能直接看到：

代码片段Text
load_order       success
check_policy     success
approval         success
refund           success
save_result      success
notify           failed

问题发生在哪里一目了然。

外部写操作与幂等

退款还涉及一个生产系统里很重要的问题：幂等。

假设支付平台已经退款成功，但应用在保存 refund_id 之前崩溃。服务恢复以后重新进入 refund Node，如果再次请求支付平台，就可能发生重复退款。

因此，调用外部写接口时通常需要业务唯一键。

例如：

代码片段Text
refund_request_id = RF_ORDER_10001

每次重新调用退款接口都携带相同的 refund_request_id。

支付平台如果支持幂等请求，就会识别出这是同一次退款，而不会再次产生资金操作。

LangGraph 可以恢复 Graph 的执行位置，但外部系统能否安全重复调用，仍然需要接口和业务层共同保证。

五、失败与人工介入

生产系统中的异常不适合统一放进一个 try/except。

不同问题对应不同处理方式：

代码片段Text
网络超时          Retry
模型参数不合理    回到 Graph 中重新处理
金额较大          Interrupt 等待人工审批
代码异常          直接失败并记录 Trace

下面用一个简化的退款流程，把 Retry、Interrupt 和 Checkpoint 放在一起。

场景是：

代码片段Text
生成退款方案
   ↓
人工审批
   ↓
调用支付平台退款

其中，interrupt() 用于暂停 Graph。

恢复时使用 Command(resume=...) 将审批结果送回之前的 interrupt()。

RetryPolicy 用于配置 Node 的自动重试规则。max_attempts=3 表示该节点最多执行三次，其中包含第一次正常执行。

代码片段Python
from typing import Literal, TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, RetryPolicy, interrupt
 
 
class RefundState(TypedDict):
    order_id: str
    amount: float
    approved: bool | None
    refund_id: str | None
    status: str
 
 
attempts = 0
 
 
def prepare_refund(state: RefundState):
    return {
        "status": "waiting_approval",
    }
 
 
def approval_node(
    state: RefundState,
) -> Command[Literal["refund", "reject"]]:
 
    approved = interrupt(
        {
            "type": "refund_approval",
            "order_id": state["order_id"],
            "amount": state["amount"],
        }
    )
 
    return Command(
        update={"approved": bool(approved)},
        goto="refund" if approved else "reject",
    )
 
 
def refund_node(state: RefundState):
    global attempts
    attempts += 1
 
    # 模拟支付平台第一次请求超时
    if attempts == 1:
        raise ConnectionError("payment service timeout")
 
    return {
        "refund_id": "RF20260917001",
        "status": "refunded",
    }
 
 
def reject_node(state: RefundState):
    return {
        "status": "rejected",
    }
 
 
builder = StateGraph(RefundState)
 
builder.add_node("prepare", prepare_refund)
builder.add_node("approval", approval_node)
 
builder.add_node(
    "refund",
    refund_node,
    retry_policy=RetryPolicy(
        max_attempts=3,
        retry_on=ConnectionError,
    ),
)
 
builder.add_node("reject", reject_node)
 
builder.add_edge(START, "prepare")
builder.add_edge("prepare", "approval")
builder.add_edge("refund", END)
builder.add_edge("reject", END)
 
graph = builder.compile(
    checkpointer=InMemorySaver()
)

这里的 Command 同时承担状态更新和路由。

代码片段Python
Command(
    update={"approved": bool(approved)},
    goto="refund",
)

update 会更新 State，goto 指定下一个 Node。

返回类型：

代码片段Python
Command[Literal["refund", "reject"]]

表示这个 Node 可能跳转到 refund 或 reject，也让类型检查工具能够获得更明确的信息。

Graph 使用 Interrupt 时需要配置 Checkpointer，因为暂停以后必须保存当前执行状态。

运行时还要指定一个 thread_id：

代码片段Python
config = {
    "configurable": {
        "thread_id": "refund-order-10001"
    }
}
 
 
first = graph.stream_events(
    {
        "order_id": "ORDER-10001",
        "amount": 899.0,
        "approved": None,
        "refund_id": None,
        "status": "created",
    },
    config=config,
    version="v3",
)
 
_ = first.output
 
print(first.interrupted)
print(first.interrupts[0].value)

运行结果如下：

代码片段Text
True
{
    'type': 'refund_approval',
    'order_id': 'ORDER-10001',
    'amount': 899.0
}

Graph 执行到 approval 后暂停。

thread_id 是这次持久化运行的标识。稍后恢复时继续使用同一个 thread_id，Checkpointer 才能找到之前保存的执行状态。

审批通过后继续：

代码片段Python
resumed = graph.stream_events(
    Command(resume=True),
    config=config,
    version="v3",
)
 
print(resumed.output["status"])
print(resumed.output["refund_id"])

运行结果如下：

代码片段Text
refunded
RF20260917001

Command(resume=True) 中的 True 会成为前面 interrupt() 的返回值。

Graph 随后进入 refund。

第一次调用模拟支付平台超时，抛出 ConnectionError。因为 refund Node 配置了 RetryPolicy，LangGraph 会自动再次执行该节点。第二次调用成功，最终得到退款编号。

这里还有一个容易忽略的细节。

Interrupt 恢复以后，包含 interrupt() 的 Node 会从函数开头重新执行。因此，不应在 interrupt() 前执行退款、扣款、发邮件等不可随意重复的操作。

示例使用的 InMemorySaver 将 Checkpoint 保存在内存中，进程结束以后数据也随之消失，适合学习和测试。

真实服务需要跨进程恢复时，应改用持久化 Checkpointer，例如 SQLite 或 PostgreSQL 类型的 Checkpointer 实现。

六、流式交互与观测

退款流程执行过程中，用户可能需要等待模型判断、订单查询、审批甚至第三方支付系统响应。

如果调用端只能等待最终结果，前端通常只能显示：

代码片段Text
退款处理中……

实际上，后端可能已经完成了大部分工作，只是在等待工作人员确认。

生产应用更适合把运行状态持续传给前端。

例如：

代码片段Text
✓ 已读取订单
✓ 已通过退款规则校验
✓ 已生成退款方案
⏸ 等待人工审批
○ 尚未发起退款

LangGraph 提供 stream() 和 Event Streaming 等接口，可以在 Graph 执行过程中持续取得事件。

包含 Interrupt 的流程中，可以通过 stream_events(..., version="v3") 获取消息、Interrupt 和最终输出等数据。

例如：

代码片段Python
stream = graph.stream_events(
    input_state,
    config=config,
    version="v3",
)
 
for message in stream.messages:
    ...
 
if stream.interrupted:
    pending = stream.interrupts
 
final_state = stream.output

这里几个属性分别对应不同信息。

stream.messages 提供模型生成的流式消息。

stream.interrupts 保存当前等待处理的 Interrupt。

stream.interrupted 表示 Graph 是否处于暂停状态。

stream.output 则提供本次执行完成后的最终输出。

这样，前端展示的是 Agent 的真实执行阶段，而不只是一个持续旋转的 Loading。

Trace

Streaming 解决当前运行状态，线上排查还需要完整的历史执行记录。

例如某个用户反馈：

代码片段Text
订单显示退款成功，
银行卡却没有收到退款。

此时开发者需要确认：

代码片段Text
退款 Node 是否实际执行
调用支付平台时发送了什么参数
支付平台返回了什么
是否发生过 Retry
refund_id 是否成功写入 State
后续保存节点是否执行成功

LangSmith Trace 可以把一次运行中的 Graph Node、模型调用和 Tool 调用关联起来。

因此，可观测性的目标不是“多记录一些日志”，而是能够沿着一次真实执行重新还原当时发生的过程。

如果线上系统无法定位到具体 Thread、Node 和 Tool 调用，后续恢复和问题分析都会变得困难。

七、部署与工程边界

规模不大的应用，可以直接把编译好的 Graph 放在现有 Web 服务中。

例如：

代码片段Text
FastAPI
   │
   ▼
CompiledGraph
   │
   ▼
PostgreSQL

API 接收到退款请求后直接调用 Graph。

如果任务执行时间较短、并发规模有限，这种结构已经足够，也方便开发和部署。

任务持续时间变长以后，可以把 API 请求和 Agent 执行分开：

代码片段Text
Client
   │
   ▼
API Server
   │
   ▼
Task Queue
   │
   ▼
Worker
   │
   ├── Graph
   ├── Model
   └── Tools
   │
   ▼
Persistence

例如用户提交退款申请后，API Server 创建任务并快速返回。

Worker 再负责查询订单、调用模型、等待审批以及执行退款。Checkpointer 保存任务状态，即使 Worker 重启，也能够根据原来的 Thread 恢复。

Agent Server 已经提供了类似的运行模型，其中包括 Graph、持久化存储和任务队列。API Server 负责接收请求，Worker 负责实际执行 Graph。

工程上还需要保持一条清晰边界：

代码片段Text
Graph
负责退款流程和状态转换

运行基础设施
负责任务队列、持久化、并发和部署

业务规则放在 Graph 中，例如：

代码片段Text
退款金额超过 500 元进入人工审批

至于哪个 Worker 执行任务、任务怎样排队、服务怎样扩容，则留给运行基础设施。

复杂业务还可以进一步使用 Subgraph。

例如退款 Agent 中，“订单风险审查”本身可能已经包含：

代码片段Text
查询用户历史
↓
检查异常退款次数
↓
分析订单行为
↓
输出风险等级

这部分可以单独组织成一个 Subgraph，再由退款主 Graph 调用。

这样主 Graph 仍然保持清晰：

代码片段Text
读取订单
   ↓
风险审查 Subgraph
   ↓
退款审批
   ↓
执行退款

Subgraph 的意义就在这里：把已经形成独立状态和执行流程的业务模块收进自己的边界中，而不是让主 Graph 无限增长。

八、LangGraph 的使用边界

经过前面这些讨论，再判断一个项目是否需要 LangGraph，会清晰很多。

如果应用只是一次模型调用：

代码片段Text
输入
↓
LLM
↓
输出

或者只有几步固定处理流程，并且没有状态恢复要求，普通函数、LangChain 或简单 Agent 往往已经足够。

当系统逐渐出现多步骤状态流转、条件分支、长时间任务、人工审批、失败恢复、跨请求状态和执行过程观测等要求时，Graph Runtime 的价值才会明显体现出来。

退款 Agent 就是一个典型例子。

刚开始可能只是：

代码片段Text
用户问题 → 查询订单 → LLM 回答

加入真实退款以后，很快就需要考虑审批、幂等、Checkpoint、Retry 和执行记录。开发工作的重点，也逐渐从“模型应该怎样回答”转向“这次任务怎样可靠地完成”。

这也是判断是否需要 LangGraph 的一个实用信号：

当开发重点开始从模型如何完成任务，转向任务执行到一半以后如何管理，就已经进入 Runtime 的问题域。

总结来看，LangGraph 各部分之间的协作关系如下：

State 保存运行过程中需要持续存在的业务事实，Node 划定执行和恢复边界，Checkpoint 保存运行现场，Retry 处理暂时性故障，Interrupt 将人工操作接入流程，Streaming 和 Trace 让外部系统能够看到 Agent 正在执行什么。

一个能够运行的 Demo 可以验证业务思路；当系统还需要处理失败、暂停、恢复、重试和人工操作时，Runtime 的设计才真正决定了这套 Agent 是否适合进入生产环境。