程云来 / 杭州
返回文章·Agent 与 Harness
·约 9 分钟

实践|Agent Loop,写出第一个可控的循环

把 main.py 接到一个小 Harness:提供模型输入,执行工具,回传结果,并分清正常回答、程序失败与调用上限。


文件失败了,循环还要不要继续?

用户说:“运行 main.py,只报告结果。”文件里只有一句:

python
print(10 / 0)

运行会失败,但 Harness 还有工作:取得错误、保存结果、让模型读到它,再把回答交给用户。程序失败和循环结束,是两个不同的时刻。

这篇把实现放到文章里走一遍。你可以先看图里每步发生的事,再展开图下的断点验证;不需要打开仓库或配置 API Key,就能观察循环怎样继续、又在哪里停下。

沿着实现走一圈

先选“报告错误”,点下一步。图中的三个设备分别表示 Harness、模型和 Python 程序;“对应实现”标出当前涉及的函数与代码位置。注意第五步:错误已经回来,模型还没有看到它。第六步保存以后,下一次调用才有这份结果。

你:运行 main.py,只报告结果。
1 / 8
先保存你的问题
Harness→记录

run_loop 建立对话记录。调用模型和执行文件都还没发生。

原问题
运行 main.py,只报告结果。
预算
最多 4 次模型调用
LOOPGOALMODELREADYPYTHONREADY模型Python 程序Harness
对应实现 · run_loopmessages = [{"role": "user", "content": QUESTION}]

图按实现推演;下方短代码执行后回放真实轨迹。模型响应为固定教学数据。

报告错误

  1. 先保存你的问题
    Harness→记录

    run_loop 建立对话记录。调用模型和执行文件都还没发生。

    对应实现 · run_loopmessages = [{"role": "user", "content": QUESTION}]
    原问题
    运行 main.py,只报告结果。
    预算
    最多 4 次模型调用
  2. 第一次调用模型
    Harness→模型

    循环把完整记录交给 ask_claude;它把工作约定、记录和工具说明送入一次 API 请求。

    对应实现 · run_loop → ask_clauderesponse = ask(messages)
    输入
    system + messages + tools
    工具
    run_main:固定运行 main.py,参数为空
  3. 先存请求,再决定执行
    模型→Harness

    模型要求运行文件。循环保存完整响应,确认停止原因是 tool_use,才进入工具执行。

    对应实现 · run_loopmessages.append({"role": "assistant", "content": blocks})
    原因
    stop_reason = tool_use
    请求
    id: call_1 · run_main({})
  4. 把请求交给程序
    Harness→Python 程序

    run_main 检查工具和参数,用当前 Python 解释器运行文件,等待进程返回。模型没有执行它。

    对应实现 · run_loop → run_mainresult = execute(call)
    进程
    当前 Python + main.py
    限制
    文件运行超过 3 秒,返回超时错误
  5. 有了错误,还没有回传
    Python 程序→Harness

    除零让进程失败。run_main 返回工具结果,调用 ID 保留下来;它此刻还只在 result 变量里。

    对应实现 · run_main"tool_use_id": call["id"]
    对应请求
    tool_use_id: call_1 · is_error: true
    结果
    exit_code=1 · ZeroDivisionError
  6. 保存结果,准备下一次调用
    Harness→记录

    工具结果进入第三条记录。当前只调用了 1 次,上限为 4,循环还能继续;程序失败不会直接结束循环。

    对应实现 · run_loopmessages.append({"role": "user", "content": results})
    记录
    问题 → 运行请求 → 对应错误
    判断
    1 < 4,可以再次调用
  7. 把结果带回同一个模型
    Harness→模型

    ask_claude 再读完整记录,工作约定和工具说明仍然提供。模型现在才看到了运行错误。

    对应实现 · run_loop → ask_clauderesponse = ask(messages)
    第 2 次
    原问题 + 运行请求 + call_1 的错误结果
  8. 收到回答,交回控制权
    模型→Harness

    本例模型报告除零错误。循环保存回答,再以 answered 返回;它表示拿到回答,文件仍然运行失败。

    对应实现 · run_loopreturn finish("answered", text, number)
    响应
    end_turn · 运行失败:除零错误
    停止
    answered · 共 2 次调用

只允许一次调用:第 6 步改为停下

  1. 结果保存了,预算也用完了
    Harness→记录

    第 1 次调用的工具结果已经保存,但调用上限也是 1。循环返回 call_limit,不再问模型,所以没有对结果的回答。

    对应实现 · run_loopfor number in range(1, max_calls + 1):
    保留
    问题 → 请求 → 错误,三条记录仍在
    停止
    call_limit · 1 次调用已用完

响应被截断:第 3 步改为停下

  1. 截断的响应,不能继续执行
    模型→Harness

    循环保存已收到的响应后检查停止原因。max_tokens 表示输出截断,本课直接交回控制权,工具没有执行。

    对应实现 · run_loopif reason not in ("end_turn", "tool_use"):
    原因
    stop_reason = max_tokens
    停止
    model_stopped · 没有工具结果
用断点验证这条路径 · 报告错误

点保存结果的 messages.append 行号,再点“调试”:结果已在 result 中,记录还只有两条。下一步写入后变成三条。“响应被截断”会提前停下,走不到这一行。

再给 response = ask(messages) 设断点:报告错误会命中两次;只允许一次调用会在保存结果后结束。切换路径会更换输入并清除当前回放,图的步骤由你手动推进。

这段短代码保留消息块与调用 ID,直接求值替代启动文件;完整实现另有参数校验、超时和 API 异常处理。

PyPython · 浏览器本地运行
未运行
max_calls = 4
truncated = False


def ask(messages):
    # 固定响应替代模型;保留 stop_reason 和内容块结构。
    if truncated:
        return {"stop_reason": "max_tokens", "content": []}
    if len(messages) == 1:
        call = {"type": "tool_use", "id": "call_1", "name": "run_main", "input": {}}
        return {"stop_reason": "tool_use", "content": [call]}
    text = {"type": "text", "text": "运行失败:除零错误"}
    return {"stop_reason": "end_turn", "content": [text]}


def execute(call):
    # 直接求值替代文件进程;真实触发并捕获异常。
    try:
        10 / 0
    except ZeroDivisionError as error:
        return {"tool_use_id": call["id"], "is_error": True,
                "content": f"{type(error).__name__}: {error}", "type": "tool_result"}


messages = [{"role": "user", "content": "运行 main.py,只报告结果。"}]
for number in range(1, max_calls + 1):
    response = ask(messages)
    messages.append({"role": "assistant", "content": response["content"]})
    if response["stop_reason"] == "end_turn":
        print("answered:", response["content"][0]["text"])
        break
    if response["stop_reason"] != "tool_use":
        print("model_stopped:", response["stop_reason"])
        break
    result = execute(response["content"][0])
    messages.append({"role": "user", "content": [result]})
    print("保存", result["tool_use_id"], ";记录", len(messages))
else:
    print("call_limit:", number, "次调用;结果已保存")
输出

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

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

走完后,再试两条路径:“只允许一次调用”会在保存结果后停下;“响应被截断”会在执行工具前停下。图下短代码使用同一组输入,点行号设置断点,再点“调试”,可回放短代码真实执行的行前快照。图仍由你手动推进。

完整实践默认使用离线模型替身,真实运行 main.py;图是对实现的推演,浏览器短代码用直接求值替代文件进程。三者观察同一个问题,执行范围各自标清。

三个函数,分别接住什么?

入口 practice.py 先选择模型接口和工作目录,再把它们交给 run_loop。循环每次向 ask_claude 取得响应;只有有效的工具请求,才会交给 run_main 执行。结果回到同一份 messages,供下一次调用读取。

函数读什么交回什么
run_loop原问题、模型接口、工具实现和调用上限停止状态、回答、调用次数和完整记录
ask_claude工作约定、完整 messages、工具说明一次模型响应:停止原因和内容块
run_main一个工具请求及工作目录对应调用 ID、退出码、输出和错误标记

run_loop 管顺序和停止,ask_claude 负责一次 API 请求,run_main 负责启动程序。这样安排后,替换模型接口不需要重写循环;程序异常也能变成可保存的工具反馈。

这里有两个容易忽略的细节。第一,保存的是完整 assistant 响应,文字和工具请求都保留。第二,结果里的 tool_use_id 要对应请求的 id。Claude 的客户端工具协议要求把结果放入 role: user 的 tool_result 内容块;这表示应用回传执行结果,不是用户又输入了一句话。

停下以后,先看执行证据

在“报告错误”路径里,工具结果有 is_error=True,最后状态却是 answered。这并不矛盾:前者说程序运行失败,后者说循环拿到了正常结束的模型回答。

对“只报告结果”这个问题,如实报告除零错误就是回答。如果用户要求“修好并验证”,同一句回答就还不够。answered 本身也不能证明工具执行过:模型可能直接返回文字。要判断任务是否完成,需要看原问题,以及记录中实际执行的请求和结果。

“只允许一次调用”则停得更早:结果保留在第三条记录里,但没有第二次模型调用。调用上限控制的是还能不能继续请求模型,不是把失败变成成功。记录目前只在进程内存中,进程退出后不会自动恢复。

完整实现还处理哪些异常?
  1. 文件运行失败或超过 3 秒,返回带 is_error 的工具结果;循环仍可把这份反馈交给模型。
  2. API 请求异常,返回 model_error;客户端网络超时设为 20 秒,自动重试关闭。
  3. 响应截断或本课不支持的停止原因,返回 model_stopped;空回答、重复调用 ID 等矛盾格式,返回 protocol_error,避免继续猜测。

图中的截断路径采用第三种处理。正式实现先检查停止原因,再决定能否执行工具,不能把“模型停止生成”一律当成拿到了最终回答。

在自己的电脑上运行同一例子

源码在独立的公开 Python 仓库。用 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

默认文件包含 10 / 0。终端会标明模型是离线替身,文件真实执行;最后显示 answered 和“运行失败:除零错误”。加 --max-calls 1,就会在保存结果后以 call_limit 停下,退出码为 1。

想检查边界,可运行:

bash
python3 examples/s01-agent-loop/user_code/verify_practice.py
python3 examples/s01-agent-loop/user_code/verify_reading.py

前者验证真实成功、除零和超时进程,以及请求/结果关联和停止控制;后者验证本文短代码的三条路径与保存前后的记录变化。短代码原文件是 user_code/reading_demo.py;完整运行时仍在 core/runtime.py。阅读源码时,可以沿图里的函数名定位。

可选:接入真实模型,循环需要改吗?

不需要。入口把 ask 从测试替身换成 ask_claude,run_loop 和 run_main 继续使用同一份实现。ask_claude 中的一次请求是:

python
response = client.messages.create(
    model=model, system=SYSTEM, messages=messages, tools=TOOLS,
    max_tokens=1024,
)
return response.model_dump(mode="json", exclude_none=True)

SYSTEM 约定按用户要求工作、运行后如实报告、不修改文件。TOOLS 只声明 run_main 空参数工具;真正执行文件的函数留在 Harness 中。max_tokens 限制一次响应的输出长度。

本课使用 Anthropic Python SDK,验证版本为 1.13.0(2026-10-13)。在 learn-agent-harness 根目录安装:

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

verify_sdk 使用真实 SDK 和模拟 HTTP 响应检查报文,Python 文件真实执行;没有请求在线模型。在自己的终端配置 ANTHROPIC_API_KEY 和 ANTHROPIC_MODEL,型号选择账户可用且支持工具调用的模型,然后显式运行:

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

在线模型可能直接回答或提出不同请求,离线替身不能证明它一定按图中路线行动。完整程序和 SDK 片段保持静态展示,浏览器断点只执行上述独立短代码。

与参考课程和 Claude Code 的关系

参考课程第一课从一个工具和循环开始。本课沿这个最小结构增加调用上限与停止状态。Claude Code 官方说明描述了收集上下文、执行动作、验证结果的循环,还涉及上下文管理和权限;三个教学函数不能代表它的完整架构。

Claude Code 的公开仓库许可证保留全部权利,公开仓库不等于核心运行时完整开源。本课使用开放的 API 客户端 SDK,在独立公开仓库中编写教学 Harness。

当前工具固定运行 main.py,不接受任意命令;文件仍以本机权限执行,是教学入口而非隔离沙箱。当问题变成“查看文件、修改代码,再验证”,循环可以保留,工具就需要扩展。这是下一步设计自然出现的需求。