HuanCode Docs

Harness实战:从零搭建Agent Loop

Agent的核心不是模型,而是循环。用50行Python搭建最小Agent Loop,从纯对话到工具调用,拆解每一次request和response,看懂Agent到底怎么"自己干活"。

写在前面

上一篇《Harness实战:从API调用到Function Calling》把地基打好了——API 调通了,多轮对话跑通了,Function Calling 也摸清了。

但那些都是单次调用。你问一句,模型答一句。模型说"我需要调工具",你手动把结果喂回去。

真正的 Agent 不是这样工作的。

真正的 Agent 有一个循环:模型说要调工具,Harness 就去调;调完把结果喂回来,模型再想想要不要继续调;直到模型说"我搞定了",循环才结束。

这个循环,就是 Agent Loop

今天我们用不到 50 行 Python,从零搭建一个最小的 Agent Loop。


第一步:搭终端交互框架

先不管模型,搭一个能跑的输入循环:

def agent_loop(history):
    pass

BANNER = """\
\033[1;35m ▐▛███▜▌\033[0m   \033[1mHuanCode Harness s01\033[0m
\033[1;35m▝▜█████▛▘\033[0m  Agent Loop · 最小 Agent 循环
\033[1;35m  ▘▘ ▝▝\033[0m    输入 q 退出

\033[90m─────────────────────────────────────────────\033[0m"""

if __name__ == "__main__":
    print(BANNER)
    history = []
    while True:
        try:
            query = input("\033[1;35m❯\033[0m ")
        except (EOFError, KeyboardInterrupt):
            break
        if query.strip().lower() in ("q", "exit", ""):
            break
        history.append({"role": "user", "content": query})
        agent_loop(history)

模仿 Claude Code 的启动风格——紫色 logo、分隔线、 提示符。agent_loop 先留空,跑一下确认框架没问题:

终端启动效果

框架就绪,现在往 agent_loop 里填东西。


第二步:实现最小对话循环

把用户输入发给模型,拿到回复,打印出来,加入历史。核心就这几行:

import anthropic, os
from dotenv import load_dotenv

load_dotenv()

client = anthropic.Anthropic(
    api_key=os.environ["API_KEY"],
    base_url=os.environ["BASE_URL"],
)
MODEL = "claude-sonnet-4.6"
SYSTEM = "you are a helpful assistant"

def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            max_tokens=1000,
        )
        messages.append({"role": "assistant", "content": response.content})

        # 只有 tool_use 需要继续循环,其余都退出
        if response.stop_reason != "tool_use":
            return

这里有个关键设计:循环的退出条件是判断"不是 tool_use"

为什么?因为 stop_reason 有四种值:

stop_reason含义该怎么办
end_turn模型说完了退出
tool_use需要调工具继续循环
max_tokens被截断了退出
stop_sequence遇到停止序列退出

只有 tool_use 需要继续,其余全部退出。一行 if response.stop_reason != "tool_use": return 就够了。

stop_reason 四种状态

验证:多轮上下文

现在还没有工具,先验证对话历史是否正确累积:

❯ 我叫小明
❯ 我刚才说我叫什么?

模型能准确回答"小明",说明 messages 列表在正确累积对话历史,Agent Loop 的核心能力已经具备。

验证多轮上下文


第三步:加入 Bash 工具

纯对话的 Agent 没什么用。Agent 的价值在于能动手——而最万能的"手"就是一个 Bash 工具。

定义工具执行函数

import subprocess

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)"
    except (FileNotFoundError, OSError) as e:
        return f"Error: {e}"

subprocess.run 是 Python 执行 shell 命令的标准方式。capture_output=True 捕获输出,text=True 返回字符串,timeout=120 防止命令挂死。加了一层简单的危险命令过滤。

注册工具 Schema

告诉模型有一个 bash 工具可以用:

TOOLS = [{
    "name": "bash",
    "description": "Run a shell command.",
    "input_schema": {
        "type": "object",
        "properties": {"command": {"type": "string"}},
        "required": ["command"],
    },
}]

同时修改系统提示词,让模型倾向于用工具解决问题:

SYSTEM = f"You are a coding agent at {os.getcwd()}. Use bash to solve tasks. Act, don't explain."

补全循环中的工具执行

agent_loop 中加入工具执行逻辑——这才是 Agent Loop 真正"loop"的部分:

def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
            max_tokens=1000,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return

        # 执行工具并把结果回传,继续循环让模型决定下一步
        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})

整个循环的逻辑:

while True:
    调用模型 → 拿到 response
    ├─ stop_reason != "tool_use" → 模型搞定了,return
    └─ stop_reason == "tool_use" → 执行工具,回传结果,继续循环

Agent Loop 核心流程

模型自己决定要不要用工具、用哪个工具、用完之后还要不要继续。 Harness 只负责执行和传话。这就是 Agent Loop 的全部秘密。


逐帧拆解:一次完整的工具调用

运行程序,输入"当前目录下有哪些文件?",看看 Agent Loop 到底发生了什么。

第一轮:用户提问 → 模型决定调工具

Request(发给模型的):

{
  "model": "claude-sonnet-4.6",
  "system": "You are a coding agent at /Users/.../harness. Use bash to solve tasks.",
  "tools": [{"name": "bash", ...}],
  "messages": [
    {"role": "user", "content": "当前目录下有哪些文件?"}
  ]
}

Response(模型返回的):

{
  "stop_reason": "tool_use",
  "content": [
    {
      "type": "tool_use",
      "name": "bash",
      "input": {"command": "ls /Users/.../harness"}
    }
  ]
}

注意 stop_reason"tool_use"——模型没有直接回答,而是说"我要用 bash 工具跑一下 ls"。

Harness 执行 ls 命令,拿到结果,塞进 tool_result,继续循环。

第二轮:工具结果 → 模型整理回复

Request(第二次发给模型的):

{
  "messages": [
    {"role": "user", "content": "当前目录下有哪些文件?"},
    {"role": "assistant", "content": [{"type": "tool_use", "name": "bash", ...}]},
    {"role": "user", "content": [{"type": "tool_result", "content": "01-agent-loop.py\nharness.py\nimages\n..."}]}
  ]
}

三条消息:用户提问 → 模型要调工具 → 工具执行结果。完整的对话历史。

Response(模型最终回复):

{
  "stop_reason": "end_turn",
  "content": "当前目录下共有以下文件和文件夹:\n| 名称 | 类型 |\n|------|------|\n| 01-agent-loop.py | Python 文件 |\n| harness.py | Python 文件 |\n| images | 文件夹 |\n..."
}

stop_reason 变成了 "end_turn"——模型看到工具结果后,整理成表格回复给用户。循环结束。

两轮就完成了:提问 → 调工具 → 拿结果 → 回复。 这就是一个最简单的 Agent Loop 的完整生命周期。

一次完整的工具调用时序


关于 Request 和 Response 的一个细节

你可能注意到了:发给 API 的 messages 用的是 Python dict,但 response 回来的却是 SDK 的 Pydantic 对象。比如 response.content 是一个 TextBlockToolUseBlock 对象列表,访问属性用点号(.type.input)而不是方括号。

但把 response.content 直接塞进 dict 的 "content" 字段居然也能跑通:

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

这是因为 Anthropic SDK 做了兼容处理——发送请求时会自动把 Pydantic 对象序列化为 JSON。所以 dict 和 SDK 对象可以混用,不需要手动转换。


完整代码

不到 50 行核心逻辑,一个能用 Bash 干活的 Agent:

import json, os, subprocess
import anthropic
from dotenv import load_dotenv

load_dotenv()

client = anthropic.Anthropic(
    api_key=os.environ["API_KEY"],
    base_url=os.environ["BASE_URL"],
)
MODEL = "claude-sonnet-4.6"
SYSTEM = f"You are a coding agent at {os.getcwd()}. Use bash to solve tasks. Act, don't explain."

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)"
    except (FileNotFoundError, OSError) as e:
        return f"Error: {e}"

TOOLS = [{
    "name": "bash",
    "description": "Run a shell command.",
    "input_schema": {
        "type": "object",
        "properties": {"command": {"type": "string"}},
        "required": ["command"],
    },
}]

def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            messages=messages, tools=TOOLS, max_tokens=1000,
        )
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            return
        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[1;35m❯\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()

下一步

这个 Agent 已经能干活了,但它有几个明显的缺陷:

  • 只有一个工具——真正的 Agent 需要文件读写、搜索等多种工具
  • 没有计划——不会拆分任务、跟踪进度
  • 没有上下文管理——对话太长会撑爆 context window

下一篇,我们给它加上更多工具,让它从"能执行命令"进化到"能完成任务"。

On this page