☰
AI Agent从最小循环到可靠系统:工程化实战与踩坑记录
2026/10/7 18:08:10 网站建设 项目流程

做AI Agent这块也有段时间了,从最早用裸Prompt硬怼,到后面上LangChain、LangGraph这种框架,再到自己在生产环境里搭一套带监控、带限流的Agent服务,一路踩过不少坑。我最大的体会是:Agent这个事,真正值钱的部分不是那个“智能”,而是把智能变成可靠服务的过程。标题里写了“从最小循环到可靠系统”,这其实是一条非常清晰的进阶路径——先把最内核的循环跑通,再一步步把稳定性、可观测性、并发控制这些东西长上去。这篇文章我就按这个思路,把AI Agent的基础知识、主干架构和工程化要点串一遍,每个环节都附上我自己的实操经验和踩坑记录,希望能帮正在学习或者刚接手Agent项目的朋友少走弯路。

1. 先把“最小循环”讲清楚

1.1 这个标题背后的核心思路

很多人一上来就聊Agent框架、多智能体协作、记忆系统,这些当然重要,但我觉得一个新手最该先搞明白的,是Agent的最基本工作循环是什么。

所谓“最小循环”,其实就是四个节点:模型推理 → 决策是否调用工具 → 执行工具 → 把结果交回给模型。这个循环反复转起来,Agent就具备了“想一下 → 动一下 → 看一下结果 → 再想下一步”的能力。跟普通的对话机器人本质区别就在这里:普通对话是“一问一答”,Agent是“一问 → 多步行动 → 最终作答”。

我见过不少人把Agent想得很玄乎,觉得它像一个人一样在思考。实际上你把框架扒开,底层就是这么个循环。不管你是用LangGraph、AutoGen、CrewAI,还是自己用FastAPI裸写,最终都要落到这个循环上来。把这个循环的代码一笔一笔写清楚、写可控,后面所有的架构升级才有地基。

1.2 最小工作循环的四个节点

我用一个最朴素的方式来描述这个循环:

  • 推理(Think):把当前的任务、历史上下文、可用的工具信息一起丢给大模型,让模型输出下一步动作。这一步消耗的token最多,也是整个循环的“大脑中枢”。
  • 决策(Decide):模型输出的内容可以分成两类,一类是“我直接回答你”,一类是“我需要调用某个工具”。Agent要做的是解析模型的输出结构,判断当前处于哪个分支。
  • 执行(Act):如果模型决定调用工具,就把工具名和参数抽取出来,在本地或远程执行对应的函数或API,拿到一个结构化结果。
  • 回填(Observe):把工具执行结果作为新的上下文消息,拼接回对话历史,再次触发推理。这个Observe的结果会直接影响模型下一轮判断。

这四个节点形成了一个带反馈的闭环,跟传感器-控制器-执行器的逻辑很像。区别在于,传统控制系统的“控制器”是规则代码,而Agent的“控制器”是大模型,规则变成了概率,所以后面要解决可靠性的问题就会更复杂。

1.3 一套能跑起来的最小代码

市面上很多框架把Agent封装成了一行代码调用,新手看着很爽,但出了问题就抓瞎。我的建议是自己先写一个最朴素的循环,哪怕只有几十行。这个过程会让你对“Agent的原理”有脱胎换骨的理解。

用Python的伪代码来演示,大概是这个样子:

import json from llm import chat_completion # 假设的LLM调用函数 def run_agent(task, tools, max_steps=10): messages = [{"role": "user", "content": task}] for step in range(max_steps): resp = chat_completion(messages, tools=tools) msg = resp["choices"][0]["message"] messages.append(msg) # 判断模型是否发出工具调用 if msg.get("tool_calls"): for call in msg["tool_calls"]: fn_name = call["function"]["name"] fn_args = json.loads(call["function"]["arguments"]) # 执行工具 result = execute_tool(fn_name, fn_args) # 回填观察结果 messages.append({ "role": "tool", "tool_call_id": call["id"], "content": json.dumps(result) }) else: # 没有工具调用,说明Agent认为可以作答了 return msg["content"] raise TimeoutError("Agent exceeds max steps")

这个代码有几个细节值得注意:

  • max_steps限制:必须设置,否则万一模型陷入循环,你的成本和延迟会双双爆炸。
  • tool_call_id回填:多轮工具调用时,模型是靠这个ID来关联工具结果的。漏掉它,新一轮对话就会出现错乱。
  • json.dumps序列化:工具返回结果可能是dict,但模型期望的是字符串,而且最好是紧凑的JSON,能省token。

这段代码我建议所有人都亲手跑一遍。你把OpenAI或者DeepSeek的API接进去,写两个简单的工具函数,比如查天气、算加法,然后让Agent完成“把12和23相加后再查北京天气”这种多步任务。当你在终端看到模型自己决定先调加法、再调天气、最后总结回答的时候,AI Agent的核心机制你就真正吃透了。

2. 工具调用与循环的第一次升级

2.1 LLM为什么需要工具

大模型本身的训练数据有时间截止点,它不掌握实时信息;它的数学计算能力也相当有限,让它算复杂数字十有八九出错。工具就是给模型插上的“外接硬件”。

我在实际项目里常用这样一句话来跟同事解释:模型的文本生成能力是灵魂,但工具是手脚。没有灵魂是僵尸,没有手脚就只能在原地打转。

工具不仅能补齐实时性和精确性,还能补齐权限边界。比如你想让Agent查询公司内部数据库,你不可能把数据库连接串直接写进Prompt让模型去“理解”,你只会给它一个封装好的query_database()工具,参数是SQL语句。这样权限可控、错误可追踪、结果可审计。这也是Agent项目在企业落地时最被看重的一点。

所以在设计Agent系统时,工具层往往是你第一个需要认真打磨的模块。一个设计良好的工具集合,能让Agent高效地完成任务;反之,工具描述含糊、参数混乱,模型就像拿着看不懂的说明书在操作机器,错误率惨不忍睹。

2.2 ReAct是最简单的可靠路径

学术界对这个“思考-行动-观察”模式有个正式叫法:ReAct(Reasoning and Acting)。Google Brain团队2022年底发表的那篇论文《ReAct: Synergizing Reasoning and Acting in Language Models》基本奠定了今天Agent的底层范式。

ReAct的核心特点是让模型把推理过程显式输出出来,比如:

Thought: 用户让我查北京天气,我需要先调用天气查询工具。 Action: query_weather({"city": "北京"}) Observation: {"weather": "晴", "temp": 28} Thought: 天气查询完成,我可以根据结果回答用户了。 Answer: 北京今天晴,气温28摄氏度。

这个显式的推理过程非常重要。一是它能引导模型一步一步思考,减少复杂任务的出错概率;二是它天然适合日志记录,Agent每一步在想什么、做了什么,全都有迹可循。我在排查线上Agent异常时,最依赖的就是这份推理日志——没有它,你根本不知道模型哪一步“想歪了”。

虽然现在各家大模型都支持结构化的function calling(直接输出JSON格式的工具调用参数),不再需要让模型输出Action文本再解析,但ReAct的思维框架依然有效。你可以把结构化的工具调用看作ReAct中Action的机器可读版本,而Thought仍然存在于模型的隐式上下文里。

2.3 工具声明的细节怎么设计

工具声明质量,直接决定Agent的可用性上限。这里的“质量”不光是功能对不对,还包括模型能不能正确理解、能不能正确选对、能不能正确传参。我总结出三条经验:

  • 描述要写“什么场景下用”,不写“怎么实现”。比如一个send_email(recipient, subject, body)工具,描述里应该写“当用户需要发送邮件时使用”,而不是写“调用smtp服务器发送邮件”。模型理解场景比理解实现更重要。

  • 参数名要通俗、含义要准确。我曾经把一个日期参数取名dt,结果模型经常不传或者传错格式。改成date并配合描述“目标日期,格式YYYY-MM-DD”之后,正确率一下子从60%拉到95%以上。

  • 必须设计失败分支。工具不是100%成功的,调用第三方API会超时、网络会抖动、用户数据会缺失。工具本身要把错误信息格式化地返回给Agent,而不是直接抛异常把整个循环打断。比如查询天气失败,就返回{"error": "api_timeout", "message": "天气服务暂时不可用"}。模型看到这个结果后会自行决定是重试还是换个方案,还是直接告诉用户“目前查不了”。这个设计在工程上叫“失败是数据的一部分”,很多新手没意识到这一点,导致Agent一遇报错就整体卡死。

工具这块我多说一句:工具数量不是越多越好。模型在大量工具中选择正确的一个,本身是有一定概率出错的。如果业务只需要10个工具,那就别一口气上50个。真要扩展,后面可以做工具分组或两阶段选择,但这是后话了。

3. 从循环到系统:稳定性是怎么一步步长出来的

3.1 demo和系统的差距在于失败处理

很多人写的Agent Demo能跑到“看起来挺智能”,但部署到线上就开始翻车,原因不是模型变笨了,而是你没给它设计失败路径。

我把这个差距总结成一张表,新手可以对照着看:

维度Demo阶段生产系统
LLM调用试一次成功就行超时重试、熔断、降级
上下文全部塞进窗口裁剪、摘要、关键信息保留
工具执行假设一定成功超时控制、错误捕获、结果校验
并发单用户单请求限流、排队、隔离
观测终端打印链路追踪、日志、指标告警
状态内存里一次会话持久化、可恢复、可续跑

说穿了,Demo是“快乐路径”,系统是“所有能坏的环节都要提前预设坏掉以后怎么办”。

我见过最经典的翻车场景:Agent在循环第三步调用了一个耗时5秒的外部接口,结果LLM服务在等待的5秒里因为上游超时直接断开了连接。如果循环没有做超时和恢复机制,整条任务就废了,用户还莫名其妙。这个场景在Demo里几乎不会出现,在生产里天天出现。

3.2 状态、记忆与上下文管理

状态是Agent系统里最容易被忽略的问题。Demo里你用一个Python dict存messages,但生产环境里要考虑:

  • 多用户隔离:不同用户的任务状态不能互相污染,每个会话要有独立的session id。
  • 持久化:进程崩溃、服务重启之后,正在进行的Agent任务能不能恢复?如果不能,至少要把中间状态落库,以便任务追踪和人工介入。
  • 状态机:Agent处理一个复杂任务,往往经历“收集信息 → 确认意图 → 执行工具 → 汇总结果”等阶段。把这些阶段建模成状态机,比一个混沌的循环在工程上可控得多。

然后是上下文管理。LLM的上下文窗口是有限的,你每多一个loop,消息条数就膨胀一轮。如果不加处理,第20轮工具调用时就会爆token。

三个主流方案:

  • 截断:保留最早的系统提示和最近的几轮消息,中间的老消息直接丢掉。简单粗暴,但可能丢失关键信息。
  • 摘要:把超过阈值的早期对话交给LLM总结成正稿,作为压缩上下文继续保留。
  • 结构化提取:只抽取任务的关键实体和约束放到顶部,比如用户要什么、已经完成了什么、还差什么。这在复杂任务里是最实用的,因为摘要还是可能丢失细节。

我现在的偏好是组合使用:任务级关键信息结构化保存,普通对话按轮数做摘要压缩,最末端的原始消息只保留最近的3到5轮。这套方案在长耗时任务中能显著降低token成本,并保持答案质量。

3.3 并发与限流:扛住流量前先扛住自己

热搜词里有“ai agent怎么扛并发”,这问得挺实在。Agent服务跟普通API服务不一样,它有双重并发:请求并发和LLM调用并发。

一个用户的单个Agent任务,可能内部要调3到4次LLM,每次LLM调用还夹杂工具执行;如果有50个用户同时发起任务,LLM服务端的压力就是几百个并发请求。这时候你面临两个问题:

  • LLM服务限流:OpenAI、Anthropic、DeepSeek、阿里的DashScope都有速率限制,超过就会429。不加处理,你的Agent会产生大量报错。
  • 成本失控:并发一高,token消耗按倍数涨,月底账单会非常“感人”。

解决并发问题没有银弹,只有三板斧:

  1. 队列化:把Agent任务丢进消息队列(Redis Stream、RabbitMQ、Kafka),用固定数量的worker去消费。这等于给系统一个缓冲区,流量再大也不会直接压垮LLM服务和你的后端。
  2. 令牌桶限流:对每个用户或每个API key做QPS限制和并发限制,超出的请求排队或者直接返回“系统繁忙”。
  3. SLA分级:给不同业务线不同的优先级,核心交易类任务优先用高配模型、高并发额度;分析报告类任务可以降级到低配模型或者放到低谷时段跑。

还有个小细节:Agent内部对LLM的调用,强烈建议单独封装一个“LLM网关”层,统一管理API key、重试策略、超时、限额、模型路由。业务代码只负责调用agent_llm.chat(messages),底层是走GPT、走DeepSeek还是走便宜模型,由网关决定。这样你后续换模型、做降级,就不用动业务逻辑了。

3.4 可观测性:没有日志就没有debug的权利

Agent系统的调试比普通后端难得多,因为每一步都有随机性。同一个输入,今天跑可能成功,明天跑可能失败。没有日志和追踪,你根本无法判断是模型抽风、工具出错还是上下文污染。

我做Agent服务时,每一轮Agent循环至少记录以下几样东西:

  • 当前任务ID和会话ID
  • 这是第几轮循环
  • 模型本轮输出的完整内容(包括Thought和tool_calls)
  • 发送给模型的messages数组长度和token估算
  • 工具名、参数、执行结果或异常
  • 单轮耗时和累计耗时

这些日志攒下来,你可以做很多事:任务失败时,把日志回溯一遍就能定位是哪一步出了问题;跑了一批测试用例后,可以统计成功率和平均轮数来优化Prompt;甚至可以做回归对比——修改Prompt之后,同样的任务成功率是上升还是下降。

进阶一点的团队还会接入OpenTelemetry之类的链路追踪,把一次Agent任务的所有LLM调用、工具调用串成一个trace。这个在复杂多工具场景下特别好用——你一眼就能看出哪一步最慢、哪一步最贵、哪一步最容易出错。

4. 实操:用LangGraph把最小循环变成可靠系统

4.1 为什么选LangGraph

我自己写最小循环练手,但生产环境我还是会用框架。原因很简单:框架帮你把标准化的状态机、持久化、重试机制都做进去了,你只需要专注于业务节点本身。我选中的是LangGraph,理由有三个:

  • 原生支持状态机:Agent的循环本质是带条件的图,LangGraph把节点和边显式建模,天然适合描述“思考→行动→观察”循环。
  • 内置持久化和恢复:支持checkpoint机制,任务中断后可以从最近一个节点恢复。
  • 和LangChain生态衔接好:工具调用、记忆、模型适配都已经封装好了,不用自己造轮子。

网上也有很多人用AutoGen、CrewAI做多智能体协作,但如果你做的是重业务逻辑的单一Agent,LangGraph的透明度和控制力更好。它不会帮你隐藏细节,反而鼓励你显式定义状态图。

4.2 节点拆分与状态定义

我以一个“客户支持Agent”为例:用户提问后,Agent根据问题类型决定是否查询订单库,需要查询则调用工具,然后给出答复。

先定义全局状态:

from typing import TypedDict, Annotated from operator import add class AgentState(TypedDict): messages: Annotated[list, add] # 消息列表,累加更新 order_id: str # 当前处理的订单号 pending_action: str # 待执行动作 from langgraph.graph import StateGraph, END graph = StateGraph(AgentState)

这里用了Annotated[list, add]来指定消息列表的更新方式是“累加”。这是LangGraph的核心概念——节点返回的新消息会被追加到现有列表里,而不是直接覆盖。这个设计非常贴合Agent循环的特性,因为每轮工具回填本质上就是在追加消息。

4.3 编译、运行和调试

定义好节点和边之后,编译并运行:

from langgraph.checkpoint.memory import MemorySaver # 记忆持久化:用内存或Redis做checkpoint checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer) # 运行Agent任务 config = {"configurable": {"thread_id": "customer_123"}} result = app.invoke( {"messages": [{"role": "user", "content": "帮我查一下订单8888的物流状态"}]}, config=config ) print(result["messages"][-1].content)

这里thread_id就是会话ID,所有中间状态都会以它作为key保存。同一个thread再次调用时,可以从上次断点继续,这就是持久化和恢复的基本形态。

调试LangGraph有很多工具,我最常用的就是graph.get_graph().draw_mermaid()(生成ASCII版或者图片版的状态图)和逐节点打印。LangGraph SDK 1.0以后还提供了内置的调试模式,可以在Agent执行过程中实时观察每个节点输入输出的状态变化。做复杂Agent时,这个可视化能力能救命的。

5. 常见问题与排查技巧实录

5.1 Agent卡住、重复、死循环

这是最常遇到的问题。现象是Agent在同一个工具上反复调用,明明返回了错误还继续重试;或者翻来覆去执行同一个动作,没有进展。

排查顺序:

  1. 先看日志里每一步的Thought,确定模型是“理解出错了”还是“策略出错了”。
  2. 如果是反复调用同一个工具且每次都成功,很可能是模型不知道自己已经完成任务,你需要把判断任务完成的判据写清楚。比如工具返回total_done=100%,Prompt里明确要求“全部完成时直接回答,不要继续调用工具”。
  3. 如果是工具返回错误模型还在硬试,说明工具的错误信息格式不对。模型不知道“无法完成”,以为是自己参数传错,所以一直重试。把错误信息改明确点,例如“该工具不支持此操作,请勿重试,应当告知用户无法完成”。

还有一招比较实用:设置循环上限和下轮动作的唯一性校验。如果模型某次调用工具的参数跟上一轮一模一样,可以直接拦截,要求模型换一种方式或者结束。

5.2 token爆炸与上下文污染

Agent跑了几轮之后,上下文越来越长,不仅慢、贵,还会“污染”模型判断——因为模型会把一些早期的旧信息当作当前事实。

我自己踩过一次坑:让Agent做一个多步骤数据整理任务,第四步时模型突然引用第一步的一条旧数据来回答用户,完全忽略了中间步骤的更准确信息。查询了日志才发现,那是因为完整的上下文太长,模型在处理时注意力被早期内容“带跑了”。

现在的处理办法是:

  • 每轮Agent循环完成后,主动压缩历史:把任务状态和关键结论提取到顶部,保留最近一两轮详细对话。
  • 给Intent加时间戳:如果新旧信息发生冲突,一律以最新时间为准。
  • 控制工具输出大小:有的工具一口气返回几千行数据,直接全部塞进上下文非常伤。工具层要配置limit参数,Aggregate好只把必要统计结果喂给模型。

5.3 并发场景下的抖动

线上流量一上来,Agent服务各种不稳定:LLM调用超时、429限流、Redis连接数打满、worker消费不过来。

我的经验是先把“爆点”找出来。有一次我们Agent服务平均响应时间飙到30秒,分析后发现有80%的时间是耗在等待LLM响应上——因为模型网关层的并发控制没做好,所有请求都排队了。

解决方法是加一个信号量限制最大并发LLM请求数,同时把阈值调低到LLM服务商允许的80%左右,留出缓冲。另外在消费端加一个简单的“事务型任务表”,每个任务处理完后落一个状态位,进程崩了也能从表中恢复未完成的任务。

还有个小技巧:对于非实时类Agent任务(比如报表生成、批量总结),完全可以不用同步调用,而是异步跑完通过Webhook或WebSocket推送结果。这样用户体验更平滑,系统压力也小一个量级。

5.4 工具错误如何回传

工具的错误信息回传,直接决定Agent有没有“挽救”机会。很多初学写的工具是把底层异常直接抛出:

def query_order(order_id): try: return db.fetch(order_id) except Exception as e: raise e # 错误做法:直接中断Agent循环

正确的姿势是捕获后转为结构化返回:

def query_order(order_id): try: data = db.fetch(order_id) return {"status": "ok", "data": data} except Exception as e: return { "status": "error", "error_type": "db_unavailable", "message": "订单数据库暂时不可用", "suggestion": "可以提示用户稍后重试,或联系人工客服" }

模型拿到这个错误返回后,会在下一轮推理里根据suggestion给出用户一个合理的处理建议,而不是傻傻地再说一遍“调用失败”。我见过太多Agent因为工具抛异常导致整个任务链崩溃,这根本不是模型的问题,而是工程层没做好容错。

6. 一些从实战里得出的体会

文章写到最后,按照我的习惯,不做什么远大展望,就说几个实实在在的操作体会。

第一,管好循环,就管好了Agent。不管是多复杂的Agent,最终都是最小循环的组合。循环里的每一步都要有日志、有超时、有上限;有这三样,至少不会出大问题。

第二,工具的边界要划清楚。工具太多太杂,模型选择会出错;工具太抽象,模型不会用。我在实践中通常一个Agent暴露给模型的工具控制在6到8个,描述写得像“使用说明”而不是“实现文档”。

第三,可靠的系统都是先从小循环迭代出来的。不要一开始就上多Agent、记忆系统、复杂编排。把最小循环跑稳、加上日志、加上重试、加上状态持久化,再逐步增加新能力。每加一层,都要有可观测数据支撑,否则你分不清是能力增强还是混乱加剧。

说实话,Agent这个领域变化非常快,但底层的基本原理没有变过:让模型在“推理-行动-观察”的循环里,借助工具去完成一个个具体的任务。你能把这个循环打磨得多稳定,你的系统就能在多大程度上让人放心。希望这些经验和踩坑分享,能给你一点参考。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询