我最早接触 AI Agent 时犯过一个挺典型的错误以为 Agent 就是写个 prompt 让大模型做多轮对话。结果做出来的东西看起来有来有回实际上离开预设脚本就寸步难行。后来我才慢慢意识到Agent 的核心不是“多轮对话”而是“最小循环”——感知、规划、行动、观察四个动作反复执行直到任务完成。这个循环听起来简单但把它从一段演示代码变成一套能扛并发、能容错、能迭代的可靠系统中间隔着大量工程细节。这篇文章就把我从最小循环走到可靠系统这条路上的关键节点捋一遍适合刚开始学 Agent 的同学也适合已经写了几个 demo、但总觉得离“可用”还差一口气的工程师。文章的核心会围绕一条主线展开先跑通一个最小的 Agent 循环再逐个拆解它在真实场景里的失控点然后给每一类失控补上对应的工程手段最后把话题延伸到并发、可观测性和持续迭代。你不需要有很深的机器学习背景但最好写过几年代码理解 HTTP 调用、JSON 和基础的状态管理这样读起来会顺畅很多。1. 最小 Agent 循环先让模型“边走边看”地干一件实事在动手写代码之前得先把概念对齐Agent 和普通 LLM 调用的分水岭到底在哪里这决定了你后面做的所有设计。1.1 Agent 与普通 LLM 调用的分水岭一个有反馈的执行闭环普通 LLM 调用是“输入一段文本输出一段文本”本质是一次性的问答。你问“北京的天气适合穿什么”它凭训练记忆给你一个泛泛的回答这个回答可能对也可能错但它不需要承担后果。Agent 则完全不同它被给予一个任务然后在一个循环里反复决策和行动。每走一步它把外部世界的真实反馈读回来再决定下一步怎么做。我常用一个类比来解释这个区别普通调用像“问路”Agent 像“按导航开车”。问路时对方给你一个方向你听完就走了后续全靠自己按导航开车时导航给你一个路线建议你开出去几百米导航根据你的实际位置重新计算再给你下一个指令。如果前面修路封路了导航会重新规划路线而不是固执地让你继续往前撞墙。这个“根据真实结果修正下一步动作”的闭环就是 Agent 和普通 LLM 调用的本质区别。基于这个闭环的经典模式叫 ReAct即 Reasoning Acting。模型每一步的输出里既要包含它的思考Reasoning也要包含它想要执行的动作Acting动作执行后得到的结果再被喂回给模型。早期的 Agent 论文基本都在这个范式上做文章今天你看到的 LangGraph、AutoGPT、各类 Agent 框架底层也绕不开这个循环。理解了闭环再看“最小循环”就很容易了你只需要一个模型、一个工具执行函数、一个循环结构就能搭出一个最小可用的 Agent。1.2 一个 40 行的极简循环把 ReAct 跑通下面这段代码就是最小循环的完整骨架。它不依赖任何 Agent 框架只要有一个 LLM 的 SDK 就能跑。我特意把模型调用封装成了一个函数这样你可以换成任意厂商的模型。import json MAX_STEPS 5 def llm_respond(messages: list[dict]) - str: 封装任意 LLM 提供商的调用返回模型文本输出。 # 以 OpenAI SDK 为例其他模型同理 from openai import OpenAI client OpenAI() response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, response_format{type: json_object}, # 尽量要求结构化输出 ) return response.choices[0].message.content def execute_tool(name: str, args: dict): 工具函数模型决定调用什么这里负责真实执行。 if name calc: return {result: eval(args[expr])} # 演示代码生产环境别用 eval if name get_now: return {now: 2025-01-15 14:30:00} return {error: funknown tool {name}} def run_agent(task: str): messages [ {role: system, content: ( 你是一个会调用工具来完成任务的助手。 你的输出必须是 JSON格式为 {type: tool_call, tool_name: ..., tool_args: {...}} 或 {type: final, answer: ...}。 只有拿到足够信息时才输出 final。 )}, {role: user, content: task}, ] for step in range(MAX_STEPS): raw llm_respond(messages) messages.append({role: assistant, content: raw}) try: decision json.loads(raw) except json.JSONDecodeError: # 格式错了让模型看到错误后自己修正 messages.append({role: user, content: 输出不是合法 JSON请重新生成。}) continue if decision.get(type) final: print(f[{step}] 得到最终答案{decision[answer]}) return decision[answer] if decision.get(type) tool_call: result execute_tool(decision[tool_name], decision.get(tool_args, {})) print(f[{step}] 调用 {decision[tool_name]}返回 {result}) messages.append({ role: user, content: f工具返回{json.dumps(result, ensure_asciiFalse)}, }) continue # type 字段不符合预期让模型重新思考 messages.append({role: user, content: type 字段不符合预期请重新输出。}) return 达到最大步数任务未完成。 # 跑一个例子让 Agent 计算当前时间 3 小时后的时间 print(run_agent(现在几点3 小时 15 分钟之后又是几点))这段代码的核心结构只有三个部分往对话里追加消息、调用模型、执行工具并把结果追加回去。循环反复直到模型输出final或者步数耗尽。你可以试一下当你让它算一个需要两步推理的任务时它会先思考再决定调用工具看到结果后继续思考最后给出答案。这个过程中有一个细节值得注意我让模型“看到”自己的输出格式错误然后在下一次生成时自我修正。这是 Agent 循环里一个非常有用的设计思想——错误本身也是一种“观察”。模型收到“输出不是合法 JSON”的提示后通常会立刻修正格式。这比程序直接崩溃退出要优雅得多。1.3 最小循环里最容易被忽略的三件事代码跑通之后大部分人会觉得“So easy”。但我在实际项目里发现有三个细节被忽略后后面迟早要回来补课。第一系统提示词里必须明确“什么时候该收手”。没有终止条件模型很容易在一个循环里转不出来。最常见的表现是模型已经拿到了足够的信息但还在反复调用工具“确认”。上面代码里的MAX_STEPS就是兜底但更好的做法是在系统提示词里写明“一旦获得足够信息立即输出 final不要重复确认”。我在早期项目里就见过一个 Agent 为了查询一个固定值连续调了七八次同一个接口。第二工具描述要像写接口文档一样认真。模型不是靠代码注释理解工具的它靠的是你在工具注册信息里给的名称、描述、参数说明。很多人给工具起的名字模糊、描述敷衍模型自然会用错。比如你有个查询订单状态的工具叫query参数是order_id模型可能猜成orderId或id。工具描述写得好模型调用准确率会明显提升这是成本最低、收益最直接的优化。第三工具返回结果要干净。模型每轮都会把工具返回的原始内容读进上下文。如果你返回一大段日志、一堆无关字段模型会被带偏上下文也被白白浪费。正确做法是工具只返回精简后的结构化数据。比如查完数据库不要返回原始 SQL 结果集的全部字段只返回当前任务需要的几个关键字段。2. 跑通之后的第一课最小循环在真实场景里的四个失控点demo 跑通只是开始。一旦你把 Agent 放进真实业务里立刻会撞上四个失控点。这四个点不是理论推演是我和不少同行在实际项目里反复踩过的坑。2.1 上下文无限膨胀token 消耗不是线性涨是平方级涨最小循环里每一轮都会把之前所有消息重新发给模型。这意味着上下文的增长速度不是线性的而是平方级的。假设每轮新增 300 token 的消息第 5 轮时单次请求的输入约 1500 token前 5 轮累计消耗约 4500 token但到了第 10 轮单次请求输入变成 3000 token前 10 轮累计消耗高达 16500 token。如果任务到 20 轮前 20 轮累计消耗接近 63000 token。这个问题的直接后果是成本和延迟同时上涨。token 消耗翻了几倍费用也翻了几倍请求体变大模型处理时间也变长。更麻烦的是长上下文里早期的重要信息会被淹没。让模型在 3 万 token 里找到第 2 轮时的一个关键数字效果远不如让它在 3000 token 里找到准确。缓解手段大致有三类一是限制最大轮数这是最简单粗暴的二是上下文压缩每隔几轮把前面的历史消息做一次摘要用摘要替代原始内容三是只保留和当前目标相关的工具结果丢弃无关上下文。它们各有代价但都比不管不顾地膨胀要好。2.2 工具调用参数不合法模型字典里没有“校验”两个字现实中的模型返回远没有示例代码那么规整。工具名多一个空格、参数名差一个字母、类型不对、日期格式不对都是家常便饭。我见过一个 Agent 把2024-12-31解析成了2024-12-31T00:00:00.000Z下游系统完全不认也见过模型把金额12.50传成了1250因为它读错了原始数据单位。问题的根源在于模型是在做“概率生成”不是“程序执行”。它认为calculator和calc可能是同一个意思它认为传入一个字符串也能凑合当数字用。你指望它天然生成合法参数本质上是在赌概率。赌赢一百次输一次就够你排查半天。正确的思路是把工具调用的参数校验当作整个循环里不可省略的一环。模型生成 JSON 后先跑一遍 schema 校验不合法就拒绝执行并把错误信息喂回给模型让它重新生成。这条我放到下一章详细展开。2.3 幻觉在循环里被逐级放大变成“自信的胡说”单次 LLM 调用的幻觉问题行业内已经讨论很多了但在 Agent 循环里幻觉还有一个更危险的特性它会逐级放大。模型第一步把结果猜错了第二步会基于错误结果继续推理错误就像滚雪球一样越滚越大。我举个真实发生的例子。让 Agent 查询三个城市的天气然后判断要不要带伞。第一城的查询成功第二城的查询因为网络原因失败返回了空数据。模型为了“完成任务”自己编了一个第二城的温度填了进去。到了第三步它拿着这个编造的温度做“是否需要带伞”的判断最后输出了一份看似完整的结论。整个过程模型都处于“自信满满”的状态因为它确实“觉得”自己完成了任务。这种问题比单次幻觉难处理得多因为它发生在任务执行的中途你很难用一句“模型可能撒谎”来定位。真正有效的防线有两道一道是让模型在拿不到数据时明确说“数据缺失”而不是自行补全另一道是在最终输出前做校验这个我后面会讲。2.4 过程不可观测黑盒运行时出了问题连复现都难最小循环里大家都用print打日志觉得一切尽在掌握。但真实场景里工具超过十个、任务超过二十步、同时有多个用户并发调用时print日志就是一场灾难。日志会疯狂刷屏多任务日志混在一起你根本分不清哪行属于哪个任务。出了错想复现只能再跑一遍碰运气看它会不会稳定复现。我印象特别深的一次是一个 15 步的任务在第 9 步出错。日志被后续的 6 步刷过去了我为了定位问题只能重新跑一遍并在每步之间加sleep粗略算下来浪费了一个下午。这次之后我再也没有用过print当观测手段。要解决这个问题需要给循环装上“黑匣子”——结构化的、带追踪 ID 的日志系统。这部分在第五章专门讲。3. 给循环装上安全带校验、兜底和人工闭环最小循环之所以不可靠是因为整个循环里只有“模型生成”这一个决策点没有校验、没有兜底、没有边界。要让它变成可靠系统就要在模型输出和工具执行之间加一道道护栏。这一章我按重要性从高到低讲四类手段。3.1 用输出 Schema 把模型输出关进笼子里如果说只加一道护栏那一定是输出 schema 校验。模型输出JSON后先别急着解析执行先拿一个严格定义的Schema校验一遍。字段名对不对、类型对不对、可选字段有没有缺全部检查完再放行。Python 生态里最顺手的就是Pydantic。它不仅能校验还能在校验失败时给出非常明确的错误信息。这段代码演示了怎么在循环里接住校验失败from pydantic import BaseModel, ValidationError class ToolCall(BaseModel): type: str # tool_call 或 final tool_name: str | None None tool_args: dict | None None answer: str | None None # 在循环里替代原来的 json.loads 逻辑 try: decision ToolCall.model_validate_json(raw) except ValidationError as e: messages.append({ role: user, content: f你的输出格式不合法请对照要求重新生成。错误信息{e} }) continue注意我这里的处理方式校验失败不是直接退出而是把错误信息作为新的消息喂回给模型。模型看到具体的校验错误后大部分情况下会自己修正格式重新生成一个合法的调用。这等于给循环加了一个“自我修复”通道比直接raise要实用得多。校验还有一个容易被忽略的好处它让模型输出和真实业务参数之间有了一个稳定的转换层。即使模型偶尔传错单位、传错格式校验层也能拦下来不污染下游系统。3.2 有界循环最大步数之外还要有超时和熔断最小循环里的max_steps只解决了一个问题防止死循环。但真实世界里光有它远远不够。一个工具调用可能永远不返回一个已经连续失败十次的服务可能还在被反复请求这些都需要单独的兜底策略。我实际项目里的兜底层级大概长这样层级针对的问题兜底策略循环级模型反复思考、反复调用不肯收尾定义最大步数到达后强制终止并返回部分结果单次动作级单个工具调用卡住给每次工具调用设置超时例如 10 秒超时即中止工具级某个工具连续失败连续失败 3 次后熔断不再调用该工具并告知模型任务级整个任务无法完成返回“失败 已执行的中间步骤”交给人工或上层系统我最想强调的其实是熔断。模型和人都一样面对失败的反馈会产生“再试一次”的冲动。如果工具返回连续报错模型可能还是会调同一个工具因为它不知道工具是否真的挂了。如果你在工具层做熔断连续失败后直接返回“该工具已被禁用请换一种方式完成任务”模型就会另想办法——比如换工具或者老实承认失败。这比让它一遍遍撞墙好得多。3.3 双模型质检让另一个模型先审一遍答案Agent 的最终输出是给用户看的结果但模型自己生成的最终答案往往缺少“回头检查”的视角。一个很有效的手段是在最终答案返回前用一个独立的校验模型审查一遍。这个思路的关键在于“独立”和“立场”。不要让同一个模型又当运动员又当裁判它在生成时已经形成了自己的逻辑闭环让它自查很难发现自己的漏洞。正确做法是用一个更便宜的小模型站在用户需求的角度去检查答案是否满足原始约束。比如任务要求“输出三个选项”生成结果只有两个就让校验模型把差距反馈出来让主 Agent 重写。如果只能用同一个模型也至少要把校验的提示词和生成任务的提示词完全隔离让它在“审视”而不是“回忆”的语境下工作。成本上校验模型用gpt-4o-mini这类小模型就够了一次任务多花几厘钱但能拦下大量低级错误。3.4 人工介入分级不是所有动作都该让 Agent 自动执行最后一道护栏也是最容易被忽略的一道明确哪些动作允许自动、哪些必须人工确认。很多团队一上来就把 Agent 接进了发邮件、写文档、操作数据库的权限出了问题才知道后悔。我建议把所有工具操作按影响程度分四个等级操作类型示例处理策略只读操作查天气、搜索资料、读数据库自动执行全程记录日志低风险写操作创建草稿、修改沙箱配置自动执行事后可审计高影响写操作发邮件、提交订单、对外发布必须人工确认后才执行不可逆操作删除数据、覆盖文件、转账默认禁止单独授权这个分级不是凭空设计的。我的判断标准很简单操作产生的后果如果错了你花五分钟能挽回可以自动如果错了要花一小时以上善后绝对要人工。我在一个项目里让 Agent 自动回复邮件它把一封外部投诉信回复得相当客气但完全答非所问从那以后所有对外消息一律人工确认。4. 扛并发从单循环到可横向扩展的任务系统可靠性问题解决后下一个被问得最多的就是“AI Agent 怎么扛并发”。很多人的第一反应是优化 LLM 调用速度但说实话这个问题更核心的瓶颈不在模型延迟而在你整个系统的架构设计。4.1 先把瓶颈想清楚慢的是外部副作用不只是模型延迟一个 Agent 任务是串行执行的要完成一个任务通常要经历 5 到 10 轮循环每轮包含一次 LLM 调用和若干次工具调用。假设单次 LLM 调用 2 秒、单次工具调用 0.5 秒一个 8 轮的任务大概需要 20 秒才能完成。这里的“慢”不是单次调用的问题而是串行链条本身很长。所以“扛并发”的核心诉求不是让单次任务变快当然能快更好而是让大量任务能够并行推进彼此互不干扰。每个任务的状态必须独立保存每个任务的执行最好能独立调度。如果所有任务都塞在进程内存里跑一旦进程崩溃所有任务全部丢失这是不可接受的。4.2 一个实用的异步架构任务接口 后台 worker 状态存储我最近在项目里验证下来比较好用的架构本质上是经典的任务队列模式。你不需要引入特别复杂的 Agent 框架核心只有三块接收任务的 API、执行循环的后台 worker、保存任务状态的存储。流程是这样的用户请求进来先创建任务记录把初始消息写入状态存储立刻返回一个task_id然后任务被塞进队列。后台 worker 从队列里取出任务 ID执行 Agent 循环每完成一步就把最新状态写回存储。用户侧通过task_id轮询进度或订阅结果。用 FastAPI 写出来大概是这种感觉app.post(/api/tasks) async def create_task(body: CreateTaskRequest): task_id generate_id() await state_store.set( task_id, {status: queued, messages: initial_messages, step: 0}, ) await work_queue.put(task_id) return {task_id: task_id} app.get(/api/tasks/{task_id}) async def get_task(task_id: str): return await state_store.get(task_id) # worker 侧 async def worker_loop(): while True: task_id await work_queue.get() state await state_store.get(task_id) # 执行多步循环每步后更新 state_store # 完成时更新 status: completed / failed这套架构的价值在于worker 可以水平扩展。任务多的时候加几个 worker 进程任务少的时候减几个互不影响。状态存储建议用 Redis它能保证多进程读到一致的状态而且天然带过期时间可以避免任务状态堆积。数据库也可以但要注意别把高频的状态更新打到一个库上。4.3 幂等设计没有幂等键一次超时重试就是一次重复操作并发系统和可靠性系统合流后最大的隐藏杀手是重复操作。Agent 在某一步调用了一个有副作用的工具——比如“发送邮件”——邮件发出去了但网络超时模型没收到工具返回值。这时模型会认为工具调用失败了重新发起一次调用结果用户收到两封一模一样的邮件。解决这个问题必须引入幂等机制。每个工具执行前都带一个幂等键比如task_id:step:tool_name。工具执行前先查询这个幂等键是否存在存在就直接返回上次结果不存在才执行真实操作。def execute_with_idempotency(task_id, step, tool_name, args): key fidem:{task_id}:{step}:{tool_name} if redis.exists(key): return json.loads(redis.get(key)) result execute_tool(tool_name, args) redis.set(key, json.dumps(result), ex3600) # 保存 1 小时 return result这个方案几乎零成本但能避免最高频的重试事故。凡是设计到发消息、写文件、改数据库这类操作的 Agent我都建议把幂等键当成标配而不是等到出了事故再补。4.4 显式状态机把内存里的循环变成可恢复的图最小循环运行在内存里任务状态只存在于 Python 的变量中。一旦进程崩溃所有中途任务全部消失。要让系统真正可靠就要把“循环”改造成“状态机”——每一轮结束状态都持久化每执行一步之前都重新加载状态。显式状态机的好处有三个一是可恢复worker 挂了新 worker 可以接着跑二是可观测每一步的状态都可以随时查看三是支持更复杂的控制流比如条件分支、并行节点。这也是为什么 LangGraph 这类基于图的 Agent 编排框架会流行——它把 Agent 循环抽象成“节点 边 状态对象”本质上就是在做显式状态机。我没有让你一上来就上框架是因为先用手写循环跑通以后你会更容易理解框架里每个概念是解决什么问题的。框架不是魔法它就是把这些工程兜底封装成了现成的 API。5. 让系统持续变好观测、回归集和护栏沉淀可靠系统跑起来之后真正的挑战才刚开始你怎么确认下一次改动没有让系统变差改了一个 prompt你怎么知道是变好了还是变坏了这部分靠的不是运气而是三件事结构化观测、回归测试集、错误模式沉淀。5.1 给循环装黑匣子结构化日志是全链路排查的前提print日志在并发场景下基本不可用。你需要的是结构化日志每条日志带上trace_id记录步骤序号、动作类型、工具名称、输入输出、耗时和 token 消耗。长这样{ trace_id: t-001, task_id: task-123, step: 3, action: tool_call, tool: calc, input: {expr: 11}, output: 2, llm_tokens: 234, latency_ms: 820 }有了这类日志你可以随时按trace_id拉出整个任务的全链路模型每一步说了什么、调了哪个工具、工具返回了什么、哪一步耗时最长、哪一步 token 消耗最多。排查问题从“碰运气复现”变成了“查日志定位”。我自己还会额外记录一个“决策快照”每一轮的系统提示词、本轮的消息列表、模型的原始输出。这个快照覆盖了完整上下文能解答很多事后疑问。缺点是多占一点存储但和排查问题节省的时间比完全不值一提。5.2 建一个最小回归集改 prompt 不再是掷骰子没有回归集的 Agent 项目改 prompt 基本靠感觉。改完你觉得“好像变聪明了”实际上可能只是这个例子上变聪明了换一批例子直接拉胯。这个问题怎么解决建一个固定的评估集把典型任务收集起来每次改动都跑一遍。评估集不需要很大十几个任务就能起到很大作用。每个任务除了任务描述最好带上预期结果或关键验收字段EVAL_CASES [ { task: 计算 23*17并告诉我结果, expected: 391, tags: [calc], }, { task: 查询上海的当前天气并判断是否需要带伞, expected_fields: [precipitation, verdict], tags: [tool, judgment], }, { task: 统计订单列表中金额超过 100 元的订单总数, expected: 3, tags: [data_extraction], }, ]每次改完 prompt、换完模型、调完工具描述都跑一遍评估集记录完成率、正确率、工具调用失败率、平均轮数和平均耗时。对比前后两轮的数据就能说清楚这次改动到底是正向还是负向的。这个方法听起来简单但绝大多数团队都没做原因只是“还没疼到那个程度”。5.3 高频错误最终沉淀成护栏规则、校验器和硬编码保护观测和评估的最终目的是把高频错误转化为结构性的“护栏”。所谓护栏就是在错误再发生之前先用规则把它拦住。我自己的习惯是每个月复盘一次日志里的失败模式然后逐条变成护栏。看两个例子高频失败模式护栏方案模型经常把金额单位“元”误传为其他单位工具描述里显著标注单位并在校验层拒绝超出合理范围的数值模型调用工具时缺少必填字段用 schema 强制校验 反馈错误信息给模型重新生成模型在数据缺失时自行编造结果系统提示词明确“数据缺失时必须说明缺失不得猜测” 最终答案质检从错误里提炼护栏本质上是在给模型边界。模型的能力上限由模型决定但系统的行为下限由护栏决定。一个 Agent 系统可靠不可靠最后往往不是看模型有多强而是看护栏有多密。说到这想起很多刚开始学 Agent 的朋友问过我学习路线的问题。你要让我给建议我觉得顺序特别简单先拿今天这个最小循环手写一遍跑到你对“循环”有了肌肉记忆然后往里加校验、加超时、加幂等感受一下工程兜底的分量再然后换到 LangGraph 这类框架去理解显式状态机如何让系统可恢复最后花和写代码一样多的时间做观测和评估集。工具框架都是其次真正值钱的不是那些花哨的编排而是你手里有一套能稳定复现、能度量好坏的循环。我踩过最贵的坑就是在系统已经跑得很热闹的时候发现自己根本没法判断它是变好了还是变坏了。现在每次动手改 Agent我都会先问自己一句如果我改坏了我怎么知道这个问题有了答案你就已经走在从最小循环到可靠系统的路上了。