主题
原文：https://www.mstx.cn/harness/418.html · 创建：2026-09-04 · 更新：2026-09-04 · 阅读：1192
从本节开始，你不再只是"写代码"
在1.5节中，你写了一个能跑的Agent。那段代码的核心是一个while循环，加上几次OpenAI API调用。它确实能工作，但它是不可复用的、不可测试的、不可观测的——它只是一段脚本。
从本节开始，你要做一件不同的事情：把那段脚本改造成一个最小但完整的Harness。
什么是最小Harness？它至少具备以下特征：
- 有清晰的模块边界（Runtime、Tool Registry、Memory），而非一堆散落的代码
- 有基本的错误处理和终止条件管理
- 有执行轨迹的记录（哪怕只是结构化日志）
- 可以通过配置而非修改代码来调整行为
本节的目标不是构建一个功能完备的Harness，而是搭建一个正确的骨架。这个骨架在后续章节中会逐步长出血肉——工具系统、记忆管理、评估体系、可观测性——但骨架的结构从第一天起就应该是合理的。
项目结构
先创建项目目录：
my-harness/
├── harness/
│ ├── __init__.py
│ ├── runtime.py # 核心循环
│ ├── tools.py # 工具注册表
│ ├── memory.py # 记忆系统
│ └── config.py # 配置管理
├── examples/
│ └── basic_agent.py # 使用Harness的示例
└── requirements.txt这个结构的核心思想：Harness是独立于具体Agent的。你可以在examples/中写不同的Agent（天气助手、客服机器人、数据分析员），但它们都使用同一套harness/中的基础设施。
第一步：配置管理（config.py）
Harness的所有可调参数都应该集中在配置中，而不是散落在代码各处：
python
# harness/config.py
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class LLMConfig:
"""LLM Provider的配置"""
base_url: str = "https://api.openai.com/v1"
api_key: str = ""
model: str = "gpt-4o-mini"
temperature: float = 0.0 # Agent场景建议用0，降低随机性
max_tokens: int = 1024 # 单次LLM调用的最大输出token数
@dataclass
class RuntimeConfig:
"""Agent循环的配置"""
max_iterations: int = 10 # 最大循环次数
timeout_seconds: int = 60 # 总超时时间
retry_attempts: int = 2 # 工具调用失败时的重试次数
@dataclass
class HarnessConfig:
"""Harness的总体配置"""
llm: LLMConfig = field(default_factory=LLMConfig)
runtime: RuntimeConfig = field(default_factory=RuntimeConfig)使用dataclass的好处：类型清晰、默认值明确、易于扩展。当后续章节加入记忆配置、评估配置时，你可以在HarnessConfig中添加新的配置类。
第二步：工具注册表（tools.py）
工具注册表的最小实现只需要三个能力：定义工具、注册工具、执行工具。
python
# harness/tools.py
import json
import inspect
from typing import Any, Callable, Dict, Optional
from dataclasses import dataclass
@dataclass
class ToolResult:
"""工具执行的统一返回格式"""
success: bool
content: str
error: Optional[str] = None
tool_name: str = ""
duration_ms: int = 0
def to_message(self) -> Dict[str, Any]:
"""转换为LLM可读的消息格式"""
if self.success:
return {"role": "tool", "content": self.content}
else:
return {"role": "tool", "content": f"工具执行失败：{self.error}"}
class ToolRegistry:
"""工具注册表：管理Agent可用的所有工具"""
def __init__(self):
self._tools: Dict[str, Dict[str, Any]] = {}
def register(self, name: str, description: str, func: Callable,
parameters: Dict[str, Any]):
"""注册一个工具"""
self._tools[name] = {
"name": name,
"description": description,
"func": func,
"parameters": parameters,
}
def get_schemas(self) -> list:
"""获取所有工具的Schema，用于传递给LLM"""
schemas = []
for tool in self._tools.values():
schemas.append({
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["parameters"],
}
})
return schemas
def execute(self, name: str, arguments: Dict[str, Any]) -> ToolResult:
"""执行一个工具"""
import time
start = time.time()
if name not in self._tools:
return ToolResult(
success=False,
content="",
error=f"未知工具：{name}",
tool_name=name
)
tool = self._tools[name]
try:
result = tool["func"](**arguments)
duration = int((time.time() - start) * 1000)
return ToolResult(
success=True,
content=str(result),
tool_name=name,
duration_ms=duration
)
except Exception as e:
duration = int((time.time() - start) * 1000)
return ToolResult(
success=False,
content="",
error=str(e),
tool_name=name,
duration_ms=duration
)这个实现有几个值得注意的设计决策：
第一：ToolResult是统一的返回格式。 无论工具成功还是失败，都返回一个ToolResult对象。这让Runtime的处理逻辑变得简单——它不需要关心具体工具的行为，只需要检查success字段。
第二：工具执行永远不抛异常。 所有异常都在execute中被捕获，转换为ToolResult(success=False, error=...)。这是Harness的错误隔离原则：单个工具的失败不应该导致整个Agent崩溃。
第三：get_schemas()方法将工具定义转换为LLM所需的格式。 这是Tool Registry与LLM之间的接口。工具的定义（描述、参数Schema）直接决定了LLM能否正确使用它们。
第三步：记忆系统（memory.py）
最小记忆系统只需要管理短期记忆——当前会话的对话历史：
python
# harness/memory.py
from typing import List, Dict, Any
class MemorySystem:
"""记忆系统：管理Agent的对话历史"""
def __init__(self, system_prompt: str):
self.system_prompt = system_prompt
self.messages: List[Dict[str, Any]] = []
def initialize(self, user_input: str):
"""初始化一次新的会话"""
self.messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_input},
]
def add_llm_message(self, message: Dict[str, Any]):
"""添加LLM的响应到对话历史"""
self.messages.append(message)
def add_tool_result(self, result):
"""添加工具执行结果到对话历史"""
self.messages.append(result.to_message())
def get_context(self) -> List[Dict[str, Any]]:
"""获取当前完整的对话上下文"""
return self.messages
def get_stats(self) -> Dict[str, Any]:
"""获取记忆系统的统计信息"""
total_tokens_estimate = sum(
len(str(m.get("content", ""))) // 4 # 粗略估算：中文约1.5-2字符/token，英文4字符/token
for m in self.messages
)
return {
"message_count": len(self.messages),
"estimated_tokens": total_tokens_estimate,
}这个最小实现做了三件事：初始化会话（放入系统提示词和用户输入）、累积消息（LLM响应和工具结果）、提供上下文（给Runtime调用LLM时使用）。
注意get_stats()方法：它提供了一个粗略的token估算。在后续章节中，我们会用更精确的token计数替换这个估算。但在最小版本中，有一个"大致消耗了多少token"的概念已经很有价值。
第四步：核心循环（runtime.py）
这是Harness的心脏：
python
# harness/runtime.py
import json
import time
from typing import Optional
from openai import OpenAI
from .config import HarnessConfig
from .memory import MemorySystem
from .tools import ToolRegistry
class AgentResult:
"""Agent一次执行的完整结果"""
def __init__(self, status: str, content: str = "",
iterations: int = 0, duration_ms: int = 0,
termination_reason: str = ""):
self.status = status # "success" | "max_iterations" | "timeout" | "error"
self.content = content
self.iterations = iterations
self.duration_ms = duration_ms
self.termination_reason = termination_reason
def to_dict(self):
return {
"status": self.status,
"content": self.content,
"iterations": self.iterations,
"duration_ms": self.duration_ms,
"termination_reason": self.termination_reason,
}
class AgentRuntime:
"""Agent运行时：驱动核心循环"""
def __init__(self, config: HarnessConfig,
tool_registry: ToolRegistry,
memory: MemorySystem):
self.config = config
self.tools = tool_registry
self.memory = memory
self.client = OpenAI(
base_url=config.llm.base_url,
api_key=config.llm.api_key,
)
def run(self, user_input: str) -> AgentResult:
"""执行一次完整的Agent循环"""
start_time = time.time()
# 初始化记忆
self.memory.initialize(user_input)
for iteration in range(self.config.runtime.max_iterations):
# 检查超时
elapsed = time.time() - start_time
if elapsed > self.config.runtime.timeout_seconds:
return AgentResult(
status="timeout",
iterations=iteration,
duration_ms=int(elapsed * 1000),
termination_reason=f"超时（>{self.config.runtime.timeout_seconds}秒）"
)
# 调用LLM（感知 + 思考）
try:
response = self.client.chat.completions.create(
model=self.config.llm.model,
messages=self.memory.get_context(),
tools=self.tools.get_schemas(),
temperature=self.config.llm.temperature,
max_tokens=self.config.llm.max_tokens,
)
except Exception as e:
elapsed = time.time() - start_time
return AgentResult(
status="error",
content=f"LLM调用失败：{e}",
iterations=iteration,
duration_ms=int(elapsed * 1000),
termination_reason="LLM调用异常"
)
message = response.choices[0].message
# 判断：LLM是否要调用工具
if message.tool_calls:
# 将LLM的决策加入记忆
self.memory.add_llm_message(message)
# 执行所有工具调用（行动）
for tool_call in message.tool_calls:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
# 执行工具（带重试）
result = self._execute_with_retry(name, args)
# 将工具结果加入记忆（感知）
self.memory.add_tool_result(result)
else:
# LLM给出了最终回答
elapsed = time.time() - start_time
return AgentResult(
status="success",
content=message.content or "",
iterations=iteration + 1,
duration_ms=int(elapsed * 1000),
termination_reason="正常终止（LLM给出最终回答）"
)
# 达到最大迭代次数
elapsed = time.time() - start_time
return AgentResult(
status="max_iterations",
content="已达到最大循环次数，Agent未能完成任务",
iterations=self.config.runtime.max_iterations,
duration_ms=int(elapsed * 1000),
termination_reason=f"达到最大迭代次数（{self.config.runtime.max_iterations}次）"
)
def _execute_with_retry(self, name: str, args: dict):
"""执行工具，失败时重试"""
result = self.tools.execute(name, args)
if result.success:
return result
# 重试逻辑
for attempt in range(self.config.runtime.retry_attempts):
print(f" [重试 {attempt + 1}/{self.config.runtime.retry_attempts}] 工具 {name} 执行失败：{result.error}")
result = self.tools.execute(name, args)
if result.success:
return result
return result # 重试后仍失败，返回最后一次的结果让我们仔细理解这个Runtime的设计：
第一：AgentResult是一个结构化的结果对象。 它记录了执行的状态（成功/超时/达到上限/错误）、迭代次数、耗时、终止原因。这些信息对于后续的评估和可观测性至关重要。没有结构化的结果，就无法系统性地度量Agent的表现。
第二：终止条件被显式管理。 循环有四个可能的终止路径：LLM给出最终回答（正常）、超时、达到最大迭代次数、LLM调用异常。每条路径都返回一个带有明确termination_reason的AgentResult。
第三：工具调用带重试。 _execute_with_retry方法在工具失败时自动重试（次数由配置决定）。这是Harness对"工具可能偶发失败"这一现实的基本应对。注意：重试是在同一个工具和参数上进行的，不会改变策略——改变策略是LLM的职责（如果重试后仍失败，LLM会看到错误信息并决定下一步）。
第四：循环的每次迭代都包含了完整的"感知-思考-行动"。 LLM调用是"思考"，工具执行是"行动"，将结果加入记忆是"感知"。这个结构在代码中清晰可见。
第五步：组装并运行
现在把这些组件组装起来：
python
# examples/basic_agent.py
import os
from harness.config import HarnessConfig, LLMConfig, RuntimeConfig
from harness.tools import ToolRegistry
from harness.memory import MemorySystem
from harness.runtime import AgentRuntime
# 1. 配置
config = HarnessConfig(
llm=LLMConfig(
api_key=os.getenv("OPENAI_API_KEY", ""),
model="gpt-4o-mini",
temperature=0.0,
),
runtime=RuntimeConfig(
max_iterations=10,
timeout_seconds=60,
retry_attempts=2,
)
)
# 2. 定义工具
def get_weather(city: str) -> str:
"""查询指定城市的天气（模拟数据）"""
mock_data = {
"北京": "晴，26°C",
"上海": "多云，29°C",
"深圳": "雷阵雨，31°C",
}
return mock_data.get(city, f"暂无{city}的天气数据")
def calculate(expression: str) -> str:
"""计算数学表达式"""
try:
result = eval(expression, {"__builtins__": {}}, {})
return f"计算结果：{result}"
except Exception as e:
return f"计算出错：{e}"
# 3. 注册工具
registry = ToolRegistry()
registry.register(
name="get_weather",
description="查询指定城市的天气",
func=get_weather,
parameters={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
)
registry.register(
name="calculate",
description="计算数学表达式",
func=calculate,
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}
)
# 4. 创建记忆系统
memory = MemorySystem(
system_prompt="你是一个有用的助手。你可以查询天气和计算数学表达式。"
)
# 5. 创建Runtime
runtime = AgentRuntime(
config=config,
tool_registry=registry,
memory=memory,
)
# 6. 运行
if __name__ == "__main__":
result = runtime.run("北京今天多少度？这个数字乘以3再加5等于多少？")
print("=" * 50)
print("Agent执行结果：")
print(f" 状态：{result.status}")
print(f" 迭代次数：{result.iterations}")
print(f" 耗时：{result.duration_ms}ms")
print(f" 终止原因：{result.termination_reason}")
print(f" 回答：{result.content}")
print("=" * 50)
# 查看记忆统计
stats = memory.get_stats()
print(f"记忆统计：{stats}")运行：
shell
export OPENAI_API_KEY="sk-..."
python examples/basic_agent.py输出示例：
shell
==================================================
Agent执行结果：
状态：success
迭代次数：3
耗时：4200ms
终止原因：正常终止（LLM给出最终回答）
回答：北京今天26°C。26 × 3 + 5 = 83。
==================================================
记忆统计：{'message_count': 6, 'estimated_tokens': 850}这个最小Harness与1.5节的脚本有什么区别？
| 维度 | 1.5节的脚本 | 本节的Harness |
|---|---|---|
| 配置管理 | 硬编码在代码中 | 集中在HarnessConfig中 |
| 工具管理 | 散落的字典和函数映射 | 独立的ToolRegistry |
| 记忆管理 | 一个消息列表 | 独立的MemorySystem，带统计 |
| 执行结果 | 直接返回文本 | 结构化的AgentResult |
| 错误处理 | 基本没有 | 工具异常捕获、LLM异常捕获、重试机制 |
| 终止条件 | 只有最大迭代次数 | 超时、最大迭代、异常、正常终止 |
| 可复用性 | 每次写新Agent都要复制代码 | Harness是独立的，可被多个Agent复用 |
这些区别的本质是：从"写一段能跑的代码"到"构建一个可维护的系统"。 每个区别单独看都不大，但它们合在一起，意味着你可以在这个Harness上继续构建更复杂的能力，而1.5节的脚本则会随着功能增加而迅速失控。
本节小结
你现在有了一个最小但结构正确的Harness。它包含：
config.py：集中管理LLM和Runtime的配置tools.py：ToolRegistry管理工具的定义、注册和执行，ToolResult统一工具返回格式memory.py：MemorySystem管理对话历史，提供上下文和统计runtime.py：AgentRuntime驱动核心循环，管理终止条件和错误处理
这个骨架的关键特征是模块边界清晰。当后续章节加入更多能力时——工具沙箱（第4章）、长期记忆（第5章）、评估体系（第7章）、可观测性（第9章）——它们会作为新的模块加入，而不是在现有代码上打补丁。
在下一节中，我们将为这个Harness添加结构化输出与解析能力，让Agent不仅能调用工具，还能以受控的格式返回结果。