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 个 token,1000 个汉字通常要消耗 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 里跟业务相关的字段单独存到 Redis,Graph 负责执行,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(model="gpt-4o-mini", temperature=0) 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 不是模型排行榜的附属品,它是一门控制边界的工程。控制好工具的边界、状态的边界、成本的边界,剩下的就是给模型足够的自由度让它自己折腾。先跑通,再优化,复杂度和可控性永远要同步上升。