Harness 设计：不是外壳，而是 Agent 系统的控制结构

1. 为什么要重新理解 Harness

很多人第一次接触 harness，会把它理解成：

  • 一个运行时外壳
  • 一个更大的 prompt
  • 一个会话续接层
  • 一个帮 Agent 跑起来的脚手架

这些理解不是完全错，但都偏窄。如果只这样理解，你会看不见 harness 真正关键的部分：谁控制系统、任务一轮轮怎么推进、哪些动作被允许、出错后如何恢复、上下文预算如何治理。

有一个发人深省的数据点：在 Terminal Bench 2.0 上，Claude Opus 4.6 跑在 Claude Code 里的得分，远低于同一个模型跑在定制 Harness 里的得分。有团队仅通过优化 Harness 设计，就把一个 coding agent 从排行榜第 30 名提升到第 5 名。Harness 改变的不是模型能力，而是模型能力被激活和约束的方式。

这也意味着：当 Agent 在真实工程环境失败时，大多数情况下不是"模型不够强"，而是 Harness 没有设计好。

2. 一个核心公式：Agent = Model + Harness

这是理解 Harness Engineering 的起点：

# Agent = Model + Harness
# 裸模型只能预测 token，Harness 赋予它 Agent 能力

class Agent:
    def __init__(self, model, harness):
        self.model = model      # LLM 模型（负责推理）
        self.harness = harness  # 控制结构（负责执行）
    
    def run(self, task):
        state = self.harness.initialize(task)
        while not state.is_finished():
            action = self.model.decide(state)
            result = self.harness.execute(action)
            state = self.harness.update(state, result)
        return state.output

一个裸的模型不是 Agent。它只能预测下一个 token。

变成 Agent，需要 Harness 给它：

  • 状态（知道自己在哪个阶段）
  • 工具执行（能对世界产生实际影响）
  • 反馈循环（能根据结果调整行动）
  • 可强制执行的约束（在边界内而非无限自由地行动）
# Agent = Model + Harness
# 裸模型只能预测 token，Harness 赋予它 Agent 能力

class Agent:
    def __init__(self, model, harness):
        self.model = model      # LLM 模型（负责推理）
        self.harness = harness  # 控制结构（负责执行）
    
    def run(self, task):
        state = self.harness.initialize(task)
        while not state.is_finished():
            action = self.model.decide(state)
            result = self.harness.execute(action)
            state = self.harness.update(state, result)
        return state.output

模型负责"想什么"，Harness 负责"怎么做和做不做"。

3. Harness 到底是什么

结合 Claude Code 这类系统更准确的语境，一个实用定义是：

Harness = 让 Agent 系统保持有界、可问责、可恢复的控制结构

它通常包含以下几部分：

所以 Harness 不是"一个组件"，更像是一整套 runtime skeleton（运行时骨架）。

4. 七层控制结构详解

4.1 第一层：控制平面（Control Plane）

Prompt 不是人格设定，而是控制平面的组成部分。

很多人把 system prompt 当成"给模型的任务说明"。但在工程语境里，prompt 承担的是更接近控制规则的角色：

interface Harness {
  // 状态管理
  state: AgentState;           // 知道自己在哪个阶段
  initialize(task: Task): AgentState;
  
  // 工具执行
  tools: ToolRegistry;         // 能对世界产生实际影响
  execute(tool: string, args: any): Result;
  
  // 反馈循环
  feedback(result: Result): void;  // 能根据结果调整行动
  
  // 可强制执行的约束
  constraints: Constraint[];   // 在边界内而非无限自由地行动
  check(action: Action): boolean; // 每次执行前检查约束
}

一个工程化的控制平面示例：

# System Prompt 作为控制平面
你在回答前必须确认：
1. 当前正在执行的操作是什么？
2. 这个操作是否在允许的区域内？
3. 有没有更好的实现方式？
4. 如果失败，回退策略是什么？

你只能使用提供的工具，不能直接修改文件系统。
如果遇到边界情况，必须先解释你的分析。

关键洞察：控制平面的每一行规则，最好都能追溯到一次具体的失败。

4.2 第二层：Query Loop（主循环）

Harness 的第二个核心，是系统如何一轮一轮运转。

这是 Agent 的心跳——不在模型里，而在 Harness 驱动的循环里：

# 工程化的控制平面：可编程、可审计、可测试

class ControlPlane:
    def __init__(self):
        self.rules = []
        self.audit_log = []
    
    def add_rule(self, condition: str, action: str):
        """每条规则都对应一次真实项目失败"""
        self.rules.append({
            "condition": condition,
            "action": action,
            "added_by": "incident-0425",
        })
    
    def evaluate(self, context) -> list[str]:
        triggered = []
        for rule in self.rules:
            if self._matches(rule["condition"], context):
                triggered.append(rule["action"])
                self.audit_log.append({
                    "rule": rule,
                    "context": context,
                    "timestamp": now()
                })
        return triggered

这个循环看起来简单，但真实系统的大量稳定性由它决定。循环里的每一个判断节点，都是 Harness 的工作，不是模型的工作。

4.3 第三层：工具、权限与中断

工具不是"给模型一双手"，而是"在权限策略下借给它一双手"。

参考 Claude Code 的三层权限模型：

# Agent 的心跳——Harness 驱动的主循环

async def agent_loop(agent, task, max_steps=50):
    state = await initialize(task)
    step = 0
    
    while step < max_steps:
        step += 1
        
        # 1. 上下文构建（Harness 的工作）
        messages = build_context(state)
        
        # 2. 模型推理（模型的工作）
        response = await model.generate(messages)
        
        # 3. 解析输出（Harness 的工作）
        actions = parse_actions(response)
        
        # 4. 权限检查（Harness 的工作）
        allowed = []
        for action in actions:
            if check_permission(action, state):
                allowed.append(action)
            else:
                await notify_user(f"Action blocked: {action}")
        
        # 5. 执行工具（Harness 的工作）
        for action in allowed:
            result = await execute_tool(action)
            state = apply_result(state, result)
        
        # 6. 终止条件检查（Harness 的工作）
        if should_terminate(state):
            break
    
    return state

实现权限分级的代码结构：

三层权限模型 (Claude Code style):
┌─────────────────────┬─────────────────┬─────────────────┐
│     Level 1         │   Level 2       │   Level 3       │
│   ⌘ Enter          │   Auto-approve  │   Always allow   │
│  (手动确认)         │  (安全操作)      │  (只读操作)      │
├─────────────────────┼─────────────────┼─────────────────┤
│  bash write         │   ls, cat       │   git pull      │
│  file write         │   grep, find    │   npm install   │
│  package install    │   file read     │   build         │
└─────────────────────┴─────────────────┴─────────────────┘

关键设计原则：权限检查必须在 Harness 层执行，不能依赖模型自我约束。被妥协的模型无法通过"讲道理"绕过权限门。

4.4 第四层：Context Governance（上下文治理）

不是"省 token"，而是对系统注意力做预算分配。

Harness 视角下的上下文不只是一段聊天记录，而是一套分层管理的信息体系：

class PermissionGuard:
    """权限检查必须在 Harness 层执行，不能依赖模型自我约束"""
    LEVEL3 = {"read", "ls", "cat", "grep", "find", "stat"}
    LEVEL2 = {"write_file", "edit_file", "mkdir", "cp", "mv"}
    LEVEL1 = {"bash", "package_install", "delete", "chmod"}
    
    def check(self, tool_name: str, context) -> Permission:
        if tool_name in self.LEVEL3:
            return Permission.ALLOW
        if tool_name in self.LEVEL2:
            return Permission.ALLOW if self._is_safe(context) else Permission.ASK
        if tool_name in self.LEVEL1:
            return Permission.ASK
        return Permission.DENY  # 默认拒绝

上下文治理的核心机制：

上下文分层管理体系：
┌──────────────────────────────────────┐
│ Layer 1: 核心指令（系统 prompt）       │ ← 始终保留
│ 角色定义、核心约束                      │
├──────────────────────────────────────┤
│ Layer 2: 当前任务上下文                │ ← 最近 N 轮
│ 当前目标、中间结果                      │
├──────────────────────────────────────┤
│ Layer 3: 长时记忆（摘要/记忆）         │ ← 压缩后注入
│ 历史事实、关键决策                      │
├──────────────────────────────────────┤
│ Layer 4: 参考材料                     │ ← 按需加载
│ 项目文件、API 文档                     │
└──────────────────────────────────────┘

4.5 第五层：错误与恢复路径

真正成熟的 Harness，不以"不会失败"为前提，而以"失败后还能继续工作"为目标。

常见失败类型及对应恢复策略：

interface Harness {
  // 状态管理
  state: AgentState;           // 知道自己在哪个阶段
  initialize(task: Task): AgentState;
  
  // 工具执行
  tools: ToolRegistry;         // 能对世界产生实际影响
  execute(tool: string, args: any): Result;
  
  // 反馈循环
  feedback(result: Result): void;  // 能根据结果调整行动
  
  // 可强制执行的约束
  constraints: Constraint[];   // 在边界内而非无限自由地行动
  check(action: Action): boolean; // 每次执行前检查约束
}

恢复路径的实现：

class ContextGovernance:
    MAX_TOKENS = 100_000
    MIN_BUDGET = 10_000
    
    def build(self, state):
        messages = []
        budget = self.MAX_TOKENS
        
        # Layer 1: 核心指令
        messages.append(self.get_system_prompt())
        budget -= self.count(messages[-1])
        
        # Layer 2: 最近上下文（保留 N 轮）
        for msg in state.messages[-10:]:
            if budget <= 0:
                break
            messages.append(msg)
            budget -= self.count(msg)
        
        # 自动摘要
        if budget < self.MIN_BUDGET:
            summary = self.summarize(state.older_messages)
            messages.insert(1, summary)
        
        # 工具结果预算控制
        for result in state.tool_results:
            if budget <= 0:
                result = self.truncate(result)
            messages.append(result)
        
        return messages

检查点策略：在哪些节点保存检查点？

你在回答前必须确认：
1. 当前正在执行的操作是什么？
2. 这个操作是否在允许的区域内？
3. 有没有更好的实现方式？
4. 如果失败，回退策略是什么？

你只能使用提供的工具，不能直接修改文件系统。
如果遇到边界情况，必须先解释你的分析。

4.6 第六层：验证与分工

长任务系统的危险，不只是模型会犯错，还因为它会把错误一路带下去。

验证机制可以分几个层级：

常见失败类型及恢复策略：
┌──────────────┬──────────────────────────────┐
│  工具超时     │ 重试 + 指数退避               │
│  API 错误    │ 切换模型 / 重试               │
│  权限拒绝     │ 通知用户 + 要求授权            │
│  token 超限  │ 压缩上下文 + 继续              │
│  模型幻觉     │ L2 验证 + 纠正                │
│  循环崩溃     │ 最大步数限制 + 回退检查点       │
│  状态丢失     │ 从最新检查点恢复               │
└──────────────┴──────────────────────────────┘

用 Hook 实现 L2 验证：

class RecoveryManager:
    def __init__(self):
        self.policies = {
            "tool_timeout": RetryPolicy(3, "exponential"),
            "api_error": RetryPolicy(2, "linear"),
        }
        self.checkpoints = CheckpointStore()
    
    async def run(self, agent, task):
        cp = await self.checkpoints.load(task.id)
        state = cp.state if cp else await agent.initialize(task)
        
        try:
            return await agent.execute(state)
        except ToolTimeoutError:
            return await self._retry(agent, state, "tool_timeout")
        except TokenLimitError:
            return await agent.execute(await self._compress(state))
        except LoopDetectedError:
            fb = await self.checkpoints.load_latest(task.id)
            return await agent.execute(fb.state)
    
    async def save_checkpoint(self, task_id, state, node):
        await self.checkpoints.save(
            Checkpoint(task_id, node, state, timestamp=now()))

多 Agent 分工不是功能炫技，而是对不稳定模型的一种治理方式：

# 工程化的控制平面：可编程、可审计、可测试

class ControlPlane:
    def __init__(self):
        self.rules = []
        self.audit_log = []
    
    def add_rule(self, condition: str, action: str):
        """每条规则都对应一次真实项目失败"""
        self.rules.append({
            "condition": condition,
            "action": action,
            "added_by": "incident-0425",  # 可追溯到具体 incident
        })
    
    def evaluate(self, context) -> list[str]:
        """评估当前上下文，返回触发的规则"""
        triggered = []
        for rule in self.rules:
            if self._matches(rule["condition"], context):
                triggered.append(rule["action"])
                self.audit_log.append({
                    "rule": rule,
                    "context": context,
                    "timestamp": now()
                })
        return triggered
    
    def test_coverage(self, scenarios):
        """每条控制规则都应该有测试覆盖"""
        pass

4.7 第七层：团队落地与本地治理

Harness 真正成熟的标志，不是你自己会用，而是团队可以稳定复用。

团队落地的四个层级：

验证机制层级：
┌──────────────┬──────────────────────────────┐
│  L0: 编译检查  │ 语法正确性、类型安全           │
│  L1: 静态分析  │ Lint、代码风格、安全扫描        │
│  L2: 功能验证  │ 测试通过、diff 合理性          │
│  L3: 集成检查  │ 不破坏已有功能、API 兼容        │
│  L4: 人工审核  │ 关键变更需要人工确认            │
└──────────────┴──────────────────────────────┘

一个可执行的团队清单示例：

class ValidationHook:
    def __init__(self):
        self.validators = []
    
    def add(self, name, fn, level="L2"):
        self.validators.append({"name": name, "fn": fn, "level": level})
    
    async def before_apply(self, changes):
        for v in self.validators:
            result = await v["fn"](changes)
            if not result.passed and result.severity == "blocking":
                return result
        return ValidationResult(passed=True)
    
    async def after_apply(self, changes):
        for v in self.validators:
            result = await v["fn"](changes)
            if not result.passed:
                return result
        return ValidationResult(passed=True)

5. 从零手写一个最小 Harness

理论已经足够，现在来看一个可运行的最小 Harness 实现：

多 Agent 分工：对不稳定模型的一种治理方式
┌──────────────┬──────────────┬──────────────┐
│  Planner     │  Coder       │  Reviewer     │
│  拆解任务     │  生成代码     │  审查输出     │
│  制定方案     │  执行改动     │  验证质量     │
│  分配资源     │  跑测试       │  报告问题     │
└──────────────┴──────────────┴──────────────┘
每个 Agent 在各自的 Harness 约束内工作。

运行后你会看到 Harness 的完整工作流：控制平面约束行为 → Query Loop 驱动推进 → 权限门把关执行 → 结果写回继续下一轮。

6. Claude Code 的 Harness 架构解析

Claude Code 是目前最好的 Harness Engineering 教学案例。当开发者反编译分析它时，发现的不是"带文件权限的聊天机器人"，而是一套精密的控制结构：

团队落地四个层级：
┌──────────────┬──────────────────────────────┐
│  L1: 个人实验  │ 开发者自己写 prompt + 工具      │
│  L2: 团队共享  │ 共享 AGENTS.md + 工具集        │
│  L3: 平台化    │ 统一 Harness 框架 + 监控       │
│  L4: 持续改进  │ 失败自动入库 + 规则持续迭代      │
└──────────────┴──────────────────────────────┘

Claude Code 的核心设计哲学（来自论文）：

模型负责推理，Harness 负责执行动作。

模型永远不会直接接触文件系统——Harness 决定读取是否被允许、结果如何处理、多少内容能进入下一次 prompt。

在 src/query.ts:365 之后，每次模型调用前，循环还会处理：消息切片、工具结果预算、历史截断、microcompact、上下文折叠和自动 compact。这些都在 Harness 里，不在模型里。

7. Harness 的"棘轮机制"：把每一次失败变成规则

Harness Engineering 中最重要的工作习惯：把 Agent 的每次失败当作永久信号，而不是一次性事故。

# Agent 的心跳——Harness 驱动的主循环

async def agent_loop(agent, task, max_steps=50):
    state = await initialize(task)
    step = 0
    
    while step < max_steps:
        step += 1
        
        # 1. 上下文构建（Harness 的工作）
        messages = build_context(state)
        
        # 2. 模型推理（模型的工作）
        response = await model.generate(messages)
        
        # 3. 解析输出（Harness 的工作）
        actions = parse_actions(response)
        
        # 4. 权限检查（Harness 的工作）
        allowed = []
        for action in actions:
            if check_permission(action, state):
                allowed.append(action)
            else:
                await notify_user(f"Action blocked: {action}")
        
        # 5. 执行工具（Harness 的工作）
        for action in allowed:
            result = await execute_tool(action)
            state = apply_result(state, result)
        
        # 6. 终止条件检查（Harness 的工作）
        if should_terminate(state):
            break
    
    return state

# 循环里的每一个判断节点，都是 Harness 的工作

实际操作示例：

┌─────────────────────────────────────────────┐
│ 三层权限模型 (Claude Code style)             │
├─────────────┬───────────────┬───────────────┤
│   Level 1   │   Level 2     │   Level 3     │
│  ⌘ Enter    │   Auto-approve│  Always allow │
│  (手动确认)  │  (安全操作)    │  (只读操作)    │
├─────────────┼───────────────┼───────────────┤
│  bash write │   ls, cat     │   git pull    │
│  file write │   grep, find  │   npm install │
│  package    │   file read   │   build       │
│  install    │   (非敏感)     │   test        │
└─────────────┴───────────────┴───────────────┘

这就是 Harness Engineering 是一种工程纪律而不是一个框架的原因——最好的 AGENTS.md 是根据你自己的失败历史写出来的，不能直接下载别人的。

8. Harness vs Workflow vs Framework vs Skill

这几个概念常常被混用，实际上各自解决不同层次的问题：

更准确的层级关系：

# 权限分级的实现：检查必须在 Harness 层执行

class PermissionGuard:
    ALLOWLIST_LEVEL3 = {"read", "ls", "cat", "grep", "find", "stat"}
    ALLOWLIST_LEVEL2 = {"write_file", "edit_file", "mkdir", "cp", "mv"}
    LEVEL1_TOOLS = {"bash", "package_install", "delete", "chmod"}
    
    def check(self, tool_name: str, context) -> Permission:
        # 不能依赖模型自我约束
        # 被妥协的模型无法通过"讲道理"绕过权限门
        
        if tool_name in self.ALLOWLIST_LEVEL3:
            return Permission.ALLOW
        
        if tool_name in self.ALLOWLIST_LEVEL2:
            if self._is_safe_context(context):
                return Permission.ALLOW
            return Permission.ASK_USER
        
        if tool_name in self.LEVEL1_TOOLS:
            return Permission.ASK_USER
        
        return Permission.DENY  # 默认拒绝
    
    def _is_safe_context(self, context) -> bool:
        """判断当前上下文是否安全"""
        return context.danger_level < THRESHOLD

所以：

  • Workflow 讲顺序
  • Skill 讲局部方法
  • Framework 讲实现抽象
  • Harness 讲秩序与治理

9. 实战：为一个代码 Agent 设计完整 Harness

场景：为一个用于重构 Python 项目的 Agent 设计 Harness。

Step 1：定义控制平面

# AGENTS.md - 团队 Harness 规范
## 控制平面
- [ ] 所有约束规则有明确的失败案例出处
- [ ] 每个 rule 有对应测试
## 工具权限
- [ ] Level 1-3 工具列表已定义
- [ ] 自动批准规则明确
## 上下文治理
- [ ] 最大 token 预算已设置
- [ ] 自动摘要触发条件已配置
## 错误恢复
- [ ] 重试策略已配置
- [ ] 检查点策略已定义

Step 2：设计 Query Loop

"""
最小 Harness 实现：可运行的 Agent 控制结构
"""
import asyncio
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Any

class Permission(Enum):
    ALLOW = "allow"
    ASK = "ask_user"
    DENY = "deny"

@dataclass
class Rule:
    condition: str
    action: str

class ControlPlane:
    """控制平面：约束、规则、边界"""
    def __init__(self):
        self.rules = []
    
    def add_rule(self, condition: str, action: str):
        self.rules.append(Rule(condition, action))
    
    def evaluate(self, context: dict) -> list[str]:
        return [r.action for r in self.rules if r.condition in str(context)]

@dataclass
class Tool:
    name: str
    fn: Callable
    permission: Permission = Permission.ALLOW

class ToolRegistry:
    def __init__(self):
        self.tools = {}
    def register(self, tool: Tool):
        self.tools[tool.name] = tool
    def execute(self, name: str, **kwargs):
        return self.tools[name].fn(**kwargs)

class ContextManager:
    MAX_TOKENS = 10000
    def __init__(self):
        self.messages = []
        self.token_count = 0
    def add(self, msg: str):
        self.messages.append(msg)
        self.token_count += len(msg) // 2
    def build(self) -> list[str]:
        if self.token_count <= self.MAX_TOKENS:
            return self.messages
        system = self.messages[0]
        recent = self.messages[-3:]
        summary = f"[Summarized {len(self.messages)-4} msgs]"
        return [system, summary] + recent

class CheckpointStore:
    def __init__(self):
        self.store = {}
    def save(self, key, state):
        self.store[key] = dict(state)
    def load(self, key) -> dict | None:
        return self.store.get(key)

@dataclass
class AgentState:
    task: str = ""
    messages: list = field(default_factory=list)
    step: int = 0
    result: Any = None
    finished: bool = False

class Harness:
    def __init__(self, model_fn: Callable):
        self.model = model_fn
        self.control = ControlPlane()
        self.tools = ToolRegistry()
        self.context = ContextManager()
        self.checkpoint = CheckpointStore()
    
    async def run(self, task: str, max_steps: int = 20) -> AgentState:
        state = AgentState(task=task)
        for step in range(max_steps):
            state.step = step
            context = self.context.build()
            triggered = self.control.evaluate({"task": task, "step": step})
            if triggered:
                context = triggered + context
            response = self.model(context)
            action = response.strip()
            result = self.tools.execute(action, state=state)
            state.result = result
            self.context.add(f"Step {step}: {result}")
            self.checkpoint.save(f"step-{step}", {
                "task": task, "step": step, "result": result
            })
            if result == "DONE":
                state.finished = True
                break
        return state

# 运行示例
async def main():
    harness = Harness(lambda ctx: "search")
    harness.control.add_rule("search", "verify_source")
    harness.tools.register(Tool("search", lambda **k: "DONE"))
    result = await harness.run("Find info about X")
    print(f"Finished: {result.finished}")

if __name__ == "__main__":
    asyncio.run(main())

Step 3：设置验证 Hook

Claude Code 的 Harness 架构：
┌──────────────────────────────────────────┐
│            Claude Code Harness            │
├──────────────────────────────────────────┤
│  Control Plane: system prompt + AGENTS.md │
│  Query Loop: src/query.ts:365             │
│  ├─ 消息切片 + 工具结果预算               │
│  ├─ 历史截断 + microcompact              │
│  └─ 上下文折叠 + auto compact             │
│  Permission: ⌘ Enter / Auto / Always      │
│  Tool Execution: 模型不直接接触文件系统    │
└──────────────────────────────────────────┘

Step 4：定义团队规范

Harness Engineering 工作习惯：
每一次 Agent 失败 → AGENTS.md → 永久规则

10. 常见症状与 Harness 诊断

当 Agent 系统出现以下问题，往往不是"模型不够强"，而是 Harness 出了问题：

11. 设计 Harness 时最先要回答的 6 个问题

在动手写代码之前，先问清楚这 6 个问题：

# 操作示例
# 失败: Agent 删除了 node_modules
deny: delete node_modules/

# 失败: Agent 修改了 src/config/database.ts
ask: modify src/config/
review_required: src/config/database.ts

# 失败: Agent 升级了依赖导致兼容性问题
deny: upgrade dependencies without test
require: explicit version pinning

如果这 6 个问题没有答案，系统通常还没真正进入 Harness 设计阶段。

12. 本节总结

核心要点

什么是 Harness：

让 Agent 系统保持有界、可问责、可恢复的控制结构。不是给模型包一层，而是把一个天然不稳定的模型放进有控制平面、权限边界、上下文治理、恢复路径和验证纪律的工程秩序里。

七层结构：

关键判断：

当 Agent 反复失败时，先问"是 Harness 问题"，再问"是模型问题"。

今天模型之间的差距，远小于 Harness 设计好坏带来的差距。

参考资料：