1. 别被循环两个字骗了Agent Loop 的真实复杂度很多人第一次接触 Agent 开发看到Agent Loop这个词脑子里浮现的就是一个 while 循环调模型、拿结果、执行工具、把结果塞回去、再调模型直到模型说我完成了。写出来大概二十行代码跑个 demo 看着挺像那么回事。我当初也是这么想的直到我把这套东西接到真实项目里才发现这个循环背后藏着一堆让人头皮发麻的细节。先说结论Agent Loop 的本质不是循环而是一个状态机驱动的决策-执行-反馈系统。循环只是它最外层的表现形式。你真正要处理的是每一轮对话里模型输出的意图如何被解析、工具调用如何被调度、执行结果如何被裁剪和回注、上下文如何被管理、异常如何被恢复、以及整个过程中如何保证不失控。这些东西demo 里一个都不会教你。这篇文章适合谁看如果你已经写过最基础的 Agent 循环能跑通模型调用工具的流程但一到真实场景就各种翻车——工具调错、上下文爆炸、循环停不下来、SSE 流断在半路——那这篇就是写给你的。我会把 Agent Loop 从看起来简单到实际复杂之间的那层窗户纸捅破把每个环节的真实坑点和处理思路讲清楚。核心关键词先摆出来Agent、Agent Loop、Tool System、SSE、Claude Code。这几个词基本覆盖了当前 Agent 开发的主战场。下面我按实际开发中会遇到的顺序一层一层拆。2. Agent Loop 的整体设计与思路拆解2.1 为什么调模型-执行工具这个朴素循环不够用最朴素的 Agent Loop 长这样把用户输入和工具定义一起发给模型模型返回一个工具调用请求你执行工具把结果拼回消息列表再发给模型如此往复。这个结构在单轮、单工具、无异常的场景下确实能跑。但真实项目里你马上会遇到几个问题。第一模型可能一次返回多个工具调用。比如你让它查一下北京天气顺便看看明天适不适合出门它可能同时发起天气查询和日历查询两个工具调用。你的循环如果只处理一个另一个就被丢了。第二工具执行可能失败。网络超时、参数错误、权限不足这些失败信息怎么回传给模型让它自己决定重试还是换方案是个设计问题。第三循环可能停不下来。模型陷入调工具-看结果-再调同一个工具的死循环或者一直说我需要更多信息但永远不给出最终答案。所以一个能用的 Agent Loop必须在朴素循环之上加三层东西工具调度层处理多工具、并发、超时、状态管理层维护对话历史、工具结果、中间状态、终止控制层判断什么时候该停、什么时候该强制停。2.2 状态机视角把 Loop 拆成可管理的阶段我习惯把 Agent Loop 拆成五个阶段每个阶段有明确的输入输出和退出条件。这样拆的好处是每个阶段可以单独测试、单独优化出问题的时候能快速定位是哪一层挂了。阶段职责关键输出常见问题意图解析解析模型输出提取工具调用或最终答案结构化意图对象解析失败、格式不符工具调度决定调哪些工具、并发还是串行、超时设置执行计划并发冲突、超时未处理结果处理裁剪、格式化、截断工具返回可回注的结果上下文爆炸、信息丢失上下文管理决定哪些历史保留、哪些压缩、哪些丢弃新的消息列表超长截断、关键信息被删终止判断判断是否继续循环、是否强制结束继续或结束信号死循环、提前终止这个拆法不是教科书上的标准答案是我在实际项目里踩坑之后总结出来的。每个阶段的边界清晰了你才能针对性地加日志、加监控、加兜底。2.3 为什么选 SSE 而不是 WebSocketAgent 场景下前端需要实时看到模型的输出流和工具执行状态。这里有个选型问题用 WebSocket 还是 SSE我的选择是SSEServer-Sent Events。原因很简单Agent 的输出是单向的——服务端推给客户端客户端不需要在这个通道上往回发消息用户输入走普通 HTTP 请求就行。SSE 基于 HTTP天然支持断线重连、自带事件 ID、实现简单不需要像 WebSocket 那样维护双向连接状态。对于 Agent 这种请求-流式响应的模式SSE 是更贴合的选择。但 SSE 也有它的坑。最常见的就是stream disconnected before completion: idle timeout waiting for SSE——连接空闲超时被断开。这个问题的根源在于Agent 在执行工具的时候可能几十秒都不产生新的输出SSE 连接因为长时间没有数据而被中间层网关、负载均衡、浏览器掐断。解决办法后面会详细讲。3. 核心细节解析与实操要点3.1 工具系统设计别把工具当函数很多人设计 Tool System 的时候就是把一堆函数注册进去写个 JSON Schema 描述参数完事。这在工具少的时候没问题工具一多就乱套。我总结下来工具系统设计要关注四个维度。第一个维度是工具粒度。工具太细模型要调很多次才能完成一件事循环轮数暴涨工具太粗参数复杂模型容易填错。我的经验是一个工具对应一个明确的动作参数控制在 5 个以内。比如查询天气是一个工具发送邮件是一个工具但处理用户请求这种就不该是工具。第二个维度是工具描述。模型选工具全靠描述。描述写得含糊模型就乱选。我见过有人把工具描述写成用于处理数据结果模型什么数据都往这个工具里塞。好的描述应该包含这个工具做什么、什么时候用、参数含义、返回什么、有什么限制。描述里最好带一两个使用示例。第三个维度是错误处理。工具执行失败时不要把异常直接抛给模型看堆栈。要把错误转成模型能理解的自然语言比如天气查询失败原因是城市名称无法识别请确认城市名是否正确。这样模型才能根据错误信息调整策略。第四个维度是权限和副作用。有些工具是只读的查询类有些是有副作用的写入、发送、删除。有副作用的工具要格外小心最好加确认机制或者限制调用频率。Claude Code 里那个每次都要点确认的机制本质上就是在有副作用的工具上加人工确认。3.2 上下文管理Agent Loop 里最容易爆的地方Agent Loop 跑着跑着上下文就爆了这是几乎所有 Agent 项目都会遇到的问题。原因很简单每一轮工具调用的结果都会追加到消息列表里几轮下来消息列表长度轻松超过模型的上下文窗口。处理这个问题有几种策略我按推荐程度排序。策略一工具结果截断。工具返回的内容如果很长比如读了一个大文件、查了一大段日志只保留关键部分。具体做法是设置一个字符上限超过就截断并在截断处加一句内容已截断如需完整内容请指定更精确的查询范围。这样既控制了长度又给了模型继续探索的线索。策略二历史消息压缩。当消息列表超过一定长度把早期的对话压缩成摘要。比如把前 10 轮的工具调用和结果压缩成一段之前已经完成了 X、Y、Z的描述。这个操作要小心压缩得太狠会丢失关键信息导致模型重复劳动。策略三滑动窗口。只保留最近 N 轮的消息更早的直接丢弃。这个最简单粗暴但风险也最大——如果早期有重要的工具结果丢了之后模型就失忆了。我的实际做法是组合使用工具结果先截断消息列表超过阈值时做压缩压缩后如果还是超长再用滑动窗口兜底。三层防护下来基本不会爆。3.3 SSE 流式输出的实现要点SSE 在 Agent 场景下的实现有几个细节必须处理好否则用户体验会很差。事件类型设计。不要只推一种事件。我一般会定义这几类事件message_start开始输出、content_delta文本增量、tool_call_start工具调用开始、tool_call_result工具执行结果、message_end输出结束、error错误。前端根据事件类型做不同的 UI 展示用户能看到模型正在思考正在调用工具工具返回了结果这些状态。心跳保活。前面提到的idle timeout问题解决办法是定期发送心跳事件。即使没有实际内容也每隔 15-30 秒推一个ping事件保持连接活跃。这个间隔要根据你的网关超时设置来定一般比网关超时短一半比较安全。断线重连。SSE 自带Last-Event-ID机制客户端重连时会带上最后收到的事件 ID服务端可以从这个 ID 之后继续推。但 Agent 场景下服务端的状态是有状态的消息列表在内存里重连后要能恢复到正确的状态。我的做法是把 Agent 的执行状态持久化重连时根据会话 ID 恢复。错误传播。工具执行出错、模型调用失败这些都要通过 SSE 推给前端而不是让连接静默断掉。前端收到error事件后可以展示错误信息并决定是否重试。4. 实操过程与核心环节实现4.1 从零搭一个可用的 Agent Loop下面我用 Python 伪代码的方式把核心循环写出来。这不是完整可运行的代码但结构是完整的你可以照着这个骨架往里面填具体实现。class AgentLoop: def __init__(self, model_client, tool_registry, max_iterations20): self.model model_client self.tools tool_registry self.max_iterations max_iterations def run(self, user_input, session): session.add_message(user, user_input) for i in range(self.max_iterations): # 1. 调用模型 response self.model.chat( messagessession.get_messages(), toolsself.tools.get_schemas() ) # 2. 解析意图 intent self.parse_intent(response) if intent.type final_answer: session.add_message(assistant, intent.content) return intent.content # 3. 执行工具 if intent.type tool_calls: session.add_message(assistant, response.raw) for call in intent.calls: result self.execute_tool(call) session.add_message(tool, result, tool_call_idcall.id) # 4. 上下文管理 session.compact_if_needed() # 5. 强制终止 return self.force_finalize(session)这个骨架里parse_intent负责把模型输出转成结构化意图execute_tool负责执行单个工具并处理异常compact_if_needed负责上下文压缩force_finalize是循环次数用尽后的兜底。4.2 工具执行器的实现细节工具执行器看起来简单实际上要处理的东西不少。我把它拆成几个步骤。def execute_tool(self, call): # 1. 参数校验 tool self.tools.get(call.name) if not tool: return f错误工具 {call.name} 不存在 try: params json.loads(call.arguments) except json.JSONDecodeError: return f错误参数格式不正确请检查 JSON 格式 # 2. 参数 schema 校验 validation_error tool.validate(params) if validation_error: return f错误参数校验失败 - {validation_error} # 3. 执行带超时 try: result tool.execute_with_timeout(params, timeout30) except TimeoutError: return f错误工具 {call.name} 执行超时30秒 except Exception as e: return f错误工具执行失败 - {str(e)} # 4. 结果截断 return self.truncate_result(result, max_chars4000)这里有几个关键点。参数校验要在执行前做不要等工具内部报错。超时必须有否则一个卡住的工具会拖死整个循环。错误信息要转成自然语言让模型能理解。结果要截断防止上下文爆炸。4.3 SSE 服务端的实现SSE 服务端我用 FastAPI 举例核心是返回一个StreamingResponse里面是一个生成器不断 yield 事件。from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio, json app FastAPI() async def event_stream(session_id, user_input): agent get_agent(session_id) queue asyncio.Queue() # 启动 Agent 执行把事件推到 queue asyncio.create_task(agent.run_streaming(user_input, queue)) last_heartbeat time.time() while True: try: event await asyncio.wait_for(queue.get(), timeout15) yield format_sse(event) last_heartbeat time.time() except asyncio.TimeoutError: # 心跳保活 yield format_sse({type: ping}) last_heartbeat time.time() def format_sse(event): return fevent: {event[type]}\ndata: {json.dumps(event)}\n\n app.get(/agent/stream) async def stream(session_id: str, input: str): return StreamingResponse( event_stream(session_id, input), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, # 禁用 nginx 缓冲 } )这里的关键是X-Accel-Buffering: no这个头。如果你的服务前面有 nginx不加这个头nginx 会缓冲 SSE 响应导致前端收不到实时数据。这个坑我踩过排查了大半天。4.4 终止控制的实现终止控制分两种正常终止和强制终止。正常终止是模型主动给出最终答案。判断逻辑是模型返回的消息里没有工具调用只有文本内容且文本内容不是我需要调用工具这类中间态。这个判断要小心有些模型会输出我将要调用工具这样的文本但实际没有工具调用这种要当成最终答案处理。强制终止是循环次数用尽或者检测到死循环。死循环的检测方法是记录最近几轮的工具调用签名工具名参数如果连续三轮完全相同就判定为死循环强制终止并返回当前已有的信息。def detect_loop(self, session, window3): recent session.get_recent_tool_calls(window) if len(recent) window: return False signatures [f{c.name}:{c.arguments} for c in recent] return len(set(signatures)) 1强制终止时不要直接返回空要把当前已经收集到的信息整理一下返回给用户并说明由于达到最大执行轮数以下是目前已完成的部分。5. 常见问题与排查技巧实录5.1 SSE 相关问题的排查stream disconnected before completion: idle timeout waiting for SSE这个报错我见过太多次了。排查思路是这样的。先确认心跳有没有在发。在服务端加日志看ping事件是不是按预期间隔发出。如果没有检查你的asyncio.wait_for超时设置是不是比网关超时还长。再确认中间层有没有缓冲。如果服务前面有 nginx、网关、CDN每一层都可能缓冲 SSE。nginx 要加X-Accel-Buffering: no网关要确认支持流式响应CDN 一般要关掉对text/event-stream的缓冲。最后确认客户端有没有正确处理。有些 HTTP 客户端库默认会缓冲整个响应要设置成流式读取。浏览器端的EventSource默认就是流式的但如果你用fetch手动处理要注意用response.body.getReader()而不是response.text()。5.2 工具调用相关问题的排查模型调错工具、参数填错、工具执行失败这些问题排查起来有个通用思路把完整的消息列表打出来看。很多时候问题出在工具描述不清楚或者历史消息里有干扰信息。我整理了一个常见问题速查表。现象可能原因排查方向模型不调用工具工具描述不清、系统提示没强调检查工具描述和 system prompt模型调错工具工具之间描述太相似区分工具描述加使用场景说明参数格式错误Schema 定义不严、缺少示例补充参数示例和格式说明工具执行超时工具本身慢、网络问题加超时、优化工具实现循环停不下来终止条件不明确、工具结果没给足信息检查终止逻辑、补充工具返回信息上下文爆炸工具结果太长、历史没压缩加截断和压缩策略5.3 实操心得几个让我少走弯路的经验第一个经验日志要打全。Agent Loop 出问题的时候你需要的不是哪一步错了而是每一步的输入输出是什么。我习惯在每一轮循环开始时打一条日志包含当前消息列表长度、本轮模型输出、工具调用详情、工具返回结果。这样出问题的时候直接看日志就能定位。第二个经验给模型留退路。工具执行失败的时候不要只返回错误要返回错误建议。比如查询失败城市名可能拼写错误请尝试使用标准城市名如北京上海。模型看到建议更容易自我纠正。第三个经验限制工具调用轮数。我一般设 15-20 轮上限。超过这个轮数还没完成的任务要么是任务本身太复杂需要拆解要么是模型陷入了某种循环。强制终止比无限循环好。第四个经验工具结果要结构化。工具返回的内容尽量用结构化的格式JSON、Markdown 表格而不是一大段自然语言。结构化内容模型更容易解析也更容易截断。第五个经验测试要覆盖异常路径。正常路径谁都能跑通真正考验 Agent Loop 健壮性的是异常路径工具超时、参数错误、模型返回格式不对、上下文超长、SSE 断连。这些场景要专门写测试用例。5.4 关于 Claude Code 和 Agent 框架的一些观察Claude Code 这类工具之所以好用很大程度上是因为它在 Agent Loop 之上做了很多工程优化。比如它的工具系统设计得很克制每个工具职责单一它的上下文管理很激进会主动压缩和丢弃它的终止控制很严格有副作用的操作都要确认。如果你在自研 Agent 框架我建议先不要追求功能全面而是把 Agent Loop 的核心环节做扎实工具调度要可靠、上下文管理要稳健、终止控制要明确、SSE 推送要实时。这四个做好了再往上加功能。关于harness 和 agent 的区别我的理解是harness 是承载 Agent 运行的基础设施模型调用、工具执行、状态管理agent 是基于 harness 构建的具体智能体有特定的提示词、工具集、行为模式。两者是分层关系不是替代关系。6. 把 Agent Loop 做稳的几个关键决策6.1 并发工具调用的处理模型一次返回多个工具调用时要不要并发执行我的答案是看工具性质。只读工具查询类可以并发有副作用的工具写入类要串行。并发执行能显著降低延迟但要注意结果顺序——并发执行完的结果要按模型请求的顺序回注到消息列表里否则模型可能对不上号。async def execute_tools_concurrent(self, calls): read_calls [c for c in calls if self.tools.get(c.name).is_readonly] write_calls [c for c in calls if not self.tools.get(c.name).is_readonly] # 只读并发 read_results await asyncio.gather(*[self.execute_tool_async(c) for c in read_calls]) # 写入串行 write_results [] for c in write_calls: write_results.append(await self.execute_tool_async(c)) # 按原始顺序合并 result_map {c.id: r for c, r in zip(read_calls write_calls, read_results write_results)} return [result_map[c.id] for c in calls]6.2 上下文压缩的具体策略上下文压缩不是简单截断要有策略。我的做法是分三档。第一档工具结果截断。单个工具结果超过 4000 字符就截断保留头部和尾部中间用省略号。头部通常是关键信息尾部通常是总结或错误信息。第二档历史轮次压缩。消息列表超过 30 条时把最早的 10 条压缩成一条摘要消息。摘要内容包含用户原始需求、已完成的关键操作、当前状态。第三档滑动窗口。压缩后还是超长就丢弃最早的消息但保留第一条用户消息原始需求和最近 10 条消息。这三档策略组合使用基本能应对绝大多数场景。关键是每一档都要有日志方便排查为什么模型失忆了这类问题。6.3 模型输出的解析容错模型输出格式不对是 Agent Loop 最常见的故障之一。模型可能返回不完整的 JSON、多余的文本、格式错误的工具调用。解析器要足够健壮。我的做法是先尝试严格解析失败后尝试宽松解析比如用正则提取 JSON 片段再失败就把原始输出作为文本处理并给模型一个提示你的输出格式不正确请按以下格式重新输出。这个重试机制能解决大部分格式问题。def parse_intent(self, response): # 严格解析 try: return self.strict_parse(response) except ParseError: pass # 宽松解析 try: return self.lenient_parse(response) except ParseError: pass # 兜底当成文本处理 return Intent(typefinal_answer, contentresponse.text)6.4 监控和可观测性Agent Loop 上线之后你需要知道它在干什么。我一般会埋这几类指标每轮循环的耗时、工具调用的成功率、上下文长度的分布、循环轮数的分布、终止原因的分布。这些指标能帮你发现潜在问题比如某个工具成功率突然下降、上下文长度持续增长、循环轮数异常偏高等。日志方面我建议每个会话一个 trace ID所有相关日志都带上这个 ID。这样排查问题时能快速拉出一个会话的完整执行链路。7. 一些踩坑之后的个人体会Agent Loop 这个东西入门容易精通难。我见过太多人写了个 demo 就觉得不过如此结果一上真实场景就各种翻车。真实场景和 demo 的区别在于demo 里模型总是乖乖听话真实场景里模型会犯错、工具会失败、网络会抖动、上下文会爆炸。我的建议是不要一开始就追求功能全面而是把核心环节做扎实。工具调度要可靠上下文管理要稳健终止控制要明确SSE 推送要实时。这四个做好了再往上加功能。另外测试要覆盖异常路径正常路径谁都能跑通真正考验健壮性的是异常路径。最后分享一个小技巧如果你在调试 Agent Loop把每一轮的完整消息列表 dump 到文件里出问题的时候直接看文件。比在控制台里翻日志高效得多。这个习惯帮我省了无数排查时间。 Agent Loop 实战：从朴素循环到状态机驱动的工程化设计
1. 别被循环两个字骗了Agent Loop 的真实复杂度很多人第一次接触 Agent 开发看到Agent Loop这个词脑子里浮现的就是一个 while 循环调模型、拿结果、执行工具、把结果塞回去、再调模型直到模型说我完成了。写出来大概二十行代码跑个 demo 看着挺像那么回事。我当初也是这么想的直到我把这套东西接到真实项目里才发现这个循环背后藏着一堆让人头皮发麻的细节。先说结论Agent Loop 的本质不是循环而是一个状态机驱动的决策-执行-反馈系统。循环只是它最外层的表现形式。你真正要处理的是每一轮对话里模型输出的意图如何被解析、工具调用如何被调度、执行结果如何被裁剪和回注、上下文如何被管理、异常如何被恢复、以及整个过程中如何保证不失控。这些东西demo 里一个都不会教你。这篇文章适合谁看如果你已经写过最基础的 Agent 循环能跑通模型调用工具的流程但一到真实场景就各种翻车——工具调错、上下文爆炸、循环停不下来、SSE 流断在半路——那这篇就是写给你的。我会把 Agent Loop 从看起来简单到实际复杂之间的那层窗户纸捅破把每个环节的真实坑点和处理思路讲清楚。核心关键词先摆出来Agent、Agent Loop、Tool System、SSE、Claude Code。这几个词基本覆盖了当前 Agent 开发的主战场。下面我按实际开发中会遇到的顺序一层一层拆。2. Agent Loop 的整体设计与思路拆解2.1 为什么调模型-执行工具这个朴素循环不够用最朴素的 Agent Loop 长这样把用户输入和工具定义一起发给模型模型返回一个工具调用请求你执行工具把结果拼回消息列表再发给模型如此往复。这个结构在单轮、单工具、无异常的场景下确实能跑。但真实项目里你马上会遇到几个问题。第一模型可能一次返回多个工具调用。比如你让它查一下北京天气顺便看看明天适不适合出门它可能同时发起天气查询和日历查询两个工具调用。你的循环如果只处理一个另一个就被丢了。第二工具执行可能失败。网络超时、参数错误、权限不足这些失败信息怎么回传给模型让它自己决定重试还是换方案是个设计问题。第三循环可能停不下来。模型陷入调工具-看结果-再调同一个工具的死循环或者一直说我需要更多信息但永远不给出最终答案。所以一个能用的 Agent Loop必须在朴素循环之上加三层东西工具调度层处理多工具、并发、超时、状态管理层维护对话历史、工具结果、中间状态、终止控制层判断什么时候该停、什么时候该强制停。2.2 状态机视角把 Loop 拆成可管理的阶段我习惯把 Agent Loop 拆成五个阶段每个阶段有明确的输入输出和退出条件。这样拆的好处是每个阶段可以单独测试、单独优化出问题的时候能快速定位是哪一层挂了。阶段职责关键输出常见问题意图解析解析模型输出提取工具调用或最终答案结构化意图对象解析失败、格式不符工具调度决定调哪些工具、并发还是串行、超时设置执行计划并发冲突、超时未处理结果处理裁剪、格式化、截断工具返回可回注的结果上下文爆炸、信息丢失上下文管理决定哪些历史保留、哪些压缩、哪些丢弃新的消息列表超长截断、关键信息被删终止判断判断是否继续循环、是否强制结束继续或结束信号死循环、提前终止这个拆法不是教科书上的标准答案是我在实际项目里踩坑之后总结出来的。每个阶段的边界清晰了你才能针对性地加日志、加监控、加兜底。2.3 为什么选 SSE 而不是 WebSocketAgent 场景下前端需要实时看到模型的输出流和工具执行状态。这里有个选型问题用 WebSocket 还是 SSE我的选择是SSEServer-Sent Events。原因很简单Agent 的输出是单向的——服务端推给客户端客户端不需要在这个通道上往回发消息用户输入走普通 HTTP 请求就行。SSE 基于 HTTP天然支持断线重连、自带事件 ID、实现简单不需要像 WebSocket 那样维护双向连接状态。对于 Agent 这种请求-流式响应的模式SSE 是更贴合的选择。但 SSE 也有它的坑。最常见的就是stream disconnected before completion: idle timeout waiting for SSE——连接空闲超时被断开。这个问题的根源在于Agent 在执行工具的时候可能几十秒都不产生新的输出SSE 连接因为长时间没有数据而被中间层网关、负载均衡、浏览器掐断。解决办法后面会详细讲。3. 核心细节解析与实操要点3.1 工具系统设计别把工具当函数很多人设计 Tool System 的时候就是把一堆函数注册进去写个 JSON Schema 描述参数完事。这在工具少的时候没问题工具一多就乱套。我总结下来工具系统设计要关注四个维度。第一个维度是工具粒度。工具太细模型要调很多次才能完成一件事循环轮数暴涨工具太粗参数复杂模型容易填错。我的经验是一个工具对应一个明确的动作参数控制在 5 个以内。比如查询天气是一个工具发送邮件是一个工具但处理用户请求这种就不该是工具。第二个维度是工具描述。模型选工具全靠描述。描述写得含糊模型就乱选。我见过有人把工具描述写成用于处理数据结果模型什么数据都往这个工具里塞。好的描述应该包含这个工具做什么、什么时候用、参数含义、返回什么、有什么限制。描述里最好带一两个使用示例。第三个维度是错误处理。工具执行失败时不要把异常直接抛给模型看堆栈。要把错误转成模型能理解的自然语言比如天气查询失败原因是城市名称无法识别请确认城市名是否正确。这样模型才能根据错误信息调整策略。第四个维度是权限和副作用。有些工具是只读的查询类有些是有副作用的写入、发送、删除。有副作用的工具要格外小心最好加确认机制或者限制调用频率。Claude Code 里那个每次都要点确认的机制本质上就是在有副作用的工具上加人工确认。3.2 上下文管理Agent Loop 里最容易爆的地方Agent Loop 跑着跑着上下文就爆了这是几乎所有 Agent 项目都会遇到的问题。原因很简单每一轮工具调用的结果都会追加到消息列表里几轮下来消息列表长度轻松超过模型的上下文窗口。处理这个问题有几种策略我按推荐程度排序。策略一工具结果截断。工具返回的内容如果很长比如读了一个大文件、查了一大段日志只保留关键部分。具体做法是设置一个字符上限超过就截断并在截断处加一句内容已截断如需完整内容请指定更精确的查询范围。这样既控制了长度又给了模型继续探索的线索。策略二历史消息压缩。当消息列表超过一定长度把早期的对话压缩成摘要。比如把前 10 轮的工具调用和结果压缩成一段之前已经完成了 X、Y、Z的描述。这个操作要小心压缩得太狠会丢失关键信息导致模型重复劳动。策略三滑动窗口。只保留最近 N 轮的消息更早的直接丢弃。这个最简单粗暴但风险也最大——如果早期有重要的工具结果丢了之后模型就失忆了。我的实际做法是组合使用工具结果先截断消息列表超过阈值时做压缩压缩后如果还是超长再用滑动窗口兜底。三层防护下来基本不会爆。3.3 SSE 流式输出的实现要点SSE 在 Agent 场景下的实现有几个细节必须处理好否则用户体验会很差。事件类型设计。不要只推一种事件。我一般会定义这几类事件message_start开始输出、content_delta文本增量、tool_call_start工具调用开始、tool_call_result工具执行结果、message_end输出结束、error错误。前端根据事件类型做不同的 UI 展示用户能看到模型正在思考正在调用工具工具返回了结果这些状态。心跳保活。前面提到的idle timeout问题解决办法是定期发送心跳事件。即使没有实际内容也每隔 15-30 秒推一个ping事件保持连接活跃。这个间隔要根据你的网关超时设置来定一般比网关超时短一半比较安全。断线重连。SSE 自带Last-Event-ID机制客户端重连时会带上最后收到的事件 ID服务端可以从这个 ID 之后继续推。但 Agent 场景下服务端的状态是有状态的消息列表在内存里重连后要能恢复到正确的状态。我的做法是把 Agent 的执行状态持久化重连时根据会话 ID 恢复。错误传播。工具执行出错、模型调用失败这些都要通过 SSE 推给前端而不是让连接静默断掉。前端收到error事件后可以展示错误信息并决定是否重试。4. 实操过程与核心环节实现4.1 从零搭一个可用的 Agent Loop下面我用 Python 伪代码的方式把核心循环写出来。这不是完整可运行的代码但结构是完整的你可以照着这个骨架往里面填具体实现。class AgentLoop: def __init__(self, model_client, tool_registry, max_iterations20): self.model model_client self.tools tool_registry self.max_iterations max_iterations def run(self, user_input, session): session.add_message(user, user_input) for i in range(self.max_iterations): # 1. 调用模型 response self.model.chat( messagessession.get_messages(), toolsself.tools.get_schemas() ) # 2. 解析意图 intent self.parse_intent(response) if intent.type final_answer: session.add_message(assistant, intent.content) return intent.content # 3. 执行工具 if intent.type tool_calls: session.add_message(assistant, response.raw) for call in intent.calls: result self.execute_tool(call) session.add_message(tool, result, tool_call_idcall.id) # 4. 上下文管理 session.compact_if_needed() # 5. 强制终止 return self.force_finalize(session)这个骨架里parse_intent负责把模型输出转成结构化意图execute_tool负责执行单个工具并处理异常compact_if_needed负责上下文压缩force_finalize是循环次数用尽后的兜底。4.2 工具执行器的实现细节工具执行器看起来简单实际上要处理的东西不少。我把它拆成几个步骤。def execute_tool(self, call): # 1. 参数校验 tool self.tools.get(call.name) if not tool: return f错误工具 {call.name} 不存在 try: params json.loads(call.arguments) except json.JSONDecodeError: return f错误参数格式不正确请检查 JSON 格式 # 2. 参数 schema 校验 validation_error tool.validate(params) if validation_error: return f错误参数校验失败 - {validation_error} # 3. 执行带超时 try: result tool.execute_with_timeout(params, timeout30) except TimeoutError: return f错误工具 {call.name} 执行超时30秒 except Exception as e: return f错误工具执行失败 - {str(e)} # 4. 结果截断 return self.truncate_result(result, max_chars4000)这里有几个关键点。参数校验要在执行前做不要等工具内部报错。超时必须有否则一个卡住的工具会拖死整个循环。错误信息要转成自然语言让模型能理解。结果要截断防止上下文爆炸。4.3 SSE 服务端的实现SSE 服务端我用 FastAPI 举例核心是返回一个StreamingResponse里面是一个生成器不断 yield 事件。from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio, json app FastAPI() async def event_stream(session_id, user_input): agent get_agent(session_id) queue asyncio.Queue() # 启动 Agent 执行把事件推到 queue asyncio.create_task(agent.run_streaming(user_input, queue)) last_heartbeat time.time() while True: try: event await asyncio.wait_for(queue.get(), timeout15) yield format_sse(event) last_heartbeat time.time() except asyncio.TimeoutError: # 心跳保活 yield format_sse({type: ping}) last_heartbeat time.time() def format_sse(event): return fevent: {event[type]}\ndata: {json.dumps(event)}\n\n app.get(/agent/stream) async def stream(session_id: str, input: str): return StreamingResponse( event_stream(session_id, input), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, # 禁用 nginx 缓冲 } )这里的关键是X-Accel-Buffering: no这个头。如果你的服务前面有 nginx不加这个头nginx 会缓冲 SSE 响应导致前端收不到实时数据。这个坑我踩过排查了大半天。4.4 终止控制的实现终止控制分两种正常终止和强制终止。正常终止是模型主动给出最终答案。判断逻辑是模型返回的消息里没有工具调用只有文本内容且文本内容不是我需要调用工具这类中间态。这个判断要小心有些模型会输出我将要调用工具这样的文本但实际没有工具调用这种要当成最终答案处理。强制终止是循环次数用尽或者检测到死循环。死循环的检测方法是记录最近几轮的工具调用签名工具名参数如果连续三轮完全相同就判定为死循环强制终止并返回当前已有的信息。def detect_loop(self, session, window3): recent session.get_recent_tool_calls(window) if len(recent) window: return False signatures [f{c.name}:{c.arguments} for c in recent] return len(set(signatures)) 1强制终止时不要直接返回空要把当前已经收集到的信息整理一下返回给用户并说明由于达到最大执行轮数以下是目前已完成的部分。5. 常见问题与排查技巧实录5.1 SSE 相关问题的排查stream disconnected before completion: idle timeout waiting for SSE这个报错我见过太多次了。排查思路是这样的。先确认心跳有没有在发。在服务端加日志看ping事件是不是按预期间隔发出。如果没有检查你的asyncio.wait_for超时设置是不是比网关超时还长。再确认中间层有没有缓冲。如果服务前面有 nginx、网关、CDN每一层都可能缓冲 SSE。nginx 要加X-Accel-Buffering: no网关要确认支持流式响应CDN 一般要关掉对text/event-stream的缓冲。最后确认客户端有没有正确处理。有些 HTTP 客户端库默认会缓冲整个响应要设置成流式读取。浏览器端的EventSource默认就是流式的但如果你用fetch手动处理要注意用response.body.getReader()而不是response.text()。5.2 工具调用相关问题的排查模型调错工具、参数填错、工具执行失败这些问题排查起来有个通用思路把完整的消息列表打出来看。很多时候问题出在工具描述不清楚或者历史消息里有干扰信息。我整理了一个常见问题速查表。现象可能原因排查方向模型不调用工具工具描述不清、系统提示没强调检查工具描述和 system prompt模型调错工具工具之间描述太相似区分工具描述加使用场景说明参数格式错误Schema 定义不严、缺少示例补充参数示例和格式说明工具执行超时工具本身慢、网络问题加超时、优化工具实现循环停不下来终止条件不明确、工具结果没给足信息检查终止逻辑、补充工具返回信息上下文爆炸工具结果太长、历史没压缩加截断和压缩策略5.3 实操心得几个让我少走弯路的经验第一个经验日志要打全。Agent Loop 出问题的时候你需要的不是哪一步错了而是每一步的输入输出是什么。我习惯在每一轮循环开始时打一条日志包含当前消息列表长度、本轮模型输出、工具调用详情、工具返回结果。这样出问题的时候直接看日志就能定位。第二个经验给模型留退路。工具执行失败的时候不要只返回错误要返回错误建议。比如查询失败城市名可能拼写错误请尝试使用标准城市名如北京上海。模型看到建议更容易自我纠正。第三个经验限制工具调用轮数。我一般设 15-20 轮上限。超过这个轮数还没完成的任务要么是任务本身太复杂需要拆解要么是模型陷入了某种循环。强制终止比无限循环好。第四个经验工具结果要结构化。工具返回的内容尽量用结构化的格式JSON、Markdown 表格而不是一大段自然语言。结构化内容模型更容易解析也更容易截断。第五个经验测试要覆盖异常路径。正常路径谁都能跑通真正考验 Agent Loop 健壮性的是异常路径工具超时、参数错误、模型返回格式不对、上下文超长、SSE 断连。这些场景要专门写测试用例。5.4 关于 Claude Code 和 Agent 框架的一些观察Claude Code 这类工具之所以好用很大程度上是因为它在 Agent Loop 之上做了很多工程优化。比如它的工具系统设计得很克制每个工具职责单一它的上下文管理很激进会主动压缩和丢弃它的终止控制很严格有副作用的操作都要确认。如果你在自研 Agent 框架我建议先不要追求功能全面而是把 Agent Loop 的核心环节做扎实工具调度要可靠、上下文管理要稳健、终止控制要明确、SSE 推送要实时。这四个做好了再往上加功能。关于harness 和 agent 的区别我的理解是harness 是承载 Agent 运行的基础设施模型调用、工具执行、状态管理agent 是基于 harness 构建的具体智能体有特定的提示词、工具集、行为模式。两者是分层关系不是替代关系。6. 把 Agent Loop 做稳的几个关键决策6.1 并发工具调用的处理模型一次返回多个工具调用时要不要并发执行我的答案是看工具性质。只读工具查询类可以并发有副作用的工具写入类要串行。并发执行能显著降低延迟但要注意结果顺序——并发执行完的结果要按模型请求的顺序回注到消息列表里否则模型可能对不上号。async def execute_tools_concurrent(self, calls): read_calls [c for c in calls if self.tools.get(c.name).is_readonly] write_calls [c for c in calls if not self.tools.get(c.name).is_readonly] # 只读并发 read_results await asyncio.gather(*[self.execute_tool_async(c) for c in read_calls]) # 写入串行 write_results [] for c in write_calls: write_results.append(await self.execute_tool_async(c)) # 按原始顺序合并 result_map {c.id: r for c, r in zip(read_calls write_calls, read_results write_results)} return [result_map[c.id] for c in calls]6.2 上下文压缩的具体策略上下文压缩不是简单截断要有策略。我的做法是分三档。第一档工具结果截断。单个工具结果超过 4000 字符就截断保留头部和尾部中间用省略号。头部通常是关键信息尾部通常是总结或错误信息。第二档历史轮次压缩。消息列表超过 30 条时把最早的 10 条压缩成一条摘要消息。摘要内容包含用户原始需求、已完成的关键操作、当前状态。第三档滑动窗口。压缩后还是超长就丢弃最早的消息但保留第一条用户消息原始需求和最近 10 条消息。这三档策略组合使用基本能应对绝大多数场景。关键是每一档都要有日志方便排查为什么模型失忆了这类问题。6.3 模型输出的解析容错模型输出格式不对是 Agent Loop 最常见的故障之一。模型可能返回不完整的 JSON、多余的文本、格式错误的工具调用。解析器要足够健壮。我的做法是先尝试严格解析失败后尝试宽松解析比如用正则提取 JSON 片段再失败就把原始输出作为文本处理并给模型一个提示你的输出格式不正确请按以下格式重新输出。这个重试机制能解决大部分格式问题。def parse_intent(self, response): # 严格解析 try: return self.strict_parse(response) except ParseError: pass # 宽松解析 try: return self.lenient_parse(response) except ParseError: pass # 兜底当成文本处理 return Intent(typefinal_answer, contentresponse.text)6.4 监控和可观测性Agent Loop 上线之后你需要知道它在干什么。我一般会埋这几类指标每轮循环的耗时、工具调用的成功率、上下文长度的分布、循环轮数的分布、终止原因的分布。这些指标能帮你发现潜在问题比如某个工具成功率突然下降、上下文长度持续增长、循环轮数异常偏高等。日志方面我建议每个会话一个 trace ID所有相关日志都带上这个 ID。这样排查问题时能快速拉出一个会话的完整执行链路。7. 一些踩坑之后的个人体会Agent Loop 这个东西入门容易精通难。我见过太多人写了个 demo 就觉得不过如此结果一上真实场景就各种翻车。真实场景和 demo 的区别在于demo 里模型总是乖乖听话真实场景里模型会犯错、工具会失败、网络会抖动、上下文会爆炸。我的建议是不要一开始就追求功能全面而是把核心环节做扎实。工具调度要可靠上下文管理要稳健终止控制要明确SSE 推送要实时。这四个做好了再往上加功能。另外测试要覆盖异常路径正常路径谁都能跑通真正考验健壮性的是异常路径。最后分享一个小技巧如果你在调试 Agent Loop把每一轮的完整消息列表 dump 到文件里出问题的时候直接看文件。比在控制台里翻日志高效得多。这个习惯帮我省了无数排查时间。