> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/04_runtime/4.1_agent_loop.md). # 4.1 智能体循环的工程实现 智能体循环是运行时的核心执行模式，定义了 Agent 如何在思考、行动和观察三个阶段间反复迭代。本节对比 Claude Code 和 OpenClaw 两种实现模式，分析循环的终止条件和消息流演化。 ## 4.1.1 概念模型：思考-行动-观察循环 智能体循环（Agent Loop）是智能体系统的核心执行模式。经典的“思考-行动-观察”(Think-Act-Observe)循环定义如下： 1. **思考(Think)**：Agent 基于当前上下文（历史、状态、工具信息）进行推理，生成意图和下一步行动计划 2. **行动(Act)**：模型发出工具调用意图，运行时验证、调度并由工具层执行，必要时改变外部世界状态 3. **观察(Observe)**：Agent 获得运行时返回的工具执行结果，更新内部认知模型，为下一轮思考做准备 如果工具调用失败或返回意外结果，观察阶段会将其作为“负面证据”反馈给思考模块，触发错误恢复或备选方案。 **思考-行动-观察循环流程：** ```mermaid graph LR A["LLM 推理"] --> B["工具执行"] B --> C["结果分析"] C --> D{"是否
完成?"} D -->|否| A D -->|是| E["返回最终结果"] style A fill:#e1f5ff style B fill:#fff3e0 style C fill:#f3e5f5 style D fill:#e8f5e9 style E fill:#c8e6c9 ``` 图 4-1：思考-行动-观察循环 **为什么是循环而不是链** 传统的链式智能体(Chain-of-Thought)将思考和行动分开，形如：提示→思考→规划→工具调用→输出。这种设计的缺点是： * **缺乏适应性**：一旦执行遇到问题（工具调用失败、输出超限），难以动态调整策略 * **低效率**：规划阶段必须预测所有可能的工具调用，容易过度规划或不足规划 * **不可恢复**：一旦规划失败，需要重新启动整个流程 **循环设计** 允许智能体逐步细化目标，在每个循环迭代中获得反馈，动态调整策略。这更符合人类解决问题的过程。 ## 4.1.2 工程实现模式对比 ### Claude Code 的异步生成器模式 Claude Code 采用异步生成器模式实现智能体循环，具体如下： ```mermaid flowchart TD subgraph Generator["QueryEngine.submitMessage() Generator"] A["submitMessage(userInput)"] --> B["yield agent_start"] B --> C subgraph C["Async Turn Loop (while hasWorkToDo)"] D["yield turn_start"] --> E["Build & Validate Context
(考虑缓存边界)"] E --> F["Stream Model Response
(yield text_delta / tool_use)"] F --> G["StreamingToolExecutor.execute()
(并发执行工具)"] G --> H["yield message_end"] H -->|"继续下一轮"| D end C --> I["yield agent_end, result"] end ``` **特点**： * 异步生成器模式，客户端可以逐个处理事件 * 支持流式响应实时推送(text\_delta) * StreamingToolExecutor 支持工具的并发执行 * 上下文构建时考虑缓存边界(SYSTEM\_PROMPT\_DYNAMIC\_BOUNDARY) * AppState 管理会话状态（150+ 字段） **Claude Code 的核心代码骨架** （伪代码）： ```python class QueryEngine: async def submitMessage(self, user_input: str): """异步生成器循环,每次 yield 一个事件""" yield AgentStartEvent() # 初始化会话状态 app_state = self.init_app_state(user_input) while app_state.has_work_to_do(): yield TurnStartEvent() # 构建完整的消息和上下文 # 考虑缓存边界,分离系统提示词和动态上下文 system_prompt = self.get_system_prompt() messages = await self.build_messages(app_state) # 检查Token预算 if self.estimate_tokens(system_prompt + messages) > self.token_budget: messages = self.auto_compact_history(messages) # 流式调用模型 stream = await self.model_client.create_message_stream( messages=messages, system=system_prompt, tools=self.get_available_tools() ) # 逐个处理流中的事件 # 按 Anthropic 流式协议增量累积内容块,在消息终止时才执行工具 assembler = MessageAssembler() async for event in stream: if event.type == "content_block_start": # 新的内容块开始(文本或工具使用),登记到累积器 assembler.start_block(event.index, event.content_block) yield event elif event.type == "content_block_delta": # 把增量累积进当前块(文本/工具入参 JSON) assembler.append_delta(event.index, event.delta) if event.delta.type == "text_delta": yield TextDeltaEvent(event.delta.text) elif event.delta.type == "input_json_delta": yield ToolInputDeltaEvent(event.delta.partial_json) elif event.type == "content_block_stop": # 当前块完成,定稿(如解析工具入参 JSON) assembler.finish_block(event.index) elif event.type == "message_delta": # 顶层增量:仅携带 stop_reason 与累计 usage,不含 content assembler.set_stop_reason(event.delta.stop_reason) assembler.add_usage(event.usage) elif event.type == "message_stop": # 消息终止信号(不含 content);此时内容块已累积完毕 message = assembler.build_message() # 异步执行工具(StreamingToolExecutor) tool_results = await self.execute_tools( message.content, app_state ) # 更新应用状态 app_state.add_message(message) app_state.add_tool_results(tool_results) yield MessageEndEvent(message) # 检查是否需要继续 if not tool_results: app_state.mark_done() # 否则继续下一轮(while 循环) # 返回最终结果 yield AgentEndEvent( result=app_state.final_response, message_count=len(app_state.messages) ) ``` ### OpenClaw 的线性流水线模式 OpenClaw 采用线性流水线模式，在单个事件循环中顺序执行，具体如下： ```mermaid flowchart TD subgraph SessionLoop["Session Loop(单线程)"] A["Intake
(user msg)"] --> B["Context Assembly"] B --> C["Inference
(Claude)"] C --> D["Response Streaming
+ Tool Detection"] D --> E["Tool Execution Loop
(顺序/不阻塞UI)"] E --> F["Persistence
(会话存储)"] F -->|"继续下一轮(如需要)"| A end ``` **特点**： * 每轮循环是独立的阶段：input → assembly → inference → execution → persist * 单线程执行，一次一个会话，不支持会话间并发 * 工具执行与流式响应紧耦合 * 错误作为“观察”(ToolResultBlock)反馈回消息流 **OpenClaw 的核心代码骨架** （伪代码）： ```python class SessionRuntime: def run_session(self, session_id: str): """单个会话的运行时主循环""" while True: # 1. Intake:获取用户输入或等待事件 user_message = self.intake_queue.get(timeout=30) if user_message is None: break # 会话结束 # 2. Context Assembly:组装上下文（每条新用户消息只组装一次） context = self.context_engine.assemble( session_id=session_id, user_input=user_message, system_prompt=self.system_prompt, tools_schema=self.get_tools_schema(), memory=self.memory_store.load(session_id) ) # Initialize messages from context messages = context.messages.copy() # 3-4. 推理与工具调用循环：直到模型不再请求工具（4.1.3 的"工具调用耗尽"终止条件） while True: # 3. Inference:调用模型 response = self.model_client.create_message( messages=messages, system=context.system_prompt, tools=context.tools, max_tokens=context.token_budget ) # 4. Tool Execution Loop:处理本轮 response 中的所有工具调用 tool_results = [] for tool_use in response.content: if isinstance(tool_use, ToolUseBlock): result = self.execute_tool( tool_use.name, tool_use.input ) tool_results.append(ToolResultBlock( tool_use_id=tool_use.id, content=result )) # 没有工具调用 → 推理完成，跳出内层循环 if not tool_results: break # 有工具调用 → 把助手回复与工具结果包装成消息追加到 messages，继续内层推理 # 注意：tool_result 块按 Anthropic Messages 约定属于 user 角色消息 messages.append({"role": "assistant", "content": response.content}) messages.append({"role": "user", "content": tool_results}) # 5. Persistence:持久化（仅在内层循环结束、得到最终回复后执行一次） self.message_store.save(session_id, response) self.memory_store.maybe_consolidate(session_id) # 把最终响应交付给用户，回到 Intake 等待会话的下一条输入 self.deliver(session_id, response) ``` ## 4.1.3 循环的终止条件 两个系统都需要明确定义何时停止循环，避免无限循环： ### 终止条件的类型 1. **工具调用耗尽**：Agent 推理出最后一条消息，不包含工具调用，直接回复用户 2. **最大轮数限制**：防止无限循环的安全机制，通常设为 10-30 轮 3. **Token 预算耗尽**：当前轮推理会导致上下文溢出，无法继续 4. **显式停止信号**：用户取消、超时、或智能体发出停止标记 5. **目标达成**：可选的高层目标追踪（自驱型智能体） ### 实现示例 以下是循环终止检查器的实现示例： ```python class LoopTerminationChecker: def __init__(self, max_turns: int = 20, token_limit: int = 100000): self.max_turns = max_turns self.token_limit = token_limit def should_continue(self, state: AgentState) -> bool: """判断是否应该继续循环""" # 检查1:轮数上限 if state.turn_count >= self.max_turns: state.termination_reason = "max_turns_reached" return False # 检查2:是否有工具调用 last_message = state.messages[-1] if not self._has_tool_calls(last_message): state.termination_reason = "no_tool_calls" return False # 检查3:Token预算 next_turn_tokens = self._estimate_tokens_for_next_turn(state) if next_turn_tokens > self.token_limit: state.termination_reason = "token_limit_exceeded" return False # 检查4:显式停止信号 if state.stop_signal_received: state.termination_reason = "user_stop" return False return True def _has_tool_calls(self, message: Message) -> bool: """检查消息中是否有工具调用块""" for block in message.content: if isinstance(block, ToolUseBlock): return True return False def _estimate_tokens_for_next_turn(self, state: AgentState) -> int: """估计下一轮推理的Token使用""" # 简化版本,实际需要调用计数器 history_tokens = len(state.messages) * 50 # 粗略估计 new_input_tokens = 100 # 工具结果的平均大小 return history_tokens + new_input_tokens ``` ## 4.1.4 消息流的演化过程 一个完整的 智能体循环中，消息流如何演化： ``` 轮次 1: Input: [UserMessage("Find files modified in last 7 days")] Output: [AssistantMessage([ TextBlock("I'll find recent files for you."), ToolUseBlock(name="bash", input="find /data -mtime -7") ])] 轮次 2(工具执行): Input: [ UserMessage("Find files modified in last 7 days"), AssistantMessage([...]), ToolResultBlock(tool_use_id=xxx, content="file1.txt\nfile2.log\n...") ] Output: [AssistantMessage([ TextBlock("Found 3 recent files: file1.txt, file2.log, file3.csv") ])] # 不再有工具调用,循环终止 最终消息历史: [UserMessage, AssistantMessage(Tool 1), ToolResultBlock, AssistantMessage(Final)] ``` 这种结构保证了： * 完整的推理链条可追踪 * 每次工具调用都有对应的结果 * 智能体的决策过程对用户透明 ## 4.1.5 Codex 智能体循环协议规范 Codex Harness 采用了 JSON-RPC 作为循环协议的基础，通过严格的通信原语确保循环的可靠性和可观测性。本节介绍 Codex 循环的协议层设计。 ### 三层通信原语 Codex Agent Loop 定义了三个核心的 JSON-RPC 通信原语，从底层到顶层分别为： 1. **Item** （项）：最小的原子通信单元，代表单次输入/输出操作。每个 Item 都有明确的生命周期： * `item/started`：Item 开始处理 * 可选的 `item/{type}/delta`：内容流式增量到达（适用于文本、工具执行结果等） * `item/completed`：Item 完成并返回最终 payload Item 的类型包括用户消息、助手消息、工具执行、审批请求、差异(diff)等，类型化设计使客户端能够进行强类型校验。 2. **Turn** （轮）：由用户输入发起的单位工作。一个 Turn 包含一个有序的 Item 序列，表示从用户请求到最终输出的整个执行过程： * `turn/start`：接收用户输入（例如“运行测试并总结失败原因”） * 中间步骤：思考、工具调用、获取结果等（各自作为 Item） * `turn/completed`：最终助手消息返回，Turn 结束 3. **Thread** （线程）：持久化的会话容器，包含多个 Turn。Threads 支持创建、恢复、分叉和归档，历史记录持久化确保客户端可以重连并恢复一致的状态。 ### 前缀一致性与缓存优化 Codex 的 JSON-RPC 协议强制执行 **前缀一致性** 原则：每一轮推理的输入都是上一轮输出的精确前缀。这一设计直接启用了模型推理的 **提示词缓存** (Prompt Caching)，使多轮对话的成本从二次方复杂度降至线性复杂度。 具体地，如果第一轮推理的完整消息列表为 `[msg1, msg2, ..., msgN]`，第二轮新增工具结果后的列表必须为 `[msg1, msg2, ..., msgN, toolResult, assistantMsg]`，旧消息部分保持不变。这对循环的构建有两个影响： * **工具枚举顺序的重要性**：MCP 工具在初始化时必须以 **确定的顺序** 枚举，否则工具列表顺序变化会破坏前缀，导致缓存未命中(Cache Miss)。Codex 发现过一个 bug：MCP 工具枚举的顺序不确定，在长会话中每一轮都造成缓存未命中，性能严重下降。 * **配置变更的处理**：若在对话中途需要改变权限设置、沙箱配置或工作目录，Codex 不会修改历史消息，而是追加一个新的`role=developer`消息声明配置变更，保持前缀的完整性。 ### 双向通信模式 App Server 的 JSON-RPC 协议是 **完全双向的**，不同于传统的单向请求-响应模式： * **客户端请求**：初始化握手(initialize)、线程和轮次创建(thread/start, turn/start)、输入提交 * **服务端通知**：进度更新（thread/started, turn/started, item/started 等）、Item 的增量内容（delta 事件）、Item 完成(item/completed) * **服务端请求**：当需要用户审批（例如“是否允许执行 pnpm test”）时，服务端主动向客户端发起请求，转为 **暂停** 状态，直到客户端响应批准或拒绝 这种双向特性使循环成为一个 **响应式的流程**：客户端 UI 可以在 Item 开始时立即开始渲染，流式地接收 delta 事件进行增量更新，在 Item 完成时进行最终定稿。服务端也可以在关键决策点（如权限检查）暂停转移控制权给用户。 ## 4.1.6 本节小结 智能体循环的工程实现有两种主要模式： * **Claude Code 的异步生成器** 追求低延迟与高并发，通过事件驱动的方式支持流式响应，适合交互式的任务场景 * **OpenClaw 的线性流水线** 强调确定性与可追踪性，通过单线程阶段化执行保证顺序，适合需要完全控制的自驱型场景 在生产级 Codex 系统中，通过 **JSON-RPC 协议的三层原语** (Item/Turn/Thread)、 **前缀一致性缓存优化** 和 **双向通信模式**，实现了高效可靠的循环编排。这些设计模式的核心是“思考-行动-观察”循环，但将其具体化为可观测、可缓存、可控制的 protocol-level 操作。 理解这两种实现模式和协议规范，是构建生产级 Harness 运行时的基础。