第 2 章 Agent Loop —— 一切的起点

Agent 的全部秘密浓缩为一个公式：while stop_reason == "tool_use": execute → append → call LLM。无论项目规模、语言选择、架构风格如何变化，所有 coding agent 的内核都是同一段循环。

2.1 最小循环的解剖

30 行 Python 实现完整 Agent

用不到 30 行核心代码即可实现一个完全可工作的 coding agent。以下是基于 Anthropic Messages API 的最小实现：

# Python 实现：最小 Agent Loop
def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        # Append assistant turn
        messages.append({"role": "assistant", "content": response.content})
        # If the model didn't call a tool, we're done
        if response.stop_reason != "tool_use":
            return
        # Execute each tool call, collect results
        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = run_bash(block.input["command"])
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": output})
        messages.append({"role": "user", "content": results})

三个角色

循环中只有三个角色，严格交替出现：

1. User message — 用户的原始请求，或工具执行结果（以 tool_result 的形式伪装成 user 消息）
2. LLM response — 模型返回的 content 数组，可能包含 text block 和 tool_use block
3. Tool execution — harness 侧执行工具，将 stdout/stderr 封装回 tool_result

graph LR
    A[User Prompt] --> B[LLM]
    B -->|stop_reason = tool_use| C[Tool Execute]
    C -->|tool_result| B
    B -->|stop_reason = end_turn| D[Final Response]

退出条件：模型决定何时停止

循环的退出判断是 response.stop_reason != "tool_use"。这意味着模型拥有完全的控制权——它决定是否继续调用工具，harness 只负责执行。这是 agent loop 与传统 pipeline 的本质区别：控制流从硬编码的 if-else 转移到了模型的推理能力。

2.2 消息协议：messages[] 作为唯一状态

append-only 列表

Agent 的全部状态就是 messages: list。每一轮循环向其中追加两条记录：一条 assistant 消息，一条 user 消息（包含 tool_result）。没有数据库、没有状态机、没有中间变量——整个对话历史就是一个只增不删的列表。

# 循环前：messages = [{"role": "user", "content": "创建 hello.py"}]
# 第 1 轮后：
#   messages[1] = {"role": "assistant", "content": [TextBlock, ToolUseBlock]}
#   messages[2] = {"role": "user", "content": [{"type": "tool_result", ...}]}
# 第 2 轮后：
#   messages[3] = {"role": "assistant", "content": [TextBlock]}  ← stop_reason=end_turn, 循环结束

tool_use 与 tool_result 的配对

Anthropic API 中，模型请求调用工具时返回一个 tool_use block，其中包含 id、name、input。harness 执行后必须返回一个 tool_result block，通过 tool_use_id 字段与对应的 tool_use 配对。这种基于 ID 的配对机制允许模型在一次响应中发起多个并行工具调用。

# tool_use block（模型返回）
{"type": "tool_use", "id": "toolu_abc123", "name": "bash",
 "input": {"command": "ls -la"}}

# tool_result block（harness 返回）
{"type": "tool_result", "tool_use_id": "toolu_abc123",
 "content": "total 32\ndrwxr-xr-x ..."}

Anthropic API vs OpenAI API 的消息格式差异

两套 API 在消息协议上有本质区别：

维度	Anthropic Messages API	OpenAI Chat Completions API
工具调用触发	stop_reason == "tool_use"	finish_reason == "tool_calls"
工具调用位置	在 content[] 数组内作为 block	在 message.tool_calls[] 独立字段
工具结果角色	role: "user" + tool_result block	role: "tool" 独立消息
工具定义格式	顶层 tools[]，直接声明 input_schema	顶层 tools[]，包裹在 function 对象内
配对机制	tool_use_id ↔ block id	tool_call_id ↔ tool_calls[].id

使用 OpenAI API 格式时，tool 结果是独立的 role="tool" 消息，对比 Anthropic 格式可以清楚看到差异：

// Rust 实现：OpenAI 格式的 tool 结果序列化
// tool 结果是独立的 role="tool" 消息
HistoryEntry::Tool { tool_call_id, tool_name, content, .. } => {
    messages.push(json!({
        "role": "tool",
        "tool_call_id": tool_call_id,
        "name": tool_name,
        "content": content
    }));
}

graph TD
    subgraph "Anthropic Messages API"
        A1["role: user<br/>content: '创建文件'"]
        A2["role: assistant<br/>content: [text_block, tool_use_block]"]
        A3["role: user<br/>content: [tool_result_block]"]
    end
    subgraph "OpenAI Chat Completions API"
        O1["role: user<br/>content: '创建文件'"]
        O2["role: assistant<br/>content: '...', tool_calls: [...]"]
        O3["role: tool<br/>tool_call_id: 'xxx', content: '...'"]
    end
    A1 --> A2 --> A3
    O1 --> O2 --> O3

尽管格式不同，语义完全等价：都是 请求 → 调用 → 结果 → 再请求 的循环。这意味着 agent loop 的核心逻辑可以跨 API 迁移，只需替换消息序列化层。

2.3 最小 Loop 的工程化四件套

最小 Python 实现虽然能工作，但距离生产环境差四样东西。一个典型的 Rust 教学项目（约 850 行代码量级）恰好逐一补齐了这四项。

1. 流式输出 (Streaming)

教学级实现通常使用 "stream": false（同步请求），但在等待期间用 spinner 提供视觉反馈。生产级系统使用 SSE 流式输出来降低用户感知延迟——即使实际计算时间不变，逐 token 呈现让用户"感觉快"。

// Rust 实现：同步请求（教学版）
fn build_request_body(config: &LlmConfig, messages: Vec<Value>, enable_tools: bool) -> Value {
    let mut body = json!({
        "model": config.model,
        "messages": messages,
        "stream": false
    });

2. 重试 (Retry)

网络请求必然面临 transient error。典型实现是固定间隔重试，最多 7 次，每次间隔 1.5 秒：

// Rust 实现：固定间隔重试
const API_RETRIES: usize = 7;
const API_RETRY_DELAY_MS: u64 = 1_500;

for attempt in 1..=API_RETRIES {
    let outcome = /* ... send request ... */;
    match outcome {
        Ok(text) => return Ok(text),
        Err(err) => {
            if attempt < API_RETRIES {
                std::thread::sleep(Duration::from_millis(API_RETRY_DELAY_MS));
            }
        }
    }
}

生产系统通常使用指数退避（exponential backoff）并区分可重试错误（429、5xx）和不可重试错误（401、403）。

3. 审批 (Approval)

最小实现的 run_bash 有一个简陋的危险命令黑名单。工程化版本将其升级为交互式审批门控：

// Rust 实现：交互式审批门控
if !self.auto_approve && !prompt_for_approval(&mut self.auto_approve)? {
    return Ok(CommandOutcome {
        success: false,
        tool_content: format_tool_content(
            &request.command, &workdir.display().to_string(),
            false, "command rejected by user".to_string(),
        ),
    });
}

用户可以通过 /auto on 和 /auto off 在运行时切换审批模式。被拒绝的命令不会中断循环——其结果以 "command rejected by user" 的形式回传给模型，让模型自行决定下一步。

4. 命令校验 (Workspace-scoped Validation)

防止工具调用逃逸出工作目录是安全基线。通过 canonicalize + starts_with 校验确保所有命令在 workspace 范围内执行：

// Rust 实现：工作目录校验
fn resolve_workdir(workspace_root: &Path, requested: Option<&str>) -> Result<PathBuf> {
    let root = workspace_root.canonicalize()?;
    let candidate = match requested {
        None | Some("") | Some(".") => root.clone(),
        Some(path) => root.join(path),
    };
    let candidate = candidate.canonicalize()?;
    if !candidate.starts_with(&root) {
        bail!("workdir escapes workspace: {}", candidate.display());
    }
    Ok(candidate)
}

这四件套——streaming、retry、approval、validation——是从 toy 到 production 的最小增量。后续章节会看到 Codex 和 DeepAgents 如何将每一项做到极致。

2.4 跨项目的循环实现对比

Rust 教学实现：loop { match } + session resume + compaction

一个典型 Rust 教学项目的核心循环在 run_agent_loop 方法中，是一个带步数上限的 for 循环：

// Rust 实现：带步数上限的 Agent Loop
fn run_agent_loop(&mut self) -> Result<()> {
    for _ in 0..LOOP_LIMIT {           // LOOP_LIMIT = 64
        self.compact_history_if_needed(CompactionMode::MidTurn)?;
        let messages = build_messages(&self.config.workspace_root, &self.history.entries);
        let reply = call_model(&self.client, &self.config.llm, messages, true)?;
        self.history.push_assistant(reply.content.clone(), assistant_tool_calls);
        self.save_history()?;

        if !reply.tool_calls.is_empty() {
            self.handle_tool_calls(reply.tool_calls)?;
            self.save_history()?;
            continue;                  // ← 有工具调用，继续循环
        }
        return Ok(());                 // ← 无工具调用，结束
    }
    Err(anyhow!("agent loop exceeded {LOOP_LIMIT} steps"))
}

与最小 Python 实现的 while True 相比，工程化差异在三处：

1. 步数上限：LOOP_LIMIT = 64，防止模型陷入无限循环
2. History compaction：每次迭代前检查 token 使用量，超过阈值（80%）时让模型生成摘要替换旧历史
3. Session persistence：每次 save_history() 将完整对话写入 JSON 文件，支持 resume 恢复

退出条件从 Anthropic 的 stop_reason 变成了 OpenAI 的 tool_calls.is_empty()——语义相同，表达不同。

Codex (Rust)：事件驱动的异步循环 + SandboxPolicy

OpenAI 官方 Codex 的循环位于核心模块的 run_turn 函数（约 500 行），是一个 async 的 loop 块：

// Rust 实现：Codex 事件驱动异步循环
loop {
    // ... 构建 sampling request input ...
    match run_sampling_request(/* ... */).await {
        Ok(SamplingRequestResult { needs_follow_up, last_agent_message }) => {
            if token_limit_reached && needs_follow_up {
                run_auto_compact(/* ... */).await;
                continue;
            }
            if !needs_follow_up {
                // 运行 stop hooks、after_agent hooks
                break;
            }
            continue;  // needs_follow_up=true → 继续循环
        }
        Err(CodexErr::TurnAborted) => break,
        Err(e) => { /* 发送错误事件 */ break; }
    }
}

与教学级 Rust 实现的核心区别：

维度	教学级 Rust 实现	Codex
并发模型	同步阻塞	tokio async
循环退出	tool_calls.is_empty()	!needs_follow_up（封装在 sampling result 中）
Hook 系统	无	SessionStartHook / StopHook / AfterAgentHook
沙箱策略	resolve_workdir 路径校验	SandboxPolicy + Landlock + NetworkProxy
Compaction	同步调用模型生成摘要	支持 local 和 remote 两种 compact 策略
Pending input	不支持	循环内处理用户在模型运行时提交的新消息

Codex 的循环虽然复杂，但骨架仍然是 loop { call_model → if no_follow_up break → execute_tools → continue }。

OpenCode (TypeScript)：provider-agnostic 的消息驱动循环

OpenCode 的核心循环使用 Vercel AI SDK 的 streamText 处理 LLM 调用：

// TypeScript 实现：OpenCode 的 provider-agnostic 循环
let step = 0
while (true) {
    if (abort.aborted) break
    let msgs = await MessageV2.filterCompacted(MessageV2.stream(sessionID))
    // ... 查找 lastUser, lastAssistant, lastFinished ...

    // 退出条件：最后的 assistant 消息不是 tool-calls
    if (lastAssistant?.finish &&
        !["tool-calls", "unknown"].includes(lastAssistant.finish) &&
        lastUser.id < lastAssistant.id) {
        break
    }
    step++
    // ... 构建 tools, system prompt, 调用 LLM ...
}

OpenCode 的独特之处在于 provider-agnostic 设计——通过 Vercel AI SDK 抽象层，同一循环代码支持 Anthropic、OpenAI、Gemini、DeepSeek 等多家模型。退出条件 finish !== "tool-calls" 是 AI SDK 对各家 API 的统一抽象。

另一个特点是 agent 多态：循环内通过 Agent.get(lastUser.agent) 动态获取当前 agent 配置（build / plan / explore / compaction），每种 agent 有不同的 permission ruleset 和 prompt。但循环本身不变。

DeepAgents (Python)：LangGraph 编译的状态机循环

DeepAgents 的方式最为独特——它不手写循环，而是通过 LangGraph 的 create_agent 编译出一个状态机：

# Python 实现：LangGraph 编译的状态机循环
return create_agent(
    model,
    system_prompt=final_system_prompt,
    tools=tools,
    middleware=deepagent_middleware,
    checkpointer=checkpointer,
    # ...
).with_config({
    "recursion_limit": 1000,
    # ...
})

create_agent 内部构建了一个 LangGraph CompiledStateGraph，其中包含：

• agent node：调用 LLM，决定是否使用工具
• tool node：执行工具调用
• 条件边：如果 LLM 返回 tool_calls，转到 tool node；否则结束

循环不再是显式的 while True，而是隐含在图的拓扑结构中。recursion_limit: 1000 充当步数上限。

DeepAgents 的独特价值在于 middleware 栈。在循环外围层层叠加了：

TodoListMiddleware → SkillsMiddleware → FilesystemMiddleware
→ SubAgentMiddleware → SummarizationMiddleware → PatchToolCallsMiddleware
→ AnthropicPromptCachingMiddleware → MemoryMiddleware → HumanInTheLoopMiddleware

每层 middleware 在 agent node 执行前后注入行为（修改 prompt、注入工具、拦截审批）。循环本身的 call_model → execute_tool → repeat 逻辑被 LangGraph runtime 完全托管。

四种实现的统一模式

项目	语言	循环形式	退出信号	步数上限
最小 Python 实现	Python	while True	stop_reason != "tool_use"	无
教学级 Rust 实现	Rust	for _ in 0..64	tool_calls.is_empty()	64
Codex	Rust	loop {}	!needs_follow_up	无显式上限
OpenCode	TypeScript	while (true)	finish ∉ ["tool-calls","unknown"]	agent.steps（可配置）
DeepAgents	Python/LangGraph	隐式状态机	无 tool_calls 的条件边	recursion_limit: 1000

形式千变，本质不变。

2.5 循环不变量

在基础循环上逐步叠加新机制，可以构建出完整的 production harness：

• 多工具路由
• 对话历史持久化
• 流式输出
• 审批门控
• Context injection
• ...直到完整 production harness

这些机制属于 harness 层——包围循环的脚手架，而不是循环本身。无论加多少层：

Tool Router → Approval Gate → Streaming → Retry → Compaction → Hook System
      ↓              ↓            ↓         ↓          ↓           ↓
      └──────────────────── Agent Loop ──────────────────────────────┘
                    while has_tool_calls:
                        execute → append → call LLM

核心循环那三行逻辑从未改变。Codex 的 run_turn 虽然膨胀到 500 行，但其中 490 行是 hooks、compaction、pending input 处理、错误恢复——剥离这些后剩下的仍然是 call_model → check follow_up → execute tools → continue。

The loop belongs to the agent. The mechanisms belong to the harness.

Agent 拥有循环（决定何时停止），harness 拥有机制（决定如何执行）。理解这个分界线，就理解了所有 coding agent 的架构。