1. 从一段脚本说起为什么单文件跑不通了最开始接触 LangGraph 的时候我和大多数人一样写一个main.py里面定义 StateGraph、加几个节点、graph.compile()然后graph.invoke()跑通了心里挺美。但真到了要给别人用、要接到前端、要跑在服务器上的时候问题就来了——脚本是脚本服务是服务这中间隔着的不是一行if __name__而是一整套工程化的东西。我踩过的第一个坑特别典型本地跑得好好的图部署到服务器上之后同事调接口一直超时。排查了半天发现是图里有个节点调用了外部模型接口本地有缓存所以快服务器上冷启动每次都要等十几秒而 FastAPI 默认没有设置合理的超时和并发策略请求全堵在那儿。这件事让我意识到LangGraph 的部署不是把脚本包一层 FastAPI这么简单它涉及到状态管理、并发模型、进程生命周期、依赖隔离这一整套问题。所以这篇内容我想聊的是当你手里已经有一个能跑的 LangGraph 图之后怎么把它变成三种不同形态的服务——本地开发调试用的轻量服务、单机生产可用的 FastAPI 服务、以及容器化的可移植部署。这三条路径不是递进关系而是针对不同场景的三种选择我会把每条路径的适用边界、核心代码、以及我实际踩过的坑都摊开讲。适合谁看如果你已经写过 LangGraph 的基本 demo知道 StateGraph、Node、Edge 是什么但一到怎么让别人用上就卡壳那这篇就是给你准备的。如果你还没接触过 LangGraph建议先把官方 quickstart 跑一遍再回来不然有些概念会接不上。先明确一个前提本文讨论的部署指的是把 LangGraph 编译后的图作为服务对外提供能力而不是讨论模型本身的部署。模型你可以用本地推理也可以调云端 API这不影响部署路径的选择。三条路径的核心差异在于运行形态和运维复杂度下面逐个拆。2. 路径一脚本内嵌模式最快跑通但边界清晰2.1 什么场景下应该停在脚本层很多人一上来就想上 Docker、上 K8s其实没必要。如果你的 LangGraph 图只是自己用、或者给一两个同事临时跑个批处理任务那脚本内嵌模式就是最优解别过度工程化。我判断的标准很简单调用方是不是人。如果调用方是你自己在终端敲命令或者是一个定时任务脚本那就不需要 HTTP 服务。这种情况下你的 LangGraph 图直接compile()之后invoke()就行最多加个命令行参数解析。# run_graph.py import argparse from my_graph import build_graph def main(): parser argparse.ArgumentParser() parser.add_argument(--input, requiredTrue) args parser.parse_args() graph build_graph() result graph.invoke({messages: [args.input]}) print(result) if __name__ __main__: main()这段代码没什么技术含量但我想强调的是边界脚本模式的图生命周期和进程绑定进程结束图就销毁没有状态持久化没有并发处理。一旦你的需求里出现多个人同时用需要保留对话历史要能被其他系统调用中的任何一条就该考虑往路径二迁移了。2.2 脚本模式里最容易被忽略的 checkpoint 问题即便是脚本模式有一个东西我强烈建议你从一开始就加上checkpointer。LangGraph 的 checkpointer 机制是用来保存图执行过程中的状态的很多人以为只有做多轮对话才需要其实不是。我遇到过一个场景一个数据处理图中间有个节点要调用比较慢的接口跑一次要五分钟。有一次跑到第四分钟服务器网络抖了一下整个图崩了前面四分钟白跑。如果当时配了 checkpointer就可以从最后一个成功的 checkpoint 恢复而不是从头再来。from langgraph.checkpoint.memory import MemorySaver graph build_graph() graph graph.compile(checkpointerMemorySaver()) config {configurable: {thread_id: task-001}} result graph.invoke({messages: [...]}, config)注意thread_id这个参数它是 checkpoint 的索引键。同一个thread_id的多次 invoke 会共享状态历史不同的thread_id互相隔离。脚本模式下用MemorySaver就够了进程结束状态就没了但至少单次执行过程中能抗住中断。提示MemorySaver是内存级的进程一退就清空。如果你需要跨进程恢复得换成SqliteSaver或PostgresSaver这就自然过渡到路径二了。2.3 从脚本到服务的那个临界点在哪我给自己定的临界点是当第二个非本人的调用方出现时。第一个调用方可以是你自己第二个如果是同事、是前端、是另一个服务那就该上服务了。这个判断背后是有道理的。脚本模式的调用方式是我告诉你命令怎么敲这要求调用方懂你的代码、懂你的环境。而服务模式的调用方式是我给你一个接口调用方只需要知道接口契约。从共享代码到共享接口这是工程协作的分水岭。还有一类情况也会触发这个临界点你需要控制图的执行资源。比如你的图会调用付费 API你希望限制并发数、做请求排队、记录调用日志。这些在脚本模式里做起来很别扭但在服务模式里是天然的能力。3. 路径二FastAPI 单机服务生产可用的最小闭环3.1 为什么是 FastAPI 而不是 Flask选 FastAPI 不是跟风是有具体原因的。LangGraph 的图执行天然是异步的——节点里经常要调模型接口、调外部 API这些都是 IO 密集操作。FastAPI 原生支持async/await而 Flask 是同步框架虽然也能跑异步代码但需要额外的 WSGI 适配容易出问题。更关键的是 FastAPI 的依赖注入和生命周期管理。LangGraph 的图对象应该在整个服务生命周期内只构建一次而不是每个请求都重新 compile。FastAPI 的lifespan机制正好干这个事。from contextlib import asynccontextmanager from fastapi import FastAPI from my_graph import build_graph graph None asynccontextmanager async def lifespan(app: FastAPI): global graph graph build_graph().compile(checkpointerMemorySaver()) yield # 清理逻辑 app FastAPI(lifespanlifespan) app.post(/invoke) async def invoke(payload: dict): config {configurable: {thread_id: payload.get(thread_id, default)}} result await graph.ainvoke({messages: payload[messages]}, config) return {result: result}这里有个细节值得说用ainvoke而不是invoke。在 async 路由里调同步的invoke会阻塞事件循环导致其他请求排队。我见过有人这么写压测的时候 QPS 上不去排查半天才发现是这个问题。3.2 图对象的生命周期别在每个请求里 compile上面代码里graph是全局变量在 lifespan 里初始化一次。这个设计不是随便定的compile()这个操作本身有开销——它要构建执行计划、校验图结构。如果每个请求都 compile 一次等于每次请求都付一遍这个成本。我做过一个粗略的对比测试同一个图每次请求 compile 和全局 compile 一次在简单图上的差距大概是 20-30ms 每次请求。图越复杂差距越大。对于高并发场景这个开销是纯浪费。但全局变量也有代价图对象必须是无状态的。如果你的图在节点里往某个全局变量写数据多请求并发就会串数据。正确的做法是把所有请求相关的状态都通过config和输入传进去图本身保持纯净。注意如果你用了MemorySaver做 checkpointer它本身是线程安全的但状态是按thread_id隔离的。千万别用固定的thread_id否则所有用户的对话历史会混在一起这是个很隐蔽的 bug。3.3 流式输出LangGraph 的 astream 怎么接LangGraph 有个很好用的能力是astream可以流式产出每个节点的执行结果。对于需要实时反馈的场景比如对话式应用这个能力很关键。但把它接到 FastAPI 上有个坑。from fastapi.responses import StreamingResponse app.post(/stream) async def stream(payload: dict): async def event_generator(): config {configurable: {thread_id: payload[thread_id]}} async for chunk in graph.astream({messages: payload[messages]}, config): yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这里用的是 SSEServer-Sent Events格式。坑在于astream产出的 chunk 结构取决于你的图怎么定义。默认情况下它产出的是每个节点的输出但如果你想要 token 级别的流式需要在节点内部用astream_events粒度更细但代码更复杂。我个人的经验是先跑通节点级流式再考虑 token 级。节点级流式已经能满足大部分让用户看到进度的需求token 级流式主要用在纯对话场景实现成本高不少。3.4 单机服务的并发与超时配置FastAPI 默认跑在 uvicorn 上单进程单事件循环。对于 IO 密集的 LangGraph 图单进程其实能扛不少并发因为大部分时间在等 IO。但如果你的图里有 CPU 密集的节点比如本地做向量计算单进程就会成为瓶颈。我的配置经验是这样的场景worker 数说明纯 IO 密集调模型 API1-2事件循环足够多了反而增加内存混合型CPU 核数 / 2留一半给系统和其他进程CPU 密集CPU 核数但要配合--workers多进程启动命令大概是uvicorn app:app --host 0.0.0.0 --port 8000 --workers 2 --timeout-keep-alive 30--timeout-keep-alive这个参数容易被忽略。如果你的图执行时间较长客户端连接可能被中间层比如 Nginx掐断适当调大这个值能减少这类问题。但注意这个值不是越大越好太大会占用连接资源。4. 路径三Docker 容器化可移植性的代价与收益4.1 什么时候真的需要 DockerDocker 不是银弹它解决的是环境一致性和部署可移植性问题。如果你的服务只跑在一台固定的服务器上环境已经配好了那 Docker 带来的收益有限反而增加了构建和调试的复杂度。我判断需要 Docker 的场景有这么几个第一服务要部署到多台机器手动配环境容易不一致第二依赖里有系统级的库比如某些图像处理、科学计算库pip 装不干净第三团队协作新人拉下来就能跑不用配半天环境。如果你的场景不满足这些路径二的裸机 FastAPI 服务就够了别为了 Docker 而 Docker。4.2 Dockerfile 怎么写才不臃肿LangGraph 项目的 Dockerfile最常见的错误是把整个开发环境都塞进去。我见过一个镜像 3 个 G 的里面装了 jupyter、pytest、各种调试工具生产环境根本用不上。我的写法是多阶段构建把构建依赖和运行依赖分开# 构建阶段 FROM python:3.11-slim AS builder WORKDIR /app COPY requirements.txt . RUN pip install --user --no-cache-dir -r requirements.txt # 运行阶段 FROM python:3.11-slim WORKDIR /app COPY --frombuilder /root/.local /root/.local COPY . . ENV PATH/root/.local/bin:$PATH EXPOSE 8000 CMD [uvicorn, app:app, --host, 0.0.0.0, --port, 8000]用slim基础镜像而不是alpine是因为 alpine 用的是 musl libc某些 Python 包的预编译 wheel 不兼容装的时候要现场编译反而更慢更容易出错。slim是 Debian 系兼容性好很多。--no-cache-dir这个参数一定要加不然 pip 的缓存会让镜像大一圈。--user装到用户目录方便在运行阶段复制。4.3 容器里的状态持久化checkpointer 怎么选这是容器化部署里最关键的一个决策点。容器是无状态的一重启里面的数据就没了。如果你的 LangGraph 用了 checkpointer就必须把状态存到容器外面。三个选择SqliteSaver存到挂载的 volume 里。适合单机、低并发场景。优点是零依赖缺点是并发写有锁竞争。PostgresSaver存到外部数据库。适合多实例、高并发场景。优点是可扩展缺点是要维护数据库。RedisSaver存到 Redis。适合需要快速读写的场景。优点是快缺点是持久化配置要小心。我个人的选择逻辑是单机 Docker 用 Sqlite volume多机部署用 Postgres。Redis 我一般不用来做 checkpointer因为它的持久化语义不如数据库清晰状态数据丢了很难排查。from langgraph.checkpoint.sqlite import SqliteSaver # 挂载到 /data 目录 with SqliteSaver.from_conn_string(/data/checkpoints.db) as checkpointer: graph build_graph().compile(checkpointercheckpointer)注意from_conn_string是个上下文管理器要在 lifespan 里正确管理它的生命周期别在请求里临时开。4.4 镜像构建的缓存优化与体积控制Docker 构建有个特性每一层都会缓存只要该层的输入没变就复用缓存。利用这个特性把变化频率低的操作放前面。COPY requirements.txt .然后RUN pip install放在COPY . .之前这样只要你没改依赖改代码的时候 pip install 那层直接走缓存构建秒级完成。这个顺序千万别搞反我见过把COPY . .放前面的每次改一行代码都要重装所有依赖构建十分钟起步。体积控制上几个实用技巧.dockerignore里排除.git、__pycache__、*.pyc、测试数据、本地虚拟环境目录用pip install --no-cache-dir避免缓存构建完可以用docker image prune清理悬空镜像如果依赖里有编译型包考虑用python:3.11-slim而不是自己从源码编一个优化到位的 LangGraph 服务镜像控制在 500MB 以内是合理的。如果超过 1G基本可以确定有冗余。5. 三条路径的横向对比与选型决策5.1 用一张表把差异说清楚维度脚本内嵌FastAPI 单机Docker 容器化适用调用方本人/定时任务团队/前端/其他服务多机部署/跨环境状态持久化内存/无Sqlite/Postgres外部存储并发能力无中高高可水平扩展部署复杂度极低低中调试难度低中中高环境一致性依赖本机依赖本机强迁移成本高中低这张表不是让你照着选而是让你看清每个选择的代价在哪。脚本模式省事但迁移成本高Docker 可移植但调试麻烦。没有银弹只有权衡。5.2 我实际项目里的迁移路径说说我自己的真实经历。一个 LangGraph 项目最开始是脚本跑批处理任务。后来产品要加个界面就包了 FastAPI。再后来要部署到客户的服务器上客户环境五花八门就上了 Docker。这个迁移过程不是一次到位的是需求推着走的。每次迁移都只解决当前最痛的问题不提前做过度设计。我见过有人一开始就上 K8s结果项目还没上线运维成本先把人拖垮了。我的建议是从脚本开始遇到第一个非本人调用就上 FastAPI遇到第一个环境不一致就上 Docker。每一步都有明确的触发条件不凭感觉。5.3 那些三条路径都会遇到的共性问题不管走哪条路径有几个问题是绕不开的第一日志。LangGraph 的图执行过程默认不打日志出问题的时候你根本不知道卡在哪个节点。我的做法是在每个节点的入口和出口加日志用logging模块别用print。容器化之后日志要输出到 stdout让 Docker 的日志驱动去收集。第二错误处理。图里某个节点抛异常整个图就挂了。LangGraph 支持在节点里 try/except把错误转成状态的一部分让图能继续走。这个设计很关键尤其是调用外部 API 的节点失败是常态。第三配置管理。API key、数据库连接串这些绝对不能硬编码。脚本模式用环境变量FastAPI 用 pydantic-settingsDocker 用--env-file或 secrets。我见过把 key 写死在代码里提交到仓库的这是大忌。第四版本锁定。LangGraph 迭代很快不同版本 API 有变化。requirements.txt里一定要锁版本号别用。我吃过这个亏本地跑得好好的服务器上装了个新版本API 变了直接崩。6. 部署之后那些文档里不会写的运维细节6.1 健康检查接口不是可选项FastAPI 服务上线之后第一件事是加健康检查接口。这不是给用户用的是给负载均衡、容器编排、监控系统用的。app.get(/health) async def health(): return {status: ok, graph_ready: graph is not None}graph_ready这个字段很关键。服务刚启动的时候lifespan 可能还没执行完图还没构建好这时候如果健康检查返回 ok流量打进来就会报错。加上这个字段编排系统会等到图真正就绪才放流量。Docker 里对应的是HEALTHCHECK指令HEALTHCHECK --interval30s --timeout3s --start-period10s \ CMD python -c import urllib.request; urllib.request.urlopen(http://localhost:8000/health)--start-period是给服务启动留的缓冲时间LangGraph 图构建如果比较慢这个值要调大。6.2 内存泄漏的排查思路LangGraph 服务跑久了内存涨这是个常见问题。原因通常有几个checkpointer 里积累的历史状态没清理、图里某个节点缓存了数据没释放、或者 Python 本身的循环引用。排查方法用tracemalloc或者memory_profiler定位增长点。如果是 checkpointer 的问题定期清理旧的thread_id数据。如果是节点缓存检查有没有用全局字典做缓存但没设上限。我遇到过一次是MemorySaver在长跑服务里无限积累因为每个请求都用了新的thread_id状态全留在内存里。后来换成SqliteSaver并加了定期清理任务才解决。6.3 版本升级时的平滑过渡LangGraph 升级版本的时候最怕的是正在执行的图被中断。如果你的服务有长任务升级前要先 drain 掉正在跑的请求。Docker 环境下docker stop默认发 SIGTERM然后等 10 秒再 SIGKILL。你需要在 FastAPI 里捕获 SIGTERM停止接受新请求等正在执行的图跑完再退出。这个逻辑写在 lifespan 的清理部分。import signal asynccontextmanager async def lifespan(app: FastAPI): global graph, shutting_down graph build_graph().compile(...) shutting_down False def handle_sigterm(*args): global shutting_down shutting_down True signal.signal(signal.SIGTERM, handle_sigterm) yield # 等待正在执行的请求完成配合docker stop --time 60给足退出时间就能做到升级不丢请求。6.4 我踩过的最坑的一次时区问题最后分享一个特别隐蔽的坑。有次部署到海外服务器日志时间戳全乱了排查了半天发现是容器时区默认 UTC而代码里用了本地时间做逻辑判断导致某些基于时间的条件判断出错。解决方案是在 Dockerfile 里设置时区或者代码里统一用 UTC 时间只在展示层转本地时区。我现在的习惯是代码里一律用 UTC展示层再转这样跨时区部署不会出问题。ENV TZAsia/Shanghai RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime echo $TZ /etc/timezone这个坑不大但很典型——部署环境的差异往往不在代码逻辑里而在这些环境细节里。脚本模式在本机跑时区、编码、路径分隔符都是对的一进容器全变了。这也是为什么我说 Docker 的收益和代价是并存的它带来一致性的同时也引入了新的变量。三条路径说到底是三种不同的工程权衡。脚本模式赢在简单FastAPI 赢在灵活Docker 赢在一致。选哪条取决于你当前最痛的是什么。我个人的经验是别提前优化让需求推着你走每一步迁移都有明确的触发条件这样既不会过度设计也不会在需要扩展的时候手忙脚乱。
