01 - Agent 核心循环：ReAct 模式与多轮推理

学习目标

理解 CowAgent 如何通过 AgentStreamExecutor 实现基于 tool-call 的多轮 ReAct 推理循环，包括取消机制、失败保护、空响应处理和流式输出。

前置知识

CowAgent 的 Agent 核心循环遵循经典的 ReAct（Reasoning + Acting） 模式。通用原理见 Agent 核心循环设计。本章聚焦 CowAgent 的具体实现方案。

项目实践

核心循环架构

CowAgent 的推理循环由 AgentStreamExecutor.run_stream() 方法实现，核心流程如下：

关键代码结构

AgentStreamExecutor 的 run_stream() 方法实现了一个带保护的多轮循环：

# 伪代码
def run_stream(self, user_message: str) -> str:
    # 1. 添加用户消息（Claude content blocks 格式）
    self.messages.append({"role": "user", "content": [{"type": "text", "text": user_message}]})

    # 2. 裁剪上下文（仅在循环开始前，不在 tool 步骤中裁剪）
    self._trim_messages()
    self._validate_and_fix_messages()

    # 3. 推理循环
    while turn < self.max_turns:
        self._check_cancelled()  # 每个 turn 开始时检查取消

        # 调用 LLM（支持流式 + 自动重试）
        assistant_msg, tool_calls = self._call_llm_stream(retry_on_empty=True)

        # 无 tool_calls → 循环结束
        if not tool_calls:
            if not assistant_msg and turn > 1:
                # 空响应保护：明确要求 LLM 生成文本回复
                self.messages.append({"role": "user", "content": "请向用户说明结果"})
                assistant_msg, tool_calls = self._call_llm_stream(retry_on_empty=False)
            break

        # 执行工具（带失败保护）
        for tool_call in tool_calls:
            self._check_cancelled()  # 工具间也检查取消
            result = self._execute_tool(tool_call)

            if result["status"] == "critical_error":
                return result["result"]  # 严重错误 → 立即终止

        # 注入 tool_result 到消息历史
        self.messages.append({"role": "user", "content": tool_result_blocks})

        # 循环保护：相同工具+参数连续 3+ 次成功 → 提示 LLM 停止

取消机制

CowAgent 通过 cancel_event（threading.Event）实现用户发起的取消：

• 检查点：每个 turn 开始时、工具调用之间、LLM 流式每 8 个 token
• 优雅终止：取消时注入 orphaned tool_result 块，保持消息历史格式合法
• 部分响应保留：已生成的文本回复被保留并返回

取消处理的关键设计：

def _handle_cancelled(self, partial_response: str) -> None:
    # 场景 (a): 最后一条是 assistant 包含 tool_use 但无匹配的 tool_result
    # → 注入 is_error=True 的 tool_result 保持格式合法
    # 场景 (b)/(c): 消息格式已合法 → 仅追加 "Cancelled by user" 标记

失败保护

CowAgent 实现了三层失败保护：

保护层级	触发条件	行为
参数重复检测	相同工具 + 相同参数连续 5 次调用	停止执行，提示 LLM 换方法
连续失败检测	相同工具 + 相同参数连续 3 次失败	停止该工具，提示 LLM 换方法
工具硬停止	相同工具连续 8 次失败（任意参数）	返回 critical_error，终止整个对话

这种分层设计避免了无限循环，同时在工具偶发失败时给 LLM 重试的机会。

空响应保护

当 LLM 返回空内容且无 tool_calls 时：

1. 第一轮：直接 fallback 到预设的错误提示
2. 后续轮次：注入明确的提示消息 "请向用户说明刚才工具执行的结果"，再调用一次 LLM
3. 第二次仍空：使用 fallback 提示

这个设计解决了常见的”工具执行后 LLM 认为任务完成不再回复”的问题。

问题与规避

问题	CowAgent 的规避策略
LLM 无限循环调用同一工具	参数哈希检测 + 连续失败计数，5 次同参数/8 次同工具自动停止
上下文裁剪导致 tool_use/tool_result 断裂	仅在循环开始前裁剪一次，循环中不裁剪
取消导致消息格式损坏	_handle_cancelled() 注入 error tool_result 保持格式合法
工具结果过大撑爆上下文	当前轮截断到 50K 字符，历史轮进一步截断到 20K 字符
LLM 流式返回无效 JSON	json_repair 库自动修复转义错误，截断时提供 max_tokens 提示

设计取舍

为什么只在循环开始前裁剪上下文，而不是每个 turn 都裁剪？

CowAgent 明确注释：“Context trimming is done once in run_stream() before the loop starts, NOT here — trimming mid-execution would strip the current run’s tool_use/tool_result chains and cause LLM loops.”

这是经过实践验证的决策。如果在工具调用过程中裁剪，可能会删除当前轮次的 tool_result，导致 LLM 在下一次调用时看不到结果，从而重复调用相同的工具——这是一个难以调试的无限循环 bug。

为什么不使用 asyncio 管理异步工具？

CowAgent 使用 threading 而非 asyncio 管理多渠道并发。原因是多渠道适配（微信、飞书等 IM SDK）大多是同步 API，asyncio 会引入不必要的复杂度。代价是线程安全需要显式加锁（messages_lock、BoundedSemaphore）。

参考来源

• CowAgent AgentStreamExecutor 源码
• CowAgent Agent 核心类源码
• Agent 核心循环设计