s01 - Agent Loop 逐行拆解

核心思想

一个 while True 循环 + 一个工具 = 一个 Agent。

模型决定什么时候调工具、什么时候停止。代码只负责执行模型的要求。

---

全局变量（运行前初始化）

代码运行 agent_loop() 之前，先创建了 4 个全局变量：

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv(override=True)

client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
MODEL  = os.environ["MODEL_ID"]
SYSTEM = f"You are a coding agent at {os.getcwd()}. Use bash to solve tasks. Act, don't explain."
TOOLS  = [{ ... }]

变量	类型	是什么	类比
client	Anthropic 对象	API 客户端，发请求用的	你和 Claude 之间的电话
MODEL	字符串	模型名，如 "claude-sonnet-4-6"	打电话给谁
SYSTEM	字符串	系统提示词，定义角色	岗位说明书
TOOLS	列表[字典]	可用工具定义（只有 bash）	工具箱里唯一的工具

TOOLS 详细结构

TOOLS = [{
    "name": "bash",                  # 工具名
    "description": "Run a shell command.",  # 告诉模型这工具干嘛的
    "input_schema": {                # 参数定义（JSON Schema 格式）
        "type": "object",
        "properties": {
            "command": {"type": "string"}  # 一个参数：command，类型字符串
        },
        "required": ["command"],     # command 是必填
    },
}]

---

run_bash 函数（第 65-77 行）

模型的"手"——唯一能做的事就是执行 bash 命令。

def run_bash(command: str) -> str:       # 输入字符串，返回字符串
    # 安全检查：拦截危险命令
    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
    if any(d in command for d in dangerous):
        return "Error: Dangerous command blocked"

    # 执行命令
    r = subprocess.run(
        command,              # 要执行的命令
        shell=True,           # 通过 shell 执行（支持管道 |、重定向 > 等）
        cwd=os.getcwd(),      # 在当前目录执行
        capture_output=True,  # 捕获输出（stdout + stderr）
        text=True,            # 输出是字符串不是字节
        timeout=120           # 120 秒超时
    )
    out = (r.stdout + r.stderr).strip()    # 拼接正常输出和错误输出
    return out[:50000] if out else "(no output)"  # 最多返回 50000 字符

---

agent_loop 逐行拆解

函数签名

def agent_loop(messages: list):   # 接收对话历史列表

messages 初始状态示例：

messages = [
    {"role": "user", "content": "帮我创建一个 hello.py 文件"}
]

循环体

while True:    # 死循环，模型说不调工具才退出

第一步：调用 Claude API

response = client.messages.create(
    model=MODEL,            # 哪个模型
    system=SYSTEM,          # 系统提示词
    messages=messages,      # 对话历史
    tools=TOOLS,            # 可用工具
    max_tokens=8000,        # 最大回复长度
)

client.messages.create() 是 Anthropic SDK 的方法，本质发了一个 HTTP POST 请求。

返回的 response 对象：

属性	类型	含义
response.content	列表	模型返回的内容块（可能包含文字 + 工具调用）
response.stop_reason	字符串	"tool_use" = 想调工具，"end_turn" = 说完了

第二步：把模型回复存入历史

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

此时 messages：

[
    {"role": "user",      "content": "帮我创建 hello.py"},      # 用户
    {"role": "assistant", "content": [ ... ]}                    # 模型回复（新增）
]

第三步：判断是否继续

if response.stop_reason != "tool_use":   # 不是想调工具？
    return                                # 退出，循环结束

第四步：执行工具，收集结果

results = []                             # 存放工具执行结果
for block in response.content:           # 遍历每个内容块
    if block.type == "tool_use":         # 只处理工具调用块
        output = run_bash(block.input["command"])  # 执行命令
        results.append({
            "type": "tool_result",       # 类型标记
            "tool_use_id": block.id,     # 配对 ID（哪个调用对应哪个结果）
            "content": output,           # 实际输出
        })

block（ToolUseBlock）的结构：

block.type   == "tool_use"            # 块类型
block.id     == "toolu_xxx"           # 唯一 ID，用于配对
block.name   == "bash"                # 工具名
block.input  == {"command": "ls"}     # 传入参数

第五步：结果追加到历史，循环回去

messages.append({"role": "user", "content": results})
# 回到 while True 开头，带着完整历史再次调 API

此时 messages：

[
    {"role": "user",      "content": "帮我创建 hello.py"},        # 1. 用户提问
    {"role": "assistant", "content": [tool_use_block]},            # 2. 模型：我要执行 bash
    {"role": "user",      "content": [tool_result]}               # 3. 执行结果（新增）
]

---

完整流程图

用户输入 "帮我创建 hello.py"
        │
        ▼
   ┌─ messages ───────────────┐
   │ [user: 帮我创建 hello.py] │
   └──────────┬───────────────┘
              │  发给 Claude API
              ▼
   Claude: "我要执行 echo 'hello' > hello.py"
   stop_reason = "tool_use"
              │
              ▼
   run_bash("echo 'hello' > hello.py")  →  "(no output)"
              │
              ▼
   结果追加到 messages，再发 API
              │
              ▼
   Claude: "文件已创建完成。"
   stop_reason = "end_turn"
              │
              ▼
   退出循环 → 返回

---

关键理解

1. messages 是唯一的状态：整个循环就是在不断往这个列表里追加消息
2. 模型做决策：什么时候调工具、调什么、什么时候停，全是模型决定的
3. 代码只执行：循环体不做任何判断，只是忠实地执行模型的要求
4. tool_use_id 配对：每次工具调用有唯一 ID，结果必须带上这个 ID，API 才能对上号

---

Python 语法备忘（本项目涉及的）

语法	含义	示例
f"..."	f-string，嵌入变量	f"路径是{os.getcwd()}"
dict["key"]	字典取值	block.input["command"]
list.append(x)	列表追加元素	results.append({...})
any(... for ... in ...)	任意一个满足则为 True	any(d in cmd for d in dangerous)
a if cond else b	三元表达式	out[:50000] if out else "(empty)"
str.strip()	去首尾空白	" hello ".strip() → "hello"
str[:N]	切片，取前 N 个	"hello"[:3] → "hel"

---

Lang/Python #Domain/Agent #Type/原理 #Session/s01