06Agent Loop

L2P0五步章 · 技术章
Why · 为什么要学
你写了个 agent demo,跑通了。上线一周:3 个任务卡 4 小时不停;1 次 agent 删了文件 retry 又删了一次;100 次在第 5 步莫名 stop 没下文。Agent Loop 是 harness 工程的心脏——你 own 的不是"让 agent 跑",而是"让它稳定地停下、稳定地重启、稳定地不重复出错"。这一节决定你的 agent 能否上生产。
1 ·核心要点

Agent Loop 形式上是 think → tool_call → observe → repeat。工程上要 own 6 件事:

1. 进入条件:user 输入 + system prompt + tool registry + budget(max_steps、max_tokens)

2. 循环主体:

· 调 LLM → 检查 stop_reason

· end_turn → 模型主动收尾,返回结果

· tool_use → 执行所有 tool,把 results 塞回 messages,继续循环

· max_tokens → 模型被截断,需要 retry 或拼接策略

3. 终止条件(6 层):

条件 触发时机 响应
模型 end_turn每步正常返回
max_steps每步硬中断 + 报告
max_tokens budget每步累计中断 + cost warning
死循环检测tool_call 后注入 reflection 或中断
用户中断任意时刻优雅退出 + 持久化
致命错误tool 报错不可恢复直接抛

4. 并发:一个 response 可含多个 tool_use。read-only 工具(read_file, search)默认并发安全;write 工具(edit_file, git_commit)按需串行或加文件锁。

5. 错误处理:tool 失败 → 显式 is_error: true tool_result,模型看到会自纠;LLM API 失败 → exponential backoff retry 5xx/429,4xx 不重试。

6. 持久化:每步 messages 写 trace log,支持 #31 Replay。

2 ·最小代码示例
def run_agent(user_query, tools, max_steps=50, max_tokens=200_000):
    messages = [{"role": "user", "content": user_query}]
    total_tokens = 0
    history = []  # for loop detection

    for step in range(max_steps):
        resp = client.messages.create(
            model="claude-sonnet-4-5",
            messages=messages, tools=tools, max_tokens=4096,
        )
        total_tokens += resp.usage.input_tokens + resp.usage.output_tokens
        if total_tokens > max_tokens: raise BudgetExceeded()

        messages.append({"role": "assistant", "content": resp.content})

        if resp.stop_reason == "end_turn":
            return resp

        if resp.stop_reason == "tool_use":
            results = []
            for block in resp.content:
                if block.type != "tool_use": continue
                key = (block.name, str(block.input))
                history.append(key)
                if history[-3:].count(key) == 3:
                    results.append({"type": "tool_result", "tool_use_id": block.id,
                        "content": "重复调用 3 次,请换思路或停下。", "is_error": True})
                    continue
                try:
                    r = execute_tool(block.name, block.input)
                    results.append({"type": "tool_result", "tool_use_id": block.id, "content": r})
                except Exception as e:
                    results.append({"type": "tool_result", "tool_use_id": block.id,
                        "content": str(e), "is_error": True})
            messages.append({"role": "user", "content": results})

    raise MaxStepsExceeded(messages)
3 ·工程权衡

何时用完整 loop

  1. 终端用户面向的 agent(Claude Code、Cursor 这类)
  2. 多步任务需要回退/反思——debug、重构、规划
  3. 跨 session long-running——见 #17