把 LangGraph 的 Agent 从本地脚本挪到生产环境这一步我见过太多人卡住了。明明python agent.py跑得好好的一接真实请求就各种幺蛾子状态丢了、并发串了、工具调用超时了、用户等得想骂人了。LangGraph 的部署路径这件事看起来是个运维话题实际上决定的是整个 Agent 项目的架构走向。这篇就基于我用 FastAPI LangChain LangGraph 做 AI Agent 的实战经历聊三条从脚本到服务的部署路径直接脚本跑、自建 HTTP 服务、上托管平台。每条路我都给出可复现的代码和配置再说说它们各自的边界和适用场景。1. 为什么聊部署脚本里能跑的 Agent离生产还差多远1.1 “能跑”和“能用”之间隔着什么很多人对 LangGraph 的第一印象停留在“写一个状态图然后.invoke()一把梭”。确实你可以在 Notebook 里定义一个StateGraph塞进去几个节点和条件边把工具调用跑通最后看着它一步步回答用户问题感觉整个 Agent 已经完成了。但生产环境不是这么回事。脚本模式是“进程内调用一次”所有状态都活在内存里进程退出就什么都没有。服务模式则要面对并发请求、失败重试、长连接超时、权限隔离、日志追踪、流量波动这些事。LangGraph 本身提供了一套很好的状态管理抽象比如Checkpointer、thread_id、recursion_limit但这些能力只有在正确的部署形态下才能真正发挥出来。我见过一个项目团队花了两周把 Agent 流程调得漂漂亮亮然后直接用一个 Flask 接口包起来上线结果第二天就出问题用户 A 的对话上下文跑到了用户 B 的会话里。原因很简单Agent 的状态存在全局变量里根本没有按会话隔离。这不是 LangGraph 的锅是部署时没把状态持久化设计进去。所以聊部署路径本质上是聊你怎么管理 Agent 的状态、并发、生命周期和可观测性。1.2 三条部署路径的定位这三条路径不是互相替代的关系而是解决不同阶段的问题脚本化运行面向开发调试和离线批处理成本最低适合验证流程。FastAPI 自建服务面向私有化部署和定制接口自己掌控一切灵活性最高。托管平台面向生产级多 Agent 场景平台帮你解决存储、监控、人审等问题。我先把三条路的关键指标列个表方便你心里有个底维度脚本运行FastAPI 自建服务托管平台上手成本很低中等中等偏高并发能力几乎没有自己控制平台负责伸缩状态持久化进程重启即丢自接 SQLite/Postgres内置存储可观测性靠 print 和日志自己埋点内置追踪与监控适合场景本地调试、定时批处理私有化交付、定制 APISaaS、多人协作、人审下面一条一条拆开讲。2. 路径一脚本化运行搞定开发和批处理2.1 最小可用的 LangGraph 脚本长什么样不管最后走哪条部署路脚本跑通是第一步。给你看一个最小但完整的例子这个结构我用了很久所有复杂 Agent 都是从这里长出来的from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode, tools_condition from langchain_openai import ChatOpenAI from langchain_core.tools import tool class AgentState(TypedDict): messages: Annotated[list, append] next_step: str tool def get_weather(city: str) - str: 查询指定城市的实时天气 # 实际项目里替换成天气 API 调用 return f{city} 晴气温 26 摄氏度 tools [get_weather] llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools) def agent_node(state: AgentState): return {messages: [llm_with_tools.invoke(state[messages])]} graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, ToolNode(tools)) graph.add_edge(START, agent) graph.add_conditional_edges(agent, tools_condition, {tools: tools, END: END}) graph.add_edge(tools, agent) app graph.compile() result app.invoke({messages: [(human, 杭州天气怎么样)]}) print(result[messages][-1].content)这个脚本的关键在于两条边tools_condition判断模型输出里有没有 tool_call有就进工具节点没有就直接走向 END。这就是 LangGraph 里最经典的 ReAct 循环。Annotated[list, append]保证每轮messages都会追加到已有状态里而不是覆盖掉。跑这个脚本的时候哪个节点被调用了、模型输出了什么、工具返回了什么全都直接打到终端里。这个体验对排查逻辑问题非常友好比任何 Service 模式都直观。2.2 把脚本当“调度任务”用别急着说脚本模式没用它在批处理场景里非常好使。比如说你要做一个每天早上 9 点自动汇总竞品信息的 Agent脚本再合适不过了。做法也简单把 LangGraph 的app.invoke()包在一个函数里然后用 cron 或 APScheduler 定时触发。这里有个经验脚本的输入不要写死从命令行参数和环境变量读。我一般会这样做import argparse import os import json parser argparse.ArgumentParser() parser.add_argument(--task, requiredTrue) args parser.parse_args() # 从环境变量读取 API Key不要在代码里硬编码 os.environ.get(OPENAI_API_KEY) result app.invoke({ messages: [(human, f执行任务{args.task})], next_step: plan }) with open(foutput_{args.task}.json, w) as f: json.dump(result, f, ensure_asciiFalse, indent2)这样做的价值在于Agent 跑批处理任务的时候可以复用同一份 graph 定义换任务只换参数。我还习惯把每次调用的输入输出落盘后面调整 prompt 或工具时可以拿历史数据做回归对比比人脑记忆靠谱多了。2.3 脚本模式的三条边界但脚本模式有三个绕不开的硬伤决定它不能直接当服务用第一状态跨请求无法保留。脚本每次运行都是新进程MemorySaver里的记录全没了。你没法让用户说一句“刚才那个问题再解释详细一点”Agent 不知道“刚才”是什么。第二没有并发隔离。一旦同时进来两个请求脚本只会按顺序处理一个另一个排队排到天荒地老。就算你用多线程硬顶共享状态会互相污染出问题的时候极难排查。第三进程退出即丢。没有崩溃恢复、没有持久化机器重启一下所有运行中的 Agent 任务直接蒸发。所以我的结论是脚本模式是调试器和批处理工具不是服务。你要让 AI Agent 真正“下地干活”至少得走到第二条路。3. 路径二用 FastAPI 把 Agent 包成 HTTP 服务3.1 为什么挑 FastAPI 而不是 Flask把 Agent 暴露成 HTTP 接口框架选择上我强烈建议 FastAPI。不是说 Flask 不行而是 FastAPI 的异步支持、类型校验和自动文档这三个特性和 LangGraph 的异步 API 简直绝配。LangGraph 提供了ainvoke、astream_events这一整套异步方法你拿 Flask 的同步模型去对接Thread 调度会浪费掉大量 IO 等待时间。FastAPI 的async def直接把 event loop 打通了Agent 在等 LLM 返回的时候同一个进程还能处理其他请求。而且 FastAPI 自带 OpenAPI 文档接口调试不用另外装 Postman 之外的工具浏览器打开/docs就能直接试。这对联调阶段的帮助特别大。看一个最小实现from fastapi import FastAPI from pydantic import BaseModel from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph # 复用第 2 节里的 graph 定义 app FastAPI() # 编译时挂上 checkpointer checkpointer MemorySaver() graph_app graph.compile(checkpointercheckpointer) class ChatRequest(BaseModel): session_id: str message: str class ChatResponse(BaseModel): reply: str app.post(/chat, response_modelChatResponse) async def chat(req: ChatRequest): config {configurable: {thread_id: req.session_id}} result await graph_app.ainvoke( {messages: [(human, req.message)]}, config ) return ChatResponse(replyresult[messages][-1].content)和脚本模式最大的区别就在checkpointer和thread_id。thread_id是会话的身份证同一个 session 的请求会共享历史状态不同 session 天然隔离。用户 A 的消息永远走 A 的线程B 的线程不会串。3.2 状态管理从 MemorySaver 到 SqliteSaver上面代码里用的是MemorySaver这玩意在服务模式下只适合开发和压测因为它把状态存在内存里进程一重启全没了。生产环境至少要换成SqliteSaver。from langgraph.checkpoint.sqlite import SqliteSaver # 注意 from_conn_string 返回的是一个上下文管理器 with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: graph_app graph.compile(checkpointercheckpointer)用 SQLite 的好处是单文件、零运维小规模部署完全够用。但如果你有多个 uvicorn workerSQLite 的并发写会有锁竞争问题。这种情况我建议直接用 PostgreSQLLangGraph 官方提供了langgraph-checkpoint-postgres包用法差不多from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver async with AsyncPostgresSaver.from_conn_string( postgresql://user:passlocalhost:5432/agent ) as checkpointer: graph_app graph.compile(checkpointercheckpointer)这里有个实际教训graph.compile()的时机很重要。很多人在模块 import 的时候就把 graph 编译了checkpointer 也随之初始化。如果这时数据库还没准备好或者之后要切换数据库配置你就得重启整个服务。我的做法是把编译逻辑放到 FastAPI 的 lifespan 钩子里启动时再初始化配置改起来方便得多。3.3 核心接口与工具调用的透传把 Agent 包成服务之后接口设计就直接影响使用方的体验。我总结一个最简单的接口契约session_id负责状态隔离message负责用户输入剩下的都交给 Agent 自己判断。不要在设计接口的时候把“调用哪个工具”暴露给调用方否则你会陷入无休止的参数适配里。但工具调用本身确实需要透传一些信息比如用户在前端上传了一个文件或者给了经纬度坐标。我的做法是给消息内容做结构化包装而不是往messages列表里塞一个纯字符串class ChatRequest(BaseModel): session_id: str message: str metadata: dict {} app.post(/chat) async def chat(req: ChatRequest): user_message req.message if req.metadata: user_message f{req.message}\n附加信息{json.dumps(req.metadata, ensure_asciiFalse)}说的直白一点接口层只负责“接住”输入和“转交”给 Agent具体工具调用的路由逻辑LangGraph 的ToolNode和条件边已经处理好了。3.4 流式响应和长任务Agent 类接口最容易被吐槽的点就是“慢”。这不是接口实现的问题是 LLM 首 token 延迟加上工具调用往返时间天然就高。如果你用普通的await graph_app.ainvoke()用户会看到请求转圈十几秒体验非常差。解决办法是流式输出。FastAPI 配合 SSEServer-Sent Events可以做到用户侧像打字机一样逐字看到输出from fastapi.responses import StreamingResponse import json app.post(/chat/stream) async def chat_stream(req: ChatRequest): config {configurable: {thread_id: req.session_id}} async def event_generator(): async for event in graph_app.astream_events( {messages: [(human, req.message)]}, configconfig, versionv2 ): if event[event] on_chat_model_stream: chunk event[data][chunk].content if chunk: yield fdata: {chunk}\n\n elif event[event] on_tool_start: yield fdata: {json.dumps({tool: event[name], status: start})}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这样做还有一个额外收益你可以把工具调用的中间过程也推给前端像“正在查询天气接口…”这类状态提示用户就知道 Agent 在干活而不是卡死了。如果你不想搞流式也不想让调用方长时间占着 HTTP 连接那就把请求丢进消息队列比如 Redis Stream 或 Celery然后提供“任务提交”和“任务查询”两个接口。不过这种模式的实时性差一些适合后台异步任务。3.5 部署细节uvicorn、超时、健康检查FastAPI 服务本身部署起来不复杂但有几个参数容易踩坑。uvicorn 启动时--workers大于 1 的时候注意每个 worker 进程会各自初始化一份 checkpointer。如果用的是 SQLite 文件多进程并发写会报database is locked。我实际部署时小项目单 worker 加 SQLite 就够了一旦需要多 worker直接换 Postgres 存储别在 SQLite 上死撑。uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2 --timeout-keep-alive 60--timeout-keep-alive这个参数非常容易被忽视。默认值 5 秒会导致长连接在 Agent 处理过程中被断开。我一开始没调这个参数前端经常报网络错误排查了很久才发现是 keep-alive 超时。再一个一定要加健康检查接口。别小看这个K8s 或 Docker Compose 的探针都依赖它app.get(/healthz) async def healthz(): return {status: ok}4. 路径三托管平台与生产级 Agent 服务4.1 托管平台解决了哪些事第三条路是直接把 Agent 部署到 LangGraph 的官方托管平台LangGraph Platform 那一套。注意我不是推荐大家立刻迁移而是告诉你什么时候值得考虑。你自建 FastAPI 服务状态、监控、权限、多 Agent 编排这些事全都得自己扛。托管平台把这些全部内置了状态持久化不用自己配数据库创作者可以通过平台管理多个 agent每个请求都自带追踪日志还提供人工审核节点给关键操作加一道闸。我对托管平台最看重的其实是“人工介入”这件事。LangGraph 本身支持interrupt机制在执行到某个节点前暂停等人确认后再继续。这个能力在自建服务里要实现需要你自己设计挂起状态、通知渠道、恢复接口工作量不小。托管平台直接把这个做成平台级能力业务方只需要在 graph 里插入一个 interrupt 节点。4.2 接入一个托管平台要改什么很多人以为上托管平台要重写代码其实不用。Graph 定义还是本地那份代码平台只是负责把代码跑起来只是在接入方式上有一些约定。第一步在项目根目录放一个langgraph.json配置文件{ dependencies: [requirements.txt], graphs: { agent: ./src/agent/graph.py:build_graph }, env: .env }第二步确保graph.py里有一个build_graph函数返回编译后的 graph 或者 builder 对象。注意不要在这里直接初始化数据库连接平台会注入自己的存储方案。def build_graph(): graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, ToolNode(tools)) graph.add_edge(START, agent) graph.add_conditional_edges(agent, tools_condition, {tools: tools, END: END}) graph.add_edge(tools, agent) return graph.compile()接入平台之后平台会为每个部署的环境生成一个 API 地址。你的调用方还是通过 HTTP 请求访问只不过负载均衡、版本回滚、日志检索这些都有人管了。4.3 选择托管平台的判断标准我见过不少团队一上来就上托管平台结果发现运维是省了但定制化需求一大堆平台反而成了限制。所以我把判断标准提炼成四条满足两条以上再考虑你需要同时维护很多个 Agent每个 Agent 的 prompt 和工具都不一样自己写管理后台太累。你的 Agent 流程里有需要人工确认的环节比如交易确认、内容发布前审核。你没有专职运维也不想每天盯着日志看有没有报错。你的业务需要外部客户直接访问 Agent API而不是只在内网用。反过来如果只是公司内部一个辅助工具日请求量几百那托管平台属于过度设计自建服务更轻量。5. 三条路怎么选我的选型清单5.1 四个问题解决选择困难我在带项目的时候最常被问的就是“到底选哪条路”。我不会直接给答案而是抛四个问题第一个问题用户是谁如果是内部工具自建一个 FastAPI 服务绰绰有余如果是外部客户要考虑 API 稳定性、鉴权、限流托管平台可能更省心。第二个问题QPS 多少单机脚本顶多撑个位数并发FastAPI 单实例能撑几十到一两百左右再往上要么堆 worker要么上平台自动扩容。第三个问题状态要不要长期保存用户关了浏览器下次回来还要能继续对话那就必须上数据库持久化。MemorySaver 没法满足。第四个问题团队有没有人盯运维没人盯就选托管平台有人盯就自建。这四个问题组合起来答案基本就清晰了。5.2 组合使用效果更好三条路径不是非此即彼我实际项目中经常组合着来。新功能开发阶段先在脚本里把 graph 跑通把工具调用、状态流转这些逻辑调对。然后封装成 FastAPI 服务部署到测试环境让业务方体验。等某个 Agent 的流程相对稳定、需要正式上线了把纯逻辑部分搬到托管平台跑FastAPI 层保留内部管理接口和批处理任务。有个项目我印象很深一个是内部数据汇总 Agent要求每天定时跑这种始终留在脚本和 cron 里稳定又省钱另一个是客户咨询 Agent需要保存历史会话、支持人工介入后来就迁到了托管平台。自建的 FastAPI 服务作为导流层把所有 Agent 的入口统一管理。这种分层设计到现在运行得都很稳。6. 常见问题与排查技巧实录6.1 Agent 陷入工具调用死循环症状是日志里 agent 节点和 tools 节点来回交替输出永远不走向 END。排查方法很直接先确认tools_condition的条件映射是否正确再看模型是不是反复生成同一个 tool_call。我踩过的坑是工具返回的内容没有让模型“满意”模型就一直尝试调用工具直到逼近recursion_limit报错退出。解决方式是给工具加上详细的 docstring返回结果尽量结构化比如直接返回{status: ok, data: ...}模型拿到之后更倾向于整理答案而不是再次调用。还可以在编译时显式设置app graph.compile(checkpointercheckpointer, interrupt_before[tools])或用默认的recursion_limit兜底但不要一刀切调太高不然死循环的请求会拖着资源不放。6.2 并发一上来状态就串线这个我前面提过最普遍的原因就是没有按thread_id隔离状态。另一个隐蔽原因是你把checkpointer写成了模块级单例但不同请求复用了同一个 config 对象。我的排查习惯是给每个请求打一个唯一的session_id并且在日志里带上它logger.info(session%s user_input%s, req.session_id, req.message)一旦出现串线通过 session_id 能立刻定位是哪两个请求互相污染了。6.3 工具调用结果解析失败症状是模型生成了 tool_call但ToolNode执行时报参数缺失或格式错误。大多数情况是因为工具的参数 schema 和实际实现不一致。我一般会做两层防护第一层工具函数的 docstring写清楚每个参数含义模型是根据这个来生成参数的第二层工具内部做容错遇到异常不直接抛死而是把错误信息返回给模型tool def query_order(order_id: str) - str: 根据订单号查询订单状态参数 order_id 是字符串类型的订单编号。 if not order_id.isdigit(): return 订单号格式错误请确认后重试 return 订单已发货这样模型看到错误信息后会自己修正参数而不是整个流程崩溃。6.4 部署后响应慢甚至 504我遇到过几次这种情况一开始以为是 LangGraph 的问题后来发现是网关层超时设置太短。Agent 一个完整流程要经过 LLM、工具调用、再让 LLM 总结耗时十几秒很正常。排查顺序是先量 LLM 调用耗时再量工具调用耗时最后看 HTTP 层有没有提前断开。解决方式就三选一做流式输出把等待感降下来调大网关超时把任务改造成异步队列。除此之外把不必要的工具调用去掉或者用更快的模型版本也能明显缩短链路耗时。症状可能原因处理方式整体响应慢LLM 调用耗时高换模型、精简上下文某一步特别慢工具调用外部 API 慢给工具加超时和缓存客户端报 504代理层超时太短调整网关 timeout 或开启流式并发高时变慢SQLite 锁竞争换 Postgres 存储6.5 进程重启后历史会话丢失症状是用户第二天回来发现自己和 Agent 的对话记录没了。这不用犹豫就是MemorySaver导致的。只要换了 Postgres 或 SQLite 的 checkpoint数据才会真正落盘。这个知识点最简单但也是生产事故最多发的点。我的建议是一旦决定对外提供服务第一时间把 MemorySaver 换掉哪怕先用 SQLite 文件顶着比内存强一百倍。最后说点个人体会。部署 LangGraph 这件事我踩过最大的坑不是技术不会而是以为把 graph 编译出来就算部署完了。实际上从脚本到服务核心工作全在状态管理、并发隔离和接口设计上。如果你现在正卡在这一步别急着纠结选哪条路先拿一个最小 agent 走通“脚本 → FastAPI → checkpointer”这条链路跑通了再往上想托管平台的事。技术选型没法一步到位但“先让它稳定跑起来”这个目标任何时候都不会错。