Agent Harness 工程：从一次 LLM 调用到 Agent Loop

上一篇讲了一个核心判断：Agent = LLM + Harness。

这篇开始落到代码。我们不先搭框架，也不先讨论复杂的任务规划，而是从最朴素的一次模型调用开始：给模型一段用户输入，拿回一段回答。然后逐步加上工具定义、工具调用、工具结果回传和循环，最后得到一个最小可运行的 Agent Loop。

很多 Agent 框架看起来很神秘，但拆到最小结构，其实就是下面这条链路：

模型看到上下文 -> 模型决定下一步 -> Harness 执行动作 -> 执行结果回到上下文 -> 模型继续判断。

这条链路一旦跑通，后面再加更多工具、权限审批、上下文压缩、任务状态持久化，本质上都是在扩展这个循环。

第一步：一次最普通的 LLM 调用

最早的 ChatGPT 体验就是一问一答：用户输入问题，模型返回回答。我们先用 OpenAI 兼容接口写出这个最小过程。

这里我使用的是本地部署的 gpt-oss，服务地址是 http://localhost:8080/v1。本地部署大模型的方式，可以参考前一篇《llama.cpp 入门：在你的本地设备上运行大语言模型》。

from openai import OpenAI


client = OpenAI(
    base_url="http://localhost:8080/v1",
    api_key="no-need",
)

MODEL = "gpt-oss-20b-Q4_K_M.gguf"


def run_llm(message: str) -> str:
    response = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": message},
        ],
    )
    return response.choices[0].message.content or ""


if __name__ == "__main__":
    user_input = input("Enter your message: ")
    output = run_llm(user_input)
    print("Response from LLM:", output)

一次普通调用的核心，就是把 messages 交给模型，再从 choices[0].message 里取回回答。

这段代码很短，但里面已经出现了几个 Agent 工程里绕不开的概念。

messages 不是普通字符串

很多人第一次接触 ChatCompletion 时，会觉得 messages 只是把 prompt 换成了数组。这个理解只对了一半。

messages 更像一段结构化对话历史。每条消息都有 role 和 content。content 是具体内容，role 则表示这段内容是谁产生的，以及它在对话里应该被如何理解。

在本文这个最小 Agent 里，先重点关注四种角色：

role	生产者	作用
`system`	系统或开发者	给模型设定身份、目标和边界
`user`	用户	表达任务、问题、约束或反馈
`assistant`	模型	模型的自然语言回答或工具请求
`tool`	Harness	工具执行后的结果

你在 SDK 类型里还会看到 developer、function 等角色或兼容类型。它们各有历史和平台语义，但对于理解最小 Agent Loop 来说，先把上面四个角色想清楚就够了。

这里有一个非常关键的点：

角色不是装饰信息，而是上下文协议的一部分。

同一句话，如果放在 system 里，通常代表更高优先级的行为约束；放在 user 里，代表用户本轮输入；放在 tool 里，则代表外部环境返回的事实。Agent 能不能稳定工作，很大程度上取决于你有没有把信息放在正确的消息角色里。

choices[0] 是什么

上面的代码里有一行：

return response.choices[0].message.content or ""

为什么是 choices[0]？

因为 ChatCompletion 的响应允许一次返回多个候选结果。你可以通过 n 参数要求模型生成多个候选回答。如果没有设置，默认通常只有一个候选，所以我们取 choices[0]。

普通问答里，这个细节不太显眼。但到了 Agent 场景，choices[0].message 不一定只有自然语言内容，它还可能包含 tool_calls。也就是说，模型下一步想做的事，可能不是“回答用户”，而是“请求 Harness 调用某个工具”。

可以先记住这个判断：

assistant_message = response.choices[0].message

if assistant_message.tool_calls:
    # 模型希望调用工具
else:
    # 模型给出了最终回答

Agent Loop 的分岔点，基本就在这里。

chat_template：结构化消息如何进入模型

对本地模型来说，messages 这种结构最终还是要变成模型能读懂的 token 序列。这个转换通常由 chat_template 完成。

你可以把 chat_template 理解成一套序列化规则：它规定 system、user、assistant、tool 这些消息如何被拼接，工具定义如何被放进上下文，模型输出工具调用时应该使用什么格式。

在 OpenAI 兼容的本地服务里，这个细节尤其重要。因为模型本身并不知道 Python 里的 messages 数组长什么样，它看到的是被服务端模板化之后的文本或 token。如果模板和模型训练时使用的格式不匹配，常见问题包括：

模型忽略 system 约束；
模型不会按预期生成工具调用；
模型把工具参数写成普通自然语言；
工具结果回传后，模型无法理解那是外部执行结果。

所以，本地跑 Agent 时，不只是“接口能调通”就结束了，还要确认模型、服务端和 chat_template 的工具调用格式是一致的。

第二步：让模型知道有哪些工具

一次普通调用只能产生文字。要让 Agent 真正做事，就要让模型能够请求外部工具。

先定义一个最简单的工具：执行一条 shell 命令，并返回输出。

import os
import subprocess


def bash(command: str) -> str:
    """Execute a bash command and return the output."""
    dangerous_keywords = ["rm", "sudo", "shutdown", "reboot", "init", "poweroff"]
    if any(keyword in command for keyword in dangerous_keywords):
        return "Error: command contains dangerous keywords."

    try:
        result = subprocess.run(
            command,
            shell=True,
            cwd=os.getcwd(),
            capture_output=True,
            text=True,
            timeout=120,
        )
        output = (result.stdout + result.stderr).strip()
        return output[:5000] if output else "执行完成，但没有输出。"
    except subprocess.TimeoutExpired:
        return "Error: command execution timed out."
    except Exception as exc:
        return f"Error: an unexpected error occurred - {exc}"

这段代码只是教学用的最小示例。真实项目里不要只靠关键词过滤保护 shell 工具。关键词过滤很容易漏掉危险命令，也很容易误伤正常命令。更好的做法是命令白名单、参数结构化、沙箱隔离、超时限制、权限审批和审计日志一起上。

有了真实函数之后，还要把它描述给模型。模型不能直接读取 Python 函数签名，所以我们要通过 tools 参数告诉它：有哪些工具、叫什么、做什么、需要哪些参数。

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "bash",
            "description": "Execute a shell command and return the output.",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "The shell command to execute.",
                    }
                },
                "required": ["command"],
            },
        },
    }
]

这里的 TOOLS 不是给 Python 解释器看的，而是给模型看的。它告诉模型：如果你需要执行命令，可以生成一个名为 bash 的函数调用，并提供一个 command 字符串。

这个工具定义里也有几个细节值得注意：

name 应该短、稳定、语义明确。后面做工具路由时，Harness 会用它找到真实函数。
description 不是写给用户看的，而是写给模型看的。它会影响模型什么时候选择这个工具。
parameters 本质上是一份 JSON Schema。它让模型知道参数结构，也方便 Harness 做参数校验。
required 很重要。没有它，模型可能会生成缺字段的调用请求。

很多本地 OpenAI 兼容服务会把这份工具定义通过 chat_template 拼进模型上下文里。模型不是“天然知道”Python 里有个 bash 函数，而是通过这份工具说明知道：自己可以请求一个叫 bash 的外部动作。

第三步：模型并不会真的调用工具

这是理解 tool calling 最重要的一点：

调用工具的不是大模型，而是 Harness。模型只是在消息里生成“我想调用哪个工具以及参数是什么”。

我们先写一个只执行一次工具调用的版本：

import json
import os


def run_tool_once(message: str) -> str:
    response = client.chat.completions.create(
        model=MODEL,
        messages=[
            {
                "role": "system",
                "content": f"你是一个命令行助手，工作在当前目录：{os.getcwd()}。",
            },
            {"role": "user", "content": message},
        ],
        tools=TOOLS,
    )

    assistant_message = response.choices[0].message
    if not assistant_message.tool_calls:
        return assistant_message.content or ""

    tool_call = assistant_message.tool_calls[0]
    if tool_call.function.name != "bash":
        return f"Error: unknown tool {tool_call.function.name}"

    try:
        arguments = json.loads(tool_call.function.arguments)
    except json.JSONDecodeError as exc:
        return f"Error: invalid tool arguments - {exc}"

    command = arguments.get("command")
    if not isinstance(command, str) or not command.strip():
        return "Error: missing command."

    tool_response = bash(command)
    return f"模型请求调用 bash({command!r})\n\n工具返回：\n{tool_response}"

模型只生成工具调用意图，Harness 负责解析、校验、执行，再把结果放回对话。

工具调用返回的大致结构可以这样理解：

tool_call.id                  # 这次工具调用的唯一 ID
tool_call.type                # 通常是 "function"
tool_call.function.name       # 模型想调用的工具名，比如 "bash"
tool_call.function.arguments  # 模型生成的 JSON 字符串

这里最容易踩坑的是 arguments。它是字符串，不是已经解析好的 dict。即便你给了 parameters 结构，模型生成的参数也不应该被直接信任：JSON 可能不合法，字段可能缺失，类型也可能不对。所以 Harness 必须先解析、再校验、最后才执行真实函数。

这一步已经很接近 Agent 了：模型可以选择工具，Harness 可以执行工具。

但它还有两个明显问题。

第一，工具调用是一次性的。调用完就结束，模型没有机会继续判断下一步。

第二，模型并不知道工具执行结果。上面代码把 tool_response 直接返回给了用户，但没有把它作为 tool 消息交还给模型。这样模型无法基于真实输出继续推理，也无法把原始工具结果整理成用户真正想要的答案。

所以我们还差一个循环。

第四步：把工具结果放回 messages

一个最小 Agent Loop 的关键，就是每次工具执行后，把结果追加到 messages 里，再让模型继续看。

这时消息历史会变成这样：

user：用户提出任务。
assistant：模型返回 tool_calls，表示想调用工具。
tool：Harness 执行工具，把结果写回。
assistant：模型看到工具结果后，继续回答或继续请求工具。

Agent Loop 的本质，是让工具结果成为下一轮推理的上下文，而不是停留在 Harness 内部。

代码可以这样写：

from typing import Any


SYSTEM_PROMPT = f"""
你是一个命令行助手，工作在当前目录：{os.getcwd()}。
你可以使用 bash 工具完成任务。
调用工具前先思考目标，拿到工具结果后再判断是否需要继续。
"""


def run_bash_tool(tool_call: Any) -> str:
    try:
        arguments = json.loads(tool_call.function.arguments)
    except json.JSONDecodeError as exc:
        return f"Error: invalid tool arguments - {exc}"

    command = arguments.get("command")
    if not isinstance(command, str) or not command.strip():
        return "Error: missing command."

    return bash(command)


def run_loop(user_message: str, max_steps: int = 8) -> str:
    messages: list[dict[str, Any]] = [{"role": "user", "content": user_message}]

    for _ in range(max_steps):
        response = client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "system", "content": SYSTEM_PROMPT}] + messages,
            tools=TOOLS,
        )

        assistant_message = response.choices[0].message

        if not assistant_message.tool_calls:
            final_answer = assistant_message.content or ""
            messages.append({"role": "assistant", "content": final_answer})
            return final_answer

        messages.append(
            {
                "role": "assistant",
                "content": assistant_message.content or "",
                "tool_calls": [
                    {
                        "id": tool_call.id,
                        "type": "function",
                        "function": {
                            "name": tool_call.function.name,
                            "arguments": tool_call.function.arguments,
                        },
                    }
                    for tool_call in assistant_message.tool_calls
                ],
            }
        )

        for tool_call in assistant_message.tool_calls:
            if tool_call.function.name == "bash":
                tool_result = run_bash_tool(tool_call)
            else:
                tool_result = f"Error: unknown tool {tool_call.function.name}"

            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result,
                }
            )

    return "达到最大循环次数，任务仍未完成。"


if __name__ == "__main__":
    user_input = input("Enter your message: ")
    output = run_loop(user_input)
    print("Response from LLM:", output)

这里有几个细节值得保留。

第一，工具结果要用 role: "tool" 写回 messages。不要只在本地变量里保存结果，也不要直接把工具输出打印给用户就结束。工具结果只有回到上下文里，模型下一轮才能继续基于它推理。

第二，tool_call_id 不能写错。当模型返回一个工具调用时，这个调用会有自己的 id。Harness 把工具结果写回 messages 时，要用 tool_call_id 指向原来的调用。这样模型和服务端才能知道：这条 tool 消息对应的是哪一次工具请求。

第三，带有 tool_calls 的 assistant 消息也要写回历史。也就是说，不只是写入工具执行结果，还要先保存模型提出的工具调用请求。否则消息链会断掉，服务端或模型都可能无法正确理解后面的 tool 消息。

第四，要加 max_steps 这样的循环上限。Agent Loop 本质上是一个开放循环，如果模型一直请求工具，就可能无限跑下去。即便是教学代码，也应该保留这个保险丝。

如果上一轮有多个工具调用，就要为每个工具调用追加一条对应的 tool 消息。这一点在下一篇讲多个工具时会更明显。

这个循环为什么就是 Agent 的雏形

现在再回头看这段 run_loop，它其实已经具备了一个 Agent 的基本骨架：

观察：读取用户输入和历史消息。
决策：模型判断是直接回答，还是请求工具。
行动：Harness 按工具名和参数执行真实函数。
反馈：工具结果以 tool 消息回到上下文。
迭代：模型基于新上下文继续决策，直到输出最终答案。

这就是最小的 Agent Loop。

它还很简陋，没有任务状态持久化，没有权限审批，没有丰富的工具系统，也没有自动验证。但它已经把 Agent 最核心的结构跑通了：模型不是孤立地产生文字，而是在 Harness 提供的环境里反复观察、行动、接收反馈。

从工程角度看，Agent Loop 里真正重要的不是 while 或 for 语法，而是每一轮都维护了这份状态：

messages = [
    {"role": "user", "content": "..."},
    {"role": "assistant", "tool_calls": [...]},
    {"role": "tool", "tool_call_id": "...", "content": "..."},
    {"role": "assistant", "content": "..."},
]

只要这份消息栈是连续、可解释、可验证的，模型就能在多轮工具调用中保持任务上下文。

工程上还缺什么

从教学示例走向可用 Agent，中间还有不少工程问题要补。

工具安全。像 bash 这种工具能力很强，风险也很高。生产环境里必须有沙箱、白名单、审批、超时、资源限制和审计。

参数校验。模型生成的 JSON 不一定合法，字段也可能缺失或幻觉出来。Harness 必须验证参数，不能把模型输出直接当成可信输入。

上下文管理。循环次数多了之后，messages 会越来越长。需要压缩历史、保留关键事实、丢弃噪声，并在必要时把长期状态持久化到外部存储。

错误恢复。工具失败不是异常情况，而是 Agent 日常工作的一部分。好的 Harness 应该让模型看懂错误，并给它修复机会。

完成判定。模型说“完成了”不一定真的完成。对代码任务，可以跑测试；对网页任务，可以截图检查；对数据任务，可以校验输出格式。验证越明确，Agent 越可靠。

小结

这篇我们从最简单的一次 LLM 调用出发，逐步加上了三层能力：

用 messages 表达结构化对话历史；
用 tools 告诉模型有哪些外部动作可用；
用 tool 消息把执行结果放回上下文，并形成循环。

Agent 并不是某个神秘对象。至少在最小实现里，它就是一个持续运行的消息循环：模型负责判断下一步，Harness 负责把下一步变成真实动作，再把真实世界的反馈交还给模型。

下一篇继续沿着这段代码往前走：不再只有一个 bash，而是定义更多工具，比如读取文件、写入文件、列目录、发 HTTP 请求；然后讨论当模型返回不同工具名时，Harness 应该如何做工具注册、工具分发、参数校验，以及如何处理一轮里出现多个工具调用。