1. 从脚本到服务为什么部署这一步最容易卡住写 LangGraph 的人大概都有过这种体验本地跑一个graph.invoke()顺得不行状态流转、工具调用、条件分支全都对结果一到要给别人用这一步就懵了。脚本能跑但它不是服务——没有接口、没有并发、没有持久化、进程一挂状态全丢。这个坎几乎每个从 LangGraph 入门走向落地的人都会踩。我自己最开始也是把整张图塞进一个main.py用while True读输入跑通了就觉得很爽。直到有一次需要让前端调用才发现问题一大堆状态怎么跨请求保留多个人同时用会不会串工具调用中途报错了怎么恢复这时候才意识到LangGraph 的部署不是把脚本包一层这么简单它涉及三条完全不同的路径对应三种不同的使用场景和成熟度。这篇文章想聊的就是这三条路径本地脚本直跑、FastAPI 服务化封装、LangSmith 平台托管。它们不是互相替代的关系而是从原型到生产的一条渐进路线。我会把每条路径的适用场景、核心实现、踩过的坑都摊开讲尤其是 FastAPI 那条路因为它是绝大多数团队真正落地时会选的方案涉及 Checkpointer、状态持久化、并发处理这些绕不开的细节。如果你正在纠结我的 LangGraph 项目到底该怎么部署或者已经部署了但总觉得哪里不稳这篇应该能帮你把思路理清。先给个整体判断方便你对号入座部署路径适用阶段核心特征典型痛点脚本直跑原型验证、个人调试单进程、无接口、状态在内存无法并发、重启丢状态FastAPI 服务化团队协作、产品落地HTTP 接口、可持久化、可扩展状态管理、并发、日志LangSmith 托管快速上线、可观测优先平台托管、自带追踪定制受限、成本考量三条路径背后其实是同一个问题你的图状态存在哪、谁来管、怎么恢复。想清楚这个选哪条路就清楚了。2. 路径一脚本直跑别小看它的价值2.1 什么阶段该老老实实写脚本很多人一上来就想搞服务化觉得脚本不够专业。我的经验恰恰相反LangGraph 的调试阶段脚本直跑是效率最高的方式。原因很简单图的核心逻辑——节点函数、条件边、状态结构——在脚本里改一行就能重跑不用管服务重启、不用管序列化、不用管接口参数。你调的是图本身对不对而不是服务跑不跑得起来。这个阶段我建议把结构固定成三块状态定义、节点函数、图组装。状态用TypedDict或者 Pydantic 模型定义清楚节点函数保持纯函数风格输入状态、输出状态增量图组装单独放一段。这样等你后面要搬到 FastAPI 时这三块几乎可以原样复用只换最外层的调用方式。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] next_step: str def planner_node(state: AgentState): # 这里放你的规划逻辑 return {next_step: tool} def tool_node(state: AgentState): # 工具调用逻辑 return {messages: [(ai, 工具执行完成)]} builder StateGraph(AgentState) builder.add_node(planner, planner_node) builder.add_node(tool, tool_node) builder.add_edge(planner, tool) builder.add_edge(tool, END) builder.set_entry_point(planner) graph builder.compile() if __name__ __main__: result graph.invoke({messages: [(user, 帮我查一下天气)]}) print(result)2.2 脚本阶段就要埋好的两个伏笔脚本能跑不代表可以随便写。有两个东西我强烈建议在脚本阶段就设计好否则后面迁移会痛不欲生。第一个是状态的可序列化性。LangGraph 的状态最终要能存进 Checkpointer如果你的状态里塞了不可序列化的对象比如数据库连接、文件句柄、自定义的复杂类实例到服务化那一步就得大改。养成习惯状态里只放基础类型、消息列表、可 JSON 化的字典。需要外部资源就在节点函数内部临时获取别塞进状态。第二个是节点函数的幂等性。脚本阶段你可能不在乎同一个节点跑两次会怎样但一旦上了 Checkpointer图会在中断后从某个检查点恢复节点可能被重放。如果你的节点函数有副作用比如发邮件、扣款、写数据库重放就会出问题。我的做法是把有副作用的操作单独抽出来用状态里的标记位控制只执行一次或者干脆设计成幂等操作。提示脚本阶段别急着上 Checkpointer但状态结构要按将来要持久化的标准来设计。这个伏笔能帮你省掉后面一次大重构。2.3 脚本直跑的边界在哪脚本直跑的天花板很明显单进程、单用户、无状态保留。它适合你自己调试、适合跑批处理任务一次输入一次输出、适合做 demo。但只要出现下面任意一条就该考虑往服务化走了需要多个用户/多个请求同时用需要跨请求保留对话上下文需要前端或其他系统通过接口调用需要中断后恢复、人工介入human-in-the-loop这四条里只要中了一条脚本就不够用了。尤其是 human-in-the-loop它天然要求状态能暂停、能存、能恢复这已经是服务化的范畴了。3. 路径二FastAPI 服务化真正落地的主战场3.1 为什么是 FastAPI 而不是 Flask热词里flask 与 fastapi 比较出现频率很高说明很多人卡在这个选型上。我的结论很直接LangGraph 服务化优先选 FastAPI原因有三个而且都跟 LangGraph 的特性强相关。第一LangGraph 的调用天然是异步友好的。图执行、工具调用、模型请求大量涉及 IO 等待FastAPI 原生支持async def能把这些等待时间利用起来并发能力比 Flask 的同步模型好一大截。你用 Flask 也能跑但要么阻塞、要么自己搞线程池何必呢。第二FastAPI 的 Pydantic 集成跟 LangGraph 的状态定义思路高度一致。你的状态模型、请求体、响应体可以共用一套类型定义接口文档自动生成前端对接的时候直接看/docs就行省掉大量沟通成本。第三流式输出。LangGraph 支持astream和astream_events做打字机效果、实时推送特别方便而 FastAPI 的StreamingResponse和 SSE 支持得很顺。Flask 做流式要绕一些。当然 Flask 不是不能用如果你的团队全是 Flask 老手、项目又简单用 Flask 也没问题。但如果是新项目没有历史包袱FastAPI 是更顺的选择。3.2 一个能直接抄的项目目录结构fastapi项目目录结构是高频搜索词说明大家都在找标准答案。我把自己用了几个项目的结构贴出来这个结构的好处是图逻辑和 Web 层彻底分离图可以单独测试Web 层只负责协议转换。langgraph-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口注册路由 │ ├── api/ │ │ ├── __init__.py │ │ └── routes.py # 接口定义 │ ├── core/ │ │ ├── config.py # 配置管理 │ │ └── checkpointer.py # Checkpointer 初始化 │ ├── graph/ │ │ ├── __init__.py │ │ ├── state.py # 状态定义 │ │ ├── nodes.py # 节点函数 │ │ └── builder.py # 图组装 │ ├── schemas/ │ │ └── chat.py # 请求/响应模型 │ └── services/ │ └── agent_service.py # 业务编排 ├── tests/ ├── requirements.txt └── .env关键点是graph/目录完全不知道 FastAPI 的存在它就是一个纯粹的 LangGraph 模块。services/层负责把 HTTP 请求翻译成图调用api/层只管路由和参数校验。这样分层之后你想换 Web 框架、想加 CLI 入口、想写单元测试都不受影响。3.3 Checkpointer服务化的命门checkpointer是这批热词里技术含量最高的一个也是 LangGraph 服务化区别于普通脚本的核心。它的作用是把图的状态持久化下来让对话能跨请求延续、让中断能恢复。LangGraph 提供几种 Checkpointer选型逻辑是这样的Checkpointer存储位置适用场景注意事项MemorySaver内存本地调试、单进程重启即丢不能上生产SqliteSaver本地文件单机小规模、演示并发写入有限制PostgresSaverPostgreSQL生产环境、多实例需要建表、配连接池我的建议很明确开发用 MemorySaver 快速验证生产直接上 PostgresSaver。SqliteSaver 适合单机 demo但一旦你要多进程或者多实例部署SQLite 的写锁会成为瓶颈。用 PostgresSaver 的时候有个坑必须提前说它需要初始化表结构。很多人第一次用代码跑起来报表不存在就是因为漏了初始化这一步。from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI postgresql://user:passlocalhost:5432/langgraph pool ConnectionPool(conninfoDB_URI, max_size20) checkpointer PostgresSaver(pool) # 关键一步首次部署必须执行创建所需的表 checkpointer.setup() graph builder.compile(checkpointercheckpointer)setup()这个方法幂等重复执行不会出问题所以你可以放心地放在启动流程里。但要注意它需要数据库账号有建表权限生产环境如果权限收紧得让 DBA 提前把表建好。3.4 thread_id多用户隔离的关键Checkpointer 存状态是按thread_id隔离的。这个概念特别重要理解错了会导致用户之间串数据。你可以把thread_id理解成一个会话的身份证——同一个thread_id的多次调用共享同一份状态不同thread_id之间完全隔离。config {configurable: {thread_id: user-123-session-456}} result await graph.ainvoke({messages: [(user, 你好)]}, configconfig)thread_id的生成策略直接影响你的业务逻辑。如果是单轮问答每次请求生成一个新的 UUID 就行如果是多轮对话得用会话 ID而且要保证同一个用户的同一个会话始终用同一个thread_id。我见过有人图省事用用户 ID 当thread_id结果用户开两个对话窗口就串了——因为两个窗口共享了同一份状态。注意thread_id一旦确定同一个会话内不要变。中途换thread_id等于开了个新会话之前的状态就找不回来了。3.5 接口设计同步、流式、中断恢复FastAPI 层要暴露的接口我一般设计三个普通调用、流式调用、状态查询。普通调用适合后端到后端流式调用适合前端聊天界面状态查询用于 human-in-the-loop 场景。from fastapi import APIRouter from fastapi.responses import StreamingResponse from app.schemas.chat import ChatRequest from app.services.agent_service import run_agent, stream_agent router APIRouter() router.post(/chat) async def chat(req: ChatRequest): result await run_agent(req.message, req.thread_id) return {reply: result} router.post(/chat/stream) async def chat_stream(req: ChatRequest): async def event_generator(): async for chunk in stream_agent(req.message, req.thread_id): yield fdata: {chunk}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)流式这块有个细节值得说LangGraph 的astream_events会吐出很多类型的事件节点开始、节点结束、token 流等你不需要全推给前端按event字段过滤出真正要展示的内容就行。全推过去前端会收到一堆噪音还得自己过滤不如后端先筛一遍。4. 路径三LangSmith 托管什么时候值得上4.1 托管解决的是可观测而不是能跑很多人对 LangSmith 有误解以为它是部署平台其实它更偏向可观测性和协作。你的图还是跑在你自己的服务器上LangSmith 帮你做的是追踪每一次调用、记录每个节点的输入输出、可视化执行路径、管理数据集和评估。那什么时候值得上 LangSmith我的判断标准是当你开始问这次调用为什么慢这个节点为什么输出不对改了 prompt 之后效果是变好还是变差的时候。脚本阶段你靠 print 就能调试服务化之后请求多了、路径复杂了print 就不够用了这时候 LangSmith 的价值就出来了。接入成本其实很低配几个环境变量就行LANGCHAIN_TRACING_V2true LANGCHAIN_API_KEYyour_key LANGCHAIN_PROJECTmy-langgraph-app配好之后你的图调用会自动上报追踪数据在平台上能看到完整的执行链路。这个对排查工具调用为什么没触发条件边为什么走了这个分支特别有用。4.2 托管和自建的取舍LangSmith 不是唯一选择你也可以自建追踪比如用 OpenTelemetry 那一套。取舍点在于要快LangSmith 开箱即用配环境变量就完事自建要搭一套可观测基础设施要可控数据敏感、要求全内网的场景自建更合适要评估能力LangSmith 自带数据集管理和评估工具自建这些要自己写我的实际做法是混合开发和测试阶段用 LangSmith 快速迭代生产环境如果数据敏感就切到自建追踪但保留 LangSmith 用于离线评估。两者不冲突。4.3 三条路径的迁移关系把三条路径串起来看它们其实是一条渐进路线而且迁移成本是可控的前提是你按前面说的把图逻辑和调用层分离了。从脚本到 FastAPI你改的是最外层把graph.invoke()换成接口调用加上 Checkpointer 和thread_id。图本身、节点函数、状态定义几乎不动。从 FastAPI 到加 LangSmith你改的是配置加几个环境变量代码基本不动。这就是为什么我一直强调分层。分层做得好部署路径的切换就是换壳不是重写。分层做得差每次升级都是一次重构那才是真的痛。5. 实操中真正会咬人的那些坑5.1 uvicorn 日志丢失别以为是代码问题uvicorn fastapi 日志丢失问题是高频搜索词我踩过。现象是本地print能看到上了 uvicorn 之后日志没了或者只显示一部分。原因通常是 uvicorn 接管了日志配置把 root logger 的 handler 覆盖了。解决办法是显式配置日志别依赖默认行为import logging import sys logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, handlers[logging.StreamHandler(sys.stdout)], forceTrue, # 关键强制覆盖已有配置 )forceTrue这个参数很多人不知道它的作用是清掉已有的 handler 重新配置。不加它uvicorn 的配置会一直压着你的。另外生产环境建议日志输出到 stdout让容器平台去收集别自己写文件多实例部署时文件日志会乱。5.2 并发下的状态竞争FastAPI 是异步的同一个thread_id如果同时来两个请求会发生什么答案是状态可能错乱。因为两个请求都在读写同一份 Checkpointer 状态谁先谁后不确定。我的处理原则是同一个thread_id的请求要串行化。简单做法是在业务层加锁按thread_id维度加import asyncio from collections import defaultdict _locks defaultdict(asyncio.Lock) async def run_agent(message: str, thread_id: str): async with _locks[thread_id]: config {configurable: {thread_id: thread_id}} return await graph.ainvoke({messages: [(user, message)]}, configconfig)这个锁是进程内的单实例够用。多实例部署的话得用分布式锁比如基于 Redis或者从产品设计上避免同一会话并发——大多数聊天场景其实天然就是串行的用户不会同时发两条消息。5.3 工具调用超时把整个图拖死LangGraph 里工具调用是常见节点但工具可能慢、可能挂。如果不设超时一个卡住的工具会把整个图执行拖死请求一直不返回。我的做法是给每个工具调用包一层超时import asyncio async def safe_tool_call(tool, args, timeout10): try: return await asyncio.wait_for(tool.ainvoke(args), timeouttimeout) except asyncio.TimeoutError: return {error: 工具调用超时}超时之后返回一个错误状态让图能继续走比如走到一个处理失败的节点而不是整个卡住。这个设计在服务化环境里特别重要因为一个卡住的请求会占着连接请求多了直接把服务拖垮。5.4 常见问题速查表现象可能原因排查方向状态跨请求丢失没配 Checkpointer 或 thread_id 不一致检查 compile 时是否传 checkpointer报表不存在PostgresSaver 没执行 setup()启动时调用 checkpointer.setup()日志不输出uvicorn 覆盖了日志配置basicConfig 加 forceTrue多用户数据串了thread_id 生成策略有问题检查是否用用户 ID 当 thread_id工具调用卡死没设超时给工具调用加 asyncio.wait_for流式输出断断续续事件没过滤推了太多噪音按 event 类型过滤后再推5.5 部署形态的选择最后聊一下部署形态。FastAPI 服务化之后你可以直接跑在服务器上也可以容器化。我的建议是容器化因为 LangGraph 服务往往依赖数据库、依赖模型服务容器化之后环境一致性好迁移也方便。Dockerfile 不用复杂核心是别把开发依赖打进去启动命令用 uvicorn 指定 worker 数FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000, --workers, 2]worker 数不是越多越好。因为每个 worker 是独立进程各自持有自己的 Checkpointer 连接池和内存状态。worker 太多数据库连接会被打满。我的经验是先从 2 个 worker 起步根据 CPU 和数据库连接数再调。如果用了 PostgresSaver记得把连接池大小和 worker 数一起算别让总连接数超过数据库上限。6. 我个人的一些实际体会三条路径走下来我最大的感受是部署的难点从来不在怎么把服务跑起来而在状态怎么管。脚本阶段状态在内存里简单但脆弱服务化之后状态进了数据库复杂但可靠。这个转变是 LangGraph 从玩具走向工具的分水岭。Checkpointer 这个组件我建议每个认真做 LangGraph 的人都花时间吃透。它不只是存个状态那么简单它决定了你的图能不能中断、能不能恢复、能不能支持人工介入、能不能做多轮对话。理解了它很多部署上的困惑会自然消解。另外就是别过度设计。我见过有人原型阶段就上 PostgresSaver 加分布式锁加全套可观测结果图逻辑还没调通光在折腾基础设施。正确的顺序是脚本把图调通FastAPI 把接口跑通Checkpointer 把状态管住最后再考虑可观测和扩展。每一步都稳了再走下一步比一步到位靠谱得多。如果你现在正卡在某一步我的建议是先问自己一个问题我的状态现在存在哪重启之后还在不在。这个问题的答案基本就决定了你该走哪条路径。