Code Agent 解剖（02）：agent 是怎么一轮一轮思考和行动的？

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

# 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, ...)

def _prepare_run(self, input_text, show_raw):
    # 1. 刷新 Skills 提示词（如果 skills/ 目录有变化）
    host._refresh_skills_prompt()
    host.context_builder.set_skills_prompt(host._skills_prompt)
 
    # 2. 预处理用户输入：识别并展开 @file 引用，注入 system-reminder 提示模型去读
    preprocess_result = preprocess_input(input_text)
    processed_input = preprocess_result.processed_input
 
    # 3. 清空上一轮的 trace 事件，为本轮初始化 run_id 和 transcript 记录
    trace_logger.clear_current_run_events()
    host._run_id += 1
    host._active_transcript_run_id = f"run-{host._run_id}"
 
    # 4. 把预处理后的用户消息写入 history_manager
    #    ← 这是用户输入进入历史的唯一时机
    self._append_user_message(processed_input)
 
    # 5. 发射 run_start / user_input 事件给 trace 和 transcript
    self._emit("run_start", {...}, step=0)
    self._emit("user_input", {...}, step=0)
 
    return processed_input, trace_logger, run_id

_react_loop()
│
├─ 外层 for（step 1 → max_steps）     每次迭代 = 一个 ReAct step
│   │
│   ├─ _prepare_step_context()         构建本步的 Model View
│   │
│   ├─ 内层 while True                 单步内的模型调用与错误恢复
│   │   ├─ llm.invoke_raw()            调模型
│   │   ├─ 模型调用异常？               分类处理（压缩/重试/终止）
│   │   ├─ 响应解析（text + tool_calls）
│   │   ├─ 空响应？                     注入提示重试
│   │   └─ break                       正常响应，跳出内层
│   │
│   ├─ 有 tool_calls？                  Acting 分支
│   │   ├─ 执行工具（ToolOrchestrator）
│   │   ├─ 结果写入历史
│   │   └─ continue（下一个 step）
│   │
│   └─ 没有 tool_calls？                Reasoning 分支
│       ├─ 完成门判决（PASS/FAIL/UNVERIFIED）
│       ├─ PASS → return final_text    正常出口
│       ├─ FAIL → 注入反馈 + continue
│       └─ UNVERIFIED → return         带不确定性退出
│
└─ 超出 max_steps → return             兜底出口

# runtime/state.py
@dataclass(frozen=True)          # frozen=True：不可修改，任何"修改"都产生新对象
class LoopState:
    messages: list[dict]         # 当前 model view（不是完整历史）
    step: int                    # 当前步数
    tool_choice: str             # "auto" | "none" | 指定工具
    transition: Transition|None  # 最近一次状态转移记录
    completion_block_count: int  # 完成门拦截次数
    model_recovery_counts: dict  # 各类错误的重试计数
    last_error: str|None         # 最近一次错误信息
    # ...其余诊断字段
 
    def next(self, reason: TransitionReason, **changes) -> "LoopState":
        # 产生新对象，记录转移原因——不修改 self
        return replace(self, transition=Transition(reason=reason), **changes)

state.step = 2      # ❌ 抛出 FrozenInstanceError
state = state.next(TransitionReason.TOOLS_EXECUTED, step=2)  # ✅ 产生新对象

class TransitionReason(str, Enum):
    USER_INPUT = "user_input"                    # 新一轮开始
    MODEL_RETURNED_TOOL_CALLS = "..."            # 模型要调工具（Acting）
    TOOLS_EXECUTED = "..."                       # 工具执行完
    MODEL_RETURNED_FINAL = "..."                 # 模型给出最终回答（Reasoning）
    STOP_HOOK_BLOCKING = "..."                   # 完成门拦截，注入反馈后继续
    MODEL_RECOVERY_RETRY = "..."                 # 模型出错，恢复后重试
    MAX_STEPS_EXCEEDED = "..."                   # 超步终止
    TOKEN_BUDGET_EXCEEDED = "..."                # token 预算耗尽终止

history_manager（完整历史，append-only，永不删除）
         ↓
   build_model_view()
         ↓
   model view（token 预算内的投影，发给 LLM）

# loop.py — Acting 分支（简化）
if tool_calls:
    # 1. 确保每个 tool_call 有 id（部分模型不返回）
    for call in tool_calls:
        if not call.get("id"):
            call["id"] = f"call_{uuid.uuid4().hex}"
 
    # 2. 把 assistant 消息（含 tool_calls 列表）写入历史
    host.history_manager.append_assistant(
        content=response_text,
        metadata={"action_type": "tool_call", "tool_calls": tool_calls},
    )
 
    # 3. 执行工具（只读工具可并发，写操作强制串行）
    observations = host.tool_orchestrator.run(tool_calls, step=step)
 
    # 4. 把每个工具的执行结果写入历史，下一步模型能看到
    for obs in observations:
        host.history_manager.append_tool(
            tool_name=obs.tool_name,
            observation=obs.observation,
        )
 
    continue  # 外层 for 继续下一个 step

# runtime/completion.py — 完成门三步走
 
# 步骤 1：推断要求（从用户输入里识别"需要验证"的关键词）
requirements = infer_completion_requirements(
    user_input=pending_input,          # 扫 "pytest" / "跑测试" 等关键词
    history_messages=history_messages, # 读最新 TodoWrite 记录检查未完成项
)
 
# 步骤 2：收集证据（从历史 Bash 工具调用里找验证行为）
evidence = collect_verification_evidence(history_messages)
# 注意：Edit 之后的验证证据才算有效，Edit 之前的被标记为 invalid
 
# 步骤 3：判决
verdict = verifier.evaluate(candidate, requirements, evidence, ...)
# PASS：所有要求满足 → return final_text
# FAIL：有未完成 todo 或缺验证证据 → 注入反馈消息，继续循环
# UNVERIFIED：用户说"尽量"，证据缺失但允许跳过 → return（带标记）

# completion.py — collect_verification_evidence()
# 找到最近一次 Edit 的 step
latest_mutation_step = max(step for edit in history if edit.tool == "Edit")
 
# Edit 之前的验证证据标记为无效
for evidence in evidences:
    if evidence.step < latest_mutation_step:
        evidence.valid = False  # 已被后续 Edit 作废

invoke_raw() 抛出异常
    ↓
classify_model_error() 判断是 PROMPT_TOO_LONG
    ↓
context_engine.reactive_compact() 压缩历史
    ↓
重建 model_view，内层 continue 重试
    ↓
还是失败？→ 终止

调用成功，但 response_text 为空且没有 tool_calls
    ↓
注入提示："请在 content 中回复最终答案，或使用工具调用"
    ↓
内层 continue 重试（最多 1 次）
    ↓
还是空？→ 终止

正常出口：
  完成门 PASS             → return final_text
  完成门 UNVERIFIED       → return final_text（带标记）
 
异常出口：
  超出 max_steps          → return 错误提示
  token 预算耗尽          → return 错误提示
  模型错误无法恢复         → return 错误提示
  完成门反馈重试耗尽       → return 错误提示

git clone https://github.com/chendongqi/MyCodeAgent
cd MyCodeAgent
cp .env.example .env   # 填入你的 LLM API key
uv sync
uv run python main.py