1. 从脚本到服务LangGraph 部署到底在解决什么问题很多人第一次接触 LangGraph都是在 Jupyter Notebook 或者一个单文件 Python 脚本里跑通一个带工具调用的 Agent感觉挺爽。但一旦要把这个东西交给别人用问题就来了脚本只能你自己在终端里跑别人没法访问每次重启进程对话历史全丢多个用户同时用状态互相串想加个鉴权、限流、日志发现无从下手。这些问题的本质是你手里的是一个脚本不是一个服务。LangGraph 本身是一个编排框架它负责定义状态图、节点、边、条件跳转和工具调用逻辑。但它不负责网络通信、并发处理、进程管理、持久化存储这些“服务化”的事情。所以从脚本到服务中间需要补上几块拼图一个 Web 框架来暴露接口一个持久化层来保存状态一个部署方式来让服务稳定运行。这三块拼图的不同组合就形成了三条典型的部署路径。这篇文章面向的是已经用 LangGraph 写过 Demo、但还没把它变成正式服务的开发者。我会把三条路径逐一拆开讲清楚每条路径适合什么场景、需要哪些组件、具体怎么操作、容易踩什么坑。核心关键词包括LangGraph、部署、FastAPI、LangSmith、Checkpointer这些词会贯穿全文。读完之后你应该能根据自己的需求选出一条合适的路径并且知道每一步该怎么做。先给一个全局视角。三条路径分别是本地脚本加 Checkpointer 做轻量持久化、FastAPI 包装成 HTTP 服务、接入 LangSmith 做可观测性与云端协同。这三条路径不是互斥的而是递进的。你可以先从第一条开始跑通了再往第二条迁移最后按需接入第三条。下面逐条展开。2. 第一条路径脚本加 Checkpointer最小可用的持久化方案2.1 Checkpointer 到底是什么为什么它是第一步LangGraph 的 Checkpointer 机制说白了就是给状态图加了一个“存档点”。每次图执行到一个节点Checkpointer 会把当前的状态快照存下来。下次同一个会话继续执行时可以从最近的存档点恢复而不是从头开始。这个机制解决的是对话记忆和中断恢复两个核心问题。没有 Checkpointer 的时候你的 Agent 是无状态的。用户第一句话说“帮我查一下北京天气”Agent 查完返回结果。用户第二句话说“那上海呢”Agent 完全不知道上一轮发生了什么“那”指代的是什么它根本不知道。有了 Checkpointer每一轮的状态都被保存第二轮执行时会把第一轮的状态加载进来Agent 就能理解上下文。LangGraph 提供了几种 Checkpointer 实现。最常用的是MemorySaver存在内存里适合开发和测试。生产环境一般用SqliteSaver或者PostgresSaver把状态存到数据库里进程重启也不丢。选哪个取决于你的部署环境单机小规模用 SQLite 就够了多实例部署就得上 PostgreSQL。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph # 初始化 checkpointer指定数据库文件路径 checkpointer SqliteSaver.from_conn_string(checkpoints.db) # 编译图的时候传入 checkpointer graph builder.compile(checkpointercheckpointer) # 调用时通过 config 指定 thread_id区分不同会话 config {configurable: {thread_id: user-001}} result graph.invoke({messages: [(user, 你好)]}, config)这段代码的关键在于thread_id。它是会话的唯一标识不同用户、不同对话要用不同的thread_id否则状态会串。我见过有人所有请求都用同一个thread_id结果两个用户的对话历史混在一起排查了半天才发现是这个问题。注意SqliteSaver在多线程环境下需要额外处理连接问题。如果你的服务是多线程的建议用PostgresSaver或者给每个线程单独创建连接。2.2 从脚本到可复用模块的改造要点光加 Checkpointer 还不够脚本本身的结构也需要调整。一个典型的 LangGraph 脚本往往把所有东西写在一起定义状态、定义节点函数、构建图、编译、调用。这种结构自己跑没问题但要做成服务得把“图的定义”和“图的调用”分开。我的做法是拆成三个文件state.py定义状态结构graph.py定义节点和构建图run.py负责调用。这样拆的好处是后面用 FastAPI 包装的时候直接 importgraph.py里编译好的图就行不用改任何逻辑。状态定义这块有个容易忽略的点状态字段要尽量精简。有些人把整个消息历史、工具调用记录、中间变量全塞进状态里导致每次 Checkpointer 存档的数据量很大。如果用的是 SQLite写入频率高了之后性能会明显下降。我的经验是只把真正需要跨轮次保留的字段放进状态临时变量在节点函数内部处理就行。还有一个细节是图的编译时机。builder.compile()这个操作是有开销的不要每次请求都编译一遍。正确的做法是在模块加载时编译一次全局复用一个编译好的图实例。Checkpointer 也是同理全局初始化一次。2.3 这条路径的适用边界与局限第一条路径适合什么场景我总结下来是三类个人本地使用、内部小工具、原型验证。比如你自己写一个帮自己整理资料的 Agent或者团队内部几个人用的小助手用这条路径完全够了。部署成本几乎为零一个 Python 脚本加一个 SQLite 文件就能跑。但它有明显的局限。第一没有网络接口别人没法通过 HTTP 调用。第二没有并发处理能力多个请求同时来会排队或者出错。第三没有鉴权和限流谁拿到你的脚本都能跑。第四日志和监控基本靠 print出了问题不好排查。所以这条路径的定位是起点不是终点。它的价值在于让你用最低的成本验证核心逻辑确认 Agent 的行为符合预期。逻辑跑通了再往第二条路径迁移把服务化的部分补上。不要一上来就搞复杂的部署架构那样调试成本太高容易在细枝末节上浪费时间。3. 第二条路径FastAPI 包装把图变成 HTTP 服务3.1 为什么选 FastAPI 而不是 Flask把 LangGraph 包装成 HTTP 服务Web 框架的选择有好几个。Flask 是最老牌的上手简单生态成熟。FastAPI 是后起之秀主打异步和类型提示。我选 FastAPI 的理由有三个。第一原生异步支持。LangGraph 的很多操作是 IO 密集型的比如调用大模型 API、查询数据库、执行工具。用异步框架可以在等待 IO 的时候处理其他请求吞吐量比同步框架高不少。Flask 虽然也能配合异步库用但整体架构还是同步的不如 FastAPI 自然。第二自动生成接口文档。FastAPI 基于 Pydantic 模型写完接口自动生成 Swagger 文档前端对接、测试调试都方便。Flask 要额外装 flasgger 之类的插件才能做到。第三类型提示和校验。FastAPI 用 Pydantic 做请求体和响应体的校验类型不对直接返回 422不用自己写一堆 if 判断。这在处理 LangGraph 的输入输出时特别有用因为状态结构本身就是有类型的。当然 Flask 也有它的优势比如生态更成熟、资料更多、部署方案更丰富。如果你团队里都是 Flask 老手用 Flask 也没问题。但从长期维护的角度我倾向于 FastAPI。3.2 FastAPI 项目目录结构怎么设计一个能上生产的 FastAPI 项目目录结构不能太随意。我常用的结构是这样的project/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口创建 FastAPI 实例 │ ├── config.py # 配置管理读环境变量 │ ├── api/ │ │ ├── __init__.py │ │ ├── routes.py # 路由定义 │ │ └── schemas.py # Pydantic 请求/响应模型 │ ├── core/ │ │ ├── __init__.py │ │ ├── graph.py # LangGraph 图定义与编译 │ │ ├── state.py # 状态结构定义 │ │ └── checkpointer.py # Checkpointer 初始化 │ └── utils/ │ ├── __init__.py │ └── logger.py # 日志配置 ├── tests/ ├── requirements.txt └── .env这个结构的关键是把 LangGraph 相关的代码放在core/目录下和 API 层解耦。api/routes.py只负责接收请求、调用图、返回响应不包含任何图逻辑。这样后面换 Web 框架或者换部署方式core/里的代码不用动。config.py用 Pydantic 的BaseSettings来管理配置从环境变量读取 API Key、数据库连接串、日志级别这些。不要把敏感信息硬编码在代码里也不要提交到版本控制。3.3 核心接口设计与流式输出实现LangGraph 服务最核心的接口是对话接口。设计上有两个选择同步返回和流式返回。同步返回就是等图执行完一次性把结果返回。流式返回是图每执行一步就把中间结果推给客户端。对于对话类应用流式返回体验好得多用户不用干等。FastAPI 实现流式返回用StreamingResponse配合生成器。LangGraph 提供了astream方法可以异步流式获取图执行的每一步结果。from fastapi import FastAPI from fastapi.responses import StreamingResponse from app.core.graph import graph app FastAPI() async def event_generator(user_input: str, thread_id: str): config {configurable: {thread_id: thread_id}} async for event in graph.astream( {messages: [(user, user_input)]}, config, stream_modemessages ): # event 是 (message_chunk, metadata) 元组 if event[0].content: yield fdata: {event[0].content}\n\n app.post(/chat/stream) async def chat_stream(request: ChatRequest): return StreamingResponse( event_generator(request.message, request.thread_id), media_typetext/event-stream )这里用的是 SSEServer-Sent Events协议media_type设为text/event-stream。SSE 比 WebSocket 简单单向推送够用浏览器原生支持。注意每个数据块后面要跟两个换行符\n\n这是 SSE 的格式要求少一个客户端就收不到。stream_mode参数有好几个选项。messages模式会流式输出大模型生成的 token适合对话场景。updates模式会输出每个节点的状态更新适合调试。values模式输出完整状态数据量大。根据你的需求选。3.4 并发、超时与错误处理的实际处理方式服务化之后并发是绕不开的问题。FastAPI 用 uvicorn 跑的时候默认是单进程。要利用多核 CPU得用--workers参数起多个 worker 进程。但这里有个坑如果 Checkpointer 用的是MemorySaver多个 worker 之间内存不共享同一个thread_id的请求被分到不同 worker 就会状态不一致。所以多 worker 部署必须用数据库版的 Checkpointer。超时处理也很重要。大模型调用有时候会卡很久如果不设超时请求会一直挂着占用连接资源。我的做法是在调用图的时候加asyncio.wait_for设一个合理的超时时间比如 60 秒。超时了就返回一个友好的错误提示而不是让客户端一直等。import asyncio try: result await asyncio.wait_for( graph.ainvoke(input_data, config), timeout60.0 ) except asyncio.TimeoutError: return {error: 处理超时请稍后重试}错误处理要分层。图执行内部的错误比如工具调用失败应该在节点函数里捕获并返回错误信息给大模型让大模型决定怎么处理。图执行外部的错误比如数据库连接失败应该在 API 层捕获返回 500 并记录日志。不要把内部错误堆栈直接暴露给客户端既不安全也不友好。提示uvicorn 的日志默认输出到 stderr如果部署在容器里记得配置日志收集。我遇到过 uvicorn 日志丢失的问题后来发现是容器日志驱动配置不对改成 json-file 驱动就好了。4. 第三条路径接入 LangSmith让服务可观测、可调试4.1 LangSmith 解决的是什么问题服务跑起来之后新的问题出现了你怎么知道 Agent 每一步在干什么用户反馈“回答不对”你怎么复现和定位某个工具调用特别慢你怎么发现这些问题靠 print 日志很难解决因为 LangGraph 的执行链路是多节点、多分支的日志打出来是一团乱麻。LangSmith 就是解决这个问题的。它是 LangChain 官方出的可观测性平台能自动追踪 LangGraph 的每一次执行把每个节点的输入、输出、耗时、token 消耗都记录下来用可视化的方式展示。你可以在界面上看到完整的执行链路点开每个节点看详细数据对比不同请求的差异。接入 LangSmith 的成本很低基本上就是设几个环境变量的事。但它的价值很高尤其是在调试复杂 Agent 的时候。我自己的经验是没有 LangSmith 的时候排查一个多轮对话的问题可能要半小时有了 LangSmith五分钟就能定位到是哪个节点出了问题。4.2 环境变量配置与追踪开启步骤接入 LangSmith 需要四个环境变量LANGCHAIN_TRACING_V2true LANGCHAIN_ENDPOINThttps://api.smith.langchain.com LANGCHAIN_API_KEYyour_api_key_here LANGCHAIN_PROJECTyour_project_nameLANGCHAIN_TRACING_V2设为true开启追踪。LANGCHAIN_API_KEY在 LangSmith 平台上申请。LANGCHAIN_PROJECT是项目名不同项目的追踪数据分开管理建议按环境区分比如myagent-dev、myagent-prod。这些变量通过.env文件管理用python-dotenv加载。注意.env文件要加到.gitignore里不要提交到仓库。生产环境的 API Key 通过部署平台的密钥管理功能注入不要写在配置文件里。配置好之后不需要改任何代码LangGraph 的执行会自动被追踪。你可以在 LangSmith 界面上看到每次调用的完整链路。如果只想追踪部分请求可以用tracing_context做细粒度控制。4.3 用追踪数据反哺 Agent 优化的具体做法LangSmith 的价值不只是“看”更在于“用”。追踪数据积累起来之后可以做很多优化。第一找出慢节点。在 LangSmith 的界面上按耗时排序看看哪个节点最慢。如果是大模型调用慢考虑换更快的模型或者加缓存。如果是工具调用慢看看是不是可以并行执行。第二分析 token 消耗。LangSmith 会统计每次调用的 token 数你可以看到哪些请求消耗特别大。如果发现某个节点的 prompt 太长可以精简一下。长期来看token 优化能省不少成本。第三构建测试数据集。LangSmith 支持把线上的追踪数据直接加到数据集里用来做回归测试。每次改完 prompt 或者图结构跑一遍数据集看看效果有没有退化。这个功能在迭代 Agent 的时候特别有用。第四对比不同版本。LangSmith 支持给追踪打标签你可以给不同版本的 Agent 打不同标签然后对比它们的表现。比如 v1 版本和 v2 版本在同一个问题上的回答差异一目了然。注意LangSmith 默认会上传追踪数据到云端。如果对数据隐私有要求可以考虑自托管方案或者只上传脱敏后的数据。具体怎么取舍看你的合规要求。5. 三条路径怎么选以及迁移时的注意事项5.1 按场景选路径的决策表三条路径不是让你三选一而是根据阶段和需求递进。我整理了一个决策表帮你快速判断当前该用哪条。场景推荐路径核心组件部署成本个人本地使用路径一脚本 SqliteSaver极低团队内部小工具路径一或二脚本/FastAPI SqliteSaver低对外提供 API 服务路径二FastAPI PostgresSaver中需要调试和监控路径二 三FastAPI LangSmith中多实例高可用路径二 三FastAPI PostgresSaver LangSmith高判断的关键就两个问题有没有别人要用要不要长期稳定运行如果都是“否”路径一够了。如果有一个“是”上路径二。如果对稳定性和可观测性有要求加路径三。5.2 从路径一迁移到路径二的改造清单迁移的时候核心逻辑不用动改的是外围。我列一个改造清单照着做就行。拆分代码结构把图定义、状态定义、Checkpointer 初始化拆到独立模块。换 Checkpointer从MemorySaver或SqliteSaver换成PostgresSaver如果要多实例部署的话。加 API 层用 FastAPI 定义路由包装图的调用。加配置管理把硬编码的配置抽到环境变量。加日志配置结构化日志方便排查。加错误处理API 层捕获异常返回友好错误。加鉴权至少加个 API Key 校验别裸奔。迁移过程中最容易出问题的是 Checkpointer 的切换。不同 Checkpointer 的 API 略有差异PostgresSaver需要先建表。LangGraph 提供了建表脚本跑一下就行。另外从 SQLite 迁移到 PostgreSQL 的时候已有的状态数据不会自动迁移需要自己写脚本导。如果状态数据不重要直接清空重来也行。5.3 部署上线前必须检查的几件事上线之前有几件事必须确认不然出了问题很被动。第一Checkpointer 的连接池配置。PostgresSaver默认的连接池大小可能不够高并发下会报连接超时。根据你的并发量调整pool_size和max_overflow。我的经验是每个 worker 配 5 到 10 个连接比较稳妥。第二大模型 API 的限流和重试。大模型 API 通常有 QPS 限制超了会返回 429。要在代码里加重试逻辑用指数退避。LangChain 的一些组件内置了重试但最好自己再包一层确保可控。第三健康检查接口。加一个/health接口返回服务状态。部署平台用它做存活探针服务挂了能自动重启。第四日志级别和输出。生产环境日志级别设INFO不要用DEBUG不然日志量太大。日志输出到 stdout由部署平台收集。第五敏感信息检查。确认代码里没有硬编码的 API Key、数据库密码。用grep搜一下key、password、secret这些关键词确保都走了环境变量。6. 实操中踩过的坑与排查技巧6.1 Checkpointer 相关的典型问题问题一状态不更新。表现是第二轮对话时Agent 好像不记得第一轮的内容。排查思路先确认thread_id是否一致再看 Checkpointer 是否真的写入了数据。可以直接查数据库表看看有没有对应的记录。如果表是空的说明 Checkpointer 没生效检查编译图的时候有没有传checkpointer参数。问题二状态串了。表现是两个不相关的对话互相影响。九成是thread_id重复了。检查生成thread_id的逻辑确保每个会话唯一。如果是多用户场景thread_id里最好带上用户 ID。问题三数据库锁。SQLite 在高并发写入时会锁库报database is locked。这是 SQLite 的固有限制解决办法就是换 PostgreSQL。如果暂时不想换可以加写入重试但治标不治本。6.2 FastAPI 流式输出的常见故障故障一客户端收不到流式数据。最常见的原因是响应头不对。SSE 需要Content-Type: text/event-stream还需要Cache-Control: no-cache和Connection: keep-alive。有些反向代理会缓冲响应导致流式变成一次性返回需要配置代理关闭缓冲。故障二流到一半断了。可能是超时设置太短或者生成器抛异常了。在生成器里加 try-except捕获异常后 yield 一个错误事件而不是直接让连接断掉。故障三中文乱码。SSE 默认用 UTF-8一般不会乱码。如果乱码了检查一下是不是中间有代理做了编码转换。可以在响应头里显式指定charsetutf-8。6.3 上线后性能问题的排查顺序服务上线后如果变慢按这个顺序排查看 LangSmith 追踪哪个节点耗时最长一目了然。看数据库Checkpointer 的读写是否成为瓶颈查慢查询日志。看大模型 API是不是 API 侧变慢了看响应时间统计。看服务器资源CPU、内存、连接数是否打满。看网络客户端到服务器的网络延迟是否正常。这个顺序是从内到外先排除应用层问题再看基础设施。大部分性能问题都在应用层尤其是大模型调用和数据库操作这两块。6.4 几个能省时间的实操小技巧技巧一本地开发用 SQLite生产用 PostgreSQL。通过环境变量切换 Checkpointer代码不用改。写一个工厂函数根据配置返回不同的 Checkpointer 实例。技巧二给图执行加请求 ID。每次请求生成一个 UUID贯穿日志和 LangSmith 追踪。排查问题时用请求 ID 搜日志能快速定位到完整链路。技巧三用 LangSmith 的数据集做回归测试。每次改完 prompt 或者图结构跑一遍数据集几分钟就能知道有没有退化。比手动测试靠谱多了。技巧四流式接口加心跳。如果图执行时间较长中间没有输出客户端可能以为连接断了。可以每隔几秒发一个空事件作为心跳保持连接活跃。技巧五Checkpointer 定期清理。状态数据会越积越多定期清理过期的会话。可以按thread_id的最后更新时间来清理比如保留最近 30 天的数据。7. 关于部署路径选择我自己的几点体会三条路径走下来我最大的体会是不要过度设计但也不要欠债。一开始用最简单的方案跑通逻辑这是对的。但当服务开始有真实用户、开始承载业务的时候该补的服务化能力要补上不然技术债越积越多后面改起来更痛苦。具体来说我建议的节奏是第一周用路径一验证核心逻辑确认 Agent 的行为符合预期。第二周迁移到路径二把 API 层搭起来让团队内部先用起来。第三周接入路径三开始积累追踪数据为后续优化做准备。这个节奏不算快但每一步都踩实了后面不会返工。还有一个体会是关于 Checkpointer 的选择。我一开始图省事用MemorySaver结果服务重启一次所有对话历史全丢用户投诉了好几次。后来换成SqliteSaver单机够用但并发一上来就锁库。最后还是上了PostgreSQL才算稳定。所以如果你的服务要长期跑Checkpointer 直接上PostgreSQL别在 SQLite 上浪费时间。最后说一个容易被忽略的点LangGraph 的版本升级。LangGraph 还在快速迭代不同版本之间 API 可能有变化。升级之前一定要看 changelog在测试环境验证。我吃过一次亏升级之后 Checkpointer 的接口变了服务直接起不来。后来学乖了升级前先锁版本测试通过再上生产。