🔄 主循环设计 — ReAct / Plan-Execute / State Machine

1. 所有主循环的本质

不管哪种架构，Agent 主循环的数学本质只有一个：

2. ReAct 循环 — 9/20 项目的选择

ReAct（Reasoning + Acting）是出现频率最高的主循环模式。核心思路：每一步都让 LLM 先推理、再决定行动、观察结果、继续推理。

2.1 各项目的 ReAct 实现差异

2.2 OpenHands 的 ReAct：Event-Driven 变体（最佳实践）

OpenHands 在标准 ReAct 基础上做了两个关键创新：

# 事件层级
Event (frozen Pydantic, append-only)
  ├─ LLMConvertibleEvent → 可转为 LLM Message
  │   ├─ ActionEvent        # Agent 工具调用
  │   ├─ ObservationEvent   # 工具返回结果
  │   ├─ MessageEvent       # 用户/Agent 文本消息
  │   └─ SystemPromptEvent  # 初始系统提示词
  ├─ Condensation           # 上下文压缩事件
  ├─ CondensationRequest    # 请求压缩（超出窗口时触发）
  ├─ TokenEvent             # 流式 token
  └─ AgentErrorEvent        # 错误

class LLMSummarizingCondenser(Condenser):
    """将早期事件总结为一条摘要，保留近期事件不变"""
    
    def condense(self, events: list[Event]) -> list[Event]:
        if token_count(events) <= max_size:
            return events
        # 1. 保留最近 N 个事件
        recent = events[-keep_recent:]
        # 2. 对早期事件做 LLM 摘要
        summary = self.llm.summarize(events[:-keep_recent])
        # 3. 返回 [Condensation(summary), ...recent]
        return [Condensation(content=summary)] + recent

2.3 SWE-agent 的 ReAct：ACI + 错误重查询

SWE-agent 的 ReAct 循环有独特的 错误重查询 机制——当 LLM 输出格式错误或触发 blocklist 时，不是直接失败，而是将错误信息注入历史重新查询：

# forward_with_handling 伪代码
def forward_with_handling(self, history):
    n_format_fails = 0
    while n_format_fails < self.max_requeries:
        try:
            return self.forward(history)         # 正常 ReAct 步
        except FormatError as e:
            n_format_fails += 1
            history = requery_history(e)         # 把错误加入历史重问
        except BlocklistError as e:
            history = requery_history(e)         # 被禁命令也重问
        except BashSyntaxError as e:
            history = requery_history(e)         # Bash 语法错误也重问
        except AgentGaveUpError:
            return attempt_autosubmission()      # 放弃但尝试提交

2.4 Aider 的 ReAct：Reflection Loop

Aider 在标准 ReAct 之外增加了 Reflection Loop——每次编辑后自动检查 lint 和测试，失败则自我修正：

# Aider run_one 伪代码
def run_one(self, user_message):
    message = user_message
    while message:
        self.reflected_message = None
        self.send_message(message)           # 标准 ReAct 步
        
        if not self.reflected_message:
            break                            # 无问题，完成
        
        if self.num_reflections >= self.max_reflections:
            break                            # 超过最大反思次数
        
        self.num_reflections += 1
        message = self.reflected_message     # 用反思消息继续循环

reflected_message 可能来自：

• Lint 错误 → 自动将 lint 输出作为反思消息
• 测试失败 → 自动将测试输出作为反思消息
• 编辑格式解析失败 → 将错误信息反馈给 LLM

3. Plan-Execute — 全局规划型

Plan-Execute 模式先让 LLM 规划完整步骤列表，再逐步执行。优势是能做出全局最优决策，劣势是计划可能过时。

3.1 AutoGPT 的 Block 架构：DAG 图执行

AutoGPT 采用了 Block 流式架构，将 Plan-Execute 推向了极致——每个步骤是一个 Block，Block 之间通过 DAG 图连接：

# Block 是执行的最小单元
class Block:
    id: str
    input_schema: dict   # 输入数据定义
    output_schema: dict  # 输出数据定义
    
    def execute(self, input_data) -> dict:
        """执行 Block，返回输出数据"""

# 关键 Block 类型：
OrchestratorBlock     # ReAct 循环：LLM → 工具调用 → 观察 → 继续
AutoPilotBlock        # 嵌套子代理：创建子 Agent 执行子任务
LLMBlock              # 纯 LLM 调用（不执行工具）
CodeBlock             # 代码执行
ForLoopBlock          # 循环控制
IfElseBlock           # 条件分支

3.2 GPT Researcher 的双阶段

GPT Researcher 的 Plan-Execute 更加显式：

• Phase 1 (Research)：搜索 → 抓取 → 评估信息充足度 → 不够则继续搜索
• Phase 2 (Report)：基于收集的信息生成最终报告

4. State Machine — 精确控制流

State Machine（状态机）模式用显式的图定义控制流：每个状态是一个节点，状态之间的转移是边。LangGraph 是这一模式的集大成者。

4.1 LangGraph 的 Pregel 执行模型

LangGraph 不仅仅是一个状态机——它的执行引擎基于 Google Pregel 论文的 BSP（Bulk Synchronous Parallel） 超步模型：

# 超步循环核心
while there_are_nodes_to_run:
    # 1. READ: 从 Channel 读取
    for node in nodes_to_run:
        node.read_channels(state)
    
    # 2. COMPUTE: 并行执行（asyncio.gather）
    results = await gather(*[node.execute() for node in nodes_to_run])
    
    # 3. WRITE: 写入 Channel（Reducer 合并）
    for result in results:
        for key, value in result.items():
            channel[key] = reducer(key, value)  # Reducer 决定合并策略
    
    # 4. CHECKPOINT: 保存快照
    await checkpointer.save(state, config)
    
    # 5. 确定下一轮要执行的节点（条件边路由）
    nodes_to_run = determine_next_nodes()

4.2 StateGraph Builder 模式

from langgraph.graph import StateGraph, START, END
from typing import Annotated

class State(TypedDict):
    messages: Annotated[list, add_messages]  # Reducer: 追加消息
    next_action: str

graph = StateGraph(State)

# 添加节点
graph.add_node("analyze", analyze_fn)
graph.add_node("search", search_fn)
graph.add_node("respond", respond_fn)

# 添加边
graph.add_edge(START, "analyze")
graph.add_conditional_edges("analyze", route_fn, {  # 条件路由
    "search": "search",
    "respond": "respond",
})
graph.add_edge("search", "analyze")     # 搜索后回到分析
graph.add_edge("respond", END)

# 编译（加入 Checkpoint、中断点等）
app = graph.compile(
    checkpointer=SqliteSaver("checkpoints.db"),
    interrupt_before=["respond"],  # 人机协作中断点
)

4.3 Checkpoint 与 Time Travel（独有优势）

LangGraph 是唯一原生支持 Time Travel 的框架：

• Checkpoint：每个超步后自动保存状态快照
• Replay：从任意 checkpoint 重放执行
• Fork：从某个 checkpoint 分叉，走不同路径
• Interrupt/Resume：在指定节点前暂停，等人工确认后继续

4.4 PydanticAI 的 GraphAgent

PydanticAI 也提供了状态机模式，但更轻量：

# PydanticAI GraphAgent
class MyGraph(GraphAgent):
    node_a = Node(handler=analyze, next="node_b")
    node_b = Node(handler=execute, next=END_or_loop)

与 LangGraph 的区别：PydanticAI 没有 Pregel 并行模型，没有 Checkpoint，但集成度更高（类型安全 + 依赖注入 + result_type）。

5. Actor Model — 天然多 Agent

Actor Model 把每个 Agent 视为一个独立的 Actor，通过消息传递通信。AutoGen 和 MetaGPT 是这一模式的代表。

5.1 AutoGen 的双层架构

AutoGen 的精妙之处在于双层解耦：

• Core 层（autogen-core）：Actor Model 运行时，完全通用
• AgentChat 层（autogen-agentchat）：在 Core 之上构建 Chat Agent 抽象

# 每个 Agent 是一个 Actor
class Agent(Protocol):
    @property
    def metadata(self) -> AgentMetadata: ...
    
    async def on_message(self, message, ctx) -> Response:
        """收到消息时的处理逻辑"""

# 两种通信
result = await runtime.send_message(msg, recipient=AgentId("agent_b"))
runtime.publish_message(event, topic_id=TopicId("news"))

5.2 AutoGen 的 5 种群组编排

Actor Model 的真正威力在于群组编排——多个 Agent 如何协作完成同一个任务：

5.3 MetaGPT 的 SOP 流水线

MetaGPT 用 SOP（Standard Operating Procedure） 模式——模拟软件团队的标准化流程：

class Role:
    """每个 Role 监听特定消息，发布处理结果"""
    
    def _watch(self, actions: list[Type[Action]]):
        """订阅特定 Action 的输出"""
        self._watching = set(actions)
    
    async def _observe(self) -> int:
        """从 MessageQueue 取出自己关心的消息"""
        news = self.rc.msg_queue.pop()
        return len(news)
    
    async def run(self):
        """Role 主循环：Observe → Think → Act → Publish"""
        await self._observe()           # 1. 取消息
        reaction = await self._react()  # 2. 思考+行动
        await self._publish(reaction)   # 3. 发布结果

5.4 CrewAI 的 Process-Driven 编排

CrewAI 的模式介于 AutoGen 和 MetaGPT 之间：

• Process.sequential — 类似 MetaGPT 的流水线
• Process.hierarchical — Manager Agent 分配任务给其他 Agent
• Flow 装饰器 — 更灵活的编排（@start、@listen、@router）

6. 混合架构 — 当今趋势

2024-2025 年的 Agent 架构趋势：ReAct 内核 + 外层编排。单一模式已经不够用，混合架构成为主流。

6.1 各项目的混合架构

7. 事件驱动 — 可观测性的基础

事件驱动不是一种主循环模式，而是任何主循环都应采用的实现方式。OpenHands 和 OpenClaw 是最佳实践。

8. 循环内部：6 个关键环节

不管哪种循环模式，内部都有 6 个必须正确实现的环节：

9. 卡死检测与恢复

Agent 循环最大的运行时风险是卡死——Agent 在同一个错误上反复尝试，永远无法逃脱。只有 OpenHands 实现了完整的卡死检测系统。

9.1 OpenHands StuckDetector — 5 种检测

class StuckDetector:
    """5 种卡死模式检测"""
    
    def is_stuck(self, events: list[Event]) -> bool:
        # 1. Action-Observation Loop：同一动作重复 N 次
        #    例：反复搜索同一个文件，结果相同
        if self._detect_action_observation_loop(events):
            return True
        
        # 2. Action-Error Loop：同一失败动作重复 N 次
        #    例：反复尝试同一个语法错误的命令
        if self._detect_action_error_loop(events):
            return True
        
        # 3. Monologue：Agent 自说自话，不调用任何工具
        if self._detect_monologue(events):
            return True
        
        # 4. Alternating Pattern：两个动作来回切换
        #    例：搜索 → 编辑 → 搜索 → 编辑（编辑总被撤销）
        if self._detect_alternating_pattern(events):
            return True
        
        # 5. 上下文膨胀：Condensation 请求过于频繁
        if self._detect_excessive_condensation(events):
            return True
        
        return False

9.2 其他项目的卡死处理

10. 并行执行与原子性

当 LLM 返回多个工具调用时（Parallel Function Calling），如何并行执行？核心问题是原子性——多个工具调用应该全部成功还是允许部分失败？

10.1 OpenHands _ActionBatch — 原子批处理

@dataclass(frozen=True, slots=True)
class _ActionBatch:
    action_events: list[ActionEvent]
    has_finish: bool
    blocked_reasons: dict[str, str]     # Hook 拦截的动作
    results_by_id: dict[str, list[Event]]

    @classmethod
    def prepare(cls, action_events, state, executor, tool_runner, tools):
        # 1. 如果包含 FinishTool，截断后续调用
        # 2. 分区：blocked（Hook 拦截）vs executable
        # 3. ParallelToolExecutor 批量执行
        #    - 声明相同资源的工具串行执行
        #    - 不同资源的工具并行执行
        # 4. 返回不可变批次结果

10.2 LangGraph — Pregel 并行保证

LangGraph 的并行更根本——在 Pregel 超步模型中，所有节点天然并行执行，Reducer 保证写入合并的正确性：

class State(TypedDict):
    messages: Annotated[list, add_messages]  # 并行写入 → 追加合并
    score: Annotated[int, operator.add]       # 并行写入 → 求和

10.3 对比

11. 全景对比表

12. 选型决策树

13. 实现检查清单

不管选择哪种模式，一个完整的主循环需要覆盖以下所有项：

基础（必须）

• MUST 明确的主循环入口（run() / step() / execute()）
• MUST 消息历史组装（system + history + user → LLM input）
• MUST LLM 调用（含重试：指数退避 + 5 次重试 + 抖动）
• MUST 输出解析（Function Calling 或文本解析 + 格式错误处理）
• MUST 工具执行（含超时、错误捕获）
• MUST 终止条件（完成标志 + 最大步数硬上限）

健壮性（应该）

• SHOULD 卡死检测（重复模式识别 + 强制终止）
• SHOULD 错误重查询（LLM 输出格式错误时不直接失败，而是反馈错误重新查询）
• SHOULD 上下文窗口管理（Condenser / History Processor / 摘要压缩）
• SHOULD 流式输出（用户体验：LLM 生成过程中实时返回 token）
• SHOULD 事件日志（结构化记录每一步，便于调试和监控）

进阶（可选）

• NICE 并行工具执行（资源键锁 / Pregel Reducer）
• NICE Checkpoint / Time Travel（状态快照 + 回滚）
• NICE CriticMixin（自评机制：执行前让 LLM 评估自己的计划）
• NICE Action Sampler（多路径决策：生成多个候选动作，选择最优）
• NICE Reflection Loop（执行后自动检查，失败则自我修正）
• NICE ArchitectCoder 双模型（推理和执行分离）
• NICE 人机协作中断点（在关键步骤前暂停，等待人工确认）