第 2 周：Agent 到底是什么？一个 while 循环

{
  "model": "claude-opus-5-5",
  "max_tokens": 16000,
  "tools": [{
    "name": "run_tests",
    "description": "运行某个模块的单元测试，返回 pytest 输出。需要确认测试是否通过时调用。",
    "input_schema": {
      "type": "object",
      "properties": {"module": {"type": "string"}},
      "required": ["module"]
    }
  }],
  "messages": [{"role": "user", "content": "跑一下 login 模块的测试，有失败就查清楚原因。"}]
}

{
  "role": "assistant",
  "stop_reason": "tool_use",
  "content": [
    {"type": "text", "text": "先跑一下 login 模块的测试。"},
    {"type": "tool_use", "id": "toolu_01", "name": "run_tests", "input": {"module": "login"}}
  ]
}

{"role": "user", "content": [
  {"type": "tool_result", "tool_use_id": "toolu_01", "content": "1 failed, 2 passed ..."}
]}

[0] user       "跑一下 login 模块的测试，有失败就查清楚原因。"
[1] assistant  text + tool_use(id=toolu_01, run_tests)            stop_reason: tool_use
[2] user       tool_result(tool_use_id=toolu_01)
[3] assistant  text + tool_use(toolu_02, read_file 日志)
                    + tool_use(toolu_03, read_file 源码)          并行调用
[4] user       tool_result(toolu_02) + tool_result(toolu_03)     一条消息装两个结果
[5] assistant  text("test_login_lockout 失败：...")               stop_reason: end_turn

import json
import time

import anthropic

client = anthropic.Anthropic()  # 从 ANTHROPIC_API_KEY 读取凭证
MODEL = "claude-opus-5-5"
TOOLS = [{
    "name": "read_file",
    "description": "读取仓库里的一个文本文件并返回内容。需要看源码或日志时调用。",
    "input_schema": {
        "type": "object",
        "properties": {"path": {"type": "string", "description": "相对仓库根目录的路径"}},
        "required": ["path"],
    },
}]


def read_file(path: str) -> str:
    with open(path, encoding="utf-8") as f:
        return f.read()[:20_000]


TOOL_IMPLS = {"read_file": read_file}


def run_tool(block) -> dict:
    """执行一个 tool_use 块。无论成败，都返回 tool_use_id 相同的 tool_result。"""
    try:
        output = TOOL_IMPLS[block.name](**block.input)
        return {"type": "tool_result", "tool_use_id": block.id, "content": output}
    except Exception as e:  # 工具报错回给模型，而不是让循环崩掉
        return {"type": "tool_result", "tool_use_id": block.id,
                "content": f"{type(e).__name__}: {e}", "is_error": True}


def run_agent(task, max_turns=10, token_budget=200_000, timeout_s=300, max_repeats=3):
    messages = [{"role": "user", "content": task}]
    used, start, seen = 0, time.monotonic(), {}
    for _ in range(max_turns):
        if used >= token_budget:
            return "stopped: token_budget", messages
        if time.monotonic() - start >= timeout_s:
            return "stopped: timeout", messages
        response = client.messages.create(
            model=MODEL, max_tokens=16000, tools=TOOLS, messages=messages,
        )
        # 开了 prompt caching 时，input_tokens 不含缓存部分，还要加上
        # usage.cache_read_input_tokens 和 usage.cache_creation_input_tokens
        used += response.usage.input_tokens + response.usage.output_tokens
        messages.append({"role": "assistant", "content": response.content})  # 整个 content 原样追加

        if response.stop_reason == "end_turn":
            return "done", messages
        # 本例没有服务端工具，不会出现 pause_turn；用了 web search 等服务端工具时，
        # 要把 assistant 内容追加回去再请求一次
        if response.stop_reason != "tool_use":  # max_tokens / refusal / ... 都不算完成
            return f"stopped: {response.stop_reason}", messages

        calls = [b for b in response.content if b.type == "tool_use"]
        for b in calls:
            key = (b.name, json.dumps(b.input, sort_keys=True))
            seen[key] = seen.get(key, 0) + 1
            if seen[key] >= max_repeats:
                return f"stopped: repeated {b.name}", messages
        # 并行调用：全部结果放进同一条 user 消息，而且只放 tool_result
        messages.append({"role": "user", "content": [run_tool(b) for b in calls]})
    return "stopped: max_turns", messages


if __name__ == "__main__":
    status, history = run_agent("读一下 README.md，用三句话总结这个项目。")
    print(status, len(history))