Code Agent运行机制详解：ReAct循环的工程实现

Why Agents Need a Loop

最简单的 LLM 调用长这样：

response = llm.invoke(messages)
print(response)

这对简单的问答场景完全够用，但它无法处理"重构这个函数并运行测试"这类任务——因为这类任务涉及多个步骤：读取文件、修改代码、执行命令、根据结果决定下一步。每个步骤的输入依赖于前一个步骤的输出，而模型无法一次预测所有步骤。

ReAct（Reasoning + Acting）正是解决这个问题的模式：让模型迭代执行"思考 → 执行工具 → 观察结果 → 再次思考"的循环，直到任务完成。

MyCodeAgent 的 RuntimeRunner._react_loop() 就是这个模式的具体工程实现。本文将把它拆解开来。

Before the Loop: _prepare_run

_react_loop 并非 RuntimeRunner.run() 调用的第一个方法。在它之前还有一个前置步骤：_prepare_run()：

# runtime/loop.py — RuntimeRunner.run()
def run(self, input_text, **kwargs):
    processed_input, trace_logger, run_id = self._prepare_run(input_text, show_raw)
    response_text = self._react_loop(pending_input=processed_input, ...)

_prepare_run 做了五件事，每件都指向不同的子系统：

def _prepare_run(self, input_text, show_raw):
    # 1. Refresh Skills prompt (if the skills/ directory has changed)
    host._refresh_skills_prompt()
    host.context_builder.set_skills_prompt(host._skills_prompt)

    # 2. Preprocess user input: detect and expand @file references,
    #    inject a system-reminder telling the model to read them
    preprocess_result = preprocess_input(input_text)
    processed_input = preprocess_result.processed_input

    # 3. Clear trace events from the previous run, initialize run_id and transcript
    trace_logger.clear_current_run_events()
    host._run_id += 1
    host._active_transcript_run_id = f"run-{host._run_id}"

    # 4. Write the preprocessed user message into history_manager
    #    ← This is the only moment user input enters history
    self._append_user_message(processed_input)

    # 5. Emit run_start / user_input events to trace and transcript
    self._emit("run_start", {...}, step=0)
    self._emit("user_input", {...}, step=0)

    return processed_input, trace_logger, run_id

pending_input 从何而来：processed_input 就是 pending_input——即经过预处理的用户输入字符串。它已经在 _prepare_run 内部被写入了历史；当传入 _react_loop 时不会再被追加。它只服务于两个目的：帮助 build_model_view() 估算本轮需要预留多少 token，以及让 completion gate 推断用户想要什么。

@file 展开的作用：当用户输入 take a look at @src/main.py 时，preprocess_input() 会检测到这个 @file 引用，并向消息追加一条 <system-reminder>，指示模型先用 Read 工具读取该文件再作答——防止模型凭空编造答案。

Overall Structure: Dual-Layer Loop

_react_loop()
│
├─ outer for (step 1 → max_steps)      each iteration = one ReAct step
│   │
│   ├─ _prepare_step_context()          build Model View for this step
│   │
│   ├─ inner while True                 model invocation + error recovery
│   │   ├─ llm.invoke_raw()             call the model
│   │   ├─ model call exception?        classify and handle (compact/retry/abort)
│   │   ├─ parse response (text + tool_calls)
│   │   ├─ empty response?              inject prompt and retry
│   │   └─ break                        normal response, exit inner loop
│   │
│   ├─ has tool_calls?                  Acting branch
│   │   ├─ execute tools (ToolOrchestrator)
│   │   ├─ write results to history
│   │   └─ continue (next step)
│   │
│   └─ no tool_calls?                   Reasoning branch
│       ├─ completion gate verdict (PASS/FAIL/UNVERIFIED)
│       ├─ PASS → return final_text     normal exit
│       ├─ FAIL → inject feedback + continue
│       └─ UNVERIFIED → return          exit with uncertainty marker
│
└─ max_steps exceeded → return          fallback exit

这两层循环职责分明：外层 for 推进 ReAct 步骤，内层 while 处理单步内的模型错误。将错误处理封装在内层循环是关键设计决策——否则重试会消耗外层循环的步骤预算。

Immutable State Machine

在每个步骤开始时，状态如下：

# runtime/state.py
@dataclass(frozen=True)          # frozen=True: immutable, any "mutation" produces a new object
class LoopState:
    messages: list[dict]         # current model view (not the full history)
    step: int                    # current step number
    tool_choice: str             # "auto" | "none" | specific tool
    transition: Transition|None  # most recent state transition record
    completion_block_count: int  # number of completion gate blocks
    model_recovery_counts: dict  # retry counts per error type
    last_error: str|None         # most recent error message
    # ...additional diagnostic fields

    def next(self, reason: TransitionReason, **changes) -> "LoopState":
        # produces a new object, records the transition reason — does not modify self
        return replace(self, transition=Transition(reason=reason), **changes)

state.step = 2      # ❌ raises FrozenInstanceError
state = state.next(TransitionReason.TOOLS_EXECUTED, step=2)  # ✅ produces a new object

为什么要用不可变状态？可变状态是最难调试的东西——"这个字段到底是在哪儿被设置成这个值的？"不可变状态配合 TransitionReason 枚举，意味着每一次状态变更都有记录的变更原因。trace 系统可以重构完整的执行路径。这是函数式编程思想在工程中的应用。

TransitionReason 枚举记录了循环可能的所有走向：

class TransitionReason(str, Enum):
    USER_INPUT = "user_input"                    # new run starts
    MODEL_RETURNED_TOOL_CALLS = "..."            # model wants to call tools (Acting)
    TOOLS_EXECUTED = "..."                       # tools have been executed
    MODEL_RETURNED_FINAL = "..."                 # model gives final answer (Reasoning)
    STOP_HOOK_BLOCKING = "..."                   # completion gate blocked, inject feedback and continue
    MODEL_RECOVERY_RETRY = "..."                 # model error, retry after recovery
    MAX_STEPS_EXCEEDED = "..."                   # terminated due to exceeding max steps
    TOKEN_BUDGET_EXCEEDED = "..."                # terminated due to token budget exhaustion

Model View: The Model Doesn't See Full History

在每个步骤开始时会调用 _prepare_step_context()。它返回的 messages 并不是 history_manager 中的全部消息——而是由 context_engine.build_model_view() 投射出的有界子集。

history_manager (complete history, append-only, never deleted)
         ↓
   build_model_view()
         ↓
   model view (projection within token budget, sent to LLM)

完整历史可能有 200 条消息，但 token 预算只允许 50 条。build_model_view() 决定"发送哪 50 条"，当预算超限时触发压缩（由 LLM 对更早的轮次进行摘要）。

为什么要分离二者？历史是事实——不可删除。模型看到的是视图——可以裁剪。混淆这两个概念会破坏历史，使崩溃恢复变得不可能。本系列第 09 篇文章将详细讨论 context engineering。

Acting Branch: Model Returns Tool Calls

# loop.py — Acting branch (simplified)
if tool_calls:
    # 1. Ensure every tool_call has an id (some models don't return one)
    for call in tool_calls:
        if not call.get("id"):
            call["id"] = f"call_{uuid.uuid4().hex}"

    # 2. Write the assistant message (including the tool_calls list) to history
    host.history_manager.append_assistant(
        content=response_text,
        metadata={"action_type": "tool_call", "tool_calls": tool_calls},
    )

    # 3. Execute tools (read-only tools can run concurrently, write ops are forced sequential)
    observations = host.tool_orchestrator.run(tool_calls, step=step)

    # 4. Write each tool's result to history so the model can see it in the next step
    for obs in observations:
        host.history_manager.append_tool(
            tool_name=obs.tool_name,
            observation=obs.observation,
        )

    continue  # outer for advances to the next step

工具执行结果通过 history_manager.append_tool() 写入历史。在下一步，当 build_model_view() 构建模型视图时，这些结果会出现在消息中。模型"看到"了工具执行结果——这就是 ReAct 中的"Observation"。

Reasoning Branch: The Completion Gate

当没有 tool_calls 时，模型给出了一个文本响应。但循环不会立即返回——它首先要经过 completion gate：

# runtime/completion.py — completion gate, three steps

# Step 1: infer requirements (identify "needs verification" keywords from user input)
requirements = infer_completion_requirements(
    user_input=pending_input,          # scan for "pytest" / "run tests" etc.
    history_messages=history_messages, # read latest TodoWrite entries to check unfinished items
)

# Step 2: collect evidence (find verification behavior in historical Bash tool calls)
evidence = collect_verification_evidence(history_messages)
# Note: only verification evidence AFTER an Edit counts as valid;
#       evidence BEFORE an Edit is marked invalid

# Step 3: verdict
verdict = verifier.evaluate(candidate, requirements, evidence, ...)
# PASS:       all requirements satisfied → return final_text
# FAIL:       unfinished todos or missing verification → inject feedback, continue loop
# UNVERIFIED: user said "try to", evidence missing but skippable → return (with marker)

Completion gate 解决什么问题？模型可能说"完成了"但 todo 列表并未清空，或者在用户要求运行测试后忘记执行。Completion gate 用基于规则的检查捕获这两种情况，将"为什么还没完成"作为用户消息注入，让模型再试一次。

关于验证证据有一个反直觉的细节：只有在 Edit 之后执行的 pytest 运行才计为有效。如果先通过了测试，然后才修改代码（Edit），则更早的测试结果会被标记为无效——因为修改后的代码尚未被验证。

# completion.py — collect_verification_evidence()
# find the step of the most recent Edit
latest_mutation_step = max(step for edit in history if edit.tool == "Edit")

# mark verification evidence before that Edit as invalid
for evidence in evidences:
    if evidence.step < latest_mutation_step:
        evidence.valid = False  # invalidated by subsequent Edit

内层 while True 处理两类模型问题：

调用异常（PROMPT_TOO_LONG）：

invoke_raw() raises exception
    ↓
classify_model_error() identifies PROMPT_TOO_LONG
    ↓
context_engine.reactive_compact() compresses history
    ↓
rebuild model_view, inner continue retries
    ↓
still failing? → terminate

响应异常（EMPTY_RESPONSE）：

call succeeds, but response_text is empty and there are no tool_calls
    ↓
inject prompt: "please reply with your final answer in content, or use a tool call"
    ↓
inner continue retries (at most once)
    ↓
still empty? → terminate

两条恢复路径都有尝试次数限制（_get_model_recovery_limit()）。一旦超过限制，循环就走终止路径——不会无限重试。

Termination Conditions at a Glance

循环的退出点只有以下几种：

Normal exits:
  Completion gate PASS             → return final_text
  Completion gate UNVERIFIED       → return final_text (with marker)

Error exits:
  max_steps exceeded               → return error message
  token budget exhausted           → return error message
  model error unrecoverable        → return error message
  completion gate feedback retries exhausted → return error message

注意没有"遇到空响应立即退出"——空响应会被重试，只有在重试耗尽后才会终止。这确保了循环行为可预期，不会因为一次偶然的空响应就静默失败。

Source Code for This Series

本系列所有分析均基于开源项目 MyCodeAgent。

源代码在关键位置已按本系列文章的讲解顺序加了注释——你可以对照文章阅读代码，也可以 clone 下来运行、修改和扩展，构建你自己的 agent。

git clone https://github.com/chendongqi/MyCodeAgent
cd MyCodeAgent
cp .env.example .env   # fill in your LLM API key
uv sync
uv run python main.py

Visit PrimeSkills — a carefully curated AI Agent and skills marketplace where all content is validated against real enterprise workflows. No hype, just things that actually work.

For more practical insights and interesting products, visit my personal homepage.

For further actions, you may consider blocking this person and/or reporting abuse。