你的第一个 Harness

Core Insight: Harness 就是一个循环——调用模型、执行 Tool 调用、把结果喂回去、重复。用不到 50 行 Python 就能搞定。理解这个循环,就能看透所有 Agent 框架。

大多数 Agent 教程从框架入手——LangChain、CrewAI、AutoGen。但框架隐藏了机制。从零构建一个 Harness 能让你精确理解发生了什么:Agentic Loop、Context 组装、以及模型的决策过程。理解了这些,所有框架对你来说都是透明的。

完整的 Harness

这是一个带两个 Tool(读文件和写文件)的完整可运行 Harness。直接复制粘贴运行即可。

前置条件

pip install openai
export OPENAI_API_KEY="sk-your-key-here"

代码

#!/usr/bin/env python3
"""A complete agent harness in ~50 lines. Run: python harness.py"""

import json
import os
from openai import OpenAI

client = OpenAI()
MODEL = "gpt-4o-mini"  # Cheap and fast for learning
MAX_TURNS = 15

# --- System prompt ---
SYSTEM = """You are a helpful file assistant. You can read and write files.
When asked to work with files, use the tools provided.
Always confirm what you did after completing a task."""

# --- Tool definitions ---
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Read the contents of a file at the given path",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Path to the file"}
                },
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "Write content to a file (creates or overwrites)",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Path to the file"},
                    "content": {"type": "string", "description": "Content to write"}
                },
                "required": ["path", "content"]
            }
        }
    }
]

# --- Tool execution ---
def execute_tool(name: str, args: dict) -> str:
    try:
        if name == "read_file":
            with open(args["path"], "r") as f:
                return f.read()
        elif name == "write_file":
            os.makedirs(os.path.dirname(args["path"]) or ".", exist_ok=True)
            with open(args["path"], "w") as f:
                f.write(args["content"])
            return f"Wrote {len(args['content'])} chars to {args['path']}"
        else:
            return f"Error: Unknown tool '{name}'"
    except Exception as e:
        return f"Error: {e}"

# --- The tool loop ---
def run(user_message: str) -> str:
    messages = [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": user_message}
    ]

    for turn in range(MAX_TURNS):
        response = client.chat.completions.create(
            model=MODEL, messages=messages, tools=TOOLS
        )
        msg = response.choices[0].message
        messages.append(msg)

        # No tool calls → model is done
        if not msg.tool_calls:
            return msg.content

        # Execute each tool call
        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            print(f"  🔧 {tc.function.name}({args})")
            result = execute_tool(tc.function.name, args)
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": result
            })

    return "Max turns reached."

# --- Main ---
if __name__ == "__main__":
    print("🤖 File Agent (type 'quit' to exit)")
    while True:
        user_input = input("\nYou: ").strip()
        if user_input.lower() in ("quit", "exit"):
            break
        response = run(user_input)
        print(f"\nAgent: {response}")

试试看

python harness.py
🤖 File Agent (type 'quit' to exit)

You: Create a file called hello.txt with a haiku about programming

  🔧 write_file({'path': 'hello.txt', 'content': 'Semicolons fall\nLike rain upon the server\nCompile error: none'})

Agent: I've created hello.txt with a programming haiku!

You: Read it back to me

  🔧 read_file({'path': 'hello.txt'})

Agent: Here's the content of hello.txt:
"Semicolons fall / Like rain upon the server / Compile error: none"

Harness 的结构剖析

整个 Harness 由四个组件构成:

┌────────────────────────────────┐
│         System Prompt          │  ← Agent 的身份定义
├────────────────────────────────┤
│        Tool Definitions        │  ← 能做什么(JSON Schema)
├────────────────────────────────┤
│        Tool Execution          │  ← Tool 实际执行逻辑
├────────────────────────────────┤
│          Tool Loop             │  ← 循环:思考 → 执行 → 观察
└────────────────────────────────┘

System prompt:设定 Agent 的角色和约束。这是性价比最高的部分——改一句话就可能完全改变行为。

Tool 定义:模型读取的 JSON Schema,用来理解有哪些 Tool 可用。模型看不到你的 Python 代码——只能看到描述和参数 Schema。

Tool 执行:实际执行操作的代码。模型输出结构化 JSON,你解析并执行真正的操作。

Tool 循环:编排器。调用模型、检查有没有 Tool 调用、执行、把结果喂回去。循环直到模型返回纯文本。

添加第三个 Tool

想加 Shell 命令?只需加一个 Tool 定义和处理函数:

# Add to TOOLS list:
{
    "type": "function",
    "function": {
        "name": "run_shell",
        "description": "Run a shell command and return stdout/stderr",
        "parameters": {
            "type": "object",
            "properties": {
                "command": {"type": "string", "description": "Shell command to run"}
            },
            "required": ["command"]
        }
    }
}

# Add to execute_tool():
elif name == "run_shell":
    import subprocess
    r = subprocess.run(args["command"], shell=True, capture_output=True, text=True, timeout=30)
    return r.stdout + r.stderr

循环不需要改。模型会自动发现并使用新 Tool。

切换模型

Harness 是模型无关的。切换到 Anthropic 的 Claude 只需要换 client:

from anthropic import Anthropic

client = Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=4096,
    system=SYSTEM,
    messages=messages,
    tools=[{
        "name": t["function"]["name"],
        "description": t["function"]["description"],
        "input_schema": t["function"]["parameters"]
    } for t in TOOLS]
)

# Parse tool calls from response.content blocks
for block in response.content:
    if block.type == "tool_use":
        result = execute_tool(block.name, block.input)

同样的循环,同样的 Tool,不同的模型。

缺什么(以及后续内容)

这个 Harness 能跑,但生产级 Agent 还需要更多:

特性 当前 Harness 生产级 Harness
Memory 无(无状态) MEMORY.md + 每日日志
Context 管理 完整历史 基于优先级的窗口化
错误恢复 基础 try/catch 重试 + 升级
安全 Sandbox 执行
Tool 加载 一次性全加载 按需加载 Skill

后续章节会逐一讲解这些内容。

常见误区

  • 忘记追加 assistant 消息 —— 如果你没在 Tool 结果之前把 msg 加入 messages,模型会丢失它请求了什么的上下文。务必先追加完整的 assistant 响应。
  • Tool 结果序列化错误 —— Tool 结果必须是字符串。如果你的 Tool 返回 dict,用 json.dumps() 转一下。直接返回 Python 对象会崩。
  • 没有迭代上限 —— 没有 MAX_TURNS,模型可能会无限循环,烧 Token。一定要设上限。

延伸阅读