大模型 Agent 中的 Loop：从基本循环到 Claude Code / Codex 的工程实现

用户提出任务
    │
    ▼
模型根据指令和当前上下文决定下一步
    │
    ├── 返回最终答复 ──────────────────────┐
    │                                      │
    └── 请求调用工具                       │
            │                              │
            ▼                              │
      Harness 检查工具、权限和参数          │
            │                              │
            ▼                              │
      执行工具：读文件 / 改文件 / 跑测试     │
            │                              │
            ▼                              │
      将工具结果写回对话状态                │
            │                              │
            └────再次请求模型 ◀─────────────┘
                  │
                  ├── 还有问题：继续循环
                  └── 任务已完成：结束本轮

{
  "name": "read_file",
  "arguments": {
    "path": "src/calculator.py"
  }
}

AssertionError: expected 5, got -1

user: 修复这个测试失败的问题
assistant: 调用 read_file(path="app.py")
tool: 文件内容是 def add(a, b): return a - b
assistant: 调用 run_tests()
tool: AssertionError: expected 5, got -1
assistant: 调用 write_file(...修正后的内容...)
tool: 写入成功
assistant: 调用 run_tests()
tool: 3 passed
assistant: 测试已通过，问题已修复。

if model_requested_tools:
    execute_tools_and_continue()
else:
    return_final_answer()

history = [{"role": "user", "content": user_task}]

while True:
    # 1. 模型根据对话历史和工具定义决定下一步
    response = model.generate(history, tools=tool_schemas)

    # 2. 如果模型不再请求工具，就结束本轮
    if not response.tool_calls:
        return response.text

    # 3. 保存模型的工具请求，保持对话协议完整
    history.append(response.as_assistant_message())

    # 4. Harness 检查权限，然后执行工具
    for call in response.tool_calls:
        authorize_or_raise(call)
        result = execute_tool(call.name, call.arguments)

        # 5. 把结果交还给模型，以便下一轮决策
        history.append(make_tool_result(call.id, result))

python mini_agent_loop.py

from __future__ import annotations

from dataclasses import dataclass
from typing import Any


@dataclass
class ToolCall:
    """模型输出的结构化工具请求。"""
    name: str
    arguments: dict[str, Any]
    call_id: str


@dataclass
class ModelDecision:
    """模型的一次决策：请求工具，或者返回最终答案。"""
    tool_calls: list[ToolCall]
    final_text: str | None = None


class FakeModel:
    """用简单规则模拟模型根据工具反馈持续决策。"""

    def decide(self, history: list[dict[str, Any]]) -> ModelDecision:
        tool_results = [
            item for item in history
            if item.get("type") == "tool_result"
        ]
        completed = {
            item["name"]: item["content"]
            for item in tool_results
        }

        # 第一步：先获取源代码
        if "read_file" not in completed:
            return ModelDecision([
                ToolCall(
                    name="read_file",
                    arguments={"path": "app.py"},
                    call_id="call_read_1",
                )
            ])

        source = completed["read_file"]

        # 第二步：根据源代码生成修复操作
        if "write_file" not in completed:
            fixed_source = source.replace(
                "return a - b",
                "return a + b",
            )
            if fixed_source == source:
                return ModelDecision(
                    tool_calls=[],
                    final_text="没有找到预期的错误表达式，需要进一步人工检查。",
                )
            return ModelDecision([
                ToolCall(
                    name="write_file",
                    arguments={"path": "app.py", "content": fixed_source},
                    call_id="call_write_1",
                )
            ])

        # 第三步：运行测试验证，而不是凭感觉宣称修好了
        if "run_tests" not in completed:
            return ModelDecision([
                ToolCall(
                    name="run_tests",
                    arguments={},
                    call_id="call_test_1",
                )
            ])

        test_result = completed["run_tests"]
        if "PASS" in test_result:
            return ModelDecision(
                tool_calls=[],
                final_text="已修复 add() 的运算符错误，测试通过。",
            )

        # 这里为了保持例子简短，不再实现第二轮自动修复。
        return ModelDecision(
            tool_calls=[],
            final_text=f"测试仍然失败，需要继续调查：{test_result}",
        )


class MiniHarness:
    """最小运行时：工具路由、执行、反馈和步数限制。"""

    def __init__(self) -> None:
        self.files = {
            "app.py": "def add(a, b):\n    return a - b\n"
        }
        self.tools = {
            "read_file": self.read_file,
            "write_file": self.write_file,
            "run_tests": self.run_tests,
        }

    def read_file(self, path: str) -> str:
        if path not in self.files:
            raise FileNotFoundError(path)
        return self.files[path]

    def write_file(self, path: str, content: str) -> str:
        # 示例只允许操作 app.py，体现工具执行范围限制。
        if path != "app.py":
            raise PermissionError("只允许写入 app.py")
        self.files[path] = content
        return "写入成功"

    def run_tests(self) -> str:
        source = self.files["app.py"]
        if "return a + b" in source:
            # 代表测试断言 add(2, 3) == 5 成功。
            return "PASS: 1 test passed; add(2, 3) == 5"
        return "FAIL: expected 5, got -1"

    def execute(self, call: ToolCall) -> str:
        if call.name not in self.tools:
            raise ValueError(f"未知工具：{call.name}")
        # 真正的生产系统还需要验证 JSON Schema、权限和超时。
        return self.tools[call.name](**call.arguments)

    def run(self, task: str, model: FakeModel, max_steps: int = 8) -> str:
        history: list[dict[str, Any]] = [
            {"role": "user", "content": task}
        ]

        for step in range(1, max_steps + 1):
            print(f"\n--- Loop step {step} ---")
            decision = model.decide(history)

            if not decision.tool_calls:
                print(f"[FINAL] {decision.final_text}")
                return decision.final_text or "任务结束。"

            # 模型的工具请求也属于对话历史的一部分。
            history.append({
                "role": "assistant",
                "type": "tool_calls",
                "calls": [call.__dict__ for call in decision.tool_calls],
            })

            for call in decision.tool_calls:
                print(f"[MODEL -> TOOL] {call.name}({call.arguments})")
                try:
                    result = self.execute(call)
                except Exception as exc:
                    # 生产系统通常会将错误作为 tool_result 反馈给模型，
                    # 而不是只打印后丢弃。
                    result = f"ERROR: {type(exc).__name__}: {exc}"

                print(f"[TOOL -> AGENT] {result}")
                history.append({
                    "role": "tool",
                    "type": "tool_result",
                    "name": call.name,
                    "call_id": call.call_id,
                    "content": result,
                })

        raise RuntimeError(f"超过最大步数限制：{max_steps}")


if __name__ == "__main__":
    harness = MiniHarness()
    result = harness.run(
        task="修复 app.py 中 add 函数的错误，并运行测试验证。",
        model=FakeModel(),
    )
    print("\n任务输出：", result)
    print("最终文件：\n" + harness.files["app.py"])

--- Loop step 1 ---
[MODEL -> TOOL] read_file({'path': 'app.py'})
[TOOL -> AGENT] def add(a, b):
    return a - b

--- Loop step 2 ---
[MODEL -> TOOL] write_file({'path': 'app.py', 'content': 'def add(a, b):\n    return a + b\n'})
[TOOL -> AGENT] 写入成功

--- Loop step 3 ---
[MODEL -> TOOL] run_tests({})
[TOOL -> AGENT] PASS: 1 test passed; add(2, 3) == 5

--- Loop step 4 ---
[FINAL] 已修复 add() 的运算符错误，测试通过。

uv add anthropic
export ANTHROPIC_API_KEY="你的 API Key"

import json
import os
from pathlib import Path

import anthropic

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

# 本示例将文件访问限制在指定工作目录内；生产环境还要防范符号链接等边界。
WORKSPACE = Path("./demo-project").resolve()
WORKSPACE.mkdir(parents=True, exist_ok=True)


def read_file(path: str) -> str:
    target = (WORKSPACE / path).resolve()
    if target != WORKSPACE and WORKSPACE not in target.parents:
        raise PermissionError("路径超出工作目录")
    return target.read_text(encoding="utf-8")


def dispatch_tool(name: str, arguments: dict) -> str:
    if name == "read_file":
        return read_file(**arguments)
    raise ValueError(f"未授权或不存在的工具：{name}")


tools = [
    {
        "name": "read_file",
        "description": "读取工作目录中的文本文件。只能读取项目内文件。",
        "input_schema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"],
        },
    }
]

messages = [
    {
        "role": "user",
        "content": "请读取 app.py，并说明其中定义了什么函数。不要修改任何文件。",
    }
]

MAX_TOOL_ROUNDS = 8

for round_number in range(1, MAX_TOOL_ROUNDS + 1):
    response = client.messages.create(
        # 使用你账户当前可用、支持工具调用的模型名称。
        model=os.environ["ANTHROPIC_MODEL"],
        max_tokens=2048,
        tools=tools,
        messages=messages,
    )

    tool_uses = [block for block in response.content if block.type == "tool_use"]

    # 没有工具请求时，通常意味着模型正在给出普通答复；
    # 生产环境还应区分 end_turn、max_tokens 等停止原因。
    if not tool_uses:
        final_text = "\n".join(
            block.text for block in response.content if block.type == "text"
        )
        print(final_text or f"模型停止，stop_reason={response.stop_reason}")
        break

    # 先保存模型发出的原始 assistant 内容，再回传 tool_result。
    messages.append({"role": "assistant", "content": response.content})
    tool_results = []

    for tool_use in tool_uses:
        try:
            result = dispatch_tool(tool_use.name, tool_use.input)
            is_error = False
        except Exception as exc:
            result = f"{type(exc).__name__}: {exc}"
            is_error = True

        tool_results.append(
            {
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": result,
                "is_error": is_error,
            }
        )

    messages.append({"role": "user", "content": tool_results})
else:
    raise RuntimeError("超过最大工具轮数；任务被安全停止")

User Task
    │
    ▼
Prompt / Context Builder
    │
    ▼
Model Inference
    │
    ├── 最终消息 ──> 输出结果、结束当前任务
    │
    └── 工具请求
          │
          ▼
     Permission / Policy
          │
          ▼
     Tool Runtime / Sandbox
          │
          ▼
     Tool Result / Error
          │
          └────回到上下文，再次调用模型

while not done:
    response = model(prompt)

for item in items:
    validate(item)
    transform(item)
    save(item)

输入 Prompt -> 调用模型一次 -> 返回文本

Thought / 分析：需要查证某个信息
Action / 行动：调用搜索工具
Observation / 观测：工具返回结果
Thought / 分析：现有信息还不够
Action / 行动：再次查询
Observation / 观测：返回补充信息
Final / 答复：整合证据后给出结论

{
  "name": "get_weather",
  "arguments": {"location": "Paris"}
}

rm -rf ./build

模型提出工具请求
      │
      ▼
参数验证 / 工具白名单
      │
      ▼
权限策略检查
  ┌───┼────────┐
拒绝  自动允许  询问用户
  │      │       │
停止/反馈 执行   等待批准/拒绝

运行测试 -> 同样失败 -> 修改 -> 运行测试 -> 同样失败 -> 修改 -> ...

while not finished:
    decision = model(state, tools)
    if decision.is_tool_call:
        result = run_tool(decision.tool_call)
        state = append_result(state, result)
    else:
        finished = True

输入任务
   │
   ▼
需求分类（确定性规则）
   │
   ├── 不清晰 -> 请求补充信息
   │
   └── 清晰
        │
        ▼
  Agent 研究代码（模型驱动）
        │
        ▼
  安全检查（确定性规则）
        │
        ├── 不通过 -> 拒绝/要求修订
        │
        └── 通过
             │
             ▼
        执行修改
             │
             ▼
        测试验证
             │
             ├── 失败 -> 返回修复步骤
             └── 通过 -> 总结并结束

┌─────────────────────────────────┐
│          API / CLI / UI          │
└────────────────┬────────────────┘
                 ▼
┌─────────────────────────────────┐
│ Task Manager / Session State     │
│ 任务状态、预算、取消、检查点       │
└────────────────┬────────────────┘
                 ▼
┌─────────────────────────────────┐
│ Agent Runner / Loop              │
│ 调模型、解析决策、继续/终止       │
└──────────┬───────────────┬──────┘
           │               │
           ▼               ▼
┌────────────────┐  ┌────────────────┐
│ Policy / Approvals│ │ Context Manager│
│ 权限、规则、审批  │ │ 历史、摘要、检索 │
└────────┬───────┘  └────────────────┘
         ▼
┌─────────────────────────────────┐
│ Tool Router / Executor           │
│ Schema 校验、执行、超时、错误处理 │
└────────────────┬────────────────┘
                 ▼
┌─────────────────────────────────┐
│ Tools / Sandbox / External APIs  │
└─────────────────────────────────┘