01｜Agent 循环：让模型真正动手
@[toc]
## 本篇导读
聊天模型能够告诉你怎样修改代码，但仅仅输出建议，并不会让文件发生变化。要让它完成“创建文件、运行程序、根据报错修复”这样的任务，需要把模型的工具请求执行掉，再把执行结果交给模型。这条反馈链，就是 Agent 循环。
**学完本篇，你将能够：** 理解 messages、tool_use、tool_result 和 stop_reason 如何组成最小执行闭环。
## 设计思路
每一轮先调用模型，再保存助手响应。如果响应包含工具调用，就逐个执行，并将结果作为下一条 user 消息送回；如果模型不再请求工具，本轮任务结束。模型负责选择动作，程序负责执行与反馈。
例如用户要求创建 hello.py，模型先请求 bash 写文件；工具返回成功后，模型还可以请求运行文件，检查输出，再向用户给出结果。这比一次性生成一段代码多了“观察结果并继续行动”的能力。
## 核心代码与解释
先通过关键片段理解本篇机制；随后“完整代码”一节给出包含所有依赖定义与交互入口的程序，可整段复制保存并运行。
### 把 Shell 包装成工具
```python
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"
try:
r = subprocess.run(command, shell=True, cwd=os.getcwd(),
capture_output=True, text=True, timeout=120)
out = (r.stdout + r.stderr).strip()
return out[:50000] if out else "(no output)"
except subprocess.TimeoutExpired:
return "Error: Timeout (120s)"
```
函数捕获标准输出和错误输出，设置超时，并限制返回文本的长度。模型看到的是字符串结果，实际执行发生在 Harness 中。这里把两种输出拼接在一起，便于教学，但没有单独返回退出码。
### 把工具结果送回模型
```python
def agent_loop(messages: list):
while True:
response = client.messages.create(
model=MODEL, system=SYSTEM, messages=messages,
tools=TOOLS, max_tokens=8000,
)
# Append assistant turn
messages.append({"role": "assistant", "content": response.content})
# If the model didn't call a tool, we're done
if response.stop_reason != "tool_use":
return
# Execute each tool call, collect results
results = []
for block in response.content:
if block.type == "tool_use":
print(f"\033[33m$ {block.input['command']}\033[0m")
output = run_bash(block.input["command"])
print(output[:200])
results.append({"type": "tool_result", "tool_use_id": block.id,
"content": output})
messages.append({"role": "user", "content": results})
```
先追加 assistant 响应，再追加工具结果，才能形成完整的调用记录。tool_use_id 必须对应发起调用的 block.id，模型才能知道每个结果属于哪次请求。stop_reason 判断的是是否继续调用工具，并不是对任务正确性的独立验证。
## 跟着一次工具调用读懂消息结构
把 messages 理解为本次对话的工作记录。你输入任务后，交互入口先把它追加为 user 消息；agent_loop 再把整份记录发给模型。模型请求运行 hello.py 时，返回的是一个 tool_use 块，其中 name 是 bash，input 保存 command，id 标识这次调用。模型只是提出请求，真正启动进程的是 run_bash。
执行后，程序先保存包含这次调用的 assistant 消息，再追加带有 tool_use_id 的 tool_result。两处 ID 相同，便建立了“请求与结果”的对应关系。工具结果虽然放在 role=user 的消息中，却不是用户重新输入了一句话：这是 Anthropic Messages 协议承载工具反馈的方式。下一轮模型据此判断继续执行还是给出答复。
response.content 是内容块列表，一次响应可能同时包含文本和多个工具调用，因此程序需要遍历 block，而不能直接把响应当作字符串。这里的多个调用按 for 循环逐个执行；它们不会因为同属一次响应就自动并行。
阅读后面的完整程序时，可以先定位这些字段：
| 代码位置 | 含义与阅读重点 |
| --- | --- |
| `TOOLS / input_schema` | 向模型说明工具名称、参数格式与必填字段；这里不执行命令。 |
| `SYSTEM / messages` | SYSTEM 提供行为指导，messages 保存本次对话已经发生的事。 |
| `block.id / tool_use_id` | 把工具结果关联到发起它的那次调用，多个调用也不会混淆。 |
| `max_tokens=8000` | 限制一次模型响应的输出预算，不是整个上下文窗口的大小。 |
| `print(output[:200])` | 终端只展示前 200 个字符；送回模型的是 run_bash 返回的输出，最多 50000 个字符。 |
run_bash 中的 `shell=True` 让命令字符串交给系统 Shell 解释，cwd 决定命令的工作目录；工具名叫 bash，不代表所有操作系统都会使用 Bash。timeout 约束这次子进程等待时间，不是整个 Agent 任务的总时限。
## 动手练习
### 完整代码：复制后独立运行
本篇所需的 Python 程序全部列在下面，包含导入、模型客户端、工具定义、处理函数和交互入口。无需下载教程仓库，也无需复制前一篇的实现。你需要 Python 3.10 或更高版本，以及可用的 Anthropic 兼容接口密钥和模型名称。
#### 1. 创建目录并安装依赖
每篇使用单独目录，避免混入其他课程的任务与团队记录。以下命令用于 macOS / Linux：
```bash
mkdir -p ~/agent-harness-s01
cd ~/agent-harness-s01
python3 -m venv .venv
source .venv/bin/activate
python -m pip install "anthropic>=0.25.0" "python-dotenv>=1.0.0"
```
Windows PowerShell 用户先在用户主目录创建 `agent-harness-s01` 文件夹，再执行 `cd ~/agent-harness-s01`。将创建与激活虚拟环境的两条命令改为 `py -3 -m venv .venv` 和 `.\.venv\Scripts\Activate.ps1`，其他文件内容相同。在该终端设置 `$env:PYTHONUTF8="1"` 后运行，统一使用 UTF-8 读取练习文件。
在当前目录新建 `.env`，复制下面内容，然后把两个占位值替换为你的真实配置：
```dotenv
ANTHROPIC_API_KEY=your-api-key
MODEL_ID=your-model-id
```
`MODEL_ID` 填写接口实际支持的模型名称。使用默认 Anthropic 服务时，上面两项即可；使用兼容接口时，在 `.env` 增加 `ANTHROPIC_BASE_URL=你的接口地址`。这些程序使用 Anthropic Messages 协议，不能直接填只支持 OpenAI Chat Completions 的接口地址。网络访问、模型调用和实际费用取决于你使用的服务。
程序把启动时的当前目录作为工作区，因此运行前应进入本篇目录。文件工具中的 `hello.py`、`skills/` 和 `.tasks/` 等相对路径都以该工作区为基础；把脚本保存在正确目录，还不等于终端已经进入该目录。
#### 2. 保存完整程序
**文件名：`s01_agent_loop.py`**。在上面创建的 `~/agent-harness-s01/` 目录中新建这个文件，将下面全部代码复制进去。`.env` 和该 Python 文件保存在同一个目录。
```python
import os
import subprocess
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv(override=True)
if os.getenv("ANTHROPIC_BASE_URL"):
os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)
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 = [{
"name": "bash",
"description": "Run a shell command.",
"input_schema": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"],
},
}]
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"
try:
r = subprocess.run(command, shell=True, cwd=os.getcwd(),
capture_output=True, text=True, timeout=120)
out = (r.stdout + r.stderr).strip()
return out[:50000] if out else "(no output)"
except subprocess.TimeoutExpired:
return "Error: Timeout (120s)"
# -- The core pattern: a while loop that calls tools until the model stops --
def agent_loop(messages: list):
while True:
response = client.messages.create(
model=MODEL, system=SYSTEM, messages=messages,
tools=TOOLS, max_tokens=8000,
)
# Append assistant turn
messages.append({"role": "assistant", "content": response.content})
# If the model didn't call a tool, we're done
if response.stop_reason != "tool_use":
return
# Execute each tool call, collect results
results = []
for block in response.content:
if block.type == "tool_use":
print(f"\033[33m$ {block.input['command']}\033[0m")
output = run_bash(block.input["command"])
print(output[:200])
results.append({"type": "tool_result", "tool_use_id": block.id,
"content": output})
messages.append({"role": "user", "content": results})
if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms01 >> \033[0m")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):
break
history.append({"role": "user", "content": query})
agent_loop(history)
response_content = history[-1]["content"]
if isinstance(response_content, list):
for block in response_content:
if hasattr(block, "text"):
print(block.text)
print()
```
这段代码已包含本篇所有函数与工具 Schema；下方只需按本篇要求准备练习材料，然后启动程序。
### 进入代码目录并运行
确认完整代码已保存为 `s01_agent_loop.py`，本篇练习材料已准备，`.env` 的占位值已替换。macOS / Linux 用户在终端进入文件所在目录，激活该目录的虚拟环境，再执行程序：
```bash
cd ~/agent-harness-s01
source .venv/bin/activate
python s01_agent_loop.py
```
Windows PowerShell 用户进入相同目录，将激活命令换成 `.\.venv\Scripts\Activate.ps1`。如果你把代码保存在其他位置，将 `cd` 后的路径换成实际保存目录；运行命令中的文件名保持一致。
### 输入任务并检查结果
程序显示交互提示后，输入：
> 请创建 hello.py，输出 Hello, Agent Harness；运行它并告诉我实际输出。
**如何检查结果：** 观察终端是否出现写文件和运行文件的两次工具调用，以及最终回复是否引用真实输出。然后故意让程序抛出异常，观察模型是否能根据错误继续修复。
输入 `exit` 退出。出现缺少模块错误时，在当前虚拟环境重新执行安装命令；认证或模型名称错误则检查 `.env`。团队与后台任务需保持主程序运行，结束进程会终止守护线程。
## 工程边界与常见误区
危险命令字符串黑名单只是教学措施，不能作为可靠的 Shell 沙箱。实际系统还需要目录与进程隔离、明确权限、退出码、轮次及费用预算。循环退出也可能因为达到输出限制或其他停止原因；生产系统需要分别处理。
## 本篇小结
Agent 的最小闭环是“模型请求动作 → 工具执行 → 结果反馈 → 再次决策”。后续机制都围绕这个闭环展开。
下一篇：**02｜工具调用：设计可扩展的工具层**。