过去三个月,我一直在折腾 AI Agent 的工程落地。从前端到后端,从提示词到状态恢复,把一个从零搭起来的“会聊天”的 Demo,逐步改成了“能干活”的小系统。所谓能干活,指的是它能听懂用户一句话,自己规划要查哪个接口、调用哪几个工具,把结果整理成可读回复,甚至在某个环节中断之后还能接上进度继续跑完。如果你也准备上手做 Agent,或者已经写了几个月提示词但总觉得离生产环境差一口气,这篇笔记应该能帮你把拼图补完整。我不打算写论文式的定义,只讲真实工程视角下的 Agent:先拆成七个构成要素,再列七个我每次搭建都必须做出选择的决策点,最后给一段基于 FastAPI 和 LangGraph 的最小可运行示例,能抄作业就直接抄。
1. AI Agent 的七要素全景图:动手写代码前先画一张分工图
我最早看各种 Agent 框架的文档,最大的收获其实是一张草图:中心是大模型,周围围着一圈记忆、工具、行动、反馈。后来在项目里踩了大量坑,才发现很多问题不是模型不够聪明,而是边界没划分清楚,该由哪个模块负责的事被混在了一起。
1.1 先看全局:哪七个要素缺一不可
先给出结论。在我看来,所有 Agent 无论用什么框架、什么协议、什么部署方式,都可以拆成七个要素:目标定义、感知输入、记忆、规划、工具、行动执行、反思。这七个要素不是代码层面的模块名,而是问题域的划分。比如感知输入,在不同场景里可能是 Webhook 回调、文件读取、图片解析、数据库查询,你没法用一个统一的“感知模块”包打天下,但你必须在设计时回答“这个 Agent 能感知什么、不能感知什么”。
我把七个要素的职责和核心反问整理成了一张表:
| 要素 | 核心职责 | 设计时必须回答的问题 |
|---|---|---|
| 目标定义 | 把用户需求转成可执行的任务约定 | Agent 到底要完成什么,不做什么? |
| 感知输入 | 获取并标准化外部信息 | 输入来源有哪些?格式怎么归一化? |
| 记忆 | 保存和检索上下文、业务事实、中间结果 | 哪些信息要短期保留,哪些要长期沉淀? |
| 规划 | 将大任务拆成小步骤,决定执行顺序 | 步骤是固定的还是动态生成的?依赖关系怎么表达? |
| 工具 | 扩展模型能力边界 | 每个工具的参数、返回、错误、幂等性是否清晰? |
| 行动执行 | 与环境交互并产生真实副作用 | 执行失败怎么处理?副作用能不能回滚? |
| 反思 | 检查结果、识别错误、触发修复 | 怎么判断一次执行是成功还是需要重试? |
这七样东西少任何一样,短期可能靠提示词和经验值硬撑,但一遇到线上流量、异常输入、外部依赖抖动,就会暴露问题。
1.2 输入侧三要素:目标、感知、记忆如何协同
目标定义是第一道坎,也是最容易被一句提示词带过的地方。很多人把“目标”直接写在 System Prompt 里,比如“你是一个智能客服,要快速响应用户”,但到了执行层面,“快速”和“严格按知识库回答”可能产生冲突。我踩过的典型坑是:目标写得含糊,模型开始自由发挥,自作主张编造接口参数。正确的做法是在设计阶段就把目标拆成约束条件,比如“只调用白名单工具”“非确定信息必须标注不确定”“答案必须包含依据来源”,让模型有明确的判据,而不是靠感觉。
感知输入决定了 Agent 的信息质量。文本输入相对简单,但生产环境里更多是图片、扫描件、表格、网页抓取结果。我的经验是:感知层不要指望模型用长上下文去硬扛,应该在做入口处做预处理。比如 PDF 先转成文本再进模型,表格先做列名映射,网页先去掉导航噪音。否则模型会被无关内容干扰,工具调用时也会拿错字段。感知设计得越干净,后面的规划压力越小。
记忆这件事,很多人误以为就是“把历史消息全塞给模型”。实际上至少要分三层:短期记忆是当前会话的消息列表,长期记忆是跨会话的用户画像和业务事实,工作记忆是任务执行过程中产生的临时中间结果。三者不能混在一个列表里。短期记忆要考虑长度上限,长期记忆需要检索而不是全量加载,工作记忆要和状态管理打通。设计顺序上我建议先做长期记忆,因为短期记忆可以被上下文窗口覆盖,长期记忆才真正需要向量检索或键值存储来支撑。
1.3 执行侧四要素:规划、工具、行动、反思如何闭环
规划可以简单到一条 ReAct 规则“先想、再动、再看结果”,也可以复杂到用另一层 LLM 调用把任务拆成一组带依赖关系的 DAG。以我实测的经验,超过一定复杂度之前,不要让模型做重量级规划。用一个白名单动作集加条件分支,通常更稳也更省钱。真正值得动态规划的场景是:任务步骤事先无法确定,或者外部返回会影响下一步走向。比如查天气这种固定任务,写死流程就好;但“帮我订一个符合要求的行程”这种任务,才需要模型拆解成查景点、查交通、查住宿多个步骤。
工具是 Agent 的毛细血管,每接一个工具,都要想清楚四件事:参数 Schema 是什么、返回结构长什么样、错误语义怎么表达、是否幂等。其中幂等性最容易被忽略。用户点一次重试和系统自动重试,如果产生两个订单、两条消息,那问题就不是模型笨,而是接口设计有缺陷。我在项目里吃过这个亏,后来要求所有写操作工具必须支持 request_id 去重,重试才敢放开。
行动执行看起来只是发一个 HTTP 请求或者跑一段函数,但它藏着三个容易被低估的问题:工具副作用能否回滚,调用超时怎么办,返回数据不合法怎么办。这些问题在设计阶段就要写清楚,而不是等运行时报错再补救。反思则是闭环的最后一环,也最容易被砍掉。我建议至少保留一层轻量校验:模型每次调用工具后,对返回结果做合法性判断,发现不对就换一种方式重试。实测下来,很多“Agent 死循环”就是因为少了这个简单校验,模型拿到错误结果像没头苍蝇一样反复调用同一个工具。
2. 七个决策点:真正决定 Agent 工程成败的地方
要素是静态的认知框架,决策点是动态的工程选择。我在项目立项时,会强迫自己回答七个问题,并把答案写进设计文档。这些问题没有标准答案,但早做决定一定比晚做决定省事。
2.1 决策一:主模型怎么选,不是越贵越好
主模型选型是第一个要拍板的事。我的建议是不要只盯着跑分榜单,重点看三件事:工具调用成功率、结构化输出稳定性、长上下文下的信息保持能力。工具调用成功率直接决定了 Agent 能不能准确把意图映射到函数参数,结构化输出稳定度决定了后续解析代码要不要写一堆兜底逻辑,长上下文保持能力决定了几十轮对话后模型会不会把早期关键信息丢掉。
成本与延迟也要分场景。实时客服类 Agent 对延迟敏感,适合便宜且快的小模型;离线任务型 Agent 可以接受几十秒的思考时间,可以考虑更强推理能力的模型。我的经验是:日常工具调用场景,选择一个中端模型比如 gpt-4o-mini、DeepSeek、Qwen 系列就够用;只有多跳推理、复杂代码生成、长文档分析才需要换更强模型。结论是分层使用,不是一套模型打天下。
2.2 决策二:提示词工程怎么设计,顺序比辞藻重要
提示词怎么写,直接影响 Agent 的稳定性。我的顺序一直是:先写目标,再写边界,然后写行为公约,接着写工具说明,最后写输出格式。不要把大量笔墨花在人格化描述上,模型会把精力消耗在模仿语气上,反而忽略任务本身。系统提示词追求的是“一句话能讲清约束”,而不是文笔优美。
如果工具数量多,不要把工具说明全部塞进提示词。主流模型都支持 Function Schema 绑定,提示词里只需要写使用策略和禁忌,工具名称和参数描述交给模型平台的函数调用能力去处理。这个区分很重要,我见过有人把几十个工具的手写文档全放进 System Prompt,结果上下文被挤爆,模型反而开始幻觉。此外要防指令注入:工具返回的内容可能包含外部信息,要在提示词里明确要求模型把“数据”和“指令”区分开,不可执行的文本一律视为数据。
2.3 决策三:记忆层怎么分层存储,不能只靠一个列表
记忆实现上,我建议分两层:热数据和冷数据。热数据用 Redis,存放最近几轮对话的原文、用户 session 状态、任务中间变量;冷数据用向量数据库,存放跨会话需要检索的长期记忆和业务知识。目前我用得比较顺手的组合是 pgvector 加 Redis,pgvector 在 PostgreSQL 里做向量检索,少维护一个组件,Redis 负责热路径缓存。
上下文膨胀是个必现问题。当 Token 快满时,不要等模型报错,应该做一个滚动摘要动作:把早期对话交给一个小模型压缩成摘要,再拼上最近几轮原文,继续作为上下文输入。这个动作可以定时触发,也可以依据 token 统计触发。我在生产环境里把它做成了一个独立节点,效果比把历史全部塞给主模型好得多,因为主模型可以一直保持在它最擅长的推理工作区,而不是被迫去 processing 一堆老对话。
2.4 决策四:工具调用走哪个协议,功能调用还是 MCP
工具调用的接入方式有两种主流选择。第一种是各模型供应商原生支持的 Function Calling。它把工具描述以 JSON Schema 形式传给模型,模型生成结构化的调用请求,实现简单,走通一条链路只要半天。缺点是它和供应商平台绑定,切换模型时可能要做适配。第二种是 MCP(Model Context Protocol)这类标准化协议,它把工具能力变成可发现、可复用的服务,适合中大型项目或需要跨多个服务共享工具集的场景。
我对选型的经验是:MVP 阶段直接用 Function Calling,最快验证业务闭环;当工具数量超过十几个,或者工具由不同团队维护时,再迁移到 MCP。不要一开始就上标准化协议,除非团队里已经有多套服务要接入 Agent,否则学习成本大于收益。当然,如果项目当前就在做面向多Agent的工具市场这种基础设施,那 MCP 从第一天就要考虑进去。
2.5 决策五:任务编排用哪种模式,不是所有场景都需要 Agent
任务编排模式的选择,我总结成一句话:能用确定性流程解决的问题,别上 Agent;真需要动态决策的环节加 ReAct;需要跨步骤共享状态和失败恢复的,直接用有状态图。线性 Chain 适合流水线式处理,比如“输入 -> 清洗 -> 分类 -> 输出”,稳定高效但没有应变能力。ReAct 适合“查一下、观察结果、再查”这类动态循环,典型场景是搜索引擎问答。Plan-and-Execute 则先规划再执行,适合任务步骤较多、但每个步骤相对明确的场景。
LangGraph 这类图编排方案则把任务建模成节点和边,支持条件分支、循环、人工审批和状态持久化,几乎能覆盖上面所有模式。我的建议是:先按真实业务场景确定编排模式,再挑框架;而不是看着框架的宣传语倒推业务架构。实践中我常用混合结构,主流程用图编排,图内个别需要推理的节点用 ReAct,既保证可控,又保留灵活性。
| 编排方式 | 适合场景 | 优点 | 主要缺点 |
|---|---|---|---|
| Chain | 输入到输出的固定流水线 | 稳定、调试简单 | 无法处理分支与回退 |
| ReAct | 需要观察中间结果的问答/搜索 | 灵活、贴近模型天然推理 | 易陷入反复循环、Token 开销大 |
| Plan-and-Execute | 任务可预先拆解但步骤较多 | 全局规划更清晰 | 规划有错时整体失败率高 |
| Graph | 多任务、多状态、需要恢复的复杂流程 | 可控性最强、支持持久化 | 概念复杂,开发量稍大 |
2.6 决策六:并发与性能怎么设计,Agent 扛并发的问题在哪
“AI Agent 怎么扛并发”是很多人关心的问题,但它和普通 Web 服务扛并发有本质区别。你的服务本身可能很轻,真正的瓶颈几乎都在上游 LLM API 的 RPM 和 TPM 限制上。如果不对上游做保护,服务一扩容,限流就会立刻响应,接着就是请求堆积、超时重试、重试放大了流量,最后雪崩。所以 Agent 并发架构的第一要务是保护上游,而不是压满自己的吞吐。
我的基本架构是四件套:接口层全部异步化,用 FastAPI 的 async def 处理 HTTP 请求,遇到同步的 LangGraph 调用丢进线程池,避免阻塞事件循环;入口加令牌桶限流,按用户维度控制请求速率;任务量大的场景,把请求放入 Redis 队列,由 Worker 异步消费,客户端轮询结果;需要体验更好时,用 SSE 做流式输出,边生成边推送给用户。另外,相似请求可以做结果缓存,比如热点问题直接命中缓存,根本不打大模型。
有一个特别容易踩的坑:重试策略。无脑重试会把一次限流变成持续的重试风暴。正确做法是指数退避加抖动,并且区分可重试错误(429、超时)和不可重试错误(参数非法、认证失败)。我在项目里就是靠这个区分,把线上错误率降了一个数量级。
2.7 决策七:安全与可观测性,让 Agent 可控可查
让大模型自己决定调用所有工具,是一件很危险的事情。权限控制的核心是最小权限:每个 Agent 只能使用完成当前任务所必需的工具集合,写操作工具必须增加人工审批闸门。比如删除数据、发送消息、下单支付这类操作,流程上做成“Agent 生成意图,人工点击确认后才真正执行”。这既是对用户的保护,也是对自己系统的保护。
可观测性同样重要。每一个节点的输入输出都应该打快照,尤其是模型调用、工具调用和分支跳转。我建议从第一天就接入 trace 系统,无论是 LangSmith、Langfuse 还是自建日志链路,都要保证能回答三个问题:某次任务走到了哪个节点、模型视角看到了什么、工具真实返回了什么。没有 trace,Agent 就是一个黑盒,出了问题只能靠猜;有了 trace,大部分问题都能快速定位到是决策问题还是数据问题。
3. 最小可落地示例:用 FastAPI + LangGraph 搭一个带记忆的 Agent
理论部分讲了这么多,最后必须落到代码上。我用一个极简但完整的最小示例,展示一条能跑通的 Agent 链路:带工具调用、带记忆恢复、通过 HTTP 暴露服务。技术栈是 FastAPI 加 LangGraph,因为前者是异步友好的 Python Web 框架,后者把状态管理和条件分支做成了显式建模,非常贴近七要素里的规划与反思。
3.1 技术栈选型:为什么是 FastAPI 和 LangGraph
选择 FastAPI 是因为它天然支持 async,配合异步 HTTP 客户端和 SSE 流式输出,在 Agent 场景里比 Flask 和 Django 更顺手。选择 LangGraph 不是因为它最流行,而是它解决了其他方案很难处理的问题:状态管理。普通 ReAct 实现需要自己维护循环和上下文,LangGraph 则把状态、节点、边、检查点都做成了一等公民,尤其是检查点机制,让我可以用极少的代码做到用户断线后重连继续任务。
这个示例我会绑定一个模拟天气查询工具,模型收到“北京天气怎么样”这类问题时,会生成一次工具调用,工具节点执行完后把结果返回给模型,模型再组织语言回复。整个流程只有两个节点,但已经覆盖了七要素里的大部分内容:目标定义在系统提示词里、感知输入就是用户消息、工具和行动在同一个节点内、反思由条件边实现,记忆通过 Checkpointer 保存。
3.2 核心代码:状态图、工具调用和检查点
# app.py import asyncio from typing import TypedDict, Annotated from fastapi import FastAPI from pydantic import BaseModel from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver @tool def get_weather(city: str) -> str: """查询指定城市的当前天气。""" # 这里用模拟数据,实际项目中替换成真实天气 API return f"{city} 当前天气:晴,最高气温 28 摄氏度。" llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools([get_weather]) class AgentState(TypedDict): messages: Annotated[list, add_messages] def call_model(state: AgentState): result = llm.invoke(state["messages"]) return {"messages": [result]} def call_tool(state: AgentState): last_message = state["messages"][-1] if not last_message.tool_calls: return {"messages": []} tool_call = last_message.tool_calls[0] if tool_call["name"] == "get_weather": content = get_weather.invoke(tool_call["args"]) else: content = "未知工具" return { "messages": [ { "role": "tool", "name": tool_call["name"], "tool_call_id": tool_call["id"], "content": content, } ] } graph = StateGraph(AgentState) graph.add_node("model", call_model) graph.add_node("tool", call_tool) graph.add_edge("tool", "model") graph.add_conditional_edges( "model", lambda state: "tool" if state["messages"][-1].tool_calls else END, {"tool": "tool", END: END}, ) graph.set_entry_point("model") # MemorySaver 只能在单机内存中保存状态,生产环境建议换成 Redis 或 PostgreSQL 实现 compiled_agent = graph.compile(checkpointer=MemorySaver()) app = FastAPI() class ChatRequest(BaseModel): message: str thread_id: str = "default" @app.post("/chat") async def chat(request: ChatRequest): config = {"configurable": {"thread_id": request.thread_id}} output = await asyncio.to_thread( compiled_agent.invoke, {"messages": [HumanMessage(content=request.message)]}, config, ) return {"reply": output["messages"][-1].content}这段代码有几个关键点值得展开。bind_tools把工具的 JSON Schema 自动传给模型,这是决策四里说的 Function Calling 思路。add_messages是一个消息缩减器,LangGraph 在更新 state 时会把新旧消息正确合并,避免每次覆盖。MemorySaver是检查点实现,它会在每次节点执行后保存状态,即使中断,也能用同一个thread_id恢复上下文。
asyncio.to_thread的用法值得说明一下:compiled_agent.invoke是同步阻塞调用,直接放在 async 函数里会阻塞事件循环。用to_thread把它丢到线程池执行,HTTP 层才能继续并发处理其他请求。这是 FastAPI 集成 LangGraph 时常犯的一个错误,我在这里提前排掉。
3.3 本地跑起来:启动、调用、验证多轮记忆
pip install fastapi uvicorn langchain-openai langgraph export OPENAI_API_KEY="your-api-key" uvicorn app:app --host 127.0.0.1 --port 8000然后打开另一个终端:
curl -X POST http://127.0.0.1:8000/chat \ -H "Content-Type: application/json" \ -d '{"message": "北京天气怎么样", "thread_id": "user-123"}'第一次调用会看到模型先触发工具调用,然后返回类似“北京当前天气:晴,最高气温 28 摄氏度”的回复。接着你再发一次:
curl -X POST http://127.0.0.1:8000/chat \ -H "Content-Type: application/json" \ -d '{"message": "那上海呢", "thread_id": "user-123"}'因为用了同一个thread_id,第二次请求会自动带上第一次的对话历史,模型能理解“那上海呢”指的是查询上海天气。这就是 MemorySaver 带来的多轮记忆能力。如果把thread_id换成一个新值,就是一段全新对话,两条会话互不打扰。
4. 常见问题与排查实录:把故障变成经验库
任何 Agent 项目上线后都会遇到一批重复性故障。这里把我在实际项目中反复排查过的五个问题整理成一个速查表,再分享一些设计阶段的避坑心得。
4.1 高频故障现场:五个真实案例复盘
| 故障现象 | 根因分析 | 快速解决方案 |
|---|---|---|
| Agent 反复调用同一个工具,形成死循环 | 没有对工具返回做合法性校验,模型拿到错误结果后继续重试 | 在调用工具后增加结果校验节点,判断返回是否包含预期字段 |
| 对话进行到一半出现 Token 超限 | 上下文只增不减,没有做滚动摘要 | 在主循环里增加摘要节点,对话达到阈值时先压缩再继续 |
| 并发一上来就报上游 429 限流 | 没有限流和重试策略,请求全打到模型服务 | 加入令牌桶限流,对 429 使用指数退避加重试 |
| 用户断开后重新连接,上下文丢失 | 检查点没有持久化,状态停留在内存 | 把 MemorySaver 换成 Redis / PostgreSQL 实现的后端 |
| 工具返回了结构正确的数据但模型用错字段 | 工具返回 Schema 不够显式,模型靠猜 | 在工具返回中增加字段说明或使用严格 JSON Schema 解析 |
第一个案例特别典型。模型在对话中生成了一个工具调用,工具返回了一段格式不匹配的内容,模型没有意识到失败,又在下一个推理步骤里重复生成几乎一样的工具调用,整个流程卡死。解决办法是增加一个“结果校验节点”,在进入下一轮模型调用之前检查返回内容是否满足预期结构,不满足就重新构造工具结果并说明错误原因,逼模型换路径。
Token 超限的问题在长对话场景里几乎必现,尤其做客服和写作助手时。滚动摘要不是一次性把所有历史都删掉,而是把早期对话用一个小模型压缩成几条要点,再拼接最近两三轮的原文。这样既能保留关键信息,又能把主模型的上下文空间留出来处理最新任务。
4.2 避坑心得清单:设计阶段就避开的雷
复盘多个项目之后,我总结了几个设计阶段的实用心得。
第一,不要把全部逻辑押给模型。Agent 能动态决策不等于所有环节都要动态。固定流程用代码写死,动态决策才交给模型,这是最省钱也最稳定的组合。第二,工具的错误语义必须明确规定。返回值要区分“调用失败”“无结果”“数据不合法”三类情况,模型才能依据错误类型决定重试还是终止。第三,Trace 从第一天就接上。哪怕是个人项目,也要把每次模型调用、工具调用的输入输出记录下来,这些数据不光用于排查问题,也用于优化提示词和工具定义。第四,不要在日志里打 API Key,哪怕本地调试也要养成好习惯。第五,先跑通一条最小链路再扩展工具集。我第一次做 Agent 时就是一口气接了一堆工具,结果问题叠着问题,根本分不清是哪个环节出错。
最后再分享一个习惯
我自己做任何新 Agent 项目时,都会先用七要素画一张草图,再把七个决策点的答案填进一张表格,然后才开始写代码。这套流程帮我省掉了大量返工,也推荐你试试。折腾 Agent 的过程很像搭乐高,框架看起来谁都会,真正的差别在于你愿不愿意在每个细节上多想一层边界和异常。先想清楚目标和边界,再让它跑起来,这条经验在任何 Agent 项目里都适用。