AI Agent 这个词这两年都快被说烂了,但真到要自己动手做一个能跑的 Agent 时,很多人其实是一脸懵的。网上教程一大堆,有的讲概念,有的秀 Demo,可真到落地工程实现——要扛住并发、控制 token 成本、处理工具调用的各种意外——能说清楚的不多。这篇文章我想换个角度,把 AI Agent 拆成两条线来讲:一条是"七要素",解决的是"一个 Agent 到底由哪些部分组成";另一条是"七个决策点",解决的是"从 Demo 到生产环境,你必须在哪些地方做取舍"。我尽量按做工程的习惯来写,不堆术语,把每个环节的为什么、怎么选、怎么踩坑都讲透。不管你是刚入门想搭第一个 Agent,还是已经在用 LangChain、FastAPI 这类工具写业务代码,这篇应该都能给你一些能直接抄作业的东西。
1. 为什么把 Agent 拆成七要素和七个决策点
1.1 要素是"解剖",决策点是"施工"
我在带团队做 Agent 项目时发现一个规律:新人对 Agent 的困惑往往不在某一个具体技术上,而在脑子里的地图不清晰。他可能知道 LangChain 能调大模型,也知道 Function Calling 能让模型调用外部工具,但问他"你这个 Agent 的核心链路是什么",他答不上来。
所以我习惯先做一次解剖。七要素,就是一个 Agent 系统在逻辑上必须具备的七个模块:模型内核、记忆系统、规划能力、工具接口、行动执行器、感知上下文、反馈反思。这七个东西不是某些框架强加的,而是你只要把 Agent 当成一个"会使用工具完成目标的系统",就会自然拆出来的部分。
七要素解决的是"组成"问题,但它不解决"怎么做出来"的问题。所以我又提炼出七个决策点。工程实现拼的不是你懂多少概念,拼的是你在关键岔路口有没有做对选择。比如:你是用 LangGraph 做编排还是自己写状态机?你的记忆是全部塞进上下文还是接向量库?你的 Agent 服务是同步阻塞还是异步并发?这些决策点,每一个都直接影响系统的稳定性和成本。
1.2 两张图看懂两条主线的关系
你可以这样理解:七要素是 Agent 的"业务架构",七个决策点是 Agent 的"技术架构"和"运维架构"。业务架构告诉你需要什么,技术架构告诉你用什么实现,运维架构告诉你上线后怎么活着。
我经常用一个类比跟团队讲:七要素就像你要开一家餐厅需要的功能区域——厨房、食材仓库、点餐台、收银台。七个决策点则是你开店前要做的关键选择——选址是商场还是街边?厨房是开放还是封闭?菜品是标准化预制还是现炒?这些选择决定了你这家店能不能真的开起来,开了以后能不能赚钱。
我们现在看到的 AI Agent 主流架构,不管名字叫 ReAct、Plan-and-Execute、Multi-Agent,本质上都逃不出这七要素。区别只在于某些要素被强化了,某些要素被弱化了。比如 ReAct 架构强调的是规划和工具执行的循环,而聊天机器人可能把感知和记忆放到最大。
有了这两条主线,你再去看 LangGraph、Spring AI、Rust 乃至于扣子(Coze)这类低代码平台,就不会觉得乱——它们都是在不同层面帮你实现这七要素、帮你把七个决策点做成默认选项的工具。
2. AI Agent 的七要素拆解:从大脑到手脚
2.1 模型内核:你选的大模型决定了 Agent 的天花板
模型内核就是 Agent 的"大脑"。它负责理解用户意图、拆解任务、生成回复、判断下一步动作。市面上可选的不多但也不少,闭源有各类商用大模型,开源有 Llama、Qwen 等。选模型要看的指标很多,但工程上我认为四个最关键:上下文窗口大小、Function Calling 的稳定性、推理速度、以及价格。
上下文窗口决定了你一次能塞多少信息。比如 32K 和 128K 的模型,在工程上的处理方式完全不同——窗口小,你就得更积极地把历史对话压缩成摘要,或者把不用的工具定义踢出 prompt;窗口大,你的代码可以偷懒,但 token 成本会肉眼可见地涨。
Function Calling 稳定性尤其要重视。同样是让模型输出一个 JSON 格式的工具调用,不同模型的表现差异非常大。有的模型经常输出格式不对,你得写一整套修复逻辑去兜底;有的模型则几乎次次都规范。这个差异在开发期不明显,一旦上生产,请求量大了以后差距会被无限放大。
还有一点容易被忽略:模型不是越快越好,也不是越大越好。Agent 往往要跑多轮循环,一次任务可能调用模型十几次。这时候延迟是乘积关系,而不是加法关系。我曾经把一个 Agent 的主模型从大参数模型换成中等规模的模型,最终任务完成时间从 40 秒降到了 12 秒,效果只下降了一点点——这种事你不到真实负载下是感受不到的。
2.2 记忆系统:短期上下文和长期存储的配合
没有记忆的 Agent 就像金鱼,聊完就忘。但记忆在工程上不是简单存两个文件就行,它要分短期和长期。
短期记忆就是你当前这轮对话里的所有信息,通常以消息列表的形式存在。它有两个麻烦:一是上下文窗口有限,塞不下就得截断,截断就可能丢关键信息;二是每发一轮都要把全部历史发给模型,token 消耗是累加的。所以实际工程里,我们会做上下文压缩。压缩有两种思路:一种是滑动窗口,只保留最近 N 轮;另一种是摘要记忆,让模型把旧对话总结成几句话再塞回上下文。两种可以混用,我的经验是"滑动窗口为主,摘要为兜底"。
长期记忆则是跨会话的信息存储。最常见的做法是把关键事实抽出来放进向量数据库,下次用户提到相关话题时,用相似度检索把最相关的记忆片段带回上下文。这里要注意,向量检索不是万能的。它对"语义相似"的问题好使,但对"精确的事实查询"(比如用户上个月指定的某个具体偏好)经常答非所问。所以在设计长期记忆架构时,我建议做两层:一层是向量检索,做"泛化联想";另一层是结构化存储(比如数据库里的键值表),做"精确定位"。很多 Agent 项目死在记忆上,不是技术实现不了,而是没想清楚哪些记忆走向量、哪些走结构化。
2.3 规划与推理:让模型学会"先想后做"
规划能力是 Agent 区别于普通聊天机器人的核心特征。普通对话是你问一句我答一句,Agent 则需要把一个复杂目标分解为多个步骤,然后按顺序或按依赖关系执行。
主流实现有三种模式。第一种是 ReAct,即推理和行动交替进行——模型先思考一步,然后调一个工具,看到结果再思考下一步。这个模式实现简单,LangChain 里开箱即用,缺点是 token 费得厉害,而且容易在多步循环里走偏。第二种是 Plan-and-Execute,先把完整计划列出来,再一步步执行。这种方式省 token,但计划一旦出错,纠错能力弱。第三种是基于图的编排,代表就是 LangGraph——它把 Agent 的不同步骤定义成图里的节点,节点之间用边连接,可以显式地控制分支、循环和并行。我在生产项目里大多数情况用 LangGraph,因为它给了你一个足够底层的控制面,既保留了 ReAct 的灵活性,又能对流程做结构化约束。
规划这块最核心的坑是"无限循环"。模型在工具调用链上一旦进入到死循环,不仅任务完不成,token 还哗哗地烧。所以工程上必须加兜底机制:最大迭代次数、单次工具调用超时、以及循环检测。这些不是可选项,是必须项。
2.4 工具与接口:Function Calling 的工程化
工具是 Agent 伸向外部世界的触手,可以是搜索引擎、数据库查询、HTTP API、代码解释器。Agent 要使用工具,依赖的是模型的 Function Calling 能力——模型在生成回复时,不是直接说"我帮你查了",而是输出一个结构化指令,比如{"name": "search_products", "arguments": {"query": "xxx"}}。
工程上,工具层要做的事比大多数人想象的多。首先,你要对工具做注册和描述,描述写得好不好直接决定模型能不能正确选用工具。我踩过一个大坑:有一个查询订单的工具,描述写的是"查询用户订单信息",结果模型经常在用户问物流的时候也去调它。后来我把描述改成"查询订单的基础信息(订单号、金额、状态),不包含物流信息,物流请调用查询物流工具",问题就基本消失了。你给模型的工具描述,就是给用户看的说明书,写得不清楚不能怪模型乱来。
其次,工具调用的参数校验必须放在模型之外做。模型生成的 JSON 经常会缺字段、多字段、字段类型不对。你一旦信任了模型的输出,后面代码大概率直接崩。正确的做法是:模型输出工具调用 → 你用 Pydantic 这类库做严格校验 → 校验失败则重新生成或直接报错。还有一个细节:工具执行结果返回给模型时,一定要做截断。一个接口可能返回几十 KB 数据,全塞回上下文既费 token 又干扰模型判断。我会把返回结果做裁剪、汇总,只提取对下一步决策有用的部分。
2.5 行动执行器:从"决定"到"做成"
规划做好、工具调起来了,还需要一个行动执行器把这些散落的动作编排成一个闭环。执行器的职责是:接收模型输出的意图、调度对应的工具、拿到结果、再交还给模型做下一轮推理。
我们用 FastAPI 写过这类执行器。它内部是个状态机,维护当前会话的上下文、已执行的步骤、待执行的队列。执行循环大致是:初始化系统 prompt → 把当前状态丢给模型 → 模型决定是调用工具还是结束 → 如果调用工具则执行并返回结果 → 再丢回给模型 → 直到模型输出最终答案。
行动执行器还有一个容易被忽略的功能:异常兜底。工具可能超时、可能抛异常、可能返回脏数据。这些异常一旦发生,不能让整个 Agent 卡死。我的经验是在执行器里统一捕获异常,然后把异常信息作为"工具执行失败"的结果回传给模型,让模型决定是换一种工具、换一个参数、还是如实告诉用户执行失败。这比起直接中止任务要友好得多。
2.6 感知与上下文:Agent 的"眼睛"和"耳朵"
感知层负责把外部世界的信息转成模型能理解的形式。最基础的感知就是用户的自然语言输入和对话历史,但真正的 Agent 往往还需要感知更多:用户上传的文件、网页内容、系统日志、甚至图片。
工程上,感知层最关键的是"信息转文本"的能力。比如你接入一个图片输入,通常要先把图片转成 OCR 文本或者视觉模型的描述;你接入一个网页,要先把 HTML 清洗成纯文本或 Markdown;你接入一份 PDF,要先把内容抽取出来并做分块。这一步不做干净,后面所有环节都会被脏数据拖垮。
同时感知层要做好"目标相关性过滤"。不是所有感知到的信息都要传给模型,信息越多,模型的注意力越分散,越容易给出模棱两可的回答。我会在感知层做一个简单的相关性打分,只把跟当前任务相关的上下文送入模型。这有点像一个参谋,帮领导把材料筛过一遍再递上去。
2.7 反馈与反思:让 Agent 学会自我纠错
最后一个要素是最容易被忽略的:反馈与反思。Agent 不是一把梭跑完就结束,而是需要具备一定程度的自我监督和纠错能力。
实现方式有两种。一种是过程式反思,即在每一步工具调用后,把执行结果和预期目标做对比,出现偏差就重新规划。比如模型计划用搜索工具查资料,结果搜到的内容跟问题毫无关系,这时 Agent 应该意识到"搜索词可能不对",然后换个词再搜一次。另一种是结果式反思,即在最终答案生成后,让模型做一遍自检——换个角度重新审视自己的回答是否完整、是否互相矛盾。这个过程可以用一个独立的"Critic"模型来完成,也就是双模型互评,效果往往比单模型自评好不少。
反思层不是免费的午餐,它消耗额外 token 和延迟。所以我也建议做分级:简单任务跳过反思,复杂任务必须反思。如何判断复杂和简单?可以在执行器里对工具调用数量做个计数,超过阈值就强制走反思流程。
3. 工程实现的七个决策点:每个岔路都要选对
3.1 决策一:Agent 编排架构怎么选
第一个决策点选架构。你面前有几个大方向:一是直接用 LangChain / LangGraph 之类的现成编排框架;二是用 Spring AI 这种融入 Java 生态的方案;三是在 Rust 之类的语言里自己实现核心运行时;四是干脆用扣子这类低代码平台先验证业务逻辑。
我的建议是分场景。如果你要快速验证 Agent 产品逻辑,团队又偏中小型,那么 LangGraph 是最稳妥的选择,生态成熟、资料多、社区踩坑案例丰富,而且它对 FastAPI 这类异步服务集成很好。如果你是 Java 技术栈的企业项目,服务发现、配置中心、监控都已经围绕 Spring Boot 建好了,那 Spring AI 的 Agent 模块能让你少搭很多桥。Rust 的话,我的观点是它能给你极致的性能,但编排层的生态还不够好,适合做高吞吐的运行时内核,不适合直接拿来写复杂业务逻辑。扣子这类平台适合做原型验证,不是不能上生产,但你在灵活性、私有化、成本控制上都会受约束。
架构选型我给出一个底层逻辑:你的核心瓶颈在哪里。如果你的 Agent 要频繁多步编排、状态复杂、循环分支多,就选图编排的框架,别自己做状态机;如果你的瓶颈是单次请求延迟、并发吞吐,那就别在框架层纠结,去优化模型选型、连接池和异步模型。
3.2 决策二:模型与 token 成本怎么算
很多人第一次被 AI 账单吓到,是在把 Agent 部署上线的第二周。原因很简单:Agent 不是一次对话,而是一连串对话和工具调用。一次复杂任务,模型可能要来回十次。每次来回,累加的是用户问题、Agent 自己的思考、工具返回结果。这里面最大的坑就是"上下文膨胀"。
先说 token 是什么——它是模型处理文本的基本单位,可以是半个字、一个字或几个字符,具体要看模型的分词器。通常我们可以粗略估算:1 个中文字约等于 1 到 2 个 token,1 个英文单词约等于 1 到 1.5 个 token。计费就是按输入和输出的 token 数量分别计费,不同模型价格差异巨大。
工程上控制 token 成本有几个常规操作。第一,压缩工具返回——在工具结果进入模型前做摘要、截断,把有效信息提取出来。第二,设置迭代上限——一个任务最多允许几次工具调用,超过就强制收敛,避免空转烧钱。第三,做 prompt 缓存——系统提示词和工具定义在多次请求中是不变的,利用一些模型服务提供的上下文缓存能力,可以大比例降低成本。第四,分清主模型和辅助模型——摘要、标题生成这种简单任务,用便宜的小模型去做,大模型只处理核心推理。这四条组合起来,Agent 的推理成本通常能降到原来的三分之一不到。
3.3 决策三:记忆放本地还是向量库
记忆方案是 Agent 工程里最容易"过度设计"的地方。我一向的理念是:能用上下文窗口解决的,就不要上向量库。
很多项目一上来就接向量数据库,把历史对话、文档全部灌进去。结果相似度检索不准、数据同步麻烦、额外运维成本高,最后效果还不如直接拿上下文窗口硬扛。向量库真正应该解决的是"跨会话、跨用户的长期记忆",也就是用户的偏好、历史操作、知识库文档。如果只是一次会话里的聊天记录,那用内存里的消息列表就足够了。
当你的确需要长期记忆时,决策点在于用哪种向量库、分块多大、怎么更新。一般我们会先用轻量的开源方案(比如本地索引类工具)来支撑小规模,等数据量上来了再迁移到独立的向量数据库服务。分块大小直接影响检索质量:分块太大,检索召回内容不够精确;分块太小,语义不完整。我常用的套路是先按段落切,再根据段落长度动态合并,控制在 300 到 500 token 左右一块。
记忆更新策略同样要想清楚,不是每轮对话都往长期记忆里写。我会给记忆写入设一个"重要阈值"——只有模型判定为"用户明确表达出的偏好或关键事实"才写入,噪音数据写进去只会污染后续检索结果。
3.4 决策四:工具调用怎么设计才可靠
工具调用的设计直接决定 Agent 的可用性。这块我有几条硬经验。第一,工具函数必须是"描述优先"的设计——每个工具都要有清晰的名字、描述、参数说明和必要的示例。模型是靠着这些元信息来决定要不要调用你的工具,写得不清楚,表现一定拉胯。第二,每个工具要做超时控制,网络请求尤其要设上限,不然 Agent 会卡在某个第三方接口的响应上。第三,工具的执行结果要标准化——无论内部实现是什么,返给模型最好统一成一个包含"状态、内容摘要、原始数据引用"的数据结构。这样可以避免模型拿到乱七八糟的格式后自己瞎推理。
还有一个决策点是工具的粒度。工具定得太粗,一个工具内部逻辑太庞大,模型难以判断什么时候该调;工具定得太细,调用次数变多,token 消耗和延迟上升。我的经验是一个工具最好只做一件逻辑内聚的事,粒度控制在"一次工具调用能让 Agent 往前推进一步"。如果一步要做的事情太多,宁可把它拆成多个工具,也别做一个巨无霸。
3.5 决策五:Agent 怎么扛并发
这是全网被问烂的问题:"AI Agent 怎么扛并发?"答案不是简单的"加个线程池",而是要理清楚你的瓶颈在哪里。Agent 请求的本质是长时间占用的计算和资源——一次请求可能要好几秒、甚至几十秒,里面涉及多次模型调用和多次工具调用。这种长请求对并发模型很不友好。
我的经验是把 Agent 服务拆成异步处理链路。底层模型调用做成异步非阻塞,工具调用做成带超时控制的并发任务。Web 框架选 FastAPI 这类原生支持 async 的,而不是传统的同步框架。同时在服务层面要接入任务队列,把重请求放进去异步执行,客户端拿任务状态轮询或者通过回调拿结果。
并发这边还有一个大坑:模型服务的连接池和限流。如果你直接往模型服务发请求,不管理连接,几百个并发请求一到,连接池直接被打爆,大量请求超时重试,雪崩就来了。要给模型客户端设置合理的连接池大小和超时,另外在 Agent 入口做信号量控制,打到阈值就排队或拒绝,而不是让所有请求都挤进模型调用层。
对于那些需要彻底吃透并发问题的项目,我建议再去看看一些云厂商发布的 Agent 白皮书,里面通常有比较好的参考架构。但核心思路永远是那几条:异步化、连接池管理、排队限流、以及把长任务拆出同步接口。
3.6 决策六:部署形态和依赖怎么管
Agent 部署的决策点,很大程度上取决于你的使用场景。如果你是在已有 Django 单体项目里加一个 Agent 能力,没必要非得拆成微服务——直接在应用内集成一个 Agent 模块,对外暴露一个接口,依然可以跑得很好。如果你要用 FastAPI 单独做一个 Agent 服务,那 Docker 容器化是标配,Model 服务、向量库、Redis 这些依赖要单独部署和管理。
部署层面我特别想提醒的是环境变量和密钥管理。模型的 API Key、数据库密码、第三方服务的密钥,千万不要写进代码。我们出现过把 API Key 提交到 Git 仓库,第二天账单爆掉的事故。现在不管是大项目还是小项目,我都强制要求密钥走环境变量或者专门的密钥管理服务。
另一点是模型服务的高可用。生产环境一定不要只配置一个模型服务地址,要有主备切换或者多供应商容灾。现在很多项目在这上面吃过亏——上游模型服务一抖动,整个 Agent 就不可用了。用 FastAPI 那一层做一层代理,把模型调用统一包装,至少能做到切换供应商时,业务代码不用改。
3.7 决策七:可观测性怎么从一锅粥到明白账
Agent 应用调试的痛,做过的人都知道。模型输出是概率性的,同样的输入可能跑出完全不同的路径。出了错,你不知道是模型推理错了、工具调用错了、还是上下文丢了。所以可观测性必须从第一天就做,不是上线之后补的。
我的做法是给每个 Agent 请求生成一个 trace ID,从请求进入服务开始,到最终返回,把中间一步步的关键事件全部记录下来:模型的输入输出、工具调用参数、工具返回的摘要、上下文的 token 数量、每一步的耗时。这些日志统一汇聚到日志系统。出问题的时候,不是靠瞎猜,而是靠回放 trace。如果你在 LangGraph 上做编排,它本身有跟 LangSmith 之类的追踪工具集成,也可以用。但我更在意的是把 trace 打成结构化数据,方便自己做报表和告警——点击率、成功率、平均耗时、平均 token 数、工具调用失败率,这些都是判断 Agent 健康度的核心指标。
评估也是可观测性的一部分。你不能只靠几个手工测试用例来评判 Agent 的好坏。现在比较通用的做法是搭一个评测集,里面放几十上百条真实任务场景,每次调整 prompt、换模型、改工具后,批量跑一遍评测集,拿通过率说话。这套体系搭起来以后,你改代码的胆子都会大很多。
4. 实操:搭一个能跑起来的 Agent 工程骨架
4.1 技术栈选择和项目结构
我也不空谈,直接给一个我实际操作过的最小可运行方案。技术栈就用 FastAPI + LangGraph + LangChain 的模型接入。FastAPI 管 HTTP 层,LangGraph 管 Agent 编排,模型接口走 LangChain 的统一封装,这样以后换模型可控。
项目结构大致是这样的:
agent_service/ ├─ main.py # FastAPI 入口 ├─ agent/ │ ├─ graph.py # LangGraph 图定义 │ ├─ nodes.py # 各节点处理逻辑 │ ├─ tools.py # 工具注册与实现 │ └─ state.py # Agent 状态定义 ├─ config.py # 配置和密钥读取 └─ requirements.txt我建议把状态定义单独放一个文件。Agent 的 state 是贯穿全局的东西,它包含当前消息列表、已调用的工具记录、中间结果、任务计划。把这个数据结构定义得足够清晰,后面写节点函数就顺很多。
4.2 用 LangGraph 实现一个带工具调用的 Agent
下面是一个简化但真实可跑的示例。先定义状态:
from typing import TypedDict, List class AgentState(TypedDict): messages: List[dict] tool_calls: int final_answer: str这里我们用tool_calls记录已经调用的工具次数,专门用来做迭代上限控制。
定义工具,一个简单的天气查询工具,模拟真实中你接入外部 API 的形态:
import requests def get_weather(city: str) -> str: """查询城市天气""" # 真实场景这里会请求外部 API,注意加超时和异常处理 try: resp = requests.get( f"https://api.example.com/weather/{city}", timeout=5 ) data = resp.json() return f"{city} 当前天气:{data['desc']},温度 {data['temp']} 度" except Exception as e: return f"天气查询失败:{e}"注意我给工具返回的都是字符串,这是刻意为之——LangGraph 里工具返回给模型的本来就是一个字符串,你想返回别的结构也行,但字符串最简单直接。
然后是核心的图逻辑:
from langgraph.graph import StateGraph, END from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langchain_openai import ChatOpenAI from langchain_core.tools import tool # 用装饰器把函数注册成模型可识别的工具 @tool def get_weather(city: str) -> str: """有用的天气查询工具。输入城市名,返回该城市当前天气信息。""" return f"{city} 的天气请参考接口返回" # 模型初始化,注意这里使用环境变量注入 key llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) llm_with_tools = llm.bind_tools([get_weather]) def agent_node(state: AgentState): """Agent 核心节点:把当前消息交给模型,让模型决定是调用工具还是输出答案。""" messages = state["messages"] # 限制单次任务调用工具次数,防止死循环烧 token if state.get("tool_calls", 0) >= 5: return {"messages": [AIMessage(content="任务已完成,不再接受新工具调用。")]} response = llm_with_tools.invoke(messages) # 把模型响应追加到消息列表 updated = messages + [response] # 如果模型决定调用工具,则转换到工具执行节点;否则结束 if response.tool_calls: return {"messages": updated, "tool_calls": state.get("tool_calls", 0) + 1} return {"messages": updated, "final_answer": response.content} def tool_node(state: AgentState): """工具执行节点:解析模型的工具调用,执行对应函数,并把结果返回给模型。""" messages = state["messages"] last_ai_message = messages[-1] # 收集模型发起的工具调用 result = [] for call in last_ai_message.tool_calls: if call["name"] == "get_weather": city = call["args"].get("city") # 真正执行工具,注意这里是同步调用,生产环境要改造成异步或丢队列 observation = get_weather.invoke({"city": city}) result.append(ToolMessage(content=observation, tool_call_id=call["id"])) return {"messages": messages + result} # 构建图 graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", tool_node) graph.set_entry_point("agent") # 条件边:模型调用了工具就去 tools,否则直接结束 graph.add_conditional_edge( "agent", lambda state: "tools" if state["messages"][-1].tool_calls else END, {"tools": "tools", END: END} ) graph.add_edge("tools", "agent") app = graph.compile()这段代码里,最关键的是那个条件边。LangGraph 会把你的状态在每个节点之间传递,你不需要手动维护循环,它天然支持"模型调用工具 → 工具执行 → 再回到模型"这种循环结构。而tool_calls字段的递增,保证了哪怕模型抽风,它也不可能无限循环下去。
4.3 用 FastAPI 把 Agent 包成服务
图编译好了,剩下就是包一个 HTTP 接口。我这里直接上异步版本,这是能不能扛住并发的关键差异点:
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class QueryRequest(BaseModel): user_id: str message: str class QueryResponse(BaseModel): answer: str trace_id: str @app.post("/agent", response_model=QueryResponse) async def run_agent(req: QueryRequest): # 这里生产环境建议把 trace_id 用中间件生成并注入 trace_id = f"trace-{req.user_id}-{time.time()}" initial_state = { "messages": [HumanMessage(content=req.message)], "tool_calls": 0, } # 注意:graph.ainvoke 是异步执行,不会阻塞事件循环 result = await app.ainvoke(initial_state) return QueryResponse(answer=result.get("final_answer", ""), trace_id=trace_id)这一步做完,你就拥有了一个能通过 HTTP 调用的 Agent。这套东西离生产还差几步:任务队列、并发控制、日志采集、模型调用限流。但骨架已经是正确的。
4.4 并发参数怎么调:一个亲测的调优过程
我在本地跑这个服务时,最初一压测就发现延迟飙高。后来排查出的原因有三个:一是get_weather用的是同步requests库,会阻塞事件循环;二是模型客户端的默认连接池太小;三是我没做入口限流,所有请求都直接往后灌。
第一个问题的解法很简单,是把同步的requests换成httpx.AsyncClient,或者用asyncio.to_thread把同步调用丢到线程池,别让它卡住事件循环。第二个问题,用openai的客户端时显式设置max_connections。第三个问题,在 FastAPI 入口加一个asyncio.Semaphore(20),同时进程内排队超过一定数量就直接返回 429。
调完这三处,同样的压测场景,吞吐量大概上了一个数量级。这块我想强调:并发问题一定是"链路问题",不是某一个环节的问题。你把入口限流设得再高,底层模型连接池不够,照样倒;你把连接池调大,入口不做限流,照样雪崩。必须从入口到出口全链路做约束。
5. 常见问题排查与避坑实录
5.1 问题清单速查表
| 症状 | 可能原因 | 处理方式 |
|---|---|---|
| Agent 反复调用同一个工具停不下来 | 模型陷入了循环、工具描述有歧义 | 加最大迭代上限;优化工具描述,明确适用边界 |
| 模型输出 JSON 格式错误 | 模型 Function Calling 不稳 | 用 Pydantic 严格校验;校验失败时让模型重新生成一次 |
| 上下文爆满、token 成本飙升 | 工具返回太长、历史消息无压缩 | 对工具结果做摘要截断;历史消息做滑动窗口或摘要记忆 |
| 并发一高就大量超时 | 同步阻塞、连接池太小、入口无限流 | 全链路异步化;调大模型连接池;入口加信号量 |
| 工具返回结果模型"看不见" | tool_call_id 未正确对应 | 检查 ToolMessage 必须带正确的 tool_call_id,与模型输出严格匹配 |
| 向量库检索结果和问题无关 | 分块粒度不合适、脏数据进入记忆 | 调整分块策略;提高记忆写入阈值,减少噪音 |
| 同一个问题两次回答差异巨大 | temperature 过高、缺少评估体系 | 调低 temperature;建立评测集做回归 |
5.2 最容易被忽视的 token 黑洞
很多人以为 token 主要是聊天记录的消耗,实际上真正的 token 黑洞是工具描述和工具返回。工具描述存在每个请求里,一个 Agent 挂五个工具,每个工具描述一两百字,乘以每次调用的次数,这部分的固定消耗非常可观。工具返回更是如此——第三方接口返回一长串 JSON,你如果不过滤直接塞回去,一次调用可能就烧掉几千 token。
我现在的工程习惯是把工具描述控制在"两句话以内",必须说清楚用途、参数格式和返回内容格式。工具返回在进入上下文前,能截断就截断,能摘要就摘要。你自己人看都费劲的长文本,模型看起来也费劲,而且更加烧钱。
5.3 规划失控后的兜底机制
Agent 的规划模块是意外最多的环节。模型可能会规划出错误的步骤,可能会重复执行同一个动作,也可能会在某个分支里越走越远。我给你一个兜底的组合拳:第一,总数上限——单次任务最多 N 步,超过强制收敛;第二步数限制——同一个工具最多调用 M 次,超过则禁止再次调用;第三,时间预算——整个 Agent 任务超过 T 秒没完成,直接超时返回部分结果并注明;第四,错误路径回退——工具连续失败两次后,自动让模型重新规划而不是继续尝试。
这套组合拳我建议在写 Agent 的第一天就加进去,而不是等出了问题再回头补。生产环境里这些兜底不是束缚,而是保命绳。你的 Agent 逻辑越复杂,这些机制就越重要。
5.4 两个特别容易踩的场景
很多人会拿 Agent 做一些自动化操作,比如让它定时去某个平台发内容。我在这里想说一句:技术上完全可行,但务必要先摸清目标平台的规则。Agent 的自动化程度高、频率快、行为模式固定,一旦触发平台的风控规则,轻则功能失效,重则账号被处理。我见过不止一个团队把这类自动化做成灰色操作,最后吃大亏。正确的姿势是:只做平台规则明确允许范围内的自动化,或者把 Agent 的定位放在"辅助生成内容"而不是"代替人发布"。
另一个场景是拿 Agent 去做投资决策。个人用 Agent 做期货交易,从技术上讲就是接行情数据、跑策略模型、下委托单,能做到;但真要拿真金白银去跑,我强烈建议打消这个念头。原因不是技术不行,而是这类交易涉及资金安全和强监管,模型预测的高不确定性被 Agent 放大以后,亏钱速度比人肉操作快得多。如果要研究,就用模拟盘,千万不要实盘。这不是技术博主的说教,是我见过太多人在这个坑里摔得血肉模糊后的真心话。
6. 从入门到落地:学习路线的个人建议
6.1 给新手的逐步路线
经常有人问我 AI Agent 的学习路线。我自己的建议是分四步走。
第一步,先不碰代码,先玩明白一个低代码平台(比如扣子/Coze),在上面把"什么是工具、什么是流程编排、什么是记忆快取"体验一遍。这个阶段目标不是写生产代码,而是建立"Agent 能做什么、不能做什么"的直觉。
第二步,用 Python + LangChain 跑通一个最简 ReAct Agent。目标是把 LangChain 的模型封装、工具绑定、消息流走一遍,理解一次完整调用背后发生了什么。
第三步,切换到 LangGraph,把同一个 Agent 从链式改成图式。这一步的关键是理解状态管理和条件路由。你会开始理解为什么 Agent 需要显式的状态机,而不是一把梭的回调链。
第四步,才进入工程化,学 FastAPI、Docker、可观测性、压测。这四步走完,你再去看那些折腾 Rust Agent、Spring AI Agent 的项目,就不会觉得神秘——无非是用不同的语言和生态去实现同样的七要素。
这个学习过程没有捷径,但可以有更短的路径:一定不要一上来就研究最新最复杂的 Multi-Agent 框架。先把单 Agent 的边界和坑摸清楚,比任何花活都重要。多 Agent 协作是建立在对单 Agent 的深刻理解之上的,跳过基础直接穿梭于多个 Agent 之间,最后大概率连问题出在哪个 Agent 身上都查不明白。
6.2 别让技术热情冲昏头脑:边界意识
Agent 技术发展速度很快,但工程落地时边界意识比技术热情重要得多。边界意识包括三层:一是能力边界——你的 Agent 不是全能的,它的表现受制于模型、工具、数据质量三个环节,任何一个短板都会拖后腿;二是成本边界——一个 Agent 任务几块钱的时代还没过去,你设计的每一步都要换算成钱;三是合规边界——自动化操作外部平台、做受强监管的业务,都先在规则允许的范围内进行,别给自己惹麻烦。
我把这三条边界放在最后说,是因为我见过太多人把精力全花在模型调优上,最后项目却死在某个不起眼的现实约束上。技术人最容易犯的错是以为只要技术够强就万事大吉,但真实世界里的工程,从来是技术、成本、规则三者博弈之后的结果。
6.3 最后说一点我的体感
做 AI Agent 工程快三年,最深的感受是:Agent 真正难的从来不是"让它跑起来",而是"让它在真实环境里稳定地跑下去"。demo 谁都能做,但并发一上来就崩、token 一多就爆、上下文稍微变长模型就开始胡说,这些才是工程人真正的战场。每次有人问我该学什么框架、要不要追新出的模型,我的回答都一样:先把一套最小闭环吃透,把七要素的每个环节都亲手调一遍,把七个决策点的坑都亲自踩一遍。踩过这些坑,你手里才有真正属于你自己的工程手感,而不是停留在"看过很多教程,仍然做不好一个 Agent"的状态。希望这篇带着实操气味的拆解,能让你少走我走过的弯路。