1. 先别急着上生产LangGraph部署到底卡在哪很多人第一次把 LangGraph 项目写出来的时候感觉特别爽。状态图一画节点一接graph.invoke()一调Agent 就按照你设定的流程跑起来了该调工具调工具该回复回复逻辑清清楚楚。但爽完就懵了这玩意儿怎么给别人用怎么放到服务器上怎么扛住多个用户同时问问题我当初做第一个 LangGraph 项目——一个工单自动分类和回复生成的 Agent——也走了一遍完整的弯路。最开始就是一个python main.py在终端里跑后面要接企业微信机器人才被迫把它改成 HTTP 服务再往后几个人一起联调、测试才又补上容器化部署。整个过程走下来我发现 LangGraph 的部署路径其实是清晰的三段式对应三种不同的使用场景根本不存在一种部署方式打天下的说法。先理清一个底层认知LangGraph 本身不关心你怎么部署它。它编译出来的东西本质上是一个可调用的图对象有输入、有输出内部还维护着基于checkpointer的状态。也就是说你可以在一个普通 Python 脚本里调它也可以把它包在 FastAPI 路由里通过 HTTP 暴露出去还可以把它封装成独立服务扔进 Kubernetes。不同的部署方式难点不在 LangGraph 自身而在外部工程谁来触发调用、状态怎么持久化、并发怎么控制、日志怎么采集。所以这篇文章我把三条路径拆开讲清楚路径一脚本模式最快跑通、适合离线任务路径二FastAPI 服务化让 Agent 变成对客的 API 服务适合即时交互路径三容器化编排部署解决团队协作和生产环境的稳定性问题。每条我都贴了实测过的配置、代码和踩坑记录你可以直接照着抄。2. 路径一脚本模式——最小成本跑通全链路2.1 什么时候脚本就够了先泼一盆冷水不是所有 LangGraph 应用都必须做成服务。如果你的 Agent 是离线处理任务的比如每天晚上跑一遍销售数据的异常分析、每周生成一次周报、把一堆历史工单批量打上标签那脚本就是最优解。脚本模式的好处有三个都是实打实的一是启动成本几乎为零装好依赖直接跑不需要额外维护服务进程二是调试链条最短出问题直接看终端输出的日志甚至能breakpoint()进去打断点看每一节点的状态三是资源占用可控用完即走不会像常驻服务那样一直占着内存。我当时给一个内部知识库做文档自动清洗 Agent就是用脚本模式跑的。输入是一堆 Markdown 文件输出是清洗后的结构化条目。这个场景不需要用户实时等待跑一条命令等几分钟拿结果就行做成服务反而是给自己找麻烦——要处理超时、要设计任务队列、要考虑服务的存活监控全是徒增工作量。2.2 实测一个完整的 LangGraph 脚本骨架脚本模式的写法不复杂但有一个关键点很容易忽略状态持久化。很多人写的 LangGraph 脚本是一次性的——每次运行重新构建状态聊完就忘。这对离线批处理任务当然没问题但如果你在脚本里跑的是带多轮对话的 Agent你会发现在一次完整的invoke内状态没问题一旦脚本结束、再次启动历史上下文全丢了。所以我在脚本里会直接接入持久化 checkpointer。LangGraph 提供了几种方案最省事的是MemorySaver但它只存在内存里脚本一退出就没了如果脚本要跨多次运行保留状态建议直接用SqliteSaver或PostgresSaver。实测下来SqliteSaver在单机脚本里特别好用零额外配置一个文件搞定。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph, START, END def build_graph(): # 假设你已经定义好了节点函数 node_a, node_b g StateGraph(AgentState) g.add_node(node_a, node_a) g.add_node(node_b, node_b) g.add_edge(START, node_a) g.add_edge(node_a, node_b) g.add_edge(node_b, END) # 用 SqliteSaver 做持久化数据库文件存到本地 with SqliteSaver.from_conn_string(checkpoints.sqlite) as saver: graph g.compile(checkpointersaver) # config 里带上 thread_id用来区分不同的会话 config {configurable: {thread_id: batch_task_001}} result graph.invoke({input: 工单打印机无法连接}, configconfig) return result if __name__ __main__: print(build_graph())这段代码里有几个实用细节值得多说一句。第一thread_id是你的会话标识同一个 id 连续跑多次invokeLangGraph 会自动把上一次的最终状态作为下一次的输入实现长对话记忆不同的thread_id之间状态隔离互不污染。第二用with语句管理连接脚本跑完自动释放不会锁住 sqlite 文件。2.3 脚本模式容易翻车的 3 个细节脚本模式看着简单我还是踩过一些坑都在这里第一LLM API 的限流问题。脚本批量处理数据时通常是循环调用图。如果循环里没有限速很容易触发 API 服务的 Rate Limit。我的做法是在节点内部包装一层带退避重试的 LLM 调用捕获限流异常后按指数退避等待重试三次还失败就把当前记录写进失败清单最后统一人工处理。别小看这一步批量任务跑一半断掉的痛苦经历过的人都懂。第二日志必须带上下文。脚本跑的时候你没法盯着看所以我每个节点函数的第一行都加一句logger.info把当前输入的关键字段打出来。这样哪怕某个工单处理失败翻日志也能定位到是哪个环节、哪个输入导致的。我习惯用结构化的 keyvalue 形式打日志比如logger.info(node_a input%s, json.dumps(input_data, ensure_asciiFalse))后面 grep 起来非常方便。第三幂等性。脚本任务中断后重新跑如果是重复处理同一批数据最好先检查目标文件或数据库里是否已有结果。否则一次失误就会产生大量重复的数据条目。我在输出端加了按输入哈希建唯一索引的机制重复跑同一批数据不会产生脏数据。3. 路径二FastAPI 服务化——让 Agent 从本机走向内外网3.1 FastAPI LangGraph为什么这对组合最顺手当你需要把 Agent 能力开放给其他系统或用户时脚本就不够用了。不可能让每个调用方都去装 Python 环境、拉仓库、跑脚本——你得提供一个 HTTP 接口让对方发个请求就拿到结果。这时候服务化是必然的。服务化框架我首推 FastAPI原因有三。第一async 原生支持LangGraph 的ainvoke、astream都是异步方法FastAPI 天然契合不会出现线程阻塞问题。第二开箱即用的数据校验用 Pydantic 定义请求和响应模型外部传参不合法直接返回 422不用自己在代码里写一堆 if 判断。第三自动生成 OpenAPI 文档前后端联调的时候直接把/docs丢给对方看沟通成本降一大截。市面上也有人用 Flask 或 Django 来做但 Flask 的异步支持比较别扭Django 又偏重。对 LangGraph 这种单体服务、状态在内部、关键是 IO 密集的形态FastAPI 可以说是最省心的选择。3.2 核心实操把图挂到 HTTP 端点上服务化的核心工作就是把编译好的图封装成一个接口。先看一个最小可用的例子import uuid from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from langgraph.checkpoint.sqlite import SqliteSaver from src.graph import build_graph app FastAPI(titleAgent Service) # 应用启动时创建图对象注意要用异步的 sqlite saver 配合 async 接口 saver SqliteSaver.from_conn_string(checkpoints.sqlite) graph build_graph(saver) class QueryRequest(BaseModel): message: str Field(..., description用户消息) session_id: str | None Field(None, description会话ID不传则新开会话) class AgentResponse(BaseModel): session_id: str content: str app.post(/api/agent, response_modelAgentResponse) async def run_agent(req: QueryRequest): session_id req.session_id or uuid.uuid4().hex config {configurable: {thread_id: session_id}} try: # 这里用 ainvoke 异步执行整个图 result await graph.ainvoke({input: req.message}, configconfig) return AgentResponse( session_idsession_id, contentresult.get(output, ) ) except Exception as e: # 服务里不能让异常裸奔必须转成 HTTP 错误 raise HTTPException(status_code500, detailfagent run failed: {str(e)})这个例子能跑但如果你是照着这个去做生产得注意三个升级点。第一个升级点是耗时控制。Agent 调用 LLM 通常要几秒到几十秒HTTP 层面的超时客户端不一定等得起。我的做法是分两个接口一个同步接口适用于内部服务间调用、可以接受较长等待一个异步任务接口接收请求后立刻返回task_id背后用任务队列执行前端通过轮询或 WebSocket 拿结果。异步方案对用户体验友好得多。第二个升级点是流式输出。很多场景下用户不想等全部想完再回复希望像 ChatGPT 那样打字机式地出字。LangGraph 的astream可以逐节点逐 token 产出结果FastAPI 可以用StreamingResponse转成 SSE 流。代码长这样import json from fastapi.responses import StreamingResponse async def event_generator(): async for event in graph.astream({input: req.message}, configconfig): # 把事件转成 SSE 格式下发给客户端 yield fdata: {json.dumps(event, ensure_asciiFalse)}\n\n app.post(/api/agent/stream) async def run_agent_stream(req: QueryRequest): return StreamingResponse(event_generator(), media_typetext/event-stream)我前端伙伴拿到这个流接口后做了打字机效果整个交互质感提升很明显。第三个升级点是图对象的生命周期。注意我在代码里把build_graph放在模块层调用这个直觉是对的——图对象重则带模型实例、checkpointer 连接池每次请求都重建既浪费又容易爆连接。但这里有个隐患SqliteSaver.from_conn_string返回的 saver 在模块级初始化时依赖数据库连接在 FastAPI 的应用生命周期里最好放到startup事件中初始化避免单元测试或者多 worker 启动时连接冲突。3.3 服务的状态管理与并发控制实测服务化之后最容易被忽视的是状态管理。脚本模式下一个进程对应一个任务状态天然隔离服务模式下一个进程要服务几百个会话每个会话的thread_id都会打到同一个 checkpointer 上如果并发量大了数据库连接池和锁机制就得认真对待。我实测下来的常见问题是用默认的 SQLite checkpointer 时两个请求同时写同一个thread_id偶尔会报database is locked。原因很简单SQLite 默认只允许一个写者。解决方案有两个一是把 checkpointer 换成PostgresSaverPostgres 的并发写能力对这个场景完全是降维打击二是如果真的必须用 SQLite就把接口改成串行处理同一 session 的请求或者用 WAL 模式减少锁冲突。我当时为了快速验证就用了 WAL 模式也确实缓解了大部分问题但最后上生产还是切到了 PostgresSaver。另一个工程细节是限流与鉴权。服务一旦开放给外部系统就得考虑别人会不会把你的 Agent 当免费 API 刷。我用 FastAPI 的依赖注入做了一层简单的接口鉴权服务间通过 Header 里的X-API-Key验证调用来源同时在业务层对每个 session 的请求频次做控制。这一步花不了多少时间但能避免上线后被内部同事的测试脚本冲到限流。再补充一个并发量的参考值我用一台 4C8G 的云主机部署了带 Postgres checkpointer 的 FastAPI 服务单实例实测能扛住约 20 个并发请求瓶颈主要在大模型的响应延时和外部工具调用耗时框架本身没有明显瓶颈。如果你的并发预期是几百上千那就得进入第三阶段用容器横向扩容了。4. 路径三容器化与编排——生产环境的最后一百米4.1 Docker 镜像构建依赖、缓存、非 root服务化之后下一个问题通常是怎么部署到服务器手动 ssh 上去拉代码、装依赖、跑 uvicorn也不是不行但一旦涉及版本回滚、多机部署、环境一致性手工操作就崩了。容器化是绕不开的。LangGraph 项目的 Docker 镜像构建和普通 Python 项目没有本质区别但有一些Agent 项目特色要注意。第一依赖体积大——LangChain/LangGraph 生态会拉进来很多包pip 安装后的镜像很容易超过 2GB必须做多阶段构建用python:3.11-slim作为运行基础镜像而不是完整版。第二模型相关的凭据绝对不能打进镜像要通过环境变量或挂载 secret 文件在运行时注入。第三现代容器和 K8s 环境默认要求非 root 运行所以镜像里得建一个普通用户。我放一个能直接用的 Dockerfile 做参考# 构建阶段只用来装依赖不参与运行 FROM python:3.11-slim AS builder WORKDIR /app ENV PIP_NO_CACHE_DIR1 COPY requirements.txt . RUN pip install --prefix/install -r requirements.txt # 运行阶段尽量精简 FROM python:3.11-slim AS runner # 创建非 root 用户 RUN useradd --create-home appuser WORKDIR /app # 拷贝构建阶段安装好的依赖 COPY --frombuilder /install /usr/local # 先拷贝依赖文件再拷贝代码这样依赖没变时能命中构建缓存 COPY --chownappuser:appuser . . USER appuser EXPOSE 8000 HEALTHCHECK --interval30s --timeout5s --retries3 \ CMD python -c import urllib.request; urllib.request.urlopen(http://127.0.0.1:8000/healthz) CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000, --workers, 2]这里有两个细节是我反复调整后才稳定下来的。一是--prefix/install配合多阶段拷贝能把最终镜像从 2.3GB 压到 800MB 左右部署和拉取都快很多。二是COPY requirements.txt和COPY .分开写目的是让 Docker 的分层缓存生效——只要依赖不变后续代码改动构建时不会重跑 pip install实测构建时间从五分钟降到三十秒以内。另外HEALTHCHECK里踩了一个常见的坑用/healthz专门做健康检查不能被业务鉴权拦掉否则 K8s 会说容器一直不健康。4.2 对接微服务架构的几种模式项目一旦容器化就不可避免地要考虑和公司现有微服务架构怎么对接。我从实际项目中总结出三种常见模式按耦合度从低到高排列。模式一Agent 独立服务 内部 API 网关。把 Agent 服务当做一个标准微服务注册到公司的服务发现和网关里其他业务通过网关调用你的/api/agent接口。这种方式最简单也是我最推荐的因为它把 Agent 和业务系统完全解耦Agent 升级、回滚都不影响主业务。缺点是跨服务调用会有网络开销而且同步调用时延较长需要调用方设置合理超时。模式二事件驱动的异步订阅模式。如果你的 Agent 是处理工单创建用户消息到达这类事件的可以让 Agent 服务订阅消息队列Kafka/RabbitMQ消费到事件后异步处理再把结果发回队列。这种模式用户体验最好系统压力也最小但需要团队有消息中间件的基础设施LangGraph 的invoke调用会被封装在消费者回调里实现上并不复杂。模式三嵌入式依赖模式。直接把 LangGraph 图作为 Python 包打进其他服务的代码里通过函数调用而非网络调用。这种方式适合小团队快速迭代省去网络通信和一套鉴权逻辑但坏处是 Agent 的资源和主业务抢占同一进程而且 LangGraph 升级会牵动整个服务的构建长期看耦合太重。我在早期原型验证时用过这种模式后来规模大了还是拆了出来。4.3 落地部署时关于 K8s 与 Docker Compose 的取舍容器有了编排工具选什么我的建议很简单小规模、单机、快速验证用 Docker Compose大规模、多机、需要自动伸缩用 Kubernetes中间过渡阶段别急着上 K8s。为什么这么建议我见过好几个团队一上来就搭 K8s结果集群维护成本比 Agent 本身还高最后变成为了 K8s 而 K8s。如果你的 Agent 服务只有一两个实例Docker Compose 完全够用一条docker compose up -d就能把 FastAPI 服务、Postgres、Redis 一起拉起来。但如果你的服务访问量有波动或者要跑多副本、多环境K8s 的优势就体现出来了三个关键能力是 Agent 服务特别需要的水平自动伸缩HPA根据 CPU 或请求量自动扩缩 Pod 副本滚动更新新版本发布时逐个替换旧 Pod 而不中断服务自愈Pod 挂了自动重新拉起。我给 Agent 服务配上 HPA 后日常流量低谷只有 2 个副本晚高峰自动扩到 8 个非常省事。在 K8s 里部署 LangGraph 服务时有几个配置需要注意。一是请求与限额requests/limitsLLM 调用是内存和 CPU 密集的我给每个 Pod 配置了requests: cpu500m, memory1Gilimits: cpu2, memory4Gi防止某个异常请求把整机打死。二是探针配置K8s 的 readinessProbe 和 livenessProbe 最好指向/healthz这个接口只做简单的进程存活检查不做复杂逻辑否则探针本身就可能成为性能瓶颈。三是外部依赖我们用了 PostgresSaver 存储会话状态所以完整部署还需要一个 Postgres 服务在 Docker Compose 或 K8s 里都建议单独用 StatefulSet 管理避免数据丢失。5. 三条路径横向对比与决策清单5.1 一张表看清三条路径的差异我自己整理过一份对比放在这里不同阶段的技术选型照着对照就行维度脚本模式FastAPI 服务化容器化编排适合场景离线批处理、本地开发调试实时交互、API 对外开放生产环境、高并发、多副本并发能力单进程基本无并发单实例几十并发可水平扩展弹性伸缩状态持久化SQLite / 本地文件Postgres / RedisPostgres / Redis外部依赖部署速度最快无部署成本中需管理服务进程慢需构建镜像和编排资源占用用完即走最省常驻进程内存持续占用常驻但可按需伸缩运维复杂度几乎为零需要日志、存活监控需要完整的监控、告警、发布流程适合团队规模个人 / 小团队中小团队有平台或运维团队的团队几条路径不是非此即彼的关系。我见过很多项目是脚本启动服务化演进容器化收尾的顺序这其实很正常——需求是逐步明确的架构跟着需求走。5.2 我个人的决策流程每次接到一个新的 LangGraph 需求我会按下面这个次序判断第一版部署方式先问调用方是谁。如果只是自己或同事在命令行里跑或者嵌入在定时任务里那就脚本模式。如果调用方是外部系统、前端页面、IM 机器人那一步到位做服务化。再问实时性要求多高。用户当场等结果的走 FastAPI 同步接口能接受异步回调的走任务队列外加轮询接口。最后问并发和稳定性预期。一两个实例就能扛住的Docker Compose 足够流量不稳定、要按量扩缩的才需要考虑 K8s 全套方案。有一个很反直觉的经验不要一开始就用最重的方案。脚本模式跑通的业务逻辑在迁移到 FastAPI 服务时基本不用改图本身只需要把调用图的方式从函数调用改成接口调用。反过来如果你一上来就套 K8s调试成本和技术债会同时翻倍。部署方式是外部壳LangGraph 图本身才是核心壳可以随时换核心别乱动。6. 从脚本迁到服务的几点血泪经验6.1 项目结构一开始就按可服务化设计我从脚本迁到 FastAPI 的时候最大的调整不是代码而是项目目录结构。最开始写脚本时它是一个文件跑到底配置、图定义、节点函数、入口逻辑全塞在main.py里。迁服务时发现根本没法拆——图定义里混入了 I/O 操作节点函数里又直接读取了本地文件全部要重写。后来我学乖了新的 LangGraph 项目一律按这个结构组织agent_project/ ├─ app/ │ ├─ main.py # FastAPI 入口只管路由 │ ├─ graph.py # 图定义和编译逻辑 │ ├─ nodes/ # 每个节点一个模块只做单步处理 │ │ ├─ classify.py │ │ ├─ generate.py │ │ └─ tools.py │ ├─ config.py # 环境变量读取与配置校验 │ └─ persistence.py # checkpointer 连接管理 ├─ scripts/ │ ├─ run_offline.py # 保留脚本模式入口方便离线跑 │ └─ seed_data.py ├─ tests/ ├─ Dockerfile └─ requirements.txt这样设计的最大好处是一套图实现多个入口共用。scripts/run_offline.py直接from app.graph import build_graph调用同一个图FastAPI 的handlers里也是调同一个build_graph。以后要加 WebSocket 入口、加消息队列消费者都是之前图的复用不用再动图内部代码。拆的时候注意节点函数里不要写直接读写文件这类硬编码统一把外部依赖通过参数传入这样测起来也方便。6.2 日志、trace、成本要提前埋点没有 Run 过一段时间你不会意识到可观测性对 LangGraph 服务有多重要。我在脚本阶段只有一个终端输出感觉还行迁到服务化之后同时有几百个会话在跑任何一个环节报错如果没有日志定位排查问题就像大海捞针。我的经验是至少要做三层埋点第一层是每个节点的输入输出日志。LangGraph 的每个节点函数我都加了结构化日志记录节点名、thread_id、关键输入字段和输出字段这样哪一步出错、哪一步返回了异常数据一翻日志就知道。第二层是LLM 调用的独立 trace。每个 LLM 调用的模型名、输入 token、输出 token、耗时单独打一条日志。这层数据有两个用途一是诊断哪个环节最慢二是核算成本。用 GPT 类模型时token 成本不是小数没有这层日志月底账单来了你根本说不清楚花在哪。第三层是业务指标监控。比如单位时间处理的请求数、成功率、平均响应时长。服务化之后这些指标我直接通过 Prometheus 客户端暴露接入 Grafana 看板。配置起来也不难FastAPI 生态有现成的中间件十来行代码就能把 HTTP 指标暴露出去。6.3 最后再说一个做的时候最容易被忽略的点最后分享一个小细节是我踩过最深的坑生产环境的 Agent 服务必须把模型依赖和业务逻辑拆开抽象。我一开始在节点函数里直接硬编码from langchain_anthropic import ChatAnthropic模型提供商写死在代码里。后来客户要求换成国产模型整个图逻辑没变但每个节点的模型初始化全要改牵一发动全身。现在的做法是在config.py里做一个模型工厂通过环境变量指定用哪家模型再通过统一的接口获取llm实例。LangGraph 的节点函数只依赖能返回字符串的模型接口不关心底层是哪一家。这样换模型、换版本、A/B 测试都只改环境变量不需要动图逻辑。如果你要用 LangGraph 做长期项目这个抽象越早做越好。我现在的新项目基本是从脚本起步、第二天就套上 FastAPI 暴露接口、跑两周稳定后再进容器化。这条路径看起来很朴素但每一步都是有明确需求才推进的每一步也都踩过真实的坑。希望这篇东西能让你少走点弯路——至少在你被 SQLite 锁问题或者 K8s 探针问题恶心到的时候能想起这里有人替你趟过了。