做了大半年 AI Agent 相关项目从最早用 LangChain 写 demo到后来搭 FastAPI LangGraph 上生产中间踩了不少坑也总结出一些实在的经验。最近看到不少人在问 agent 怎么扛并发、怎么部署、token 怎么控干脆把这段时间的实践整理成一篇把我觉得真正有用的东西写出来。这篇文章不聊那些花哨的概念只讲实际干活时遇到的问题架构怎么选、并发怎么处理、token 怎么省、上下文怎么管、线上出问题了怎么排查。适合已经在接触 AI Agent、正准备从 demo 往生产环境推的开发者也适合刚入门但对Agent 到底是什么还没完全吃透的朋友。1. 先说清楚AI Agent 到底是什么1.1 别把 Agent 当成 ChatBot我见过很多项目表面上叫 AI Agent实际上就是一个 ChatBot 套了层壳——用户输入一句模型回一句最多加个 system prompt 说是角色扮演。这不算 Agent。真正的 Agent 核心在于它具备自主行动的能力拿到一个目标后能自己拆分任务、调用工具、根据结果调整下一步。比如你让它帮我查一下这周的销售数据并生成一份报告它需要先调用数据库查询接口拿到数据后可能发现数据不完整再调用另一个接口补充最后调一个生成文档的工具。这个过程中的每一步是 Agent 自己决策出来的不是提前硬编码的 if-else。我用一个简单类比来理解这件事ChatBot 是客服你问一句它答一句Agent 是员工你布置一个任务它自己规划怎么做、找什么人配合、遇到问题怎么解决最后给你交付结果。这个区别决定了架构设计的走向。做 ChatBot 时你只需要关心 prompt 和上下文窗口做 Agent 时你还得关心工具注册、状态管理、循环终止机制、错误恢复这些从来没遇到过的问题。1.2 Agent 的核心三要素规划、工具、记忆拆开来看一个能落地的 Agent 系统通常包含三块规划Planning大模型根据用户目标拆解出需要执行的步骤。这一步依赖模型的推理能力不同模型的表现差异巨大。实操中我的体会是规划能力强的模型能省掉大量重试和修正逻辑不要太指望用小模型做复杂规划。工具ToolsAgent 能调用的外部能力比如搜索引擎、数据库查询、HTTP API、文件读写。工具定义的清晰程度直接影响模型调用的成功率。记忆Memory短期记忆是当前任务上下文长期记忆则是跨会话的用户偏好、历史结果。很多 Agent 翻车就翻在记忆管理上——上下文窗口越塞越满最后模型连最初的目标都忘了。这三块缺一块Agent 就只能算半成品。我自己的项目中规划能力靠模型本身工具和记忆是代码侧可以重点优化的地方后面我会详细展开这两块的实现细节。2. 架构选型为什么我用了 FastAPI LangGraph2.1 LangChain 的问题在哪里最早我用的是 LangChain。说实话LangChain 的生态确实全各种链、各种模块都有文档看起来也很完善。但真正做项目时我发现几个问题第一抽象层太重。LangChain 把很多东西包装成了统一的接口看起来很方便但出了问题很难调试。比如某个环节报错它的堆栈信息会跨好几个抽象层排查起来特别费劲。第二流程控制不够直观。Agent 的循环、分支、条件判断在 LangChain 的 Chain 模型里很难表达。你想实现如果工具 A 返回异常则走工具 B这种逻辑得嵌套好多层 LCEL写出来的代码可读性很差。第三维护过山车式。LangChain 的 API 变更频繁我遇到过好几次升级小版本后老代码不兼容的情况。对于生产项目这很致命。2.2 LangGraph 的图状态机制后来我切到了 LangGraph。LangGraph 把 Agent 的执行流程建模成一张图节点Node是具体的处理步骤边Edge是节点之间的跳转条件。这种表达方式天然适合 Agent 的循环和分支逻辑。举个例子一个典型的 Agent 流程图里有这些节点plan_node模型接收用户输入输出计划或直接输出工具调用指令。tool_node执行工具调用返回结果。decide_node判断当前结果是最终答案还是需要继续调用工具。这三个节点之间形成循环decide_node 决定是跳到 tool_node 继续执行还是跳到输出节点结束。LangGraph 里用add_conditional_edges实现这种动态跳转比 LangChain 的链式拼接清晰得多。我实际体会最深的是它的状态机制。LangGraph 维护一个全局的 State 对象各个节点都能读和写这个 State。这意味着你在任何节点都能拿到完整的执行历史做上下文管理、做审计日志都非常方便。在 LangChain 里想拿到完整执行链路要自己想办法往 chain 里塞变量体验差很多。2.3 FastAPI 承担的角色FastAPI 在网络服务层做三件事接收用户请求、把任务投递给 Agent 执行引擎、返回结果。它本身不包含 Agent 逻辑只负责 API 层和并发管理。我选择 FastAPI 的原因很实际原生 async/await 支持协程并发处理高 IO 场景调用 LLM API 就是典型的高 IO 场景。Pydantic 做数据校验用起来顺手定义工具参数 schema 时比较方便。自动生成 OpenAPI 文档联调省事。部署方便一个 Uvicorn 进程就能跑起来。整体架构简单画一下FastAPI 接收请求 - 创建任务放入队列 - Worker 进程消费任务并执行 LangGraph 图 - 结果写回 - 前端轮询或 WebSocket 拉取。这套架构后面我会详细讲。3. 并发与性能Agent 怎么扛住真实流量3.1 串行调用的致命问题很多人第一次把 Agent 暴露给真实用户时都会懵为什么单个请求挺快并发一上来就全卡住了原因在于 Agent 的执行链路是多个模型调用 多个工具调用串联的。一次请求可能涉及 3~5 次 LLM 调用每次调用 2~5 秒再加上中间的工具执行时间一个完整任务跑完基本要 10~20 秒。如果每个请求占用一个同步 Worker 线程10 个并发请求就把线程池打满了后续请求全部排队。更麻烦的是Agent 执行期间大部分时间都在等待——等 LLM 返回、等外部 API 返回。这段时间 CPU 完全空闲但线程被占着非常浪费。3.2 异步改造Agent 引擎要设计成可暂停的解决并发问题的核心思路别让 HTTP 请求线程和 Agent 执行线程绑死在同一个生命周期里。我的做法是引入队列。FastAPI 收到请求后立刻把任务信息放入队列并返回一个 task_id前端拿着这个 task_id 去轮询结果。Agent 执行引擎作为独立的 Worker 进程从队列里取任务异步执行完把结果写入存储。这样做的直接好处是HTTP 层的并发能力不再受 Agent 执行时间限制。即使 Agent 任务平均耗时 20 秒FastAPI 也能接受成百上千的并发请求因为每个请求只花几毫秒就把任务写入队列了。在 FastAPI 里接收任务的接口长这样from fastapi import FastAPI from pydantic import BaseModel import asyncio app FastAPI() class AgentTask(BaseModel): user_id: str session_id: str message: str app.post(/agent/task) async def create_task(task: AgentTask): # 把任务投入队列立即返回 task_id task_id await task_queue.put(task) return {task_id: task_id, status: pending}注意这个接口必须是 async 的await task_queue.put(task)不阻塞事件循环。如果任务队列的写入本身很快也可以不用 async但保持统一用 async 更稳妥。3.3 Worker 进程的并发控制Worker 侧同样需要考虑并发。一个 Worker 进程可以同时跑多个 Agent 任务吗答案是要限流。LLM API 通常有每分钟请求次数限制RPM和每分钟 token 数限制TPM不同供应商限制不同。我在实际项目中踩过教训一次性开 20 个并发任务调用模型接口结果直接被限流一堆 429 错误重试之后反而把成本抬高了。我的做法是给 Worker 加信号量控制并发数import asyncio # 控制同时执行的 Agent 任务数量 semaphore asyncio.Semaphore(5) async def run_agent_task(task): async with semaphore: result await execute_agent(task) return result并发数设置多少合适取决于你的 LLM API 限额和单任务平均耗时。一个粗略的估算公式最大并发数 ≈ LLM API 每分钟限额请求数 / 单任务平均 LLM 调用次数 × 容错系数建议 0.6比如 API 限额是 60 RPM单任务平均调用 5 次 LLM理论支持 12 个并发任务乘上容错系数 0.6就设 7 个左右。3.4 Token 与成本控制既要跑得快又要有钱赚Token 是 Agent 项目里绕不开的核心成本。LangGraph 每次循环都会向模型发送完整的 State这意味着状态越长每次调用的 token 消耗就越大。我自己实测过一个简单任务初始状态只有几千 token但经过 5 轮工具调用循环后每次发送的 token 可能膨胀到 2~3 万。如果模型计价较高一次任务的成本可能从几厘飙到几毛放大到线上流量就是不小的数字。控制 token 有几个实用手法消息裁剪只保留最近 N 轮对话和工具调用结果早期的历史信息做摘要后存入摘要节点而不是完整带在上下文里。工具结果精简工具返回的数据不要原封不动塞进上下文先做摘要处理只保留模型决策需要的关键信息。模型分级规划和最终输出用更强的模型中间的工具结果判断、简单分支决策用便宜快速的小模型。流式输出最终回答用流式传输用户感知速度更快也减少超时重试的概率。其中工具结果精简是我觉得性价比最高的优化。比如一个 SQL 查询返回了 50 行数据你不需要把 50 行全给模型先让代码做统计聚合把总行数、最大值、最小值、趋势这种摘要信息给模型token 消耗直接砍掉一半以上。4. 实操搭建一个可复用的 Agent 服务4.1 环境准备与依赖安装你可以直接按照下面的步骤来搭建。我用的环境是 Python 3.11依赖如下pip install fastapi uvicorn langgraph langchain langchain-openai redis其中 Redis 用来做任务队列和结果存储。如果你不想引入 Redis也可以用内存队列比如 Python 的asyncio.Queue但要注意进程重启会丢任务生产环境建议至少用 Redis。4.2 定义 Agent 的图结构这是 LangGraph 的核心代码。我直接给出一个可运行的简化版本包含三个节点规划、工具执行、决策。from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List[dict], operator.add] task: str tools_output: List[dict] final_answer: str # 节点1调用模型决定下一步动作回答或调用工具 async def model_node(state: AgentState): messages state[messages] # 这里省略具体的 LLM 调用代码返回模型决策 decision await call_model(messages) if decision[type] tool_call: return {messages: [decision[message]]} else: return {final_answer: decision[answer]} # 节点2执行工具调用 async def tool_node(state: AgentState): last_message state[messages][-1] tool_name last_message[tool_calls][0][name] tool_args last_message[tool_calls][0][args] # 从注册表找到工具并执行 result await execute_tool(tool_name, tool_args) return {messages: [{role: tool, content: result}]} # 节点3判断继续循环还是结束 def decide_finish(state: AgentState) - str: last_message state[messages][-1] if state.get(final_answer): return end elif tool_calls in last_message: return continue else: return end # 构建图 def build_agent(): graph StateGraph(AgentState) graph.add_node(model, model_node) graph.add_node(tools, tool_node) graph.set_entry_point(model) graph.add_conditional_edges( model, decide_finish, {continue: tools, end: END} ) graph.add_edge(tools, model) return graph.compile()这里有个关键点在decide_finish函数。我见过不少人在这地方写错判断条件没有覆盖模型输出 JSON 解析失败、工具调用参数缺失这些异常情况导致图进入死循环或者静默结束。给个建议所有分支判断都要有兜底。比如模型输出了非法格式的 tool_call你宁可结束任务返回无法处理也不要让它重新循环。因为现实场景里模型输出非法格式的概率比你想的高很多。4.3 工具注册与外部系统对接工具定义的质量直接决定 Agent 的可用性。我的经验是遵循三个原则一个工具只做一件事。不要写一个万能处理工具模型会不知道传什么参数。粒度拆细一点反而调用成功率更高。参数 schema 要写清楚。字段命名要直观description 写清楚参数含义和格式要求。模型不是人它只能靠这些描述来理解你的工具。返回结果要结构化。工具返回 JSON 而不是纯文本方便模型处理也方便你自己做日志。工具注册的代码大致这样tools [ { name: query_sales_data, description: 查询指定时间段的销售数据返回汇总统计, parameters: { type: object, properties: { start_date: {type: string, description: 开始日期格式 YYYY-MM-DD}, end_date: {type: string, description: 结束日期格式 YYYY-MM-DD} }, required: [start_date, end_date] } } ] async def execute_tool(name: str, args: dict) - dict: if name query_sales_data: return await query_sales_data(args) raise ValueError(fUnknown tool: {name})4.4 记忆与上下文管理记忆模块我踩过的坑最多。初期为了方便我把所有历史消息全部保留在 State 里结果跑到第 10 轮对话时 token 开销肉眼可见地涨而且模型开始表现失忆——一开始的任务目标被大量工具结果淹没了。后来我改成两级记忆架构短期记忆当前任务的完整执行轨迹包括用户输入、模型决策、工具调用和结果。这个保留在 LangGraph 的 State 里。长期记忆跨任务的用户偏好、历史结论摘要。存到数据库或 Redis每次新任务开始时注入到 system prompt。长期记忆的写入时机很关键。我是在每个任务结束时额外调用一次模型把当前任务的要点摘要成 200 字以内的结构化文本然后入库。下次同用户发起新任务时把它作为背景信息注入。async def summarize_and_store(user_id: str, state: AgentState): messages state[messages] summary await call_model( system请将这次对话的关键信息总结为结构化文本不超过200字, messagesmessages ) await redis.set(fuser_memory:{user_id}, summary, ex7*24*3600)这样做的代价是每次任务多一次模型调用但换来的是长期记忆的精准性我认为这个成本值得。4.5 部署时的几个细节部署方面我补充几个容易被忽略的点Uvicorn 启动时要用--workers开启多进程但 Worker 数量不是越多越好。每进程有自己的事件循环进程间共享任务队列需要走 Redis所以多进程的好处有限我一般设 2~4 个进程就够。给 Agent 执行设置超时。LangGraph 里可以给图编译传入recursion_limit控制最大循环次数。我一般设 10~15 次循环上限超过就强制结束避免死循环烧钱。日志要记录完整链路。每个任务的 task_id、每个节点的输入输出、每次调用的 token 数全部结构化日志输出。线上出问题时就靠这些日志排查。5. 常见问题与排查技巧实录5.1 模型开始胡言乱语先查上下文Agent 输出的内容莫名其妙或者回答和用户问题完全对不上这是最常见的线上问题。大多数情况下不是模型傻了而是上下文被污染了。我排查这个问题的顺序是先看完整执行日志检查 messages 里是不是混入了异常内容。比如上一个工具返回了一条错误堆栈被当作正常内容喂给了模型。再看 State 里的系统提示词有没有被工具结果覆盖。我遇到过工具返回的键名和 system prompt 里的变量重名导致 prompt 被意外替换。最后看循环次数。如果任务循环了 8 次以上早期上下文里的大量重复工具结果会干扰模型判断。针对第二个原因我后来给 prompt 模板做了一层隔离用户内容、系统内容、工具结果分别用不同的命名空间避免变量冲突。5.2 工具调用陷入死循环Agent 反复调用同一个工具不停循环这是另一个高频问题。通常有两种情况一种是工具本身执行成功但结果不符合模型预期模型执意重试。比如查询天气返回晴模型想要温度于是反复查询。这种情况我在工具返回结果里加了一层结果评估如果工具返回内容里包含明确结论如无数据条件不满足就直接在结果末尾追加一句已尽力获取没有更多信息可提供引导模型停止重试。另一种是模型生成的 tool_call 参数每次都有一点随机差异导致新调用、新结果但整体目标没有推进。这种就只能靠recursion_limit兜底到上限强制终止同时记录终止原因到日志后面再针对具体场景调 prompt。5.3 并发场景下的会话隔离在线上的多用户场景里我踩过一个很深很深的坑Session 状态串了。原因是当时我把 UserID 作为 Redis key 的一部分来存储会话状态但 Agent 执行引擎单例复用时某个中间变量没有从 State 里清除干净导致用户 A 的任务残留到了用户 B 的 State 里。排查非常费劲因为错误不是每次都复现只在并发高的时候偶发。最后通过给每个任务分配一个全局唯一的 session_id并在 State 初始化时强制清空非必要字段才彻底解决。这里给个硬性建议State 字段必须显式声明生命周期。哪些字段是任务级的每个任务新建哪些是用户级的跨任务保留哪些是全局只读的共享配置在代码里写清楚不要隐式依赖。5.4 Token 超限的处理策略很多项目跑着跑着突然报错output_token_limit_exceeded或者context_length_exceeded。处理策略取决于错误发生在哪一层如果是输入上下文超限说明 State 里的历史消息太多触发消息裁剪逻辑保留最近 N 轮 摘要重新执行。如果是输出超限说明模型生成长文本时超出限制改用分块生成或者调高 max_tokens 参数或者让模型生成大纲后逐段展开。如果是工具结果太大这是最需要提前预防的。我在工具执行入口统一做结果截断超过 2000 字符自动摘要从源头避免超限。还有一个小技巧LangGraph 支持在节点里捕获模型调用的 Token 消耗我把它累加到日志里每个任务完成后统计总 token。线上跑一段时间后对照这个数据能看到哪些用户、哪些场景在烧钱有利于后续优化。5.5 前端轮询还是 WebSocket任务投递到队列后结果要如何返回给用户两种方案我都试过轮询前端每 2 秒请求一次/agent/task/{task_id}查结果。实现简单但实时性差请求频繁。WebSocketAgent 执行完服务端主动推送结果。实时性好但连接管理复杂还要处理断线重连。我的建议是MVP 阶段用轮询上线后如果用户反馈等待体验差再升级 WebSocket。轮询接口做一下优化——查一次更新一次状态状态未变化就返回 304减少不必要的数据传输。6. 根据我的经验最后再唠叨几句做 AI Agent 项目最难的不是写代码而是对整个执行链路的掌控。模型输出有随机性工具调用有不确定性状态管理有复杂性这三者叠加在一起线上问题会非常玄学。我的应对思路是把所有能结构化控制的东西全部用代码控制住模型只负责决策不负责流程。流程的每一步校验、超时、重试、终止都用确定性代码实现。另外我建议你先从一个极小的场景切入比如只做一个查库存 生成回复的 Agent跑通完整链路后再加复杂度。不要一上来就规划什么多智能体协作、知识库增强、复杂工作流编排那些都是后话。先把底层的状态管理、并发控制、成本监控做扎实再往上盖楼不然后期返工的代价非常大。上面这些经验和代码我都是在一线项目里一个个踩出来、调出来的。AI Agent 这个方向还在快速演进框架和模型隔几个月就换一轮但底层的架构思维和排查方法论是通用的。希望这篇分享能帮你少走点弯路。