1. 从脚本到服务LangGraph 部署到底在解决什么问题很多人第一次接触 LangGraph都是在 Jupyter Notebook 或者一个单文件 Python 脚本里跑通一个带状态的多轮对话 Agent。本地跑得挺爽graph.invoke()一调状态机流转、条件边跳转、工具调用全都正常。但一旦要把这个东西交给前端调用、交给同事测试、或者挂到线上让真实用户用问题就全冒出来了状态怎么持久化多个用户同时访问会不会串状态服务重启之后对话历史还在不在怎么观测每一步的执行链路这一篇就是冲着这些问题来的。标题里的“三条部署路径”说的不是三种花哨的架构而是从最简单的脚本封装到 FastAPI 服务化再到接入 LangSmith 做可观测与持久化的渐进式落地方式。核心关键词LangGraph、部署、FastAPI、LangSmith、Checkpointer会贯穿全文。如果你已经能用 LangGraph 写出一个能跑的图但卡在“怎么让它变成一个真正的服务”这一步那这篇内容就是给你准备的。哪怕你只是刚看完 LangGraph 官方教程的新手我也会把每一步的来龙去脉讲清楚让你能直接抄作业。先说清楚一个认知LangGraph 本身是一个编排框架它负责定义节点、边、状态和流转逻辑但它不是一个 Web 服务框架。它不会帮你处理 HTTP 请求、不会帮你做并发、不会帮你管理进程。所以“部署 LangGraph”这件事本质上是给 LangGraph 的图套一层服务外壳再解决状态存储和可观测性。理解了这一点三条路径的划分就顺理成章了。我先把三条路径的定位摆出来方便你对照自己的场景选路径形态适用场景核心依赖路径一脚本 CLI 调用本地验证、批处理、一次性任务LangGraph 本体路径二FastAPI 服务化前后端分离、多用户、API 调用FastAPI Uvicorn路径三服务化 持久化 可观测生产环境、多轮对话、链路追踪FastAPI Checkpointer LangSmith这三条路径不是互斥的而是递进的。你可以先跑通路径一再平滑升级到路径二最后补上路径三的持久化和观测能力。下面我逐条拆开讲重点放在路径二和路径三因为那才是真正“从脚本到服务”的跨越。2. 路径一脚本直跑先把图跑通再谈部署2.1 为什么部署之前必须先有一个干净的脚本我见过太多人一上来就搭 FastAPI 目录、配 Uvicorn、写路由结果图本身的状态定义就有问题调半天以为是服务的问题其实是StateGraph的 reducer 写错了。所以我的建议永远是先用一个最朴素的脚本把图跑通确认输入输出符合预期再考虑套服务。这个脚本的形态很简单一个graph.py定义 State、节点、边最后compile()出一个可执行对象。关键是要把图的输入契约和输出契约固定下来。所谓输入契约就是这个图接收什么样的初始状态输出契约就是invoke之后返回什么结构。这两个契约一旦稳定后面套 FastAPI 就是纯粹的搬运工作。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str def chatbot(state: AgentState): # 这里放你的模型调用逻辑 return {messages: [(assistant, 收到你的消息了)]} builder StateGraph(AgentState) builder.add_node(chatbot, chatbot) builder.add_edge(START, chatbot) builder.add_edge(chatbot, END) graph builder.compile()这段代码里有个细节值得说add_messages这个 reducer 决定了messages字段是追加而不是覆盖。如果你不加Annotated每次节点返回都会把整个 messages 列表替换掉多轮对话直接丢失历史。这是新手最常踩的坑之一我在早期调试时也栽过明明图跑通了但对话永远只有最后一轮。2.2 脚本阶段的验证清单在进入服务化之前我通常会跑一个验证清单确认图的行为是确定的单轮输入能否得到预期输出多轮输入时状态是否正确累积条件边在边界情况下是否走对分支异常输入空消息、超长文本是否会崩溃这个阶段不需要任何 Web 框架一个if __name__ __main__:加上几个测试用例就够了。脚本阶段的目标是消除图本身的 bug而不是解决服务的问题。把这两件事混在一起排查效率会低得离谱。提示脚本阶段就把user_id这类会话标识放进 State后面做多用户隔离时会省很多事。等到服务化再回头改 State 结构所有节点都要跟着动。3. 路径二FastAPI 服务化把图变成可调用的 API3.1 为什么选 FastAPI 而不是 Flask热词里同时出现了fastapi和flask 与 fastapi 比较说明这是很多人的纠结点。我的结论很直接LangGraph 服务化优先选 FastAPI。原因有三条都是实打实的工程考量。第一LangGraph 的很多能力是异步的ainvoke、astream这些异步接口在流式输出场景下几乎是刚需。FastAPI 原生支持async def而 Flask 是同步 WSGI 框架要跑异步得额外折腾。第二FastAPI 基于 Pydantic 做请求体校验而 LangGraph 的 State 本身就是类型化的两者天然契合请求模型可以直接映射到状态字段。第三FastAPI 自带 OpenAPI 文档前端联调时直接看/docs就行省掉大量沟通成本。Flask 不是不能用如果你的团队已经有一套成熟的 Flask 体系硬切 FastAPI 也有迁移成本。但如果是新项目没有理由不选 FastAPI。3.2 FastAPI 项目目录结构怎么组织热词里fastapi项目目录结构是个高频问题。LangGraph 服务化的目录结构我推荐按“图逻辑”和“服务逻辑”分离的原则来组织project/ ├── app/ │ ├── main.py # FastAPI 入口注册路由 │ ├── api/ │ │ └── routes.py # 路由定义 │ ├── graph/ │ │ ├── state.py # State 定义 │ │ ├── nodes.py # 节点函数 │ │ └── builder.py # 图构建与 compile │ ├── schemas/ │ │ └── chat.py # Pydantic 请求/响应模型 │ └── core/ │ └── config.py # 配置管理 ├── requirements.txt └── run.sh这个结构的核心思想是graph/目录里的东西不依赖 FastAPI它就是一个纯粹的 LangGraph 模块可以单独测试、单独复用。api/目录只负责把 HTTP 请求翻译成图调用把图的结果翻译成 HTTP 响应。这样分层之后你想把图迁移到别的框架比如换成 gRPC只需要重写api/层graph/层一行不用动。3.3 核心路由实现与请求模型设计请求模型用 Pydantic 定义字段和 State 对齐from pydantic import BaseModel, Field class ChatRequest(BaseModel): message: str Field(..., min_length1, max_length4000) user_id: str Field(..., min_length1) thread_id: str Field(defaultdefault)路由层拿到请求后构造初始状态调用图from fastapi import APIRouter from app.graph.builder import graph from app.schemas.chat import ChatRequest router APIRouter() router.post(/chat) async def chat(req: ChatRequest): initial_state { messages: [(user, req.message)], user_id: req.user_id, } config {configurable: {thread_id: req.thread_id}} result await graph.ainvoke(initial_state, configconfig) return {reply: result[messages][-1].content}这里有个关键点config里的thread_id。在路径二阶段如果你还没接 Checkpointer这个thread_id其实不起作用每次调用都是全新的状态。但我强烈建议你从第一天就把thread_id传进去因为路径三接上 Checkpointer 之后这个字段就是会话隔离的核心提前埋好能省掉一次重构。3.4 Uvicorn 启动与日志丢失问题热词里uvicorn fastapi 日志丢失问题是个很典型的坑。默认情况下Uvicorn 有自己的日志配置如果你在代码里用logging.getLogger(__name__)打日志可能会发现日志没输出或者格式和 Uvicorn 的不一致。解决办法是显式配置日志让应用日志和 Uvicorn 日志走同一套 handlerimport logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, )启动命令用uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 2关于--workers这里要提醒一句如果你的图里用了内存态的 Checkpointer比如MemorySaver多 worker 会导致状态不共享。因为每个 worker 是独立进程各自持有自己的内存。这种情况下要么用单 worker要么直接上持久化的 Checkpointer。这个坑我在压测时遇到过两个 worker 轮流处理同一个用户的请求对话历史时有时无排查了半天才定位到是进程隔离的问题。4. 路径三Checkpointer 持久化让对话真正“记得住”4.1 Checkpointer 到底解决了什么前面路径二的服务每次请求都是“无状态”的——图跑完状态就没了。用户第二轮提问时你得自己把历史消息拼回去而且一旦服务重启所有上下文清零。Checkpointer 就是 LangGraph 提供的状态持久化机制它在图的每个 super-step 之后把状态快照存下来下次用同一个thread_id调用时自动把历史状态恢复回来。这个机制的价值在于你不需要自己维护对话历史表不需要手动拼接 messagesLangGraph 帮你把“状态”这件事管起来了。对于多轮对话 Agent 来说这是从 demo 到可用的分水岭。4.2 内存态与持久化 Checkpointer 的选型LangGraph 提供了多种 Checkpointer选型逻辑很清晰Checkpointer存储位置适用场景注意事项MemorySaver进程内存本地开发、单进程测试重启即丢多 worker 不共享SqliteSaver本地文件单机小规模、原型验证并发写入需注意锁PostgresSaver数据库生产环境、多实例需初始化表结构开发阶段用MemorySaver完全够用但上线前一定要换成持久化方案。我个人的习惯是本地用 SqliteSaver 验证持久化逻辑生产用 PostgresSaver。切换 Checkpointer 的代码改动极小因为接口是统一的from langgraph.checkpoint.memory import MemorySaver # from langgraph.checkpoint.postgres import PostgresSaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer)编译时传入checkpointer调用时传入thread_id持久化就生效了。这里有个容易忽略的点接入 Checkpointer 后ainvoke的初始状态只需要传本轮的新消息历史消息会自动从 checkpoint 恢复。如果你还是把完整历史拼进去会导致消息重复。这个行为差异一定要在切换时验证清楚。4.3 多用户会话隔离的实操要点thread_id是会话隔离的唯一标识。实践中我建议用user_id 会话标识的组合来构造thread_id比如f{user_id}:{conversation_id}。这样同一个用户可以有多个独立会话不同用户之间天然隔离。注意thread_id不要用自增数字或者可预测的短字符串否则容易被遍历。用 UUID 或者带用户维度的组合键更稳妥。另外Checkpointer 存的是完整状态快照如果你的 State 里塞了大对象比如整个文档内容存储会迅速膨胀。我的做法是 State 里只放必要的对话消息和轻量标识大内容通过外部存储对象存储、数据库引用State 里存引用 ID。这个习惯能帮你省下大量存储成本。5. LangSmith 接入让每一步执行都看得见5.1 为什么生产环境离不开可观测图跑起来之后最怕的不是报错而是**“结果不对但不知道哪一步不对”**。LangGraph 的图可能有十几个节点、多个条件分支一次调用经过哪些节点、每个节点的输入输出是什么、耗时多少光靠打日志很难看清全貌。LangSmith 就是解决这个问题的——它把每次调用的执行链路以树状结构展示出来每个节点的输入、输出、耗时、token 消耗一目了然。接入方式简单到出乎意料设置两个环境变量即可export LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEYyour_api_key export LANGCHAIN_PROJECTlanggraph-deploy设置完之后所有 LangGraph 的调用会自动上报 trace不需要改任何业务代码。这个“零侵入”的特性是我最欣赏的地方它不会污染你的图逻辑。5.2 用 trace 定位典型问题接入 LangSmith 之后我定位过几类典型问题分享出来供你参考。第一类是条件边走错分支。图里明明写了判断逻辑但实际走了另一条路。打开 trace 一看是判断节点的返回值类型和条件函数期望的类型不一致比如返回了字符串但条件函数在比较布尔值。这种问题看 trace 一眼就能定位纯靠日志得排查很久。第二类是状态字段被意外覆盖。某个节点返回的字典里包含了不该返回的字段导致 reducer 把它覆盖了。trace 里能看到每个节点返回的增量状态对比一下就知道是哪个节点“多嘴”了。第三类是性能瓶颈。trace 会显示每个节点的耗时如果某个节点占了 80% 的时间优化目标就很明确。我遇到过一次是工具调用节点在等外部 API加了超时和缓存之后整体响应时间降了一半。5.3 流式输出与 LangSmith 的配合如果你的服务需要流式返回比如打字机效果用astream或者astream_events。这里要注意流式场景下 LangSmith 的 trace 依然是完整的因为 trace 记录的是图的执行过程和你怎么消费输出无关。但流式接口的错误处理要格外小心一旦流中途出错客户端可能已经收到部分内容了这时候要在响应里明确标识错误而不是静默断开。from fastapi.responses import StreamingResponse router.post(/chat/stream) async def chat_stream(req: ChatRequest): async def event_generator(): config {configurable: {thread_id: req.thread_id}} async for event in graph.astream_events( {messages: [(user, req.message)], user_id: req.user_id}, configconfig, versionv2, ): if event[event] on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield fdata: {chunk.content}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这段代码里versionv2是必须的否则事件结构不一样。流式接口的调试建议先在本地用 curl 测确认 SSE 格式正确再接前端。6. 常见问题与排查技巧实录6.1 部署阶段高频问题速查表问题现象可能原因排查方向对话历史丢失未接 Checkpointer 或 thread_id 不一致检查 compile 是否传 checkpointer调用是否传同一 thread_id多 worker 下状态错乱用了内存态 Checkpointer换持久化 Checkpointer 或改单 worker日志不输出Uvicorn 日志配置覆盖显式配置 logging.basicConfig消息重复累积接入 Checkpointer 后仍传完整历史初始状态只传本轮新消息流式输出中断异常未捕获在生成器内 try/except 并发送错误事件响应超时节点内同步阻塞调用改异步或加超时控制6.2 我踩过的几个坑第一个坑是在节点里做同步的耗时操作。比如在节点里直接调一个同步的 HTTP 请求整个事件循环被阻塞并发能力直接归零。解决办法是用httpx.AsyncClient或者把同步调用丢到线程池里跑。这个坑在低并发测试时发现不了一上压力就暴露。第二个坑是State 定义过于宽泛。早期我图省事State 里放了个dict类型的context字段什么杂七杂八的数据都往里塞。结果 Checkpointer 存储膨胀得飞快而且状态恢复时经常出现意料之外的字段。后来改成显式的、类型化的字段问题就消失了。State 是契约不是垃圾桶这句话我后来一直记着。第三个坑是忽略图的递归限制。LangGraph 默认有递归步数上限如果图里存在循环边某些输入可能触发无限循环。虽然框架会抛异常终止但用户体验很差。我的做法是在设计循环边时一定要有明确的退出条件并且在 State 里记录循环次数超过阈值主动跳出。6.3 性能优化的几个实操方向服务化之后如果发现响应慢按这个顺序排查先看 LangSmith trace 找耗时最长的节点再看是不是模型调用本身慢换更快的模型或加缓存然后看是不是 Checkpointer 的读写慢数据库连接池、索引最后看是不是并发配置不合理worker 数量、连接数。我实测下来Checkpointer 的读写往往是容易被忽略的瓶颈。尤其是用数据库做持久化时如果没建好索引每次状态读写都全表扫描延迟会很高。给thread_id建索引是基本操作别省这一步。7. 三条路径的选型建议与我的实际体会回到标题三条路径不是让你三选一而是让你按需递进。如果你只是本地验证一个想法路径一足够如果你要给团队提供一个可调用的接口路径二起步如果你要做真正的多轮对话产品路径三是必须的。我个人在实际项目中的做法是永远从路径一开始但写脚本时就按服务的标准组织代码——State 定义清晰、节点函数无副作用、配置外置。这样从路径一升级到路径二基本就是加一个api/目录的事图逻辑一行不用改。从路径二到路径三也就是加一个 Checkpointer 和几个环境变量。最后分享一个小技巧在开发阶段我会同时保留一个 CLI 入口和一个 FastAPI 入口两者共用同一个graph对象。CLI 用来快速验证图逻辑FastAPI 用来联调接口。这样调试时不用每次都起服务、发请求效率高很多。这个习惯坚持下来帮我省下的时间相当可观。 
长河君 · 2026/10/27 12:04:52 · 阅读 1.2k LangGraph 部署三条路径：从脚本到 FastAPI 服务化与 LangSmith 可观测
1. 从脚本到服务LangGraph 部署到底在解决什么问题很多人第一次接触 LangGraph都是在 Jupyter Notebook 或者一个单文件 Python 脚本里跑通一个带状态的多轮对话 Agent。本地跑得挺爽graph.invoke()一调状态机流转、条件边跳转、工具调用全都正常。但一旦要把这个东西交给前端调用、交给同事测试、或者挂到线上让真实用户用问题就全冒出来了状态怎么持久化多个用户同时访问会不会串状态服务重启之后对话历史还在不在怎么观测每一步的执行链路这一篇就是冲着这些问题来的。标题里的“三条部署路径”说的不是三种花哨的架构而是从最简单的脚本封装到 FastAPI 服务化再到接入 LangSmith 做可观测与持久化的渐进式落地方式。核心关键词LangGraph、部署、FastAPI、LangSmith、Checkpointer会贯穿全文。如果你已经能用 LangGraph 写出一个能跑的图但卡在“怎么让它变成一个真正的服务”这一步那这篇内容就是给你准备的。哪怕你只是刚看完 LangGraph 官方教程的新手我也会把每一步的来龙去脉讲清楚让你能直接抄作业。先说清楚一个认知LangGraph 本身是一个编排框架它负责定义节点、边、状态和流转逻辑但它不是一个 Web 服务框架。它不会帮你处理 HTTP 请求、不会帮你做并发、不会帮你管理进程。所以“部署 LangGraph”这件事本质上是给 LangGraph 的图套一层服务外壳再解决状态存储和可观测性。理解了这一点三条路径的划分就顺理成章了。我先把三条路径的定位摆出来方便你对照自己的场景选路径形态适用场景核心依赖路径一脚本 CLI 调用本地验证、批处理、一次性任务LangGraph 本体路径二FastAPI 服务化前后端分离、多用户、API 调用FastAPI Uvicorn路径三服务化 持久化 可观测生产环境、多轮对话、链路追踪FastAPI Checkpointer LangSmith这三条路径不是互斥的而是递进的。你可以先跑通路径一再平滑升级到路径二最后补上路径三的持久化和观测能力。下面我逐条拆开讲重点放在路径二和路径三因为那才是真正“从脚本到服务”的跨越。2. 路径一脚本直跑先把图跑通再谈部署2.1 为什么部署之前必须先有一个干净的脚本我见过太多人一上来就搭 FastAPI 目录、配 Uvicorn、写路由结果图本身的状态定义就有问题调半天以为是服务的问题其实是StateGraph的 reducer 写错了。所以我的建议永远是先用一个最朴素的脚本把图跑通确认输入输出符合预期再考虑套服务。这个脚本的形态很简单一个graph.py定义 State、节点、边最后compile()出一个可执行对象。关键是要把图的输入契约和输出契约固定下来。所谓输入契约就是这个图接收什么样的初始状态输出契约就是invoke之后返回什么结构。这两个契约一旦稳定后面套 FastAPI 就是纯粹的搬运工作。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str def chatbot(state: AgentState): # 这里放你的模型调用逻辑 return {messages: [(assistant, 收到你的消息了)]} builder StateGraph(AgentState) builder.add_node(chatbot, chatbot) builder.add_edge(START, chatbot) builder.add_edge(chatbot, END) graph builder.compile()这段代码里有个细节值得说add_messages这个 reducer 决定了messages字段是追加而不是覆盖。如果你不加Annotated每次节点返回都会把整个 messages 列表替换掉多轮对话直接丢失历史。这是新手最常踩的坑之一我在早期调试时也栽过明明图跑通了但对话永远只有最后一轮。2.2 脚本阶段的验证清单在进入服务化之前我通常会跑一个验证清单确认图的行为是确定的单轮输入能否得到预期输出多轮输入时状态是否正确累积条件边在边界情况下是否走对分支异常输入空消息、超长文本是否会崩溃这个阶段不需要任何 Web 框架一个if __name__ __main__:加上几个测试用例就够了。脚本阶段的目标是消除图本身的 bug而不是解决服务的问题。把这两件事混在一起排查效率会低得离谱。提示脚本阶段就把user_id这类会话标识放进 State后面做多用户隔离时会省很多事。等到服务化再回头改 State 结构所有节点都要跟着动。3. 路径二FastAPI 服务化把图变成可调用的 API3.1 为什么选 FastAPI 而不是 Flask热词里同时出现了fastapi和flask 与 fastapi 比较说明这是很多人的纠结点。我的结论很直接LangGraph 服务化优先选 FastAPI。原因有三条都是实打实的工程考量。第一LangGraph 的很多能力是异步的ainvoke、astream这些异步接口在流式输出场景下几乎是刚需。FastAPI 原生支持async def而 Flask 是同步 WSGI 框架要跑异步得额外折腾。第二FastAPI 基于 Pydantic 做请求体校验而 LangGraph 的 State 本身就是类型化的两者天然契合请求模型可以直接映射到状态字段。第三FastAPI 自带 OpenAPI 文档前端联调时直接看/docs就行省掉大量沟通成本。Flask 不是不能用如果你的团队已经有一套成熟的 Flask 体系硬切 FastAPI 也有迁移成本。但如果是新项目没有理由不选 FastAPI。3.2 FastAPI 项目目录结构怎么组织热词里fastapi项目目录结构是个高频问题。LangGraph 服务化的目录结构我推荐按“图逻辑”和“服务逻辑”分离的原则来组织project/ ├── app/ │ ├── main.py # FastAPI 入口注册路由 │ ├── api/ │ │ └── routes.py # 路由定义 │ ├── graph/ │ │ ├── state.py # State 定义 │ │ ├── nodes.py # 节点函数 │ │ └── builder.py # 图构建与 compile │ ├── schemas/ │ │ └── chat.py # Pydantic 请求/响应模型 │ └── core/ │ └── config.py # 配置管理 ├── requirements.txt └── run.sh这个结构的核心思想是graph/目录里的东西不依赖 FastAPI它就是一个纯粹的 LangGraph 模块可以单独测试、单独复用。api/目录只负责把 HTTP 请求翻译成图调用把图的结果翻译成 HTTP 响应。这样分层之后你想把图迁移到别的框架比如换成 gRPC只需要重写api/层graph/层一行不用动。3.3 核心路由实现与请求模型设计请求模型用 Pydantic 定义字段和 State 对齐from pydantic import BaseModel, Field class ChatRequest(BaseModel): message: str Field(..., min_length1, max_length4000) user_id: str Field(..., min_length1) thread_id: str Field(defaultdefault)路由层拿到请求后构造初始状态调用图from fastapi import APIRouter from app.graph.builder import graph from app.schemas.chat import ChatRequest router APIRouter() router.post(/chat) async def chat(req: ChatRequest): initial_state { messages: [(user, req.message)], user_id: req.user_id, } config {configurable: {thread_id: req.thread_id}} result await graph.ainvoke(initial_state, configconfig) return {reply: result[messages][-1].content}这里有个关键点config里的thread_id。在路径二阶段如果你还没接 Checkpointer这个thread_id其实不起作用每次调用都是全新的状态。但我强烈建议你从第一天就把thread_id传进去因为路径三接上 Checkpointer 之后这个字段就是会话隔离的核心提前埋好能省掉一次重构。3.4 Uvicorn 启动与日志丢失问题热词里uvicorn fastapi 日志丢失问题是个很典型的坑。默认情况下Uvicorn 有自己的日志配置如果你在代码里用logging.getLogger(__name__)打日志可能会发现日志没输出或者格式和 Uvicorn 的不一致。解决办法是显式配置日志让应用日志和 Uvicorn 日志走同一套 handlerimport logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, )启动命令用uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 2关于--workers这里要提醒一句如果你的图里用了内存态的 Checkpointer比如MemorySaver多 worker 会导致状态不共享。因为每个 worker 是独立进程各自持有自己的内存。这种情况下要么用单 worker要么直接上持久化的 Checkpointer。这个坑我在压测时遇到过两个 worker 轮流处理同一个用户的请求对话历史时有时无排查了半天才定位到是进程隔离的问题。4. 路径三Checkpointer 持久化让对话真正“记得住”4.1 Checkpointer 到底解决了什么前面路径二的服务每次请求都是“无状态”的——图跑完状态就没了。用户第二轮提问时你得自己把历史消息拼回去而且一旦服务重启所有上下文清零。Checkpointer 就是 LangGraph 提供的状态持久化机制它在图的每个 super-step 之后把状态快照存下来下次用同一个thread_id调用时自动把历史状态恢复回来。这个机制的价值在于你不需要自己维护对话历史表不需要手动拼接 messagesLangGraph 帮你把“状态”这件事管起来了。对于多轮对话 Agent 来说这是从 demo 到可用的分水岭。4.2 内存态与持久化 Checkpointer 的选型LangGraph 提供了多种 Checkpointer选型逻辑很清晰Checkpointer存储位置适用场景注意事项MemorySaver进程内存本地开发、单进程测试重启即丢多 worker 不共享SqliteSaver本地文件单机小规模、原型验证并发写入需注意锁PostgresSaver数据库生产环境、多实例需初始化表结构开发阶段用MemorySaver完全够用但上线前一定要换成持久化方案。我个人的习惯是本地用 SqliteSaver 验证持久化逻辑生产用 PostgresSaver。切换 Checkpointer 的代码改动极小因为接口是统一的from langgraph.checkpoint.memory import MemorySaver # from langgraph.checkpoint.postgres import PostgresSaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer)编译时传入checkpointer调用时传入thread_id持久化就生效了。这里有个容易忽略的点接入 Checkpointer 后ainvoke的初始状态只需要传本轮的新消息历史消息会自动从 checkpoint 恢复。如果你还是把完整历史拼进去会导致消息重复。这个行为差异一定要在切换时验证清楚。4.3 多用户会话隔离的实操要点thread_id是会话隔离的唯一标识。实践中我建议用user_id 会话标识的组合来构造thread_id比如f{user_id}:{conversation_id}。这样同一个用户可以有多个独立会话不同用户之间天然隔离。注意thread_id不要用自增数字或者可预测的短字符串否则容易被遍历。用 UUID 或者带用户维度的组合键更稳妥。另外Checkpointer 存的是完整状态快照如果你的 State 里塞了大对象比如整个文档内容存储会迅速膨胀。我的做法是 State 里只放必要的对话消息和轻量标识大内容通过外部存储对象存储、数据库引用State 里存引用 ID。这个习惯能帮你省下大量存储成本。5. LangSmith 接入让每一步执行都看得见5.1 为什么生产环境离不开可观测图跑起来之后最怕的不是报错而是**“结果不对但不知道哪一步不对”**。LangGraph 的图可能有十几个节点、多个条件分支一次调用经过哪些节点、每个节点的输入输出是什么、耗时多少光靠打日志很难看清全貌。LangSmith 就是解决这个问题的——它把每次调用的执行链路以树状结构展示出来每个节点的输入、输出、耗时、token 消耗一目了然。接入方式简单到出乎意料设置两个环境变量即可export LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEYyour_api_key export LANGCHAIN_PROJECTlanggraph-deploy设置完之后所有 LangGraph 的调用会自动上报 trace不需要改任何业务代码。这个“零侵入”的特性是我最欣赏的地方它不会污染你的图逻辑。5.2 用 trace 定位典型问题接入 LangSmith 之后我定位过几类典型问题分享出来供你参考。第一类是条件边走错分支。图里明明写了判断逻辑但实际走了另一条路。打开 trace 一看是判断节点的返回值类型和条件函数期望的类型不一致比如返回了字符串但条件函数在比较布尔值。这种问题看 trace 一眼就能定位纯靠日志得排查很久。第二类是状态字段被意外覆盖。某个节点返回的字典里包含了不该返回的字段导致 reducer 把它覆盖了。trace 里能看到每个节点返回的增量状态对比一下就知道是哪个节点“多嘴”了。第三类是性能瓶颈。trace 会显示每个节点的耗时如果某个节点占了 80% 的时间优化目标就很明确。我遇到过一次是工具调用节点在等外部 API加了超时和缓存之后整体响应时间降了一半。5.3 流式输出与 LangSmith 的配合如果你的服务需要流式返回比如打字机效果用astream或者astream_events。这里要注意流式场景下 LangSmith 的 trace 依然是完整的因为 trace 记录的是图的执行过程和你怎么消费输出无关。但流式接口的错误处理要格外小心一旦流中途出错客户端可能已经收到部分内容了这时候要在响应里明确标识错误而不是静默断开。from fastapi.responses import StreamingResponse router.post(/chat/stream) async def chat_stream(req: ChatRequest): async def event_generator(): config {configurable: {thread_id: req.thread_id}} async for event in graph.astream_events( {messages: [(user, req.message)], user_id: req.user_id}, configconfig, versionv2, ): if event[event] on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield fdata: {chunk.content}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这段代码里versionv2是必须的否则事件结构不一样。流式接口的调试建议先在本地用 curl 测确认 SSE 格式正确再接前端。6. 常见问题与排查技巧实录6.1 部署阶段高频问题速查表问题现象可能原因排查方向对话历史丢失未接 Checkpointer 或 thread_id 不一致检查 compile 是否传 checkpointer调用是否传同一 thread_id多 worker 下状态错乱用了内存态 Checkpointer换持久化 Checkpointer 或改单 worker日志不输出Uvicorn 日志配置覆盖显式配置 logging.basicConfig消息重复累积接入 Checkpointer 后仍传完整历史初始状态只传本轮新消息流式输出中断异常未捕获在生成器内 try/except 并发送错误事件响应超时节点内同步阻塞调用改异步或加超时控制6.2 我踩过的几个坑第一个坑是在节点里做同步的耗时操作。比如在节点里直接调一个同步的 HTTP 请求整个事件循环被阻塞并发能力直接归零。解决办法是用httpx.AsyncClient或者把同步调用丢到线程池里跑。这个坑在低并发测试时发现不了一上压力就暴露。第二个坑是State 定义过于宽泛。早期我图省事State 里放了个dict类型的context字段什么杂七杂八的数据都往里塞。结果 Checkpointer 存储膨胀得飞快而且状态恢复时经常出现意料之外的字段。后来改成显式的、类型化的字段问题就消失了。State 是契约不是垃圾桶这句话我后来一直记着。第三个坑是忽略图的递归限制。LangGraph 默认有递归步数上限如果图里存在循环边某些输入可能触发无限循环。虽然框架会抛异常终止但用户体验很差。我的做法是在设计循环边时一定要有明确的退出条件并且在 State 里记录循环次数超过阈值主动跳出。6.3 性能优化的几个实操方向服务化之后如果发现响应慢按这个顺序排查先看 LangSmith trace 找耗时最长的节点再看是不是模型调用本身慢换更快的模型或加缓存然后看是不是 Checkpointer 的读写慢数据库连接池、索引最后看是不是并发配置不合理worker 数量、连接数。我实测下来Checkpointer 的读写往往是容易被忽略的瓶颈。尤其是用数据库做持久化时如果没建好索引每次状态读写都全表扫描延迟会很高。给thread_id建索引是基本操作别省这一步。7. 三条路径的选型建议与我的实际体会回到标题三条路径不是让你三选一而是让你按需递进。如果你只是本地验证一个想法路径一足够如果你要给团队提供一个可调用的接口路径二起步如果你要做真正的多轮对话产品路径三是必须的。我个人在实际项目中的做法是永远从路径一开始但写脚本时就按服务的标准组织代码——State 定义清晰、节点函数无副作用、配置外置。这样从路径一升级到路径二基本就是加一个api/目录的事图逻辑一行不用改。从路径二到路径三也就是加一个 Checkpointer 和几个环境变量。最后分享一个小技巧在开发阶段我会同时保留一个 CLI 入口和一个 FastAPI 入口两者共用同一个graph对象。CLI 用来快速验证图逻辑FastAPI 用来联调接口。这样调试时不用每次都起服务、发请求效率高很多。这个习惯坚持下来帮我省下的时间相当可观。