实践｜Agent Loop，写出第一个可控的循环

从“知道它怎么转”到“知道代码怎么写”

用户说：“运行 main.py，只报告结果。”文件却写着 print(10 / 0)。程序一启动就失败。接下来的代码该放在哪里：运行文件的函数直接回答用户，还是把错误交回循环，再问一次模型？

上一节的交互图已经走过这次交接：问题和工具说明交给模型，模型提出请求，程序执行，结果回传，模型回答。这篇往下拆开发设计。先把三件事分开：一次模型请求、一次工具执行、整个循环的控制。

对应三个函数：ask_claude 取得响应；run_main 取得执行结果；run_loop 保存记录并决定下一步。它们是本册独立编写的教学实现，都在公开 Python 项目的同一个 core/runtime.py 中，暂时不需要一套复杂目录。下面直接展示三个完整函数，代码与 SVG 在同一行对照。点行号设置断点，再点“调试”；“下一步”或“继续”推进真实执行快照，图和说明随之变化。长函数只在代码区内滚动，代码行保持原样。

浏览器演示提供可查看的教学输入：SDK 接口和文件进程使用替身，函数正文实际执行，不发送 API 请求，也不启动本机文件。快照停在该行执行前；完整本地实践仍运行真实进程。

ask_claude：这一次，究竟交给模型什么？

只把“运行 main.py”发给模型还不够。模型需要知道自己可以请求哪个工具、参数怎么写，以及收到结果后该做什么。ask_claude 把三份材料放进同一次 API 请求：SYSTEM 是工作约定，messages 是完整对话记录，TOOLS 是工具说明。

本例的 SYSTEM 要求按用户问题工作、如实报告、不修改文件。TOOLS 声明 run_main，输入结构是空对象，不允许额外字段；真正的 Python 函数不会传给模型。首次 messages 只有原问题，第二次则带上请求和对应结果。三个材料各有职责，不能用一句“这是上下文”略过。

真实执行轨迹回放；SDK 与文件进程使用教学替身，不发 API 请求、不启动文件。

Pycore/runtime.py · ask_claude

未运行

def ask_claude(client, model, messages):
    """只管一次 API 调用，不在这里执行工具或推进循环。"""
    response = client.messages.create(
        model=model, system=SYSTEM, messages=messages, tools=TOOLS,
        max_tokens=1024,
    )
    return response.model_dump(mode="json", exclude_none=True)

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

import json
from pathlib import Path
from types import SimpleNamespace
QUESTION = '运行 main.py，只报告结果。'
SYSTEM = '按用户要求工作。需要运行 main.py 时调用 run_main；收到结果后如实报告，不修改文件。'
TOOLS = [{'name': 'run_main', 'description': '运行工作目录中的 main.py，返回退出码、标准输出和错误输出。', 'input_schema': {'type': 'object', 'properties': {}, 'additionalProperties': False}}]
_scenario = 'request'
_fixture_requests = []
_fixture_executions = []
sys = SimpleNamespace(executable="python")

class _Response:
    def __init__(self, data):
        self.data = data
    def model_dump(self, **options):
        return json.loads(json.dumps(self.data))

class _Messages:
    def create(self, **request):
        assert request["system"] == SYSTEM and request["tools"] == TOOLS
        _fixture_requests.append(json.loads(json.dumps(request)))
        if _scenario == "truncated":
            return _Response({"stop_reason": "max_tokens", "content": []})
        if len(request["messages"]) == 1:
            return _Response({"stop_reason": "tool_use", "content": [{
                "type": "tool_use", "id": "call_1", "name": "run_main", "input": {}
            }]})
        result = request["messages"][-1]["content"][0]
        text = "运行失败：除零错误" if "ZeroDivisionError" in result["content"] else result["content"]
        return _Response({"stop_reason": "end_turn", "content": [{"type": "text", "text": text}]})

class _TimeoutExpired(Exception):
    pass

class _Subprocess:
    TimeoutExpired = _TimeoutExpired
    def run(self, command, **options):
        assert command == ["python", "main.py"]
        _fixture_executions.append({"command": command, "timeout": options["timeout"]})
        if _scenario == "timeout":
            raise _TimeoutExpired(command, options["timeout"])
        success = _scenario == "success"
        return SimpleNamespace(returncode=0 if success else 1,
            stdout="ok\n" if success else "",
            stderr="" if success else "ZeroDivisionError: division by zero\n")

client = SimpleNamespace(messages=_Messages())
subprocess = _Subprocess()


# 调用入口
outcome = ask_claude(client, "教学模型", [{"role": "user", "content": QUESTION}])
print("stop_reason=" + outcome["stop_reason"] + " · id=" + outcome["content"][0]["id"])

一次模型请求

保留完整记录
首次只有用户问题。下一次调用仍用这份列表，带上已保存的请求和工具结果；不能只发送最后一条错误。
首次输入
user → 运行 main.py，只报告结果。
再次输入
user 问题 → assistant 请求 → user 结果
附上约定与工具说明
SYSTEM 约定如实报告、不修改文件。TOOLS 声明 run_main 和空参数结构；它告诉模型能请求什么，实际函数留在应用中。
工具
name=run_main · input_schema=空对象
工作约定
收到结果后如实报告，不修改文件
只请求一次模型
client.messages.create 发送工作约定、记录和工具说明。model 由入口选择；max_tokens 限制本次响应长度。这个函数不执行工具，也不决定下一轮。
一次调用
system + messages + tools → Messages API
输出预算
max_tokens=1024，响应仍可能被截断
把响应交回循环
SDK 对象转为普通字典，保留 stop_reason 与完整 content。run_loop 再据此选择分支：文字、工具请求，或需要停下检查的响应。
首次响应
stop_reason=tool_use；content=[tool_use]
请求内容
id=call_1 · name=run_main · input={}

这个函数刻意只做一次请求。SDK 返回对象后，model_dump 把它转成字典，保留 stop_reason 与完整 content。首次可能返回 tool_use 内容块，包含调用 id、工具 name 和 input；之后也可能返回文字。这里不根据返回文字启动程序，更不在内部偷偷再问一次模型。这样，调用次数和停止条件才能集中在 run_loop 中控制。

TOOLS 是模型看到的接口，run_main 是应用拥有的实现。这是 Claude 客户端工具协议的分工：模型发出结构化请求，应用执行操作并回传结果。max_tokens=1024 只是这次响应的长度预算；到限时可能截断，循环还要检查停止原因。

run_main：怎样把程序失败也变成一个结果？

接到 {"id": "call_1", "name": "run_main", "input": {}} 后，先别急着启动文件。要先记住调用 ID，再检查工具名和参数。模型看到的工具说明不能代替应用校验；否则一个意外工具名或参数就可能越过本来设计的范围。

校验通过后，用当前 Python 解释器，在指定工作目录运行固定的 main.py。subprocess.run 等待进程退出，收集标准输出、错误输出和退出码。除零错误出现在错误输出中，退出码为 1；这些都应交回模型，而不是在此处把错误吞掉。

真实执行轨迹回放；SDK 与文件进程使用教学替身，不发 API 请求、不启动文件。

Pycore/runtime.py · run_main

未运行

def run_main(call, workspace: Path, timeout=3):
    """固定执行当前 Python 解释器 + main.py；这是范围受限的工具，不是沙箱。"""
    result = {"type": "tool_result", "tool_use_id": call["id"]}
    if call.get("name") != "run_main" or call.get("input") != {}:
        return {**result, "content": "工具或参数不支持。", "is_error": True}
    try:
        process = subprocess.run(
            [sys.executable, "main.py"], cwd=workspace,
            capture_output=True, text=True, encoding="utf-8", errors="replace",
            timeout=timeout,
        )
        output = (process.stdout + process.stderr).strip() or "（无输出）"
        return {
            **result, "content": f"exit_code={process.returncode}\n{output[:2000]}",
            "is_error": process.returncode != 0,
        }
    except subprocess.TimeoutExpired:
        return {**result, "content": f"运行超时：超过 {timeout} 秒。", "is_error": True}
    except OSError as error:
        return {**result, "content": f"无法启动：{error}", "is_error": True}

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

import json
from pathlib import Path
from types import SimpleNamespace
QUESTION = '运行 main.py，只报告结果。'
SYSTEM = '按用户要求工作。需要运行 main.py 时调用 run_main；收到结果后如实报告，不修改文件。'
TOOLS = [{'name': 'run_main', 'description': '运行工作目录中的 main.py，返回退出码、标准输出和错误输出。', 'input_schema': {'type': 'object', 'properties': {}, 'additionalProperties': False}}]
_scenario = 'error'
_fixture_requests = []
_fixture_executions = []
sys = SimpleNamespace(executable="python")

class _Response:
    def __init__(self, data):
        self.data = data
    def model_dump(self, **options):
        return json.loads(json.dumps(self.data))

class _Messages:
    def create(self, **request):
        assert request["system"] == SYSTEM and request["tools"] == TOOLS
        _fixture_requests.append(json.loads(json.dumps(request)))
        if _scenario == "truncated":
            return _Response({"stop_reason": "max_tokens", "content": []})
        if len(request["messages"]) == 1:
            return _Response({"stop_reason": "tool_use", "content": [{
                "type": "tool_use", "id": "call_1", "name": "run_main", "input": {}
            }]})
        result = request["messages"][-1]["content"][0]
        text = "运行失败：除零错误" if "ZeroDivisionError" in result["content"] else result["content"]
        return _Response({"stop_reason": "end_turn", "content": [{"type": "text", "text": text}]})

class _TimeoutExpired(Exception):
    pass

class _Subprocess:
    TimeoutExpired = _TimeoutExpired
    def run(self, command, **options):
        assert command == ["python", "main.py"]
        _fixture_executions.append({"command": command, "timeout": options["timeout"]})
        if _scenario == "timeout":
            raise _TimeoutExpired(command, options["timeout"])
        success = _scenario == "success"
        return SimpleNamespace(returncode=0 if success else 1,
            stdout="ok\n" if success else "",
            stderr="" if success else "ZeroDivisionError: division by zero\n")

client = SimpleNamespace(messages=_Messages())
subprocess = _Subprocess()


# 调用入口
outcome = run_main({"id": "call_1", "name": 'run_main', "input": {}}, Path("/teaching"))
print(outcome["tool_use_id"], "· is_error=" + str(outcome["is_error"]))
print(outcome["content"])

除零错误

先记住这是哪次调用
输入是模型返回的一个 tool_use 内容块。先把 id 放进 tool_use_id；无论成功、拒绝还是超时，反馈都能对应原请求。
输入 call
id=call_1 · name=run_main · input={}
结果信封
type=tool_result · tool_use_id=call_1
校验工具名和参数
只接受 run_main 和空参数。工具说明约束模型输入，应用仍必须自己检查；这里不把模型生成的文字当命令执行。
允许
name=run_main 且 input={}
拒绝时
不启动 main.py，直接返回错误结果
在指定目录启动文件
用当前 Python 解释器运行固定文件，cwd 选择工作目录。捕获标准输出和错误输出，超过 timeout 秒就中断等待。
真实进程
当前 Python + 工作目录中的 main.py
边界
默认 3 秒；以本机权限执行
收集除零错误
程序退出后，合并 stdout 和 stderr。非零退出码表示程序失败；Python 的 Traceback 在 stderr 中，它同样是需要交给模型的执行证据。
本次文件
print(10 / 0)
进程结果
returncode=1 · stderr=ZeroDivisionError…
交回可关联的结果
返回值带调用 ID、错误标记和内容。run_main 到这里结束；谁保存结果、是否再问模型，由 run_loop 管。
tool_use_id
call_1（与输入 id 相同）
反馈
is_error=True · exit_code=1 / 除零错误

正常输出

先记住这是哪次调用
输入是模型返回的一个 tool_use 内容块。先把 id 放进 tool_use_id；无论成功、拒绝还是超时，反馈都能对应原请求。
输入 call
id=call_1 · name=run_main · input={}
结果信封
type=tool_result · tool_use_id=call_1
校验工具名和参数
只接受 run_main 和空参数。工具说明约束模型输入，应用仍必须自己检查；这里不把模型生成的文字当命令执行。
允许
name=run_main 且 input={}
拒绝时
不启动 main.py，直接返回错误结果
在指定目录启动文件
用当前 Python 解释器运行固定文件，cwd 选择工作目录。捕获标准输出和错误输出，超过 timeout 秒就中断等待。
真实进程
当前 Python + 工作目录中的 main.py
边界
默认 3 秒；以本机权限执行
收集正常输出
程序退出后，合并 stdout 和 stderr。非零退出码表示程序失败；Python 的 Traceback 在 stderr 中，它同样是需要交给模型的执行证据。
本次文件
print("ok")
进程结果
returncode=0 · stdout=ok
交回可关联的结果
返回值带调用 ID、错误标记和内容。run_main 到这里结束；谁保存结果、是否再问模型，由 run_loop 管。
tool_use_id
call_1（与输入 id 相同）
反馈
is_error=False · exit_code=0 / ok

运行超时

先记住这是哪次调用
输入是模型返回的一个 tool_use 内容块。先把 id 放进 tool_use_id；无论成功、拒绝还是超时，反馈都能对应原请求。
输入 call
id=call_1 · name=run_main · input={}
结果信封
type=tool_result · tool_use_id=call_1
校验工具名和参数
只接受 run_main 和空参数。工具说明约束模型输入，应用仍必须自己检查；这里不把模型生成的文字当命令执行。
允许
name=run_main 且 input={}
拒绝时
不启动 main.py，直接返回错误结果
在指定目录启动文件
用当前 Python 解释器运行固定文件，cwd 选择工作目录。捕获标准输出和错误输出，超过 timeout 秒就中断等待。
真实进程
当前 Python + 工作目录中的 main.py
边界
默认 3 秒；以本机权限执行
把超时变成工具反馈
subprocess.run 抛出 TimeoutExpired。这里接住它并返回可保存的错误结果，让循环继续决定如何处理。
触发条件
文件运行超过 3 秒
结果
is_error=True · 运行超时：超过 3 秒。
交回可关联的结果
返回值带调用 ID、错误标记和内容。run_main 到这里结束；谁保存结果、是否再问模型，由 run_loop 管。
tool_use_id
call_1（与输入 id 相同）
反馈
is_error=True · 超时

拒绝请求

先记住这是哪次调用
输入是模型返回的一个 tool_use 内容块。先把 id 放进 tool_use_id；无论成功、拒绝还是超时，反馈都能对应原请求。
输入 call
id=call_1 · name=other_tool · input={}
结果信封
type=tool_result · tool_use_id=call_1
校验工具名和参数
只接受 run_main 和空参数。工具说明约束模型输入，应用仍必须自己检查；这里不把模型生成的文字当命令执行。
本次输入
name=other_tool · input={}
判断
工具名不支持，不启动进程
拒绝也带回调用 ID
直接返回 is_error=True 的 tool_result。循环可以保存它，模型收到反馈后才知道请求未被执行。
tool_use_id
call_1
content
工具或参数不支持。

先调试“除零错误”，再切换“正常输出”“运行超时”或“拒绝请求”。它们最终都交回 tool_result：tool_use_id 对应输入 id，content 提供反馈，is_error 表示这次工具执行是否失败。超时和无法启动由异常分支转成错误结果；普通程序异常则通过非零退出码识别。

为什么不在这里直接返回 answered？因为 run_main 只知道这次程序怎么结束，不知道用户问题有没有回答。它把事实交回 run_loop，循环才负责保存并再次请求模型。当前工具限制为固定文件、空参数、默认 3 秒，结果文本最多保留 2000 个字符；文件仍以本机权限执行，这些限制没有提供隔离沙箱。

run_loop：顺序、记录和停止，集中在这里

前两个函数交回的数据已经确定，循环才有可靠的判断依据。run_loop 接收两个可调用对象：ask(messages) 取得模型响应，execute(call) 取得工具结果。入口负责把客户端、模型名和工作目录绑定进去。

在线入口的连接形状如下。client、model 和 workspace 由入口准备，SDK 与文件进程依赖本地环境：

python

ask = lambda messages: ask_claude(client, model, messages)
execute = lambda call: run_main(call, workspace)
outcome = run_loop(ask, execute, max_calls=4)

run_loop 不需要知道 API Key 或目录在哪里。离线验证时，只把 ask 换成固定响应；execute 继续运行真实文件。同一套控制逻辑就能被重复验证，接入在线模型时也不用重写循环。

先给保存工具结果的 messages.append({"role": "user", "content": results}) 设置断点，再点“调试”。停住时，results 已经取得错误，messages 还只有两条；下一步真正写入后，图中的记录页才增加到三条。给 response = ask(messages) 再设一个断点，正常输入会命中两次。

真实执行轨迹回放；SDK 与文件进程使用教学替身，不发 API 请求、不启动文件。

Pycore/runtime.py · run_loop

未运行

def run_loop(ask, execute, max_calls=4, emit=print):
    """ask 读完整记录，execute 读工具请求；返回状态和证据，避免把停下当作成功。"""
    if max_calls < 1:
        raise ValueError("max_calls 必须至少为 1")
    messages = [{"role": "user", "content": QUESTION}]

    def finish(status, text, calls):
        return {"status": status, "text": text, "calls": calls, "messages": messages}

    for number in range(1, max_calls + 1):
        emit(f"模型调用 {number}")
        try:
            response = ask(messages)
        except Exception as error:
            # API 异常时交回控制权；不自动重试，也不伪造工具结果。
            return finish("model_error", f"模型请求失败：{type(error).__name__}", number)
        blocks = response.get("content")
        reason = response.get("stop_reason")
        if not isinstance(blocks, list) or any(
            not isinstance(block, dict) or block.get("type") not in ("text", "tool_use")
            for block in blocks
        ):
            return finish("protocol_error", "响应不符合本课支持的格式。", number)
        messages.append({"role": "assistant", "content": blocks})
        calls = [block for block in blocks if block["type"] == "tool_use"]
        text = "\n".join(block.get("text", "") for block in blocks if block["type"] == "text")
        if reason == "end_turn" and not calls and text.strip():
            return finish("answered", text, number)
        if reason not in ("end_turn", "tool_use"):
            return finish("model_stopped", f"模型停止：{reason}；需要检查响应。", number)
        ids = [call.get("id") for call in calls]
        if reason != "tool_use" or not calls or any(
            not isinstance(call_id, str) or not call_id for call_id in ids
        ) or len(set(ids)) != len(ids):
            return finish("protocol_error", "停止原因与工具请求不一致。", number)
        # 上面已经保存完整 assistant 响应；每个结果必须关联原调用 ID。
        results = []
        for call in calls:
            emit(f"执行工具 {call.get('name')}")
            result = execute(call)
            results.append(result)
            emit("工具结果：" + result["content"].splitlines()[-1])
        messages.append({"role": "user", "content": results})
        emit("保存工具结果，供下一次调用读取")
    return finish("call_limit", f"已达 {max_calls} 次模型调用上限；保留记录，交给用户。", max_calls)

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

import json
from pathlib import Path
from types import SimpleNamespace
QUESTION = '运行 main.py，只报告结果。'
SYSTEM = '按用户要求工作。需要运行 main.py 时调用 run_main；收到结果后如实报告，不修改文件。'
TOOLS = [{'name': 'run_main', 'description': '运行工作目录中的 main.py，返回退出码、标准输出和错误输出。', 'input_schema': {'type': 'object', 'properties': {}, 'additionalProperties': False}}]
_scenario = 'report'
_fixture_requests = []
_fixture_executions = []
sys = SimpleNamespace(executable="python")

class _Response:
    def __init__(self, data):
        self.data = data
    def model_dump(self, **options):
        return json.loads(json.dumps(self.data))

class _Messages:
    def create(self, **request):
        assert request["system"] == SYSTEM and request["tools"] == TOOLS
        _fixture_requests.append(json.loads(json.dumps(request)))
        if _scenario == "truncated":
            return _Response({"stop_reason": "max_tokens", "content": []})
        if len(request["messages"]) == 1:
            return _Response({"stop_reason": "tool_use", "content": [{
                "type": "tool_use", "id": "call_1", "name": "run_main", "input": {}
            }]})
        result = request["messages"][-1]["content"][0]
        text = "运行失败：除零错误" if "ZeroDivisionError" in result["content"] else result["content"]
        return _Response({"stop_reason": "end_turn", "content": [{"type": "text", "text": text}]})

class _TimeoutExpired(Exception):
    pass

class _Subprocess:
    TimeoutExpired = _TimeoutExpired
    def run(self, command, **options):
        assert command == ["python", "main.py"]
        _fixture_executions.append({"command": command, "timeout": options["timeout"]})
        if _scenario == "timeout":
            raise _TimeoutExpired(command, options["timeout"])
        success = _scenario == "success"
        return SimpleNamespace(returncode=0 if success else 1,
            stdout="ok\n" if success else "",
            stderr="" if success else "ZeroDivisionError: division by zero\n")

client = SimpleNamespace(messages=_Messages())
subprocess = _Subprocess()

def ask_claude(client, model, messages):
    """只管一次 API 调用，不在这里执行工具或推进循环。"""
    response = client.messages.create(
        model=model, system=SYSTEM, messages=messages, tools=TOOLS,
        max_tokens=1024,
    )
    return response.model_dump(mode="json", exclude_none=True)

def run_main(call, workspace: Path, timeout=3):
    """固定执行当前 Python 解释器 + main.py；这是范围受限的工具，不是沙箱。"""
    result = {"type": "tool_result", "tool_use_id": call["id"]}
    if call.get("name") != "run_main" or call.get("input") != {}:
        return {**result, "content": "工具或参数不支持。", "is_error": True}
    try:
        process = subprocess.run(
            [sys.executable, "main.py"], cwd=workspace,
            capture_output=True, text=True, encoding="utf-8", errors="replace",
            timeout=timeout,
        )
        output = (process.stdout + process.stderr).strip() or "（无输出）"
        return {
            **result, "content": f"exit_code={process.returncode}\n{output[:2000]}",
            "is_error": process.returncode != 0,
        }
    except subprocess.TimeoutExpired:
        return {**result, "content": f"运行超时：超过 {timeout} 秒。", "is_error": True}
    except OSError as error:
        return {**result, "content": f"无法启动：{error}", "is_error": True}


# 调用入口
ask = lambda messages: ask_claude(client, "教学模型", messages)
execute = lambda call: run_main(call, Path("/teaching"))
outcome = run_loop(ask, execute, max_calls=4, emit=lambda text: None)
print(outcome["status"], "· 模型调用", outcome["calls"], "· 记录", len(outcome["messages"]))
print(outcome["text"])

报告错误

建立记录和调用预算
每次运行创建一份 messages，从原问题开始。max_calls 限制模型调用次数；finish 统一交回停止状态、文字、次数和记录。
记录
1 条：用户问题
预算
报告错误用 4 次；上限路径用 1 次
在预算内问一次模型
for 为每次模型调用计数。ask 读完整 messages；在线入口把 ask_claude 绑定为这个接口，离线测试可以换成固定响应。API 异常会返回 model_error。
number
1
返回字段
content + stop_reason
校验格式，保存完整响应
先确认内容块是本课支持的 text 或 tool_use，再整条保存 assistant 响应。混合的文字和工具请求都保留，不能只留下工具名。
记录
2 条：问题 + 模型响应
首次内容
tool_use：call_1 / run_main / {}
先看停止原因，再决定执行
tool_use 必须有完整、非空且互不重复的调用 ID。若响应被截断或格式矛盾，就交回控制权；只出现工具名不代表可以执行。
stop_reason
tool_use
检查后
有效调用 call_1，可以交给 execute
执行请求，收集每个结果
execute 由入口绑定为 run_main。即使结果 is_error=True，也先收集起来；程序失败不会自动让循环结束。多个请求各自保留对应 ID。
返回
tool_result · tool_use_id=call_1
此刻
结果在 results 中；messages 仍只有 2 条
把结果写回，再进入下一轮
以 role:user 写入 tool_result 内容块。这是应用回传工具结果的协议角色；写入后，下一次 ask 才能读到原问题、请求和错误。
记录
3 条：问题 + 请求 + 关联结果
下一轮输入
同一份 messages，包含除零错误
带着执行证据再问模型
进入第二次迭代，ask 读取刚保存的三条记录。这次模型有了运行结果，可以回答用户最初的问题。
number
2
本次响应
end_turn + 文字：运行失败：除零错误
最终文字也写入记录
保存模型的完整文字响应，然后提取 text 与 calls。保留最后回答，让返回记录能解释用户看到了什么。
记录
4 条：问题 / 请求 / 结果 / 回答
检查
reason=end_turn · calls=[] · text 非空
满足回答条件，返回 answered
end_turn、没有工具请求、文字非空，三者同时满足才返回 answered。它表示拿到了模型回答；是否完成任务，还要对照原问题和执行证据。
返回
status=answered · calls=2 · messages=4 条
本题回答
如实报告除零错误，符合“只报告结果”

只允许一次调用

建立记录和调用预算
每次运行创建一份 messages，从原问题开始。max_calls 限制模型调用次数；finish 统一交回停止状态、文字、次数和记录。
记录
1 条：用户问题
预算
报告错误用 4 次；上限路径用 1 次
在预算内问一次模型
for 为每次模型调用计数。ask 读完整 messages；在线入口把 ask_claude 绑定为这个接口，离线测试可以换成固定响应。API 异常会返回 model_error。
number
1
返回字段
content + stop_reason
校验格式，保存完整响应
先确认内容块是本课支持的 text 或 tool_use，再整条保存 assistant 响应。混合的文字和工具请求都保留，不能只留下工具名。
记录
2 条：问题 + 模型响应
首次内容
tool_use：call_1 / run_main / {}
先看停止原因，再决定执行
tool_use 必须有完整、非空且互不重复的调用 ID。若响应被截断或格式矛盾，就交回控制权；只出现工具名不代表可以执行。
stop_reason
tool_use
检查后
有效调用 call_1，可以交给 execute
执行请求，收集每个结果
execute 由入口绑定为 run_main。即使结果 is_error=True，也先收集起来；程序失败不会自动让循环结束。多个请求各自保留对应 ID。
返回
tool_result · tool_use_id=call_1
此刻
结果在 results 中；messages 仍只有 2 条
先保存最后一次工具结果
以 role:user 写入 tool_result 内容块。这是应用回传工具结果的协议角色；写入后，下一次 ask 才能读到原问题、请求和错误。
记录
3 条：问题 + 请求 + 关联错误
预算
已用完 1 次模型调用
预算用完，交回控制权
for 自然结束，返回 call_limit。最后一次工具执行和结果保存已经完成；没有第二次模型调用，也不制造最终回答。
返回
status=call_limit · calls=1 · messages=3 条
保留
最后的除零结果，可供检查；当前只在内存中

响应被截断

建立记录和调用预算
每次运行创建一份 messages，从原问题开始。max_calls 限制模型调用次数；finish 统一交回停止状态、文字、次数和记录。
记录
1 条：用户问题
预算
报告错误用 4 次；上限路径用 1 次
在预算内问一次模型
for 为每次模型调用计数。ask 读完整 messages；在线入口把 ask_claude 绑定为这个接口，离线测试可以换成固定响应。API 异常会返回 model_error。
number
1
返回字段
content + stop_reason
校验格式，保存完整响应
先确认内容块是本课支持的 text 或 tool_use，再整条保存 assistant 响应。混合的文字和工具请求都保留，不能只留下工具名。
记录
2 条：问题 + 已收到的响应
本次响应
stop_reason=max_tokens · content=[]
截断发生在执行工具之前
max_tokens 表示响应长度到限。本课返回 model_stopped，要求检查响应；不会把可能不完整的内容送给 execute。
返回
status=model_stopped · calls=1
程序
未执行 main.py，没有工具结果

这段代码最关键的顺序是：保存完整 assistant 响应 → 校验工具请求 → 执行 → 保存关联结果 → 下一次调用。不能先执行再丢掉请求，也不能只把最后的错误文字发回。请求中的 id 和结果中的 tool_use_id 把两者连起来；多个工具请求就收集多个对应结果。

工具结果放在 role: user 的 tool_result 内容块中，这是应用回传结果的协议形式。它不是用户又提了一个问题。写回前，结果只在局部变量 results 中；写回后，下一次 ask 才能看到它。因此，程序失败并不会立即让循环结束：失败也是下一轮判断需要的证据。

停止需要分清几种含义。先调试“报告错误”，再切换两种提前停止的输入：

条件	返回状态	后续动作
`end_turn`、没有工具请求、文字非空	`answered`	交回模型回答和记录
已用完 `max_calls`	`call_limit`	保留最后结果，不再请求模型
截断或本课不支持的停止原因	`model_stopped`	不执行工具，交回响应供检查
内容块格式、调用 ID 或停止原因矛盾	`protocol_error`	不猜测，不执行工具
API 请求抛出异常	`model_error`	交回控制权，不制造工具结果

这些状态是本例的设计选择。API 的 stop_reason 表示模型这次为何停止生成，应用还要据此决定动作。不能把所有停止原因都当成“任务完成”。本课只支持普通文字与客户端工具内容块；增加其他能力时，需要扩展相应分支。

answered 也不保证程序执行过，模型可能直接给文字。对“只报告结果”，如实报告除零错误足够；对“修好并验证”，同一句话就不够。完成标准要结合原问题和实际执行证据判断。调用上限控制的是模型调用次数，不限制所有工具的总数量或整个任务耗时；记录目前只在内存里，退出后不会自动恢复。

“只允许一次调用”的输入仍会写入第三条记录，然后返回 call_limit，不会再命中第二次 ask。“响应被截断”会返回 model_stopped，执行和结果写回的断点都不会命中。这里的差别来自函数真实执行后的轨迹；图不用另外手动翻步。

把同一份代码放到自己的电脑上

三个完整函数来自独立的公开 Python 仓库，本文源码摘录固定于 core/runtime.py。Python 3.10+ 可直接运行离线入口：

bash

git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s01-agent-loop/user_code/practice.py
python3 examples/s01-agent-loop/user_code/verify_practice.py
python3 examples/s01-agent-loop/user_code/verify_fixtures.py

默认 main.py 包含 10 / 0，模型为离线替身，文件真实执行。终端最后显示 answered 和“运行失败：除零错误”；加 --max-calls 1 则在保存结果后以 call_limit 结束，退出码为 1。verify_practice.py 检查真实成功、失败、超时进程与循环边界，verify_fixtures.py 验证本文三个函数与八组教学输入；准备脚本是 figure_fixtures.py。

接入在线模型与 SDK 验证

本课使用 Anthropic Python SDK，验证版本 1.13.0，日期 2026-10-13。在公开项目根目录执行：

bash

python3 -m venv /tmp/agent-loop-env
/tmp/agent-loop-env/bin/pip install -r requirements.txt
/tmp/agent-loop-env/bin/python examples/s01-agent-loop/user_code/verify_sdk.py

SDK 验证使用真实 SDK 和模拟 HTTP 响应检查报文，文件进程真实执行，没有请求在线模型。要自行调用模型，在终端配置 ANTHROPIC_API_KEY 与 ANTHROPIC_MODEL，再运行：

bash

/tmp/agent-loop-env/bin/python examples/s01-agent-loop/user_code/practice.py --live

入口将 ask 替身换成 ask_claude，并设置客户端 20 秒网络超时、关闭自动重试，避免重试次数藏在循环计数之外。在线模型可能直接回答、选择不同请求，离线验证不保证它一定走图中的路线。

当问题变成“查看文件、修改代码，再验证”，首先需要扩展工具接口与完成标准。一次模型请求、一次工具执行、循环控制这三个边界仍然可以保留，新增能力就有明确的落点。