ReAct 循环与 Agent 控制流：从原理到生产级实现

2026-06-11AgentReAct控制流生产实践

什么是 ReAct 循环

ReAct（Reasoning + Acting）是现代 Agent 的基础循环范式，也是 LangChain、LangGraph、OpenAI Agents SDK 等主流框架的核心。Agent 在每轮中交替进行"思考"和"行动"，直到任务完成或达到最大步数。

┌──────────┐     ┌──────────┐     ┌──────────────┐
│  Thought  │────▶│  Action   │────▶│ Observation  │
│  (思考)   │◀────│  (行动)   │◀────│  (观察)       │
└──────────┘     └──────────┘     └──────┬───────┘
       ▲                                  │
       └──────────── 循环 ────────────────┘
                    ↓ (任务完成)
              Final Answer

核心实现要素

1. 超时熔断机制

防止单步卡死或整体任务超时，必须在两个层面做熔断：

class AgentExecutor:
    def __init__(self, max_steps=12, max_time=60, tool_timeout=30):
        self.max_steps = max_steps          # 最大步数限制
        self.max_time = max_time             # 整体超时(秒)
        self.tool_timeout = tool_timeout     # 单步工具调用超时(秒)

    async def run(self, user_input, session_id):
        start_time = time.time()

        for step in range(self.max_steps):
            # 整体超时熔断
            if time.time() - start_time > self.max_time:
                messages.append(HumanMessage(
                    "时间已到，请基于已有信息给出最佳回答，不要继续调用工具。"
                ))
                return await self.llm.ainvoke(messages)

            # 单步工具调用超时
            try:
                tool_result = await asyncio.wait_for(
                    tool.execute(args), timeout=self.tool_timeout
                )
            except asyncio.TimeoutError:
                tool_result = "工具调用超时，请尝试简化参数或使用替代方案。"

2. 重复动作检测

Agent 有时会陷入"死循环"——反复调用同一个工具但期望不同结果。用 action_history 做哈希去重：

# 在循环中记录每一次工具调用
self.action_history = []

def _check_repetition(self, tool_name, args):
    key = (tool_name, str(args))
    self.action_history.append(key)
    count = self.action_history.count(key)
    if count >= 2:
        return f"已重试 {count} 次，结果未变。请换一种方式：使用不同工具、调整参数、或直接基于已有信息回答。"
    return None

关键细节：去重比较的是参数哈希而非仅函数名——相同工具不同参数不视为重复。另外，提示语的措辞很重要："请换一种方式"比"不要重复调用"更有效，因为正向引导优于负向禁止。

3. 工具结果截断

工具调用可能返回大量数据（网页全文、数据库查询结果等），直接塞进上下文窗口既浪费 Token 又稀释关键信息：

MAX_TOOL_RESULT_CHARS = 2000

def truncate_result(result: str) -> str:
    if len(result) <= MAX_TOOL_RESULT_CHARS:
        return result
    return result[:MAX_TOOL_RESULT_CHARS] + \
           f"\n\n...（结果已截断，原始长度 {len(result)} 字符。" + \
           f"如需完整内容，请缩小查询范围或增加过滤条件。）"

避坑：截断必须在语义边界（句子/段落结尾），不能截在 JSON 中间或代码块中间。超出部分明确告知 Agent 可如何获取更多信息。

4. 进度注入

对于多步骤任务，在每一步完成后向 Agent 注入进度信息，帮助它感知"走到哪了"：

progress_msg = (
    f"[进度] 第 {step+1}/{max_steps} 步。"
    f"已完成: {completed_count}/{total_count}。"
    f"下一步建议: {suggested_next_action}"
)
messages.append(HumanMessage(progress_msg))

架构模式选型

模式适用场景复杂度代表实现
ReAct 循环需要工具调用的交互式 Agent中LangGraph AgentExecutor
Plan-then-Execute复杂的多步骤任务，需预规划高AutoGPT、CrewAI
Router + Worker多领域多能力的 Agent 平台高OpenAI Swarm
Simple Chain确定性流程（文档处理流水线）低LangChain LCEL

生产环境 Checklist

  • 最大步数限制：10-15 步，防止无限循环
  • 整体超时：60-120 秒，取决于任务复杂度
  • 单步工具超时：30 秒，防止外部 API 卡死
  • 重复检测：≥2 次相同参数调用时触发
  • Token 预算：每步注入的上下文总 Token 数有上限
  • 优雅降级：超时后不是报错，而是基于已有信息给出最佳回答
  • 全链路日志：每一步的 thought、action、observation 完整记录
未标记