很多人写 LangGraph写完脚本就停了。本地能跑通一个 agent打印出结果然后就不知道该怎么把它变成一个能被别人调用、能长期提供服务的东西。这个问题我见过太多次而且几乎所有 LangGraph 新手都会卡在同一个地方graph 写好了接下来怎么办LangGraph 本质是状态图框架核心逻辑是一条 graph但 graph 编译完之后摆在面前的选择也就三条——当脚本跑、封装成 API 服务、打包成容器丢进生产环境。这三条路就是我说的LangGraph 的三条部署路径。这篇文章不讲怎么画节点、怎么设计状态只讲编译完之后怎么上路。我会把三条路径拆开讲清楚配合可以直接抄的代码把从本地脚本到生产服务过程中最容易踩的坑都过一遍。适合刚入门 LangGraph、想快速落地一个 agent 的人也适合已经写好 demo、正愁怎么让 AI 真正下地干活的人。1. 先搞清楚LangGraph 部署到底在部署什么聊部署之前得先把 LangGraph 的运行机制彻底理清。很多人一开始就把部署想复杂了觉得要搞 Kubernetes、要搞消息队列、要搞一堆中间件其实根本没那么玄乎。1.1 图的编译产物就是一个可调用对象LangGraph 的核心抽象是 StateGraph。你定义一个状态结构State往图里挂节点add_node用边把节点串起来add_edge最后调用 compile() 得到一个可执行对象。这个对象在 Python 里就是个普通 callable你可以同步调用它invoke、异步调用它ainvoke、也可以流式拿它吐出来的中间结果stream / astream。用个不太严谨但好懂的生活类比graph 是地图compile 是请了个司机invoke 是发车。每次 invoke 你给一个初始 state车跑完所有节点把最终的 state 带回来。部署的本质就是把这个 compile() 出来的 object 放到一个能被外部触达的位置并且保证它在被反复触发时状态正确、失败了能恢复、行为可观测。所以先别急着想我要用什么高端的部署技术先回答三个问题我的 graph 到底要被谁触发人敲命令、定时器、还是 HTTP 请求运行中的临时状态放哪里进程内存、Redis、还是数据库服务挂了谁管自己重启、容器重启、还是编排平台自动拉起这三个问题想清楚部署路径基本就定型了。1.2 三条路径的差异本质是触发方式不同脚本部署触发方式是人或者 cron运行完就退出了生命周期是一次性的。API 服务部署触发方式是 HTTP 请求进程常驻内存生命周期是持续待命的。容器化部署触发方式依然是 HTTP 或事件但多了副本数、扩缩容、健康检查这些生产属性生命周期是由平台托管的。触发方式一换后面所有东西都跟着换状态放哪里、日志怎么记、并发怎么扛、密钥怎么管理、故障怎么恢复全部是连锁反应。这也是为什么我强烈建议不要在项目一开始就选第三条路径。你连 agent 的逻辑都没在真实场景里跑过直接上容器和平台等于一边学开车一边上高速出问题了连排查都不知道从哪排查。1.3 这篇文章的三条路径分别是什么路径一裸脚本直跑。最轻量开发调试、定时任务、CI 冒烟测试的主力军。路径二FastAPI 封装成服务。让 agent 变成后端接口能被前端、机器人、其他微服务调用。路径三容器化与平台托管。Docker 镜像 编排 持久化存储生产环境的标准形态。下面逐条展开。2. 三条路径怎么选先看触发方式再看运维成本我见过很多团队一上来就照着网上模板搭了一套 Docker Kubernetes结果 agent 逻辑还天天改镜像构建一次要几分钟改一行代码等半天才能看到效果团队士气直接崩掉。选部署路径不是越高级越好是越匹配当前阶段越好。2.1 一张表看懂三条路径的差异对比维度路径一裸脚本路径二API 服务路径三容器化/平台触发方式命令行 / cron / CIHTTP 请求HTTP 负载均衡启动成本秒级python xxx.py就能跑秒级启动一个 uvicorn分钟级镜像构建 编排拉起并发能力基本没有单进程串行有异步 多进程强水平扩容状态持久化只有单次运行的内存内存可选接 DB必须接外部存储运维成本极低中等进程守护即可较高镜像、编排、监控适合场景开发调试、每天跑一次的批量任务内部系统调用、微服务之间的 agent 服务对外提供 SLA、需要扩缩容的线上服务看这张表你会发现路径一和路径三的差距是全方位的。触发方式从人主动敲变成系统被动调状态管理从内存用完即走变成持久化可恢复运维从不用管变成必须管。2.2 一个简单的选型判断逻辑如果你只是自己在本地验证 graph 逻辑或者要跑一个每天固定执行的批量任务选路径一。如果你的 agent 要提供给其他人调用不管是公司内部系统、企业微信机器人、还是前端页面要接选路径二。如果你的服务要 7x24 小时稳定在线、请求量会波动、需要自动化扩缩容选路径三。注意这是一个递进关系。绝大多数项目的正确路径是 1 到 2 到 3不要跳级。我见过最舒服的节奏是用路径一跑通逻辑用路径二验证接口最后根据真实流量再决定值不值得上路径三。2.3 项目阶段演进路径也要跟着换这里说个比较扎心的经验。很多做 AI 应用的项目死在第二阶段之前。为什么因为 agent 逻辑没验证清楚就急着上生产架构结果一半时间在调 deployment一半时间在改 prompt真正用在打磨 agent 效果上的精力微乎其微。反过来的节奏是第一天就用脚本跑通一个端到端的 agent第三天封装成 FastAPI 接口给同事试用第二周才考虑容器化和自动部署。每一步都有产出每一步都不浪费。3. 路径一裸脚本直跑最轻量的部署形态先说最常被忽略但其实最实用的路径。很多人看不起脚本觉得它太简陋但恰恰是脚本模式帮我把最多 agent 项目跑通验证了。3.1 一个最小可运行的 LangGraph 脚本假设你要写一个简单的问答 agent。你不用 LangGraph 也能做但用了 LangGraph 的好处是这个问答逻辑以后可以随时扩展成多节点、有工具调用的复杂 agent部署路径却不用变。下面这个脚本是最小可运行版本我特意把 LLM 调用占位了方便你专注看 graph 结构from typing import TypedDict from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): question: str answer: str def generate_node(state: AgentState) - dict: # 真实场景这里会调用 LLM例如 # from langchain_openai import ChatOpenAI # llm ChatOpenAI(modelgpt-4o-mini) # question state[question] # answer llm.invoke(f请回答{question}).content return {answer: f针对『{state[question]}』的回答占位} graph StateGraph(AgentState) graph.add_node(generate, generate_node) graph.add_edge(START, generate) graph.add_edge(generate, END) app graph.compile() if __name__ __main__: result app.invoke({question: 帮我写个周报模板}) print(result[answer])保存成my_agent.py命令行里直接python my_agent.py结果就出来了。注意几个细节graph 只是一个地图compile()之后才是能跑的执行器。以后你会碰到有人把 graph 和 app 混着叫无所谓知道它们是两个阶段就行。每个节点函数接收当前 state返回一个 dict 表示要更新 state 的哪些字段。invoke()是同步阻塞的适合脚本模式。后面服务模式会用异步版本。3.2 把脚本接进定时任务cron 和日志脚本最典型的真实场景是定时任务比如每天早上 9 点自动跑一个 agent 整理日报。Linux 上直接用 crontab 就能搞定0 9 * * * cd /opt/my-agent /usr/bin/python3 my_agent.py /var/log/agent.log 21这里头有个坑我要特别提醒一下千万不要在 crontab 里直接写python。因为 cron 执行时的 PATH 环境变量和交互式 shell 不一样经常会出现我本地能跑cron 里报找不到命令的诡异问题。稳妥的做法是which python3拿到绝对路径后写进 cron。同理如果脚本依赖.env文件脚本里要用load_dotenv()读取不然 cron 环境下环境变量也是空的。日志方面别用print糊弄事。脚本是没人盯着看的出了问题只能靠日志定位。建议用标准库loggingimport logging logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) logger logging.getLogger(__name__) logger.info(agent started) result app.invoke(...) logger.info(agent done, status%s, result)这样日志里能明确看到什么时候开始、什么时候结束配合 cron 的重定向问题定位效率能提升一个量级。3.3 脚本模式最容易踩的四个坑第一个坑API Key 硬编码。脚本模式最容易图方便把 key 直接写在代码里。一旦这个脚本被分享、被上传到仓库密钥就泄露了。正确做法是用.env文件 python-dotenv或者直接用系统环境变量。第二个坑幂等性。定时任务每天跑一次结果应该是可重复的、可预期的。如果 agent 内部有随机性或者依赖外部状态你得考虑这次跑挂了对下次有没有影响。别让定时任务变成定时炸弹。第三个坑重试和异常处理。LLM 调用经常不稳定超时有、限流有、返回格式错误也有。脚本里必须用 try-except 包裹外部调用加上简单的指数退避重试。不然半夜 3 点的 agent 崩一次第二天打开日志一脸懵。第四个坑并发能力为零。脚本是单进程串行执行的如果某天你批量处理 1000 条数据别直接在脚本里写 for 循环逐个调用那样会很慢而且容易触发限流。要么用 asyncio 包一层并发要么干脆换路径二做成服务后用任务队列。4. 路径二FastAPI 封装Agent 从此变成服务脚本跑通之后下一步自然就是把 agent 暴露成接口。只要你的 agent 要被别人调用无论调用方是前端页面、企业微信机器人、还是另一个后端服务路径二就是绕不过去的一环。4.1 为什么是 FastAPI 而不是 Flask这个问题我回答过几十次。核心原因只有一个异步。LangGraph 支持ainvoke和astream而 LLM 调用本质是极度 IO 密集的操作一个请求可能在等待模型响应时占着线程好几十秒。如果用 Flask 这种同步框架一个慢请求就能拖垮整个服务。FastAPI 原生支持 async/await配合 uvicorn 这种 ASGI 服务器能在单进程里高效处理大量并发请求。另外FastAPI 自带 OpenAPI 文档访问/docs就能看到接口文档再加上 Pydantic 做参数校验做微服务之间的接口约定非常方便。LangChain 生态和它也是天然搭配很多官方模板就是 FastAPI 写的。4.2 一个最小可用的 Agent API基于上一节的 graph我们包一层 FastAPIfrom typing import TypedDict, Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): question: str answer: str def generate_node(state: AgentState) - dict: return {answer: f针对『{state[question]}』的在线回答占位} graph StateGraph(AgentState) graph.add_node(generate, generate_node) graph.add_edge(START, generate) graph.add_edge(generate, END) agent graph.compile() app FastAPI(titleLangGraph Agent API) class AskRequest(BaseModel): question: str thread_id: Optional[str] None app.post(/ask) async def ask(req: AskRequest): result await agent.ainvoke({question: req.question}) return {answer: result[answer]}启动命令uvicorn main:app --host 0.0.0.0 --port 8000这里我加了一个thread_id字段虽然上面的代码暂时没用到它但它非常关键后面并发那一节会重点讲。这个字段的用途是标识一个会话同一个 thread_id 的多次请求应该共享同一个对话上下文。跑起来之后用 curl 验证一下curl -X POST http://localhost:8000/ask \ -H Content-Type: application/json \ -d {question: 帮我写个请假邮件}返回{answer: 针对『帮我写个请假邮件』的在线回答占位}服务就算通了。4.3 流式输出让回答像 ChatGPT 一样逐字出现如果 agent 的响应很长一次性返回会等得很难受尤其是接前端页面的时候。LangGraph 天然支持流式输出FastAPI 里用StreamingResponse就能把它转成 SSEServer-Sent Eventsfrom fastapi.responses import StreamingResponse app.post(/ask/stream) async def ask_stream(req: AskRequest): async def event_stream(): async for chunk in agent.astream({question: req.question}, stream_modevalues): answer chunk.get(answer, ) if answer: yield fdata: {answer}\n\n return StreamingResponse(event_stream(), media_typetext/event-stream)stream_modevalues表示在每个节点执行完成后把当时的完整 state 吐出来。这样前端可以根据事件流逐渐渲染文本体感上就是逐字输出。注意如果 agent 里加了工具调用流式输出时会有多种事件类型比如工具调用的中间结果、最终的完整响应建议一开始只用stream_modevalues拿最终的 answer 字段把交互逻辑跑通了再考虑细化事件类型。4.4 把 Agent 服务融进微服务架构很多公司现在都是微服务架构Agent 服务在里面应该扮演什么角色我的经验是它就应该是一个只干一件事的大脑服务。上游业务系统把用户的问题发过来Agent 服务负责拆解、调用工具、组织答案然后把结果返回。上游不需要知道你图里画了多少个节点、用了什么模型只需要约定一个输入输出协议。这里有几个必要的生产级配置我用伪代码给出# 简单的 API Key 鉴权 API_KEY os.getenv(AGENT_API_KEY, ) app.middleware(http) async def check_api_key(request, call_next): if request.url.path.startswith(/health): return await call_next(request) key request.headers.get(X-API-Key) if key ! API_KEY: return JSONResponse(status_code401, content{detail: unauthorized}) return await call_next(request)再加上/health健康检查接口给后面的 Docker 和负载均衡用app.get(/health) async def health(): return {status: ok}在微服务架构里这个 Agent 服务不应该直接暴露到公网而是放在内网由 API 网关统一转发和鉴权。服务内部只信任来自网关的流量Apple 和 Orange 分清楚。5. 路径三容器化与平台托管生产环境的最终归宿当服务需要 7x24 稳定运行请求量会有波动需要自动扩缩容、快速回滚路径三就来了。容器化不是目的可复制性和可运维性才是目的。5.1 Dockerfile 与依赖锁定先看一个标准的 DockerfileFROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY src ./src COPY main.py . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]这里头有个被我反复强调的点requirements.txt 一定要锁定版本。LangGraph 的 API 迭代很快今天能跑的代码一个月后可能因为依赖升级就报错了。锁版本的写法是pip freeze requirements.lock然后requirements.txt里写的就是钉死的版本号。别用langgraph0.1.0这种宽松写法迟早有一天它会自动升到一个不兼容的版本然后莫名奇妙地挂掉。5.2 状态持久化checkpointer 和外部存储路径一和路径二在开发阶段graph 的状态都存在内存里服务一重启全没了。生产环境必须把重要的会话状态持久化。LangGraph 里的概念叫 checkpointer它的作用是在每一步节点执行后把状态记录到外部存储。开发阶段可以先用内存版from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() agent graph.compile(checkpointermemory)但生产环境就必须换成数据库版本常见的是 PostgreSQLfrom langgraph.checkpoint.postgres import PostgresSaver conn_string os.getenv(DB_URL, postgresql://postgres:postgrespostgres:5432/agent) with PostgresSaver.from_conn_string(conn_string) as checkpointer: agent graph.compile(checkpointercheckpointer)注意一旦启用了 checkpointer你调用 graph 时就一定要带上thread_idconfig {configurable: {thread_id: user-001-session-a}} result await agent.ainvoke({question: 继续上一次对话}, configconfig)这也是我在路径二里留了thread_id字段的原因。没有 thread_idcheckpointer 不知道把状态存到哪个会话里并发一上来就全乱套。docker-compose 里把 PostgreSQL 并起来Agent 服务和数据库分离这样服务重启、扩容都不丢状态。5.3 健康检查、日志与可观测性容器化之后编排平台比如 Docker Swarm 或 Kubernetes需要知道服务是否健康这就要靠健康检查。/health接口在前面已经写过了在 docker-compose 里配置healthcheck: test: [CMD, curl, -f, http://localhost:8000/health] interval: 30s timeout: 10s retries: 3日志方面容器环境下不要写本地文件要输出到 stdout/stderr让 Docker 日志驱动统一收集。关键请求要打结构化日志import json, logging logger logging.getLogger(agent) logger.info(json.dumps({ event: request_start, thread_id: req.thread_id, question_len: len(req.question), ts: datetime.utcnow().isoformat(), }))另外LangGraph 生态里有个官方的可观测性工具 LangSmith做 agent 链路追踪特别好用。它能完整记录每次运行的节点顺序、每一步耗时、token 消耗、模型输出排查问题时比盲猜容易太多。如果是生产环境我强烈建议接上。5.4 官方平台和自托管怎么取舍LangGraph 官方也提供了一条完整的部署链路从本地开发服务器langgraph dev到托管平台。官方平台的好处是省运维你只管写 graph平台帮你管伸缩、监控、版本管理。缺点是成本高而且有平台锁定风险。我的态度是中小团队、想做快速验证的项目直接用官方平台是很好的选择。但如果你所在的公司已经有成熟的容器平台和运维体系那用路径三自己做容器化往往更可控。取舍的关键因素是团队里有没有愿意背运维责任的人而不是技术本身谁更高级。6. 部署实战中的常见问题与排障记录最后这一节把我在实际部署中遇到过的真实问题和排查方法整理成速查表。这些问题你迟早会遇到提前知道能省很多时间。6.1 服务起不来的排查顺序现象可能原因排查方法uvicorn启动报端口被占用8000 端口已被其他进程占用lsof -i :8000查占用进程uvicorn main:app --port 8001换端口启动成功但/ask返回 500依赖缺失或环境变量未加载看日志栈pip list核对依赖确认.env路径Docker 构建时 pip 安装失败网络问题或版本冲突换镜像源锁定版本删除本地缓存重建容器起一下就退出CMD 写错了或主模块路径不对docker logs container看报错检查main:app是否对应服务启动了但请求一直转圈外部 API 超时看日志是否有超时异常增大 timeout配置重试这里有个通用的排查顺序先看日志确认是启动阶段报错还是运行阶段报错。启动阶段报错优先检查依赖、路径、环境变量。运行阶段报错优先检查外部服务模型 API、数据库连通性和资源限制。6.2 并发状态串台最隐蔽的一个坑路径二上线后最经典的事故就是用户 A 问了个问题返回的却是用户 B 的答案。原因很简单——多个请求同时在内存里操作同一个全局 state尤其是用了一些模块级变量缓存中间结果的时候状态就串了。解决办法就是两条铁律每个请求必须带thread_id用 checkpointer 做会话隔离。graph 内部的节点函数不要依赖任何外部全局变量。所有需要传递的数据都放进 state让 state 成为唯一的共享渠道。第一点最好理解加上thread_id就能隔离。第二点很多人会忽略如果你在节点函数里用了一个全局 list 缓存了一些东西并发请求一多这个 list 就成了共享的可变状态必炸。6.3 LLM 调用超时和限流生产环境最常见的异常来源就是模型 API 本身不稳定或者是单条 prompt 太长导致响应慢。应对策略我总结为三个字等、退、换。等把 SDK 调用的 timeout 调大别用默认的 30 秒特别是复杂 agent 多轮调用工具时我见过单请求跑 2 分钟都正常。退指数退避重试第一次失败等 1 秒第二次等 2 秒第三次等 4 秒最多重试 3 次。换配置模型 fallback主力模型失败后自动切到备用模型。LangChain 里可以用ChatOpenAI之外的其他模型类组合确保单点故障不至于全挂。6.4 安全和密钥管理写脚本的时候密钥硬编码顶多算懒部署成服务之后密钥泄露就是安全事件了。几条基本规范密钥统一放环境变量不在代码、日志、docker-compose 文件里出现明文。生产环境的密钥交给密钥管理服务比如云厂商的 Secret Manager托管启动时注入环境变量。给不同的环境用不同的 key开发、测试、生产全隔离开。监控 API 调用量出现异常激增大概率是 key 被盗用了。最后说点我自己的体会。很多项目死在没想清楚部署形态就动手。你花一周写了个复杂的多 agent 编排最后只是为了在本地跑一个 print那显然不值当。反过来一上来就搞容器化和编排平台也会被运维问题拖垮节奏。我现在的习惯是先写裸脚本跑通逻辑确认 graph 的状态流转和响应质量再用 FastAPI 包一层让同事和业务方通过接口调用验证真实场景最后确认要对外提供稳定服务、需要水平扩容时再走容器化。这条路径看着多走几步实际上每步都有明确产出而且踩坑成本是最低的。如果读到这里你正卡在写完了 demo 但不知道怎么上线对照这三条路径逐个试你一定能找到适合自己的那一条。