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", required=True) 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(checkpointer=MemorySaver()) 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(checkpointer=MemorySaver()) yield # 清理逻辑 app = FastAPI(lifespan=lifespan) @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 f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream")这里用的是 SSE（Server-Sent Events）格式。坑在于：astream产出的 chunk 结构取决于你的图怎么定义。默认情况下它产出的是每个节点的输出，但如果你想要 token 级别的流式，需要在节点内部用astream_events，粒度更细但代码更复杂。
我个人的经验是：先跑通节点级流式，再考虑 token 级。节点级流式已经能满足大部分"让用户看到进度"的需求，token 级流式主要用在纯对话场景，实现成本高不少。
3.4 单机服务的并发与超时配置
FastAPI 默认跑在 uvicorn 上，单进程单事件循环。对于 IO 密集的 LangGraph 图，单进程其实能扛不少并发，因为大部分时间在等 IO。但如果你的图里有 CPU 密集的节点（比如本地做向量计算），单进程就会成为瓶颈。
我的配置经验是这样的：
| 场景 | worker 数 | 说明 |
|---|---|---|
| 纯 IO 密集（调模型 API） | 1-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 什么时候真的需要 Docker
Docker 不是银弹，它解决的是环境一致性和部署可移植性问题。如果你的服务只跑在一台固定的服务器上，环境已经配好了，那 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 --from=builder /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(checkpointer=checkpointer)注意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-settings，Docker 用--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 --interval=30s --timeout=3s --start-period=10s \ 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 TZ=Asia/Shanghai RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone这个坑不大，但很典型——部署环境的差异往往不在代码逻辑里，而在这些环境细节里。脚本模式在本机跑，时区、编码、路径分隔符都是对的，一进容器全变了。这也是为什么我说 Docker 的收益和代价是并存的，它带来一致性的同时，也引入了新的变量。
三条路径说到底，是三种不同的工程权衡。脚本模式赢在简单，FastAPI 赢在灵活，Docker 赢在一致。选哪条，取决于你当前最痛的是什么。我个人的经验是，别提前优化，让需求推着你走，每一步迁移都有明确的触发条件，这样既不会过度设计，也不会在需要扩展的时候手忙脚乱。
