
这个系列写到第38篇读者问得最多的一个问题就是你说要做AI Agent网关那这个网关到底先写什么我原本的计划是先搞路由、限流、鉴权这些看起来很像网关的能力但真正动手做claw0之后第一个跑通的其实是一个特别不起眼的while循环。这个循环就是Agent循环也就是Agent从收到请求到返回最终结果之间反复思考、调用、观察、再思考的过程。claw0这个代号取自初始之爪寓意这只网关是从零开始伸出的第一只爪子去够大模型、工具和各种外部系统。在搭它的第一天我就发现所有Agent产品不管包装得多像聊天框、多像自动化流程核心都是同一条循环。区别只是有的循环跑得快有的循环带审批、带限流、带记忆有的循环被藏得很深。所以这一节我不讲网关的协议转发也不讲负载均衡专门把循环拿出来拆开揉碎讲清楚它是什么、怎么从零手写、以及它如何变成整个网关的地基。这篇内容适合两类人一类是已经让模型开口说话、现在想让它真正动手干活的人另一类是正在评估各种Agent框架、想搞懂底层原理再决定选型的人。看完你不仅能自己写一个最小的Agent循环还能知道往网关方向扩展时那些绕不开的边界条件。1. 为什么网关的第一课是Agent循环很多人一听网关脑子里浮现的是Nginx、Kong那套东西请求转发、鉴权、限流、灰度。这套思路在传统服务架构里没错但用在AI Agent场景里会跑偏因为Agent网关要转发的不只是HTTP请求而是一段持续变化的对话过程。你说它是网关不如说它是一个运行着Agent循环的中间层其他那些路由、鉴权、配额全是这个循环的外围设施。我见过不少团队做Agent应用时第一个版本其实就是写死一个while循环模型说要用工具就用工具说回答就回答一跑就通。可一旦要把它变成一个能被多个业务方复用的网关问题就来了怎么让不同部门注册自己的工具怎么统计一次任务消耗了多少token怎么防止某个租户的Agent在循环里空转浪费钱这些问题的答案全都要落在循环的边界条件上而不是落在HTTP层。所以我把循环当作claw0的第一个课程是因为它是理解后面所有模块的前提。还有一个容易被忽略的点模型本身并不具备执行工具的能力它只会输出一段结构化的意图声明。真正调用函数、访问数据库、发请求的是循环所在的运行环境。这个分工决定了Agent循环的形态——它不是模型驱动的黑盒而是运行环境与模型之间来回协商的过程。在动手写代码之前我建议先建立一条基本认知Agent循环是一台带刹车的发动机。发动机负责让Agent持续干活刹车负责在模型乱来的时候安全停车。很多人只盯着加速忽略了刹车结果一上线就出事故。后面我会专门讲刹车怎么设计。2. Agent循环的完整运行链路一次对话从进入到结束2.1 模型的意图声明与网关的实际执行要理解Agent循环必须先理解模型输出和实际执行的鸿沟。以目前主流的对话模型为例当你问它上海现在几度模型有两种选择一是根据训练数据里的旧知识瞎编一个温度二是输出一个工具调用声明比如我要调用get_weather参数是上海。模型选择第二种时返回的内容看起来像一段普通文本但在API层面是有结构的。OpenAI兼容接口里它叫tool_callsClaude风格里叫tool_use本质上都是同一个东西一个工具名、一个参数对象、一个调用编号。这个编号非常关键它把模型的这一次请求和工具执行的这一次结果绑定起来。网关拿到这个声明之后才真正动手先去工具注册表里找到get_weather这个函数校验参数格式执行代码得到结果再把结果作为一条消息追加回对话历史重新发给模型。模型看到工具结果后才能给出最终回答。这套声明—执行—回填—再思考的链路就是Agent循环的全部内容。2.2 一次天气查询的完整消息演变我用一个具体例子说明消息列表在每个阶段长什么样。假设系统提示词是你是助手可以通过工具获取实时天气用户输入是上海现在多少度。第一步组装初始消息列表[system] 你是助手可以通过工具获取实时天气 [user] 上海现在多少度第二步发给模型模型返回一条assistant消息这条消息不包含最终回答而是包含一个工具调用声明[assistant] tool_calls[ {id: call_123, name: get_weather, arguments: {city: 上海}} ]第三步网关执行get_weather拿到结果后追加一条role为tool的消息。注意这里的tool_call_id必须等于上一步的call_123这样模型才能知道这个结果对应哪次调用[tool] tool_call_id: call_123 内容: 上海当前气温25℃多云第四步把新的消息列表发给模型模型看到工具结果输出最终回答[assistant] 上海当前气温25℃多云出门带件薄外套就行。这四步就是一个最小的完整循环。你可能会问如果模型看了工具结果之后还觉得不够怎么办比如它想再查一次湿度。那就在第三步之后继续追加一条新的tool_calls把循环再转一圈。循环的终止条件说来也简单某次模型返回不再包含任何工具调用或者外层设的安全上限到了。2.3 循环为什么必须设置终止条件我在实际使用中见过最经典的翻车案例Agent查完订单状态用户说收到货了结果Agent又开始主动调用查物流的工具说让我确认一下配送记录查完物流又想查签收人无限套娃。理论上模型不会永远不停止但套娃三轮五轮就会让用户崩溃更别说每次调用都要花钱。所以终止条件必须提前定死。最基础的两个一是模型不再请求工具自然收敛二是累计调用轮数达到上限强制结束。后者的实现是在达到上限后追加一条系统消息告诉模型已到最大步数请基于现有信息直接给最终答复再调用一次模型让它收尾。这一招我在正式环境里用了很久非常可靠。3. 手写一个最小可用的Agent循环3.1 先定义清楚三个基础数据结构我建议不要一上来就套框架先用Python写一个不依赖第三方库的最小实现把循环的骨架刻进脑子里。核心数据结构只有三个消息、工具定义、循环器。消息定义了对话历史里的一行记录它必须能表达四种角色系统、用户、助手、工具。工具消息还需要额外携带tool_call_id用来关联模型的调用声明。工具定义则把一个函数的元信息暴露给模型包括名称、描述、参数JSON Schema。这里有个设计要点描述写得越清楚模型选错工具的概率越低这段描述不是给人看的是给模型看的。3.2 循环主体从messages到messages的转换器下面这个类是我在claw0早期原型里提取出来的核心逻辑去掉了所有工程装饰只保留循环本身from dataclasses import dataclass, field from typing import Callable, Any, Optional dataclass class ChatMessage: role: str # system / user / assistant / tool content: str tool_call_id: Optional[str] None tool_calls: Optional[list] None # 仅assistant消息使用 dataclass class ToolDefinition: name: str description: str parameters: dict # JSON Schema func: Callable[..., Any] class AgentLoop: def __init__(self, chat_fn, tools: list[ToolDefinition], max_steps: int 10): # chat_fn 是模型适配函数输入消息列表输出一条assistant消息 self.chat_fn chat_fn self.tools {tool.name: tool for tool in tools} self.max_steps max_steps def run(self, messages: list[ChatMessage]) - list[ChatMessage]: for step in range(self.max_steps): assistant self.chat_fn(messages) messages.append(assistant) # 没有工具调用循环自然终止 if not assistant.tool_calls: return messages # 逐个执行模型请求的工具 for call in assistant.tool_calls: tool self.tools.get(call[name]) if tool is None: result f错误: 不存在名为{call[name]}的工具 else: try: result str(tool.func(**call[arguments])) except Exception as exc: result f工具执行失败: {exc} messages.append(ChatMessage( roletool, contentresult, tool_call_idcall[id] )) # 超过最大步数追加强制收敛提示 messages.append(ChatMessage( rolesystem, content已经达到最大推理步数请停止调用工具直接基于已有信息回复用户。 )) messages.append(self.chat_fn(messages)) return messages这段代码里有一个容易被忽略但很重要的点每执行一个工具就立刻把结果追加进messages然后进入下一轮模型调用。也就是说如果模型一次声明调两个工具先执行第一个回填结果再执行第二个再回填结果然后重新调用模型。这样模型每次只看到增量信息不容易混乱。3.3 把它接到真实模型适配器上上面的AgentLoop不关心你的模型来自哪家厂商只要有一个chat_fn函数能完成消息列表进assistant消息出的转换。这个抽象就是模型适配器的雏形。以OpenAI兼容接口为例适配器大概长这样from openai import OpenAI client OpenAI() def chat_fn(messages: list[ChatMessage]) - ChatMessage: payload [] for msg in messages: item {role: msg.role, content: msg.content} if msg.tool_calls: item[tool_calls] msg.tool_calls if msg.tool_call_id: item[tool_call_id] msg.tool_call_id payload.append(item) resp client.chat.completions.create( modelgpt-4o-mini, messagespayload, tools[{ type: function, function: { name: t.name, description: t.description, parameters: t.parameters, } } for t in tools], ) raw resp.choices[0].message return ChatMessage( roleassistant, contentraw.content or , tool_calls[ { id: tc.id, name: tc.function.name, arguments: tc.function.arguments, } for tc in (raw.tool_calls or []) ] )注意一个细节OpenAI接口返回的arguments是个字符串里面才是JSON。把字符串转成dict这一步可以在chat_fn里做也可以在执行工具时再做。我建议在chat_fn里就转换好因为ToolDefinition的func只认识真正的Python对象不认识JSON字符串。这个转换不是简单的json.loads后面我会专门讲踩坑。4. 循环里最容易踩的四个坑4.1 坑一模型在两个工具之间反复横跳我在claw0早期调试时遇到过一个真实场景用户问我的订单到哪里了模型先调用查订单接口发现订单已签收又调用查物流接口发现物流信息不完整转头又去调用查订单如此循环了好几次token消耗翻了三倍。模型不是故意捣乱而是它面对模糊信息时倾向于再多确认一步但这一步永远不够。解决这个问题的核心是两步第一步是设置max_steps兜底这是底线第二步是让工具返回结果更有结论。比如物流接口返回时主动补一句物流状态已是最新无需再查询其他信息模型往往会顺着这个结论收敛。与其跟模型的强迫症对抗不如让工具结果引导它停止。4.2 坑二工具返回结果把Prompt撑爆工具返回结果太长是最隐蔽的性能杀手。我见过有人写了一个搜索工具返回了200KB的网页正文下一轮模型调用时这些内容全部塞进上下文。不仅费用暴增模型还会被无关信息干扰开始编造网页里没有的内容。我的做法是给工具结果加一层压缩裁缝按角色区分处理。对需要模型推理的内容截断到前500到800个字符如果还有更多细节在结果末尾加一句原始内容已存入会话附件如需更多细节请再次告诉我。对纯粹的数据型结果比如订单状态、用户ID直接保留结构化字段去掉大段描述。这样既保住了关键信息又控制了消息体积。4.3 坑三模型返回的arguments不是合法JSON指望模型每次都输出标准JSON是不现实的。最常见的问题有两类一类是把JSON包在markdown代码围栏里比如返回json {city:上海}直接json.loads会抛异常另一类是参数值带了多余引号或注释看起来像JSON但解析不过去。我最终采用的策略是三级降级解析先用正则把代码围栏和首尾大括号之外的噪音剥掉再尝试json.loads失败了用json5库解析它能容忍单引号、尾逗号还不行就放弃结构化解析把原始arguments字符串当作一个参数传下去同时把解析错误写入工具结果让模型自己在下一轮修正。 grades这个方法不能保证100%成功但能覆盖我遇到的95%场景。4.4 坑四多会话并发时状态互相污染这是从单体Demo变成网关服务时最容易炸的点。在Demo里一个messages列表从头走到尾很自然但网关面对的是几十个用户同时发请求如果把messages当成全局变量A用户的工具结果就会混进B用户的对话里。claw0的解法是引入会话上下文池每个session_id对应一个独立的消息列表。循环器每次run都从上下文池里取出自己的那串messages执行完再把变更后的列表写回去。同时还要加一个会话级别的超时时间比如单个循环最多跑60秒超过就中止并把已执行到的步骤写入审计日志。并发环境下的循环本质上是把无状态的循环器变成有状态的会话沙箱。5. 从简单循环到网关循环如何变成基础设施5.1 把循环拆成五个组件单体Demo里的AgentLoop终究只是一个类要变成网关就要把它拆成可运维的组件。我在claw0里把它拆成了五块组件职责单体Demo里的对应物ModelAdapter统一各家模型接口处理tool_calls格式差异chat_fn函数ToolRegistry工具注册、参数校验、权限校验、并发控制tools字典LoopRunner负责循环编排、终止判断、超时控制AgentLoop.runContextStore按会话隔离消息列表支持持久化和恢复messages变量Auditor记录每一轮的token消耗、工具耗时、调用链路print日志这个拆分是必须的因为网关要服务多个业务方不能每个业务方都自己维护一套工具映射。ToolRegistry上线了工具之后LoopRunner拿到的是注册表里的只读快照而不是某个业务方私有的函数引用。这样既能统一治理又不影响循环核心的执行效率。5.2 循环策略要从代码里抽出来把循环变成基础设施的另一个关键点是把策略参数化。最开始的AgentLoop里max_steps是写死的后来我把最大步数、是否允许模型自行选择工具、工具结果截断长度、强制收敛提示文案全部挪到配置里。每个租户可以设置自己的额度比如免费用户最大5步付费用户最大20步。这样Admin控制台调整策略时不需要发版热加载配置即可。另外还有一个容易被忽视的策略哪些工具可以被哪些会话调用。这个校验不放在LoopRunner里而是在ToolRegistry返回工具快照时提前过滤。循环执行到某个工具却发现没有权限是一种很难看的失败因为模型的思考过程已经发生过了token已经花了。5.3 一次真实调用在claw0里的落盘记录最后看一下网关视角下一次完整循环会留下什么痕迹。每当用户请求进入网关会生成一个trace_id循环每转一圈Auditor就追加一条记录包含当前步数、调用的模型名、工具名、工具耗时、累计token、上下文长度。有了这些数据后面做成本分摊、异常告警、效果分析全都顺理成章。我见过太多Agent项目死在能跑但看不见的阶段整个循环像黑盒一样出了问题只能加日志重新跑一遍。claw0从第一课就把可观测性当成循环的一部分来设计每一轮的关键状态都落盘。这个习惯到了复杂场景会救你很多次。把最小循环跑通的那天晚上我在日志里看到模型一步一步调用工具、看到工具结果回填、看到它最终收敛出答案那时候才真正理解为什么说所有Agent框架的外壳再豪华内核都是这条循环。先把这条循环吃透再去讨论路由和限流你才会知道网关真正该守护的是什么。下一课我打算讲工具注册中心重点是怎么让一个工具安全上架、校验参数、分配权限以及当工具数量超过一百个的时候如何帮模型做预筛选。