你拿到一本讲 FastAPI 和 LangGraph 的电子书肯定不只想看概念。市面上讲 Agent 的教程太多了十篇里有八篇是跑通一个 Demo 就结束剩下两篇开始画架构图。真正缺的是从能跑到能上线这一段FastAPI 怎么和 LangGraph 的图执行引擎配合好工具调用怎么设计才不至于一报错就整个对话崩掉多 Worker 部署时状态到底存在哪里以及为什么你明明 print 了日志却什么都没看到。这篇文章我就围绕电子书的下半部分也是我认为最有含金量的部分来写——生产化。全文不糊弄全是我自己在项目里踩过的坑和验证过能用的方案你可以直接照着改。1. FastAPI 与 LangGraph 的整合模式不止是挂一个路由很多人的第一反应是写一个 FastAPI 接口然后在函数里调用graph.invoke()完事。这个小 Demo 没问题但一旦你要面对并发请求、长时间运行的 Agent 任务、以及多个用户之间的状态隔离这种写法会立刻暴露出问题。FastAPI 和 LangGraph 的整合核心不是 API 调用而是生命周期和并发模型的匹配。1.1 把 LangGraph 放进 FastAPI 生命周期而不是每次现建先说我见过的最常见的错误写法在每一个请求处理函数里实例化一个新的 Graph。这么做的后果是每次请求都要重新编译图、重新加载模型配置延迟高得离谱而且如果图里有连接池之类的资源你相当于在疯狂创建和销毁连接数据库和模型服务很快就会被拖垮。正确的做法是把图的编译和资源初始化放在 FastAPI 的lifespan里。FastAPI 的lifespan是异步上下文管理器应用启动时执行yield之前的代码关闭时执行yield之后的代码。这个机制天然适合做 LangGraph 的初始化。from contextlib import asynccontextmanager from fastapi import FastAPI from your_project.graph import build_agent_graph asynccontextmanager async def lifespan(app: FastAPI): # 启动时编译图初始化资源 app.state.graph build_agent_graph() app.state.llm_client create_llm_client() yield # 关闭时清理资源 await app.state.llm_client.aclose() app FastAPI(lifespanlifespan) app.post(/agent/run) async def run_agent(request: AgentRequest): graph app.state.graph result await graph.ainvoke({messages: request.messages}) return result这里有一个细节值得注意lifespan里初始化的资源要挂在app.state上而不是用全局变量。原因很简单app.state是 FastAPI 官方提供的存储应用状态的地方配合lifespan使用语义清晰测试时也容易替换。我自己实测下来这个改动带来的提升非常明显。之前每次请求现建图单请求延迟里光是编译图就要占到 300 到 500 毫秒改成lifespan初始化后这部分开销直接归零。如果你的图还用到了LangGraph的PersistentMemory或者外部向量库同样应该在启动时预创建好客户端而不是每次请求临时连接。1.2 异步执行模型sync 和 async 的边界要划清楚LangGraph 的invoke是同步方法ainvoke是异步方法。FastAPI 的请求处理函数如果你定义为async def它就跑在事件循环里如果你定义为普通的defFastAPI 会自动把它丢到线程池里执行。这带来一个问题很多人的图里既有同步的模型调用又有异步的工具调用混在一起最容易翻车。我的建议是一条铁律在async def的端点里只调用ainvoke。不要为了省事在异步函数里去调用同步的invoke——那会阻塞事件循环导致整个服务的并发能力急剧下降。如果你用的模型 SDK 不支持异步比如某些老的 OpenAI SDK 版本宁可把端点定义成同步def让 FastAPI 自己丢线程池也不要硬在异步里做同步调用。另外注意 LangGraph 的工具函数如果是同步的而图本身是用ainvoke驱动的工具会在 LangGraph 内部被自动执行不会阻塞你的事件循环吗实测不一定。LangGraph 的ainvoke在工具执行那一步并不总是做了线程隔离。保险的做法是工具函数内部如果涉及网络 IO就用anyio.to_thread.run_in_threadpool或者asyncio.to_thread包一层。这样能确保工具调用的阻塞操作不会殃及 FastAPI 主进程。下面是我在工具层做异步包装的一个示例顺带展示了如何给工具调用加上超时控制import asyncio from langchain_core.tools import tool tool def fetch_website_text(url: str, timeout: float 10.0) - str: 获取网页正文内容用于资料检索。 import httpx try: resp httpx.get(url, timeouttimeout, follow_redirectsTrue) resp.raise_for_status() return extract_readable_text(resp.text) except Exception as exc: return f获取失败: {exc} # 在 LangGraph 调用工具时用异步线程池包裹 async def run_tool_with_timeout(tool_node, state, timeout: float): return await asyncio.wait_for( asyncio.to_thread(tool_node, state), timeouttimeout )这个包一层的习惯我是吃过亏才养成的。之前有一个 Agent 工具调用了外部慢接口最坏情况下要等 60 秒才超时。那 60 秒里FastAPI 的 Worker 被占满所有请求全部排队用户端表现为服务卡死。后来所有工具的执行都套上了asyncio.wait_for最长等待时间硬性控制在 10 秒。1.3 用户级状态隔离别再犯共享内存的错误Agent 系统大多数是有状态的——多轮对话要记住上下文多步骤任务要记住中间结果。LangGraph 里通常用StateGraph来管理状态状态对象在图的各个节点间传递。生产环境里最危险的一个操作是把用户状态存在全局变量里。比如一个字典key 是用户 IDvalue 是对话状态。听起来很合理但在多 Worker 部署时每个 Worker 进程的内存是独立的用户第一次请求打到 Worker 1状态存在 Worker 1 的内存里第二次请求被负载均衡到 Worker 2状态直接丢失。就算你只开了一个 Worker全局字典还会面临并发写冲突的问题两个请求同时改同一个 key后写的覆盖先写的。我现在的做法是状态持久化交给外部存储FastAPI 和 LangGraph 都不持有跨请求的状态。具体用哪套存储方案第 3 部分我会细讲。这里先提一个集成模式在每次请求进入时从存储里加载该用户的最近状态序列注入到 LangGraph 的初始 State 里图执行完后把最新的消息列表和 Checkpoint 写回存储。你需要保证这个读写过程是原子的。LangGraph 官方提供了一组BaseCheckpointSaver接口配合langgraph-checkpoint库写一个自定义的 saver 并不难。这里有一个小的工程细节不要把整个 State 对象直接 pickle 存数据库。State 结构会随着代码演化而变化老的 pickle 数据在新代码里反序列化大概率报错。正确做法是只存必须的消息列表和业务字段或者用序列化后的 JSON。我自己踩过一次升级 LangGraph 版本后所有历史会话全部反序列化失败那叫一个酸爽。2. Agent 工具调用的生产级设计把会跑变成不会挂Agent 的核心能力之一是工具调用。但 Demo 里的工具调用和现实里的工具调用完全是两码事。Demo 里你调一个天气接口返回 JSON 就完事。现实里你的 Agent 可能要调用内部订单系统、信用卡扣款 API、外部爬虫任何一个环节出错Agent 的任务链就会断掉。这一部分就是讲怎么让工具调用变得健壮。2.1 工具定义的合约化输入输出都做约束很多 LangGraph 教程在定义工具时非常随意。示例代码里经常出现这种tool def search_documents(query: str): 搜索知识库。 return vector_store.search(query)这在本地跑没有任何问题。但生产环境你会立刻遇到两个麻烦第一模型返回的工具调用参数不一定符合你的预期类型比如query传进来一个嵌套的 JSON 字符串第二工具返回值五花八门有的是字符串有的是 dict有的是自定义对象LangGraph 的 State 里如果混入不可序列化的对象整个图的输出都无法被 FastAPI 正确转成 HTTP 响应。我的建议是给工具函数加上严格的数据合约。用 Pydantic 定义输入和输出模型工具只负责把外部系统的数据转换成合约模型。这样有几个好处模型调用工具时传入的参数会被 Pydantic 校验格式不对直接给出明确报错返回值经过合约约束后无论下游是数据库存储还是 API 返回都不会出幺蛾子。下面是我在项目里常用的工具定义模式from pydantic import BaseModel, Field from langchain_core.tools import StructuredTool class OrderStatusInput(BaseModel): order_id: str Field(description订单号例如 ORD20240601) include_detail: bool Field(defaultFalse, description是否返回详细物流信息) class OrderStatusOutput(BaseModel): order_id: str status: str estimated_delivery: str | None None detail: list[str] [] def _query_order_status(input_data: OrderStatusInput) - OrderStatusOutput: # 这里执行真实的 API 调用或数据库查询 ... return OrderStatusOutput(order_idinput_data.order_id, statusshipped) order_tool StructuredTool.from_function( namequery_order_status, description查询订单的最新状态。, args_schemaOrderStatusInput, func_query_order_status, return_typeOrderStatusOutput, )注意StructuredTool.from_function里我特意传了return_type。LangChain 会用它做后置校验如果函数返回的结果不符合声明的类型会主动报错而不是带着错误的数据一路往下游跑。这类编译期就能发现的问题不要留到生产环境让用户帮你发现。2.2 错误恢复与自动降级工具失败不等于任务失败Agent 工具调用失败时默认行为通常是直接抛异常。如果你没有做任何处理LangGraph 的整条执行链会被中断用户看到的就是一个 500 错误。生产环境里工具有时候就是会挂——外部接口抖动、限流、数据格式变更这些防不胜防。我更愿意把工具设计成永远返回执行结果而不是抛异常的模式失败时返回一个人可读的错误信息字符串或一个statusfailed的结构体让 Agent 自己判断下一步怎么走。from langchain_core.tools import tool tool def deduct_balance(user_id: str, amount: float) - str: 从用户余额扣除指定金额用于支付场景。 try: result payment_service.deduct(user_id, amount) return f扣款成功剩余余额: {result.balance} except InsufficientBalanceError as exc: # 给 Agent 足够判断的错误信息但不抛异常 return f扣款失败余额不足当前余额 {exc.current_balance} 元 except PaymentGatewayTimeout: return 扣款失败支付网关超时请稍后重试或联系客服 except Exception as exc: # 生产环境不要暴露内部异常细节 return 扣款失败系统暂时无法处理请稍后重试这样做的核心原因是只有 Agent 才能根据错误信息做出下一步决策。你提前把抛异常这条路封死相当于把决策权交给了编排逻辑。实测中加了这种错误返回机制之后Agent 的成功执行率提升非常明显。因为模型很擅长根据错误文本做修正——余额不足时它知道要引导用户充值超时时它知道要让用户等待——但模型对未处理的异常没有任何应对手段直接变成一条死路。2.3 工具调用的重试与幂等设计既然工具会失败重试就是不可避免的。但盲目重试是灾难。尤其是涉及扣款、创建订单、发送消息这类有副作用的工具重复执行会造成重复扣款、重复下单。所以生产级设计里重试策略必须配套两个东西可重试错误与不可重试错误的区分以及幂等键。我的实现思路是工具类函数内部自己定义错误分类。网络超时、网关 5xx、限流这些是可重试的参数错误、业务规则拒绝比如余额不足是不可重试的。工具返回给 Agent 的结果里包含一个retryable字段。Agent 的编排节点在收到retryableTrue的结果时才执行重试逻辑且最多重试 N 次使用指数退避。对写操作类工具入参里强制带上request_id下游系统用它做幂等判断。幂等键这块多说一句不要用 UUID4 随机生成要用业务键。比如扣款场景幂等键可以是用户的支付单号发消息场景可以是消息的模板 ID 加用户 ID 加发送批次号。业务键的好处是即使 Agent 重复调用下游系统也能识别出相同业务含义的请求主动去重。LangGraph 里实现重试有多个层级。工具函数内部做同步重试是最简单的但我不推荐全部在函数内部做因为模型生成一次工具调用是有 token 成本的函数内部先自己把瞬时错误消化掉再决定要不要把错误返回给模型这样能省不少重试迭代的轮次。更高级的做法是写一个自定义的ToolNode子类在它的run方法里套统一的超时和重试管护逻辑。3. 部署与监控进生产前的最后一公里Agent 应用本质上还是一个 Web 服务FastAPI 只是入口。生产环境要比拼的是当流量上来、当任务变长、当节点出问题时你的系统能不能优雅地应对。这一部分探讨部署模型、可观测性以及 Gradio 混用时的注意事项。3.1 长任务处理别再让 HTTP 请求死等Agent 任务有时候非常耗时。一次复杂的多步任务可能需要调用好几次模型、查好几个工具总耗时超过 10 秒甚至几分钟。如果你让 FastAPI 的请求同步等待图执行完成用户的浏览器或客户端大概率早就超时断开连接了而且 Worker 会被长时间占用并发能力急剧下降。这种情况下我推荐把任务提交和执行结果分离客户端 POST 请求只负责创建任务并立即返回task_id后端把任务扔进队列比如httpx Redis Streams或者简单就用 FastAPI 的BackgroundTasks在后台执行客户端通过轮询或者 WebSocket 获取执行结果。一个折中方案是用asyncio.wait_for给请求设一个较短的超时比如 30 秒超时了先返回task_id后续让客户端轮询。这样用户体验比一律后台任务要实时一些但代码复杂程度可控。下面是一个简化版方案from fastapi import BackgroundTasks from uuid import uuid4 task_store {} # 生产环境请换 Redis app.post(/agent/submit) async def submit_agent_task(request: AgentRequest, background_tasks: BackgroundTasks): task_id str(uuid4()) task_store[task_id] {status: pending, result: None} async def _run(task_id: str, request: AgentRequest): task_store[task_id][status] running try: result await app.state.graph.ainvoke({messages: request.messages}) task_store[task_id][result] result task_store[task_id][status] completed except Exception as exc: task_store[task_id][status] failed task_store[task_id][error] str(exc) background_tasks.add_task(_run, task_id, request) return {task_id: task_id, status: pending} app.get(/agent/task/{task_id}) async def get_task_result(task_id: str): return task_store.get(task_id, {status: not_found})这个方案在单 Worker 下很好使。但注意background_tasks跑在同一个进程里如果请求量大后台任务会占用事件循环影响正常的 API 响应速度。业务量再往上走就得上独立的任务队列加 Worker 进程FastAPI 只负责收发真正跑 Agent 的进程单独拉一组。这是把应用服务和业务执行解耦的关键一步。3.2 多 Worker 部署与状态持久化的选型Uvicorn 启动时常用的参数是--workers 4。这会让 Uvicorn 启动 4 个独立的进程每个进程加载一份应用、编译一份 Graph、各自持有自己的内存状态。好处是能利用多核 CPU 并发处理请求坏处就是前面提到的——内存态完全隔离。多 Worker 场景下如果你用 LangGraph 的MemorySaver做 Checkpoint 持久化你马上会遇到一个让人头皮发麻的问题用户 A 的请求打到 Worker 1对话 Checkpoint 存在 Worker 1 的内存里下一次请求打到 Worker 2完全找不到之前的 Checkpoint对话直接失忆。所以生产环境我建议至少用 Redis 或 Postgres 做 Checkpoint Saver。LangGraph 社区现在有现成的实现——langgraph-checkpoint-redis和langgraph-checkpoint-postgres都支持并发读写的场景。from langgraph.checkpoint.postgres import PostgresSaver # 注意PostgresSaver 通常需要连接池 with PostgresSaver.from_conn_string(postgresql://user:passdbhost:5432/agentdb) as saver: graph build_agent_graph(checkpointersaver)这里有个容易被忽视的坑Checkpoint Saver 的序列化和并发写入时序。多个 Worker 同时处理同一个用户的不同请求会导致 Checkpoint 冲突。LangGraph 目前在这种场景下并不能保证严格的串行化所以如果你的业务模型是同一个用户的请求可能并发进来比如用户快速连点你就需要在业务入口做用户级的串行化——每个用户同一时间只允许一个 Agent 任务在跑。这在产品上通常体现为一个正在思考中的加载状态。另外Uvicorn 的--workers并不是越多越好。Worker 数量建议取CPU 核心数的 1~2 倍同时要观察每个 Worker 的内存占用。Agent 应用通常要加载模型 tokenizer 甚至本地模型稍微吃紧。我自己一般先开2 * CPU 核心数压测后如果内存吃紧再往下调。3.3 日志、追踪与指标出问题时你才能救命Agent 应用排错难度比普通 Web 应用高得多。普通 Web 应用出问题看 HTTP 状态码、看异常堆栈基本能定位。Agent 应用不一样——同一个 HTTP 请求可能调了五六次模型中间穿插了三四次工具调用每次模型调用的输入输出都不同有的还调用失败被重试。没有一套完整可观测体系排查问题基本靠猜。日志方面首要解决的问题是结构化。不要只打字符串要打 JSON。下面是我常用的格式{ timestamp: 2024-06-01T12:00:00.123Z, level: INFO, event: tool_call_start, trace_id: abc123, request_id: req456, tool_name: query_order_status, input: {order_id: ORD20240101}, latency_ms: 235 }想要在日志里关联上trace_id最简单的方式是 FastAPI 中间件里生成并注入contextvars。LangGraph 的节点执行是在异步循环里的contextvars能安全地跨 task 传递这一点比用全局变量可靠得多。追踪层面有条件的话上 OpenTelemetry。LangGraph 生态里已经有一些实验性的自动埋点FastAPI 也有现成的 OTel instrumentation。我用下来感觉最直接的价值是能看到整条调用链的时间分布模型调用占比多少、工具调用占比多少、网络等待占多少。很多性能问题一眼就能看出来。指标方面至少要有这几个关键数请求量按端点拆分请求延迟的 P50/P95/P99模型调用成功率各工具调用的成功率与平均耗时运行中的 Agent 任务数量Checkpoint 读写耗时这些指标对一个 Agent 服务的容量规划和健康判断足够用了。比如你发现工具平均耗时从 200ms 涨到 2 秒大概率是外部依赖出问题了即使当前请求还没报错也需要及时告警。3.4 Gradio 和 FastAPI 放一起的问题别名冲突是重灾区现在很多 Agent Demo 喜欢用 Gradio 作为交互界面因为拖出一个 Chat UI 太方便了。但把 Gradio 和 FastAPI 混在一个服务里坑不少。最常见的问题是路由别名冲突。Gradio 启动时会挂载一个根路径/FastAPI 通常也会有自己的/路由或者/docs、/openapi.json。如果你用 FastAPI 的app.mount(/gradio, gradio_app)来挂载表面上没问题但 Gradio 内部会有很多静态资源路径挂载到子路径时经常出现资源 404 的问题。我的处理方法要么 Gradio 单独跑一个进程、单独端口FastAPI 通过 iframe 或者跳转过去要么让 Gradio 跑主端口FastAPI 通过子路径提供服务。两者二选一不要盘根错节地硬揉在一起。另外Gradio 的queue机制默认是阻塞的它会自己开线程跑任务。如果你把 Gradio 和 FastAPI 放在同一个事件循环里可能会因为循环嵌套问题导致极难排查的死锁。我的经验是能拆就拆各跑各的进程用 HTTP 互通。4. 常见问题与排查技巧实录最后一部分值得单独拎出来写。很多问题在文档里不会提只有自己踩到才会有深刻教训。我把高频的几个记在这里省得你重复走弯路。4.1 Uvicorn 日志丢失不是丢了是日志配置被覆盖了热词里有一个uvicorn fastapi 日志丢失问题这个我非常熟。很多人发现自己用logging.info()打的日志在 uvicorn 启动后不显示了或者显示得很诡异。原因很简单uvicorn有自己的日志配置默认会接管 root logger 的 handler。如果你的日志记录器用的是名字是uvicorn或者你在代码里调用了logging.basicConfig()uvicorn的--log-level参数和log_config可能覆盖了你的配置。解决思路在应用启动前显式配置自己的 logger而不是依赖logging.basicConfig()。给 logger 一个独特的名字比如app_logger logging.getLogger(my_agent_app)。设置 handler 和 formatter再确保不把propagate设成True否则事件会传到 root logger 再被 uvicorn 接管造成重复或丢失。排查的时候可以跑一次uvicorn看看它的日志格式有没有严格控制。如果没显示先确认--log-level是不是设成了warning以上很多人顺手设成critical就把所有 info 日志淹了。4.2 LangGraph 状态持久化丢上下文检查 Checkpoint Saver 的刷新时机如果你发现多轮对话里 Agent 偶尔会忘记之前的消息大概率不是模型问题而是 Checkpoint 没及时写入。LangGraph 的 Checkpoint 写入时机跟图的执行流程有关不同 Saver 的写入策略不同。比如MemorySaver是在内存里直接更新几乎没有延迟而 PostgresSaver 的写入是走连接池的如果连接池设置太小或者事务未提交就会出现看起来没存上的错觉。我的经验PostgresSaver 的from_conn_string会自动用一个连接池但默认的池大小可能不够你的并发写需求。等保代码里显式传入pool参数以控制上限。另外一个常被忽略的问题是LangGraph 的 State 里如果有大量冗余数据比如整个网页的文本内容序列化到 Postgres 时可能单条记录就几 MB导致写入超时。State 设计时就该把大字段排除在外或者压缩存储。4.3 Gradio 和 FastAPI 混用时的端口占用和 CORS刚才说了 Gradio 尽量单独进程但如果实在跑在同一个进程里还有两个小问题需要注意。第一是端口冲突Gradio 默认监听7860FastAPI 是8000实际上能共存但如果你用FastAPI()加了docs_urlNone但还是互相干扰建议直接给 Gradio 指定server_name和server_port并挂载在子路径下。第二是 CORS如果前端是独立页面请求 FastAPI 的接口同时 Gradio 的 iframe 也在同源页面里一定要给 FastAPI 配好CORSMiddleware不然浏览器层的跨域会非常折腾。我自己手头的一个项目FastAPI 跑在8001Gradio 跑在7860中间通过子域名区分CORS 策略如下from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins[https://your-gradio-domain.com], allow_credentialsTrue, allow_methods[*], allow_headers[*], )这样虽然丑但稳定排错容易。你要知道把这些 Web UI 和业务 API 解耦日志排查时你会感谢自己当初的决定。4.4 FastAPI 压力测试别只测 200 OK要测长时间任务和慢工具很多人上线前压测就压测一个简单接口返回个 JSON 或者跑个一步推理看着 QPS 挺高就觉得万事大吉。但 Agent 应用最怕的不是普通流量而是长任务和慢工具的叠加。一次调用模型需要 5 秒这时候来 10 个并发你的 Worker 池就已经被占满了如果再加外部工具响应慢整个系统吞吐会断崖式下跌。我压测时通常设计三类场景短请求单步推理模拟简单对话。中请求两步工具调用模拟查天气、查订单。长请求多步工具加模型推理模拟规划类任务。分别测它们的 P99 延迟和系统最大并发数。你会发现长请求的并发能力可能只有短请求的十分之一这很正常。但你必须提前知道这个数据才能决定要不要引入任务队列要不要限制用户并发。如果你发现 P99 延迟随着并发上升而线性恶化大概率不是模型变慢了而是你的 FastAPI 应用在处理并发时存在事件循环阻塞。排查方法很简单在压测时观察进程的 CPU 占用和事件循环延迟指标利器是asyncio的 debug 模式或者loop.slow_callback_duration。如果发现事件循环阻塞回头检查图里有没有同步阻塞调用鱼龙混杂在async def里那是这种问题的头号嫌疑。5. 一些我能给的实操建议都是踩坑换来的写到这里正文内容差不多到了尾声。与其给一个总结我更想把自己在多个 Agent 项目里反复验证过的一些习惯留下来你以后做生产级系统时可以直接拿用。第一做一个 Agent 服务先把 tracing 做出来再写业务。很多人是业务写完才想起来加日志导致线上出问题时毫无头绪。我现在的习惯是第一天就要在 FastAPI 里加好trace_id中间件和 JSON 日志后面每加一个新功能都自动带上 tracer。这不是锦上添花是生产环境的标配。第二所有外部调用都要设超时没有例外。模型调用、工具调用、Checkpoint 写入每一个都设上超时。Agent 系统比传统 Web 系统更容易出现级联超时一个工具慢了整条 Agent 链路被拖死所有请求都排在那里。超时是唯一能斩断级联的利刃。第三State 设计要轻。不要把大段文本、完整网页内容、长历史消息全塞进 State不仅序列化慢还容易把 Checkpoint 存储撑爆。只保留必需字段。我见过把整个对话历史原封不动存进 Postgres 的客户一个月后存储增长了十几个 GB。第四Agent 的结果要有解释性。就算你用的是黑盒模型至少也要在返回结果里带上模型思考的摘要、调用了哪些工具、结果来自哪个工具。这对用户信任和问题排查都有巨大帮助。将来出了纠纷你至少知道 Agent 当时是依据什么做判断的。FastAPI 和 LangGraph 这套组合现在生态已经成熟不少但真正在生产里跑起来看的还是这些细节。希望这篇文章能给你省一点踩坑的时间。