AI Agent 是这两年被定义得最乱的技术名词没有之一。团队A把带工具调用的聊天机器人叫 Agent团队B把多角色编排系统叫 Agent还有人把一段循环加函数调用的脚本也贴上 Agent 标签。名字叫什么其实无所谓真正的问题是它能不能在你的业务里稳定地下地干活。我在把 Agent 从 demo 推向生产的过程中把主流实现方式重新捋了一遍最后沉淀出一套比较顺手的拆解法——先看七要素再落七个决策点。这套方法不解决“怎么让模型更聪明”的问题只解决“怎么让 Agent 真正落地”的问题适合已经调过大模型 API、准备自己搭 Agent、或者正在做技术选型的读者。如果你正在找 AI Agent 学习路线我的建议是别一头扎进框架源码。先跟着本文把最小闭环跑通再看 LangGraph、Spring AI 或者各个框架的源码你会顺畅很多。工程实现这件事绝大多数坑都出在你不熟悉的地方而不是模型本身。下面从七要素讲起。1. 七要素到底是什么拆开 Agent 这个“网红名词”1.1 我为什么把 Agent 拆成七个要素而不是三件套很多文章把 Agent 拆成“大脑 工具 记忆”三件套用来讲概念确实够了但拿去指导写代码完全不够用。比如“记忆”这个词用户聊天记录是记忆任务执行到哪一步也是记忆用户偏好和知识库内容还是记忆三者的存储方式、读写时机、成本模型完全不一样。如果只笼统地说是“记忆”你根本没法设计数据表。所以我从工程落地角度拆把 Agent 拆成七个要素模型、规划、记忆、工具、行动、反馈、边界。这七个要素不是理论推导出来的是我在线上系统里被坑出来的。没有反馈环节Agent 会反复调用同一个错误工具没有边界一个简单循环可能把整月预算烧光。概念派喜欢做加法工程派必须做减法七要素就是把“能跑起来”和“能稳定跑下去”之间的坑都圈出来。1.2 七要素各自对应什么工程组件先说我经常被问的 token 是什么意思因为后面算成本和并发都要靠它。token 是模型处理文本的最小单位中文场景下粗略估算1 个汉字约等于 1 到 2 个 token1000 个汉字通常要消耗 1500 到 2500 个 token。对话越长每轮调用消耗的 token 越多钱和响应时间都堆在这上面。理解了这个下面七要素就好办了。要素工程对应物常见落地方案模型推理内核负责理解、决策、生成GPT 系列、Claude、开源模型部署规划拆解目标和步骤的机制ReAct 循环、Plan-and-Execute 节点记忆短期窗口、长期存储、任务状态Redis、Postgres、向量库工具外部能力的抽象接口Function Calling、MCP、OpenAPI 注册行动真正执行外部操作的部分工具执行器、沙箱、权限网关反馈执行结果回传给模型继续决策工具返回值、校验器、人工确认边界权限、限额、审计、超时预算控制、RBAC、可观测平台这七个要素在你写代码时不是平等的。模型、记忆、工具最显眼大多数人在这三块花了很多精力行动、反馈、规划、边界则属于“不亮眼但决定生死”的部分。举个真实场景你的 Agent 调了一个发送短信的工具模型发出了调用请求但工具服务器超时了。如果你没有把超时当成一个反馈结果传回给模型Agent 会以为短信已经发出去了然后进入下一轮用户收不到短信也不知道发生了什么。这就是“行动”和“反馈”这两个要素存在的意义让 Agent 的每一次决策都有闭环。2. 七个决策点真正写代码时绕不开的岔路口七要素讲的是 Agent 由什么组成七个决策点讲的是你在落地时要在哪些岔路口做选择。每一个决策点选错后面都要返工。我在做技术评审时基本只问这七个问题答得越清楚方案越靠谱。2.1 决策点一模型选型别只看榜单分数模型选型最大的误区是把排行榜分数当成唯一标准。线上 Agent 对模型的要求和做 benchmark 完全不一样我排优先级的话是工具调用可靠性、输出结构稳定性、推理能力、成本、延迟最后才是综合榜单分数。工具调用可靠性怎么理解就是你 bind_tools 之后模型能不能正确地决定“该不该调用工具”“该调哪个工具”“参数填得对不对”。有些模型总分很高但 function call 经常不触发或者参数类型传错这在 Agent 工程里是灾难。我实际测下来GPT-4o-mini 这类带 function calling 优化的小模型在工具调用场景经常比大模型还好用因为延迟低、成本低、行为稳定。输出结构稳定性也很关键。如果你的 Agent 后面挂了校验器模型输出 JSON 时老是多一个逗号、少一个引号你只能反复重试最后成本全耗在修正格式上。选型时一定要拿你自己真实的工具定义去测测 50 次调用统计失败率而不是看几个公开榜单。如果你处理的是高并发、低延迟的在线请求模型的首 token 延迟和吞吐量甚至比单个回答质量更重要因为用户可以等用户多但不能一起等。2.2 决策点二推理范式ReAct 还是 Plan-and-Execute推理范式决定了 Agent 的主循环长什么样。现在主流架构基本是两条路ReAct 和 Plan-and-Execute。ReAct 是让模型边推理边行动每一轮都思考“下一步做什么”然后调用工具拿到结果后再思考。它的好处是灵活能根据中间结果随机应变坏处是循环轮数不可控延迟高token 消耗大。简单任务比如“查一下上海明天的天气再告诉我”ReAct 要来回至少两轮实际上是杀鸡用牛刀。Plan-and-Execute 则是先让模型生成一个完整计划再按计划一步步执行执行过程中不轻易改计划。它的好处是过程可控、成本可预估坏处是遇到计划外的结果不会拐弯。我的个人建议在线交互场景默认 ReAct但要设置最大迭代次数并且把迭代数压到 5 以内离线批量任务、流程非常固定的场景用 Plan-and-Execute。还有一个很实用的路子是混合范式——先用一个规划节点拆出子任务然后让 ReAct 只负责执行单个子任务遇到异常再回到规划节点重新调整。这个混合方案在复杂业务里最好用代价是流程节点变多状态管理要更仔细。2.3 决策点三记忆设计短期和长期要分开记忆是最容易被低估的决策点。很多团队把所有历史消息一股脑塞进上下文结果对话到第十轮就开始超过模型窗口或者输出质量明显下降。记忆必须分层设计。短期记忆就是当前上下文窗口里的内容控制在最近 5 到 10 轮超过的部分用摘要压缩而不是继续往里堆。长期记忆分成两类一类是用户偏好、业务事实这种稳定信息可以结构化地存到 Postgres 或 Redis按用户维度查询另一类是历史会话事件适合放到向量库做语义检索只在需要的时候把相关片段捞回上下文。任务状态也必须算记忆的一部分比如“订票流程进行到哪一步了”这种数据如果只存在模型上下文里服务一重启就全丢了。我在项目里习惯把状态信息和聊天历史分开存。聊天历史是流水方向是只读状态是当前任务的断面需要频繁读写。这样设计之后Agent 的执行过程可以随时中断、恢复用户刷新页面重新进来也不会蒙圈。记忆设计不是追求存得多而是追求在正确的时机把正确的信息放回上下文同时把 token 预算压住。2.4 决策点四工具定义参数越多翻车概率越大工具是 Agent 和真实世界交互的窗口工具定义得好不好直接决定模型会不会乱来。我总结了几条工具定义的硬规则每个工具的参数越少越好能用一个字符串参数解决就不要拆成三个字段工具描述里必须写清楚“什么时候该用这个工具”而不是只写“这是什么工具”参数取值能用枚举约束就不要开放自由文本同一时间暴露给模型的工具不要太多五到八个以内最佳不然模型容易选错。还有一条很容易忽略的规则工具调用必须在服务端做二次校验。模型填的参数不可信比如用户输入了“明天”模型解析日期可能错你在工具执行前必须做格式校验、范围校验甚至权限校验。涉及发送短信、转账、下单这类敏感操作的工具必须加人工确认节点模型只能发起请求不能直接执行。工具设计得越克制Agent 越稳定什么都往里塞最后就变成一个谁也不知道会调什么的外部接口聚合器。另外工具执行要有超时和幂等设计同样的请求执行两次不应该产生两笔订单这就要求你在工具接口侧用请求 ID 去重。2.5 决策点五状态管理Graph 和状态机的取舍Agent 工程和普通后端接口最大的区别就在状态管理。普通接口是无状态的请求来了算完就走Agent 是多轮决策中间状态散落在各个节点里必须显式管理。现在主流架构是图编排LangGraph 是里面最典型的代表。Graph 的好处是你能把 Agent 的决策过程拆成节点和边每个节点只做一件事状态在节点间流转整体行为容易观察和回放。用 LangGraph 的时候State 的设计要特别注意。它用的是 TypedDict 来定义状态结构每个节点可以增删改状态里的字段。如果多个节点往同一个字段写入要用 Reducer 声明合并规则是覆盖、累加还是列表追加。很多人刚开始用 LangGraph 时忽略 Reducer结果两个节点同时写 messages 字段后一个把前一个覆盖了对话记录就断了。State 还要能序列化因为生产环境里 Agent 通常跑在异步任务里服务重启、Pod 重建都要能从外部存储恢复状态。我一般会把 State 里跟业务相关的字段单独存到 RedisGraph 负责执行Redis 负责记忆两边职责分开。2.6 决策点六并发与限流Agent 的瓶颈往往不在推理这是我最经常被问到的问题AI Agent 怎么扛并发。先说一个残酷的事实Agent 的并发瓶颈通常不在模型推理本身而在模型供应商的速率限制、工具 API 的 QPS、以及你自己的状态存储 IO。每个 Agent 任务要跑 10 到 30 秒内部还要调两三次外部 API如果在 HTTP 请求里同步阻塞你的服务很快就会被拖死。我的建议是三板斧异步任务队列、流式响应、限流熔断。把 Agent 执行过程丢到队列里由 worker 消费接口立刻返回任务 ID前端再通过 SSE 或 WebSocket 接收结果这样用户不需要干等服务的线程池也不会被打满。同时一定要给每个用户、每个会话做并发限制模型和外部工具的速率配额要提前查清楚设置本地信号量控制并发上限。我举个算并发量的例子方便你理解。假设模型供应商的输出速率限制是每分钟 600k tokens你的 Agent 任务平均消耗 15k tokens理论上每分钟最多能跑 40 个任务。再假设每个任务耗时 40 秒一个 worker 串行执行每分钟最多完成 1.5 个任务。要吃掉这 40 个任务/分钟的吞吐你需要大约 27 个并发 worker再留 50% 的余量就要准备 40 个 worker。很多人上来问代码怎么写其实先把这个账算明白代码怎么写都清楚。另外经常有人问“用 Rust 写 Agent 是不是性能更好”“Spring AI 能不能直接撑住高并发”我统一回复语言和框架都不是瓶颈瓶颈在上游速率限制和外部 API 响应时间选你团队最熟的栈反而靠谱。2.7 决策点七容错与兜底Agent 要学会认怂Agent 的失败不是小概率事件是默认事件。模型可能给出非法输出工具可能超时外部 API 可能返回乱七八糟的数据所以容错设计必须从一开始就做进去。重试要讲策略LLM 调用失败的瞬时错误可以退避重试但重试次数超过两次就要降级到备用模型工具调用超时也要区分幂等和非幂等非幂等的操作绝对不可以盲目重试。更重要的一点是给 Agent 设计“认怂”路径。当它尝试了三次以上还是失败或者连续几轮拿不到有效结果应该主动停止循环并输出一个明确的消息告诉用户“这个任务需要人工介入”而不是装作一切都好。我见过太多线上事故都是 Agent 在死循环里把费用刷爆最后才被告警捞出来。所以迭代上限、token 上限、成本上限这三道闸必须在 Agent 启动时就带上。还有可观测性Agent 每一步的输入输出都要有 trace出了问题能回放整个过程。没有 trace 的 Agent 项目线上出了问题你只能对着日志猜猜完还不一定对。3. 实操一个 FastAPI LangGraph 的最小 Agent 工程长什么样理论讲再多不如跑一个最小闭环。下面这套工程是 FastAPI LangGraph 的组合这也是我最近最常用的一套方案足够支撑你从零搭出第一个能下地干活的 Agent。3.1 工程骨架和依赖先把目录搭出来结构刻意保持精简app/ main.py # FastAPI 入口 config.py # 模型配置、限额配置 agent/ __init__.py state.py # 图状态定义 graph.py # 图编排 tools.py # 工具注册 memory.py # 记忆读写依赖只需要几个核心库不追求花哨fastapi uvicorn[standard] langgraph langchain-openai pydantic-settings redisLangGraph 负责编排LangChain 的 ChatOpenAI 负责模型对接Redis 负责会话记忆和状态存储FastAPI 只薄薄地包一层接口。注意一点LangChain 对整个框架是可选的你可以完全不用 LangChain直接用 LangGraph 配合 OpenAI SDK 自己封装模型调用但那样样板代码会多一点。我不建议一开始就陷入框架之争LangChain 在这里只是胶水核心逻辑还是 LangGraph 的图。3.2 状态、节点与工具注册的核心代码先定义状态。这里用 LangGraph 的 TypedDict 方式messages 字段要用 add_messages 这个 reducer确保多个节点往消息列表追加时不是互相覆盖。# app/agent/state.py from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] step: int max_steps: int然后定义工具。工具的描述一定要写清楚触发条件参数尽量精简# app/agent/tools.py def get_weather(city: str) - str: 查询中国城市的实时天气城市必须是完整中文名例如北京、上海 return weather_api.query(city) def get_stock_quote(code: str) - str: 查询A股行情输入6位股票代码例如600519 return stock_api.query(code) tools [get_weather, get_stock_quote]接下来是图编排的核心逻辑。Agent 节点负责让模型决定是否调用工具工具节点负责执行执行结果回到 Agent 节点继续循环# app/agent/graph.py from langchain_openai import ChatOpenAI from langgraph.graph import END, START, StateGraph from langgraph.prebuilt import ToolNode, tools_condition from app.agent.state import AgentState from app.agent.tools import tools llm ChatOpenAI(modelgpt-4o-mini, temperature0) agent_model llm.bind_tools(tools) def run_agent(state: AgentState): if state.get(step, 0) state.get(max_steps, 5): return { messages: [{ role: assistant, content: 我已经尝试了足够多次这个任务需要人工介入。 }] } result agent_model.invoke(state[messages]) return {messages: [result], step: state.get(step, 0) 1} builder StateGraph(AgentState) builder.add_node(agent, run_agent) builder.add_node(tools, ToolNode(tools)) builder.add_edge(START, agent) builder.add_conditional_edges(agent, tools_condition, {tools: tools, END: END}) builder.add_edge(tools, agent) app builder.compile()这里关键点有两个。第一step 字段用来限制最大迭代轮数超过 5 轮就让 Agent 输出“需要人工介入”这就是前面说的认怂路径。第二tools_condition 是老熟人它会检查最后一条 AI 消息里有没有 tool_calls有就路由到 tools 节点没有就直接走到 END整个循环自动结束。最后是 FastAPI 入口把图包成一个 HTTP 接口# app/main.py from fastapi import FastAPI, BackgroundTasks from fastapi.concurrency import run_in_threadpool from pydantic import BaseModel from app.agent.graph import app as agent_app from app.agent.memory import load_history, save_history app FastAPI() class ChatRequest(BaseModel): user_id: str session_id: str message: str app.post(/agent/chat) async def chat(req: ChatRequest): messages load_history(req.user_id, req.session_id) inputs { messages: messages [{role: user, content: req.message}], step: 0, max_steps: 5, } result await run_in_threadpool(agent_app.invoke, inputs) save_history(req.user_id, req.session_id, result[messages]) reply result[messages][-1].content return {reply: reply}注意我用的是 run_in_threadpool而不是直接把 agent_app.invoke 放进 async 函数。因为 LangGraph 的同步调用会阻塞事件循环如果每个用户请求都占用事件循环几秒钟整个服务就废了。生产环境我还是推荐异步任务队列但这个最小工程用 run_in_threadpool 已经能把并发地基打牢。3.3 把并发、记忆和兜底补全上面这套代码能跑通但离生产还差三块拼图并发控制、记忆分层、兜底降级。并发控制最简单的方式是用 asyncio.Semaphore 限制同时执行的 Agent 数量。比如模型允许每分钟 60k 输出 token单任务平均 10k token每分钟只能跑 6 个任务你就把并发压到 4 到 5 个留出余量。工具调用也要设超时用 httpx 的话把 timeout 设为 10 秒连接池开大一点避免每次调外部 API 都重新建 TCP 连接。记忆分层在最小工程里做到两点就够了历史消息只保留最近 10 条超出部分交给一个摘要节点压缩成一句话放在系统提示词里任务状态也就是 step 这类字段如果服务重启需要恢复就把它序列化到 Redis。别一开始就上向量检索先把短期记忆和状态持久化做扎实。兜底降级分两路。模型这一路可以准备一个便宜的开源模型作为备用主模型连续失败两次就切换。工具这一路只需要记住一条工具调用返回错误不是失败是反馈你要把错误信息拼回消息列表让模型看到也许它会换个方式再试一次。如果错误连续出现三次就直接输出需要人工介入的提示停止循环。3.4 用一张表复审七个决策点工程写完之后我会用这张表把决策点重新过一遍确保每个岔路口都有明确答案决策点本项目选择选择理由模型选型gpt-4o-mini工具调用稳定成本低延迟可接受推理范式ReAct交互场景需要灵活性靠 max_steps 兜底记忆设计最近 10 条 摘要 Redis 存状态token 预算可控状态可恢复工具定义两个工具参数最少化减少模型误选概率状态管理LangGraph State add_messages消息累加语义正确状态可序列化并发与限流run_in_threadpool 信号量事件循环不阻塞速率可控容错与兜底max_steps 人工介入提示避免死循环烧钱4. 常见问题与排查技巧实录4.1 高频问题速查表我在帮团队排查 Agent 线上问题时发现大家遇到的坑高度重复。整理成一张速查表你对照着症状找方案就行症状根因解决方案Agent 反复调用同一个工具工具结果没有正确反馈给模型或者模型陷入确认循环把每次工具返回值完整拼回消息设置最大迭代数模型传错工具参数工具定义太宽松参数缺少校验服务端二次校验参数用枚举和正则约束对话越长越笨上下文塞满token 被无效信息占掉摘要压缩历史只保留最近 N 轮并发一高全部超时模型速率限制或 worker 数超过令牌桶信号量限流退避重试备用模型服务重启后会话丢状态State 只存在内存里State 存 Redis启动时恢复钱烧得很快迭代轮数无上限工具重试太激进设置 max_steps、token 上限、成本告警工具执行两次没有幂等设计重试导致重复请求请求 ID 去重调用侧加超时判断这些坑我基本都踩过一遍。最意外的是第一个模型反复调用同一个工具看起来像是模型笨实际是你在工具返回给模型的消息里没有保留“这个工具已经调用过”的痕迹模型每轮都像失忆一样重新选择。把历史工具调用记录整理成结构化摘要给模型这个问题一般立刻缓解。4.2 几个值得记住的工程教训最后分享几条从项目里磨出来的经验希望你能少交点学费。第一Agent 的每一步都要留痕。我见过太多项目出问题时只能看到最终的输出中间模型说了什么、调了哪些工具全部一片空白。从第一天就把每一步的消息记录落库或者接上 trace后面排查问题能省十倍时间。这个成本花得绝对值。第二先让“认怂”成为设计而不是补救。给 Agent 一个明确的退出通道比让它在失败里死磕更有价值。用户不会因为 Agent 说“我需要人工介入”而失望但会因为 Agent 假装成功然后悄悄做错事而彻底失去信任。第三工具是 Agent 的能力边界也是安全边界。每加一个工具都要问三个问题这个工具会不会产生副作用模型有没有可能误用最坏情况下一次误用会造成什么影响回答不了这三个问题这个工具就先别上。我自己把 Agent 接到业务里最深的一个体会是Agent 不是模型排行榜的附属品它是一门控制边界的工程。控制好工具的边界、状态的边界、成本的边界剩下的就是给模型足够的自由度让它自己折腾。先跑通再优化复杂度和可控性永远要同步上升。