LLM - Harness Engineering

本文档系统阐述 Harness Engineering 的完整技术体系，涵盖 Harness 概念架构、环境隔离、工具 Mock、状态检查点、错误注入、执行控制、批量回归、影子灰度、轨迹复现、Skills 管理、安全与性能 Harness 及完整 Test Harness 的全链路知识

1. Harness 概念与架构

1.1 定义

Harness 是包裹 Agent 的"测试与运行框架"--它不改变 Agent 本身逻辑，而是在外部提供受控环境、监控、错误处理和可复现性保障。Harness 是 Agent 的"保险带 + 仪表盘 + 飞行记录器"，让 Agent 可测试、可监控、可回溯、可恢复。

1.2 为什么需要 Harness

Agent 上线前的核心挑战:

  挑战1: 不可预测性
    LLM 输出非确定,同一输入可能产生不同结果
    -> 需要: 可复现的测试环境

  挑战2: 工具副作用
    Agent 调用外部工具(发邮件/写数据库)可能产生不可逆操作
    -> 需要: 工具 Mock 与沙箱隔离

  挑战3: 长任务中断
    深度任务可能运行数小时,中断后无法恢复
    -> 需要: 状态检查点与恢复机制

  挑战4: 性能不可控
    Token 消耗、延迟、成本可能失控
    -> 需要: 预算控制与性能监控

  挑战5: 安全风险
    Agent 可能生成有害内容或泄露 PII
    -> 需要: 安全 Harness 过滤与审计

  Harness 的角色: 不改 Agent 逻辑,在外部解决以上所有挑战

1.3 九种 Harness 类型

┌──────────────────────────────────────────────────────────────┐
│                    Harness 类型体系 (9 种)                     │
├──────────────┬───────────────────────────────────────────────┤
│ ENVIRONMENT  │ 隔离执行环境、沙箱、路径控制、资源管理          │
│ TOOL         │ Mock 工具、注入行为、场景预设                   │
│ TASK         │ 任务定义、成功标准、验收条件                    │
│ EXECUTION    │ 超时、取消、步进、迭代限制                      │
│ STATE        │ 检查点、快照、回滚、时间旅行                    │
│ RECOVERY     │ 错误恢复、重试、熔断器                          │
│ SAFETY       │ 输出过滤、PII 脱敏、注入防护、审计              │
│ EVALUATION   │ 质量评估、轨迹评测、指标计算                    │
│ REPLAY       │ 执行复现、确定性回放、回归测试                  │
└──────────────┴───────────────────────────────────────────────┘

1.4 BaseHarness 抽象基类

from abc import ABC, abstractmethod

class BaseHarness(ABC):
    """Harness 抽象基类:所有 Harness 类型的父类"""

    def __init__(self, config: HarnessConfig):
        self.config = config
        self.enabled = config.enabled

    @abstractmethod
    def setup(self, context: ExecutionContext) -> None:
        """执行前设置(如创建沙箱、注入 Mock)"""
        pass

    @abstractmethod
    def teardown(self, context: ExecutionContext) -> None:
        """执行后清理(如销毁沙箱、恢复状态)"""
        pass

    @abstractmethod
    def wrap(self, agent_func: Callable) -> Callable:
        """包装 Agent 函数,注入 Harness 逻辑"""
        pass

1.5 Harness 编排器

┌──────────────────────────────────────────────────────────┐
│                HarnessOrchestrator（编排器）               │
│                                                          │
│   管理多个 Harness 的生命周期:                            │
│   1. 按顺序执行 setup()                                   │
│   2. 包装 Agent 函数                                      │
│   3. 执行 Agent                                           │
│   4. 按逆序执行 teardown()                                │
│                                                          │
│   ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐   │
│   │环境Harness│→│工具Harness│→│执行Harness│→│安全Harness│   │
│   │ setup    │ │ setup    │ │ setup    │ │ setup    │   │
│   └──────────┘ └──────────┘ └──────────┘ └──────────┘   │
│        ↓            ↓            ↓            ↓          │
│   ┌──────────────────────────────────────────────────┐   │
│   │          Agent 执行 (被所有 Harness 包裹)          │   │
│   └──────────────────────────────────────────────────┘   │
│        ↓            ↓            ↓            ↓          │
│   ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐   │
│   │环境Harness│←│工具Harness│←│执行Harness│←│安全Harness│   │
│   │ teardown │ │ teardown │ │ teardown │ │ teardown │   │
│   └──────────┘ └──────────┘ └──────────┘ └──────────┘   │
└──────────────────────────────────────────────────────────┘

class HarnessOrchestrator:
    """Harness 编排器:管理多 Harness 的执行顺序"""

    def __init__(self):
        self.harnesses: list[BaseHarness] = []

    def register(self, harness: BaseHarness):
        """注册 Harness"""
        self.harnesses.append(harness)

    async def run(self, agent_func: Callable, task: str) -> ExecutionResult:
        """执行被所有 Harness 包裹的 Agent"""
        context = ExecutionContext(task=task)

        # 1. 按顺序 setup
        for h in self.harnesses:
            if h.enabled:
                h.setup(context)

        # 2. 逐层包装 Agent 函数
        wrapped = agent_func
        for h in reversed(self.harnesses):
            if h.enabled:
                wrapped = h.wrap(wrapped)

        # 3. 执行
        try:
            result = await wrapped(context)
        finally:
            # 4. 逆序 teardown
            for h in reversed(self.harnesses):
                if h.enabled:
                    h.teardown(context)

        return result

1.6 与其他概念的关联

• -> 环境 Harness：提供隔离的执行环境

• -> 工具 Mock 与注入：控制工具行为

• -> 执行控制：管理执行边界

• -> 完整 Test Harness：所有 Harness 的集成

---

2. 环境 Harness

2.1 定义

环境 Harness（EnvironmentHarness）负责为 Agent 执行创建隔离的运行环境，包括沙箱目录、环境变量、资源限制和路径控制，确保 Agent 操作不会影响宿主系统。

对应 Demo: demos/06_HarnessEngineering/02_环境Harness.py

2.2 核心能力

┌──────────────────────────────────────────────────────────┐
│              EnvironmentHarness 核心能力                   │
├──────────────┬───────────────────────────────────────────┤
│ 沙箱目录      │ 创建临时工作目录,Agent 文件操作限制在此    │
│ 路径控制      │ 所有文件路径重定向到沙箱,防止越权访问       │
│ 环境变量      │ 注入测试用环境变量,隔离生产配置             │
│ 资源限制      │ CPU/内存/磁盘/网络 限制,防止资源耗尽        │
│ 清理机制      │ 执行完成后自动清理沙箱,不留痕迹             │
└──────────────┴───────────────────────────────────────────┘

2.3 沙箱实现

import tempfile
import os
import shutil

class EnvironmentHarness(BaseHarness):
    """环境 Harness:沙箱隔离与资源管理"""

    def setup(self, context: ExecutionContext) -> None:
        """创建沙箱环境"""
        # 1. 创建临时工作目录
        self.sandbox_dir = tempfile.mkdtemp(prefix="agent_sandbox_")
        context.env_vars["SANDBOX_DIR"] = self.sandbox_dir

        # 2. 注入环境变量
        context.env_vars["AGENT_MODE"] = "test"
        context.env_vars["API_TIMEOUT"] = "5"

        # 3. 设置资源限制
        context.resources = ResourceLimit(
            max_memory="512MB",
            max_cpu_time=60,
            max_file_size="10MB",
            network_enabled=False  # 测试环境默认禁网
        )

    def teardown(self, context: ExecutionContext) -> None:
        """清理沙箱环境"""
        if hasattr(self, "sandbox_dir") and os.path.exists(self.sandbox_dir):
            shutil.rmtree(self.sandbox_dir)  # 递归删除沙箱

2.4 资源限制类型

class ResourceLimit(Enum):
    """资源限制类型"""
    MAX_MEMORY = "max_memory"        # 最大内存
    MAX_CPU_TIME = "max_cpu_time"    # 最大 CPU 时间
    MAX_FILE_SIZE = "max_file_size"  # 最大文件大小
    NETWORK_ENABLED = "network"      # 是否允许网络
    MAX_PROCESSES = "max_processes"  # 最大进程数

2.5 与其他概念的关联

• <- Harness 概念与架构：环境 Harness 是九种类型之一

• -> 工具 Mock 与注入：沙箱中工具行为可被 Mock

• -> 安全 Harness：沙箱是安全隔离的基础

---

3. 工具 Mock 与注入

3.1 定义

工具 Mock 与注入（ToolMock）是在测试环境中替换 Agent 调用的真实工具，用预设的行为和返回值模拟工具响应，避免产生真实副作用（如发送邮件、修改数据库），同时支持场景预设和故障模拟。

对应 Demo: demos/06_HarnessEngineering/03_工具Mock与注入.py

3.2 五种 Mock 行为

┌──────────────┬───────────────────────────────────────────────┐
│ RETURN       │ 直接返回预设值(最常用)                         │
│              │ mock("search").return("结果")                 │
├──────────────┼───────────────────────────────────────────────┤
│ RAISE        │ 抛出指定异常(测试错误处理)                     │
│              │ mock("search").raise(TimeoutError())          │
├──────────────┼───────────────────────────────────────────────┤
│ DELAY        │ 延迟指定时间后返回(测试超时)                   │
│              │ mock("search").delay(5).return("结果")        │
├──────────────┼───────────────────────────────────────────────┤
│ SEQUENCE     │ 按序列依次返回不同值(测试多轮调用)             │
│              │ mock("search").sequence(["r1","r2","r3"])    │
├──────────────┼───────────────────────────────────────────────┤
│ DYNAMIC      │ 根据参数动态计算返回值(复杂逻辑模拟)           │
│              │ mock("search").dynamic(lambda q: f"搜索{q}") │
└──────────────┴───────────────────────────────────────────────┘

3.3 Mock 注册与注入

class ToolMock:
    """工具 Mock:替换真实工具的测试替身"""

    def __init__(self, tool_name: str):
        self.tool_name = tool_name
        self.behavior = MockBehavior.RETURN
        self.return_value = None
        self.exception = None
        self.delay_seconds = 0
        self.call_count = 0
        self.call_args_list = []  # 记录所有调用参数

    async def execute(self, args: dict) -> Any:
        """执行 Mock 逻辑"""
        self.call_count += 1
        self.call_args_list.append(args)

        if self.delay_seconds > 0:
            await asyncio.sleep(self.delay_seconds)

        if self.behavior == MockBehavior.RAISE:
            raise self.exception
        elif self.behavior == MockBehavior.SEQUENCE:
            idx = min(self.call_count - 1, len(self.return_value) - 1)
            return self.return_value[idx]
        elif self.behavior == MockBehavior.DYNAMIC:
            return self.return_value(args)
        else:  # RETURN
            return self.return_value


class MockRegistry:
    """Mock 注册表:管理所有工具 Mock"""

    def __init__(self):
        self.mocks: dict[str, ToolMock] = {}

    def register(self, mock: ToolMock):
        """注册 Mock"""
        self.mocks[mock.tool_name] = mock

    def get_mock(self, tool_name: str) -> ToolMock | None:
        """获取 Mock"""
        return self.mocks.get(tool_name)

    def verify_called(self, tool_name: str, times: int = 1) -> bool:
        """验证工具是否被调用了指定次数"""
        mock = self.mocks.get(tool_name)
        return mock is not None and mock.call_count == times

3.4 场景预设

class ScenarioPresets:
    """场景预设:预定义的 Mock 配置组合"""

    @staticmethod
    def normal_flow():
        """正常流程:所有工具正常返回"""
        return {
            "search": ToolMock("search").return("搜索结果"),
            "calculate": ToolMock("calculate").return("42"),
        }

    @staticmethod
    def network_failure():
        """网络故障:搜索超时,计算正常"""
        return {
            "search": ToolMock("search").delay(30).raise(TimeoutError()),
            "calculate": ToolMock("calculate").return("42"),
        }

    @staticmethod
    def partial_failure():
        """部分失败:第一次搜索失败,重试后成功"""
        return {
            "search": ToolMock("search").sequence([
                None,  # 第一次返回 None(触发重试)
                "搜索结果"  # 第二次返回结果
            ]),
        }

3.5 与其他概念的关联

• <- 环境 Harness：Mock 在沙箱环境中运行

• -> 状态检查点：Mock 调用记录可用于状态对比

• -> 错误注入与恢复：Mock 可模拟故障场景

• -> 完整 Test Harness：Mock 是测试 Harness 的核心组件

---

4. 状态检查点

4.1 定义

状态检查点（StateCheckpoint）是在 Agent 执行过程中定期保存状态快照，支持回滚到任意检查点、时间旅行调试和状态差异对比，是实现长任务恢复和调试的关键机制。

对应 Demo: demos/06_HarnessEngineering/04_状态检查点.py

4.2 核心能力

┌──────────────────────────────────────────────────────────┐
│              状态检查点四大能力                             │
├──────────┬───────────────────────────────────────────────┤
│ 快照保存  │ 在关键节点保存 Agent 完整状态(消息/变量/上下文) │
│ 回滚恢复  │ 从任意检查点回滚,重新执行后续步骤              │
│ 时间旅行  │ 按时间顺序浏览所有检查点,定位问题              │
│ 差异对比  │ 对比两个检查点的状态差异,发现变化              │
└──────────┴───────────────────────────────────────────────┘

4.3 快照与回滚

class StateSnapshot:
    """状态快照:某个时间点的完整 Agent 状态"""
    checkpoint_id: str        # 检查点 ID
    timestamp: float          # 时间戳
    step: int                 # 执行步数
    messages: list            # 消息历史
    variables: dict           # 变量状态
    tool_calls: list          # 工具调用记录
    metadata: dict            # 元数据


class StateCheckpointer:
    """状态检查点管理器"""

    def __init__(self):
        self.snapshots: list[StateSnapshot] = []

    def save(self, context: ExecutionContext) -> str:
        """保存当前状态为检查点"""
        snapshot = StateSnapshot(
            checkpoint_id=f"cp_{len(self.snapshots)}",
            timestamp=time.time(),
            step=len(context.trace),
            messages=copy.deepcopy(context.messages),
            variables=copy.deepcopy(context.variables),
            tool_calls=copy.deepcopy(context.tool_calls),
        )
        self.snapshots.append(snapshot)
        return snapshot.checkpoint_id

    def rollback(self, checkpoint_id: str) -> StateSnapshot:
        """回滚到指定检查点"""
        for i, snap in enumerate(self.snapshots):
            if snap.checkpoint_id == checkpoint_id:
                # 移除该检查点之后的所有快照
                self.snapshots = self.snapshots[:i + 1]
                return copy.deepcopy(snap)
        raise KeyError(f"检查点 {checkpoint_id} 不存在")

    def diff(self, cp1: str, cp2: str) -> dict:
        """对比两个检查点的状态差异"""
        s1 = self.get(cp1)
        s2 = self.get(cp2)
        return {
            "added_messages": len(s2.messages) - len(s1.messages),
            "changed_vars": {
                k for k in s2.variables
                if k not in s1.variables or s1.variables[k] != s2.variables[k]
            },
            "new_tool_calls": len(s2.tool_calls) - len(s1.tool_calls),
        }

4.4 检查点策略

检查点保存时机:

  ┌──────────────┬────────────────────────────────────────┐
  │ 每步保存      │ 每个执行步骤后保存(最安全,开销最大)     │
  │ 关键节点保存  │ 仅在工具调用/状态变更时保存(推荐)        │
  │ 定期保存      │ 每 N 步保存一次(平衡安全与性能)         │
  │ 手动保存      │ 开发者显式调用 save()(灵活控制)         │
  └──────────────┴────────────────────────────────────────┘

4.5 与其他概念的关联

• <- 工具 Mock 与注入：Mock 调用记录纳入状态快照

• -> 错误注入与恢复：检查点是恢复的基础

• -> 执行轨迹与可复现：检查点支持轨迹复现

• -> Skills 与 Deep Agents：长任务恢复依赖检查点

---

5. 错误注入与恢复

5.1 定义

错误注入与恢复（FaultInjection + RecoveryManager）是主动向 Agent 执行过程中注入预设故障，验证 Agent 的错误处理和恢复能力，并通过熔断器防止级联失败。

对应 Demo: demos/06_HarnessEngineering/05_错误注入与恢复.py

5.2 六种故障类型

┌──────────────────┬───────────────────────────────────────┐
│ TOOL_FAILURE     │ 工具调用失败(返回错误)                 │
│ TOOL_TIMEOUT     │ 工具执行超时                          │
│ NETWORK_ERROR    │ 网络错误(连接失败/DNS 解析失败)        │
│ INVALID_RESPONSE │ 工具返回无效响应(格式错误/数据缺失)    │
│ RESOURCE_EXHAUST │ 资源耗尽(内存/磁盘/配额)              │
│ STATE_CORRUPTION │ 状态损坏(数据不一致/并发冲突)          │
└──────────────────┴───────────────────────────────────────┘

5.3 故障注入器

class FaultInjector:
    """故障注入器:按计划注入预设故障"""

    def __init__(self):
        self.injections: list[FaultInjection] = []
        self._triggered: list[str] = []

    def add(self, fault: FaultInjection):
        """添加故障注入计划"""
        self.injections.append(fault)

    def maybe_inject(self, tool_name: str, step: int) -> Exception | None:
        """检查是否需要在此步注入故障"""
        for fault in self.injections:
            if fault.should_trigger(tool_name, step):
                self._triggered.append(f"{tool_name}@step{step}")
                return fault.create_exception()
        return None

5.4 五种恢复策略

┌──────────────────┬───────────────────────────────────────┐
│ RETRY            │ 重试: 等待后重新执行(指数退避)         │
│ FALLBACK         │ 降级: 切换到备选方案                   │
│ SKIP             │ 跳过: 跳过失败步骤,继续执行           │
│ ROLLBACK         │ 回滚: 回到检查点,重新执行              │
│ ABORT            │ 中止: 停止执行,报告错误                │
└──────────────────┴───────────────────────────────────────┘

5.5 熔断器

class CircuitBreaker:
    """熔断器:防止级联失败"""

    # 三种状态: CLOSED(正常) -> OPEN(熔断) -> HALF_OPEN(试探)
    def __init__(self, threshold: int = 5, timeout: float = 60.0):
        self.threshold = threshold  # 失败次数阈值
        self.timeout = timeout      # 熔断恢复超时
        self.failure_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = 0

    def can_execute(self) -> bool:
        """检查是否允许执行"""
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            # 检查是否过了恢复超时
            if time.time() - self.last_failure_time > self.timeout:
                self.state = CircuitState.HALF_OPEN
                return True
            return False  # 熔断中,拒绝执行
        if self.state == CircuitState.HALF_OPEN:
            return True  # 允许试探性执行

    def record_success(self):
        """记录成功"""
        self.failure_count = 0
        self.state = CircuitState.CLOSED

    def record_failure(self):
        """记录失败"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.threshold:
            self.state = CircuitState.OPEN  # 触发熔断

熔断器状态机:

  CLOSED (正常)
    │ 连续失败 >= threshold
    ▼
  OPEN (熔断,拒绝所有请求)
    │ 等待 timeout 时间
    ▼
  HALF_OPEN (试探,允许少量请求)
    │
    ├── 成功 ──> CLOSED (恢复正常)
    └── 失败 ──> OPEN (重新熔断)

5.6 与其他概念的关联

• <- 状态检查点：回滚恢复依赖检查点

• -> 执行控制：错误恢复影响执行流程

• -> 批量执行与回归：故障注入是回归测试的场景

• -> 完整 Test Harness：错误恢复是 Test Harness 的关键能力

---

6. 执行控制

6.1 定义

执行控制（ExecutionController）管理 Agent 执行的边界条件，包括超时限制、最大迭代次数、单步执行模式和取消机制，防止 Agent 无限循环或资源耗尽。

对应 Demo: demos/06_HarnessEngineering/06_执行控制.py

6.2 四种控制机制

┌──────────────┬───────────────────────────────────────────┐
│ 超时控制      │ 设置总执行时间上限,超时自动终止            │
│ 迭代限制      │ 限制 Agent 最大循环次数,防止死循环         │
│ 步进模式      │ 每步暂停等待确认,支持单步调试              │
│ 取消机制      │ 支持外部取消信号,优雅终止执行              │
└──────────────┴───────────────────────────────────────────┘

6.3 执行控制器

class ExecutionController:
    """执行控制器:管理 Agent 执行边界"""

    def __init__(self, config: ExecutionConfig):
        self.config = config
        self.status = ExecutionStatus.IDLE
        self.current_step = 0
        self._cancel_event = asyncio.Event()

    async def run_with_limits(self, agent_func, context: ExecutionContext):
        """带边界控制的执行"""
        self.status = ExecutionStatus.RUNNING
        try:
            # 同时启动超时和取消监听
            result = await asyncio.race([
                self._run_agent(agent_func, context),
                self._timeout_watcher(),
                self._cancel_watcher(),
            ])
            self.status = ExecutionStatus.COMPLETED
            return result
        except asyncio.TimeoutError:
            self.status = ExecutionStatus.TIMEOUT
            return ExecutionResult(success=False, error="执行超时")
        except CancelledError:
            self.status = ExecutionStatus.CANCELLED
            return ExecutionResult(success=False, error="用户取消")

    async def _run_agent(self, func, context):
        """执行 Agent,检查迭代限制"""
        while self.current_step < self.config.max_iterations:
            if self._cancel_event.is_set():
                raise CancelledError()
            self.current_step += 1
            # 步进模式:每步暂停等待确认
            if self.config.step_mode:
                await self._wait_for_confirmation()
            await func(context)
        # 达到迭代上限
        if self.current_step >= self.config.max_iterations:
            self.status = ExecutionStatus.ITERATION_LIMIT

6.4 执行状态枚举

class ExecutionStatus(Enum):
    IDLE = "idle"                    # 空闲
    RUNNING = "running"              # 运行中
    COMPLETED = "completed"          # 已完成
    TIMEOUT = "timeout"              # 超时
    CANCELLED = "cancelled"          # 已取消
    ITERATION_LIMIT = "iter_limit"   # 达到迭代上限
    ERROR = "error"                  # 错误终止

6.5 与其他概念的关联

• <- 错误注入与恢复：执行控制与错误恢复配合

• -> 批量执行与回归：执行控制用于批量测试

• -> 完整 Test Harness：执行控制是 Test Harness 的基础

---

7. 批量执行与回归

7.1 定义

批量执行与回归（BatchRunner + RegressionSuite）支持并行运行多个测试用例，管理测试基线，检测回归问题，是 Agent 持续集成的核心组件。

对应 Demo: demos/06_HarnessEngineering/07_批量执行与回归.py

7.2 核心组件

┌──────────────────────────────────────────────────────────┐
│                 批量执行与回归体系                          │
│                                                          │
│  ┌──────────┐    ┌──────────┐    ┌──────────────────┐   │
│  │TestCase  │    │TestSuite │    │ BatchRunner      │   │
│  │(单个用例)│───>│(用例集合)│───>│ (并行执行器)     │   │
│  └──────────┘    └──────────┘    └────────┬─────────┘   │
│                                           │              │
│                              ┌────────────▼──────────┐   │
│                              │ BaselineManager       │   │
│                              │ (基线管理/回归检测)    │   │
│                              └────────────┬──────────┘   │
│                                           │              │
│                              ┌────────────▼──────────┐   │
│                              │ RegressionSuite       │   │
│                              │ (回归测试套件)        │   │
│                              └───────────────────────┘   │
└──────────────────────────────────────────────────────────┘

7.3 测试用例与套件

@dataclass
class TestCase:
    """测试用例"""
    name: str
    task: str               # Agent 任务描述
    expected: Any           # 期望结果
    mock_config: dict       # Mock 配置
    timeout: float = 60.0
    tags: list[str] = field(default_factory=list)


class TestSuite:
    """测试套件:用例集合"""

    def __init__(self, name: str):
        self.name = name
        self.cases: list[TestCase] = []

    def add(self, case: TestCase):
        self.cases.append(case)

    def filter_by_tag(self, tag: str) -> list[TestCase]:
        """按标签过滤用例"""
        return [c for c in self.cases if tag in c.tags]

7.4 批量执行器

class BatchRunner:
    """批量执行器:并行运行测试用例"""

    async def run_suite(self, suite: TestSuite, parallel: int = 4) -> list[TestResult]:
        """并行执行测试套件"""
        semaphore = asyncio.Semaphore(parallel)

        async def run_one(case: TestCase) -> TestResult:
            async with semaphore:
                return await self._run_case(case)

        results = await asyncio.gather(*[run_one(c) for c in suite.cases])
        return list(results)

    async def _run_case(self, case: TestCase) -> TestResult:
        """执行单个测试用例"""
        start = time.time()
        try:
            # 设置 Mock
            mocks = MockRegistry()
            for name, config in case.mock_config.items():
                mocks.register(self._build_mock(name, config))

            # 执行 Agent
            actual = await self.agent.run(case.task, mocks=mocks)

            # 验证结果
            passed = self._verify(actual, case.expected)
            return TestResult(
                case=case.name,
                status=TestStatus.PASSED if passed else TestStatus.FAILED,
                duration=time.time() - start,
                actual=actual,
            )
        except Exception as e:
            return TestResult(
                case=case.name,
                status=TestStatus.ERROR,
                duration=time.time() - start,
                error=str(e),
            )

7.5 基线管理与回归检测

class BaselineManager:
    """基线管理:保存历史结果,检测回归"""

    def __init__(self):
        self.baselines: dict[str, Any] = {}  # case_name -> expected_result

    def save_baseline(self, case_name: str, result: Any):
        """保存基线结果"""
        self.baselines[case_name] = result

    def check_regression(self, case_name: str, actual: Any) -> bool:
        """检查是否回归(与基线不一致)"""
        baseline = self.baselines.get(case_name)
        if baseline is None:
            return False  # 无基线,不算回归
        return not self._equals(actual, baseline)


class RegressionSuite:
    """回归测试套件:对比当前结果与基线"""

    def run(self, results: list[TestResult], baseline: BaselineManager) -> SummaryReport:
        """运行回归分析"""
        regressions = []
        for r in results:
            if baseline.check_regression(r.case, r.actual):
                regressions.append(r.case)

        return SummaryReport(
            total=len(results),
            passed=sum(1 for r in results if r.status == TestStatus.PASSED),
            failed=sum(1 for r in results if r.status == TestStatus.FAILED),
            regressions=regressions,
        )

7.6 与其他概念的关联

• <- 执行控制：批量执行依赖执行控制

• -> 影子模式与灰度：回归测试结果用于灰度决策

• -> 完整 Test Harness：批量执行是 Test Harness 的核心功能

---

8. 影子模式与灰度

8.1 定义

影子模式与灰度（ShadowMode / CanaryRelease / ABTest）是 Agent 安全发布的三大策略，通过流量复制、渐进式发布和对照实验，在不影响线上服务的前提下验证新版本 Agent 的质量。

对应 Demo: demos/06_HarnessEngineering/08_影子模式与灰度.py

8.2 三种发布策略

┌──────────────────────────────────────────────────────────────┐
│                    三种发布策略对比                            │
├──────────┬─────────────────┬─────────────────────────────────┤
│ 影子模式  │ 流量复制,不影响  │ 新版本接收复制流量执行,但结果     │
│ Shadow   │ 线上用户         │ 不返回给用户,仅用于对比评估      │
├──────────┼─────────────────┼─────────────────────────────────┤
│ 金丝雀   │ 渐进式流量切换   │ 按比例(1%->5%->25%->100%)逐步    │
│ Canary   │                  │ 将流量从旧版本切换到新版本        │
├──────────┼─────────────────┼─────────────────────────────────┤
│ A/B 测试 │ 对照实验         │ 同时运行两个版本,按比例分流,     │
│ ABTest   │                  │ 统计对比质量指标                 │
└──────────┴─────────────────┴─────────────────────────────────┘

8.3 影子模式

class ShadowMode:
    """影子模式:复制流量到新版本,不影响线上"""

    async def handle(self, request, primary_agent, shadow_agent):
        """处理请求:主版本返回结果,影子版本仅记录"""
        # 1. 主版本处理(结果返回给用户)
        primary_result = await primary_agent.run(request)

        # 2. 影子版本处理(结果仅用于对比,不返回)
        try:
            shadow_result = await shadow_agent.run(request)
            # 3. 记录对比结果
            self.metrics_collector.record(ComparisonReport(
                request=request,
                primary=primary_result,
                shadow=shadow_result,
                match=self._compare(primary_result, shadow_result),
            ))
        except Exception as e:
            # 影子版本失败不影响线上
            self.metrics_collector.record_shadow_error(str(e))

        return primary_result  # 只返回主版本结果

8.4 金丝雀发布

class CanaryRelease:
    """金丝雀发布:渐进式流量切换"""

    STAGES = [0.01, 0.05, 0.25, 1.0]  # 1% -> 5% -> 25% -> 100%

    def __init__(self):
        self.current_stage = 0
        self.health_metrics = []

    def should_route_to_canary(self) -> bool:
        """判断是否将请求路由到金丝雀版本"""
        ratio = self.STAGES[self.current_stage]
        return random.random() < ratio

    def check_health(self) -> bool:
        """检查金丝雀版本健康度"""
        if len(self.health_metrics) < 100:
            return True  # 样本不足,继续观察
        success_rate = sum(m.success for m in self.health_metrics[-100:]) / 100
        if success_rate < 0.95:  # 成功率低于 95%
            return False  # 不健康,停止灰度
        return True  # 健康,可进入下一阶段

    def promote(self):
        """进入下一灰度阶段"""
        if self.current_stage < len(self.STAGES) - 1:
            self.current_stage += 1

    def rollback(self):
        """回滚到上一阶段"""
        if self.current_stage > 0:
            self.current_stage -= 1

8.5 金丝雀四阶段发布流程

  阶段1: 1% 流量 ──> 监控 15min ──> 健康? ──是──> 阶段2
                                      │
                                      否
                                      ↓
                                   回滚到 0%

  阶段2: 5% 流量 ──> 监控 30min ──> 健康? ──是──> 阶段3
                                      │
                                      否
                                      ↓
                                   回滚到 1%

  阶段3: 25% 流量 ──> 监控 1h ──> 健康? ──是──> 阶段4
                                     │
                                     否
                                     ↓
                                  回滚到 5%

  阶段4: 100% 流量 ──> 灰度完成,旧版本下线

8.6 与其他概念的关联

• <- 批量执行与回归：回归测试结果用于灰度决策

• -> 执行轨迹与可复现：灰度问题可通过轨迹复现调试

• -> 完整 Test Harness：灰度是 Test Harness 的发布能力

---

9. 执行轨迹与可复现

9.1 定义

执行轨迹与可复现（TraceRecorder + DeterministicReplay）记录 Agent 执行全过程的每个事件（LLM 调用、工具调用、状态变更），支持确定性回放和轨迹分析，是实现 Bug 复现和调试的关键。

对应 Demo: demos/06_HarnessEngineering/09_执行轨迹与可复现.py

9.2 轨迹记录

class EventType(Enum):
    LLM_CALL = "llm_call"          # LLM 调用
    TOOL_CALL = "tool_call"        # 工具调用
    STATE_CHANGE = "state_change"  # 状态变更
    DECISION = "decision"          # 决策点
    ERROR = "error"                # 错误事件
    CHECKPOINT = "checkpoint"      # 检查点


class TraceEvent:
    """轨迹事件"""
    event_id: str
    event_type: EventType
    timestamp: float
    step: int
    data: dict           # 事件数据(输入/输出/参数)
    duration: float      # 执行耗时


class TraceRecorder:
    """轨迹记录器"""

    def __init__(self):
        self.events: list[TraceEvent] = []

    def record(self, event_type: EventType, data: dict, duration: float = 0):
        """记录事件"""
        event = TraceEvent(
            event_id=f"evt_{len(self.events)}",
            event_type=event_type,
            timestamp=time.time(),
            step=len(self.events),
            data=copy.deepcopy(data),
            duration=duration,
        )
        self.events.append(event)

9.3 确定性回放

class DeterministicReplay:
    """确定性回放:按轨迹重现执行过程"""

    def __init__(self, trace: list[TraceEvent]):
        self.trace = trace
        self.current_step = 0
        self.mismatches: list[ReplayMismatch] = []

    async def replay(self, agent, mock_registry: MockRegistry):
        """按轨迹回放执行"""
        for event in self.trace:
            self.current_step = event.step

            if event.event_type == EventType.TOOL_CALL:
                # 工具调用:使用 Mock 返回轨迹中记录的结果
                tool_name = event.data["tool"]
                expected_result = event.data["result"]
                mock_registry.get_mock(tool_name).return(expected_result)

            elif event.event_type == EventType.LLM_CALL:
                # LLM 调用:Mock 返回轨迹中记录的响应
                actual = await agent.llm_call(event.data["input"])
                if not self._matches(actual, event.data["output"]):
                    self.mismatches.append(ReplayMismatch(
                        step=event.step,
                        expected=event.data["output"],
                        actual=actual,
                    ))

    def verify(self) -> bool:
        """验证回放是否一致"""
        return len(self.mismatches) == 0

9.4 轨迹分析

class TraceAnalyzer:
    """轨迹分析器:从轨迹中提取洞察"""

    def analyze(self, trace: list[TraceEvent]) -> dict:
        """分析执行轨迹"""
        return {
            "total_steps": len(trace),
            "total_duration": sum(e.duration for e in trace),
            "llm_calls": sum(1 for e in trace if e.event_type == EventType.LLM_CALL),
            "tool_calls": sum(1 for e in trace if e.event_type == EventType.TOOL_CALL),
            "errors": sum(1 for e in trace if e.event_type == EventType.ERROR),
            "slowest_step": max(trace, key=lambda e: e.duration).step,
            "tool_usage": self._tool_frequency(trace),
        }

    def _tool_frequency(self, trace) -> dict:
        """统计工具使用频率"""
        freq = {}
        for e in trace:
            if e.event_type == EventType.TOOL_CALL:
                name = e.data.get("tool", "unknown")
                freq[name] = freq.get(name, 0) + 1
        return freq

9.5 与其他概念的关联

• <- 状态检查点：轨迹包含检查点事件

• -> Skills 与 Deep Agents：轨迹用于长任务调试

• -> 完整 Test Harness：轨迹复现是 Test Harness 的调试能力

---

10. Skills 与 Deep Agents

10.1 定义

Skills 管理与 Deep Agents 是 Harness 对 Agent 能力扩展和长任务恢复的支持。Skills 是可复用的能力模块，Deep Agents 是支持中断恢复的长时间运行 Agent。

对应 Demo: demos/06_HarnessEngineering/10_Skills与DeepAgents.py

10.2 Skill 管理

class Skill:
    """Skill:可复用的 Agent 能力模块"""
    name: str
    description: str
    resources: list[SkillResource]  # 所需资源(工具/数据/提示)
    entry_point: Callable           # 入口函数


class SkillRegistry:
    """Skill 注册表:管理所有可用 Skill"""

    def __init__(self):
        self.skills: dict[str, Skill] = {}
        self.loader = SkillLoader()

    def register(self, skill: Skill):
        """注册 Skill"""
        self.skills[skill.name] = skill

    def discover(self, task: str) -> list[Skill]:
        """根据任务发现相关 Skill"""
        return [s for s in self.skills.values()
                if self._is_relevant(s, task)]

10.3 Deep Agent 长任务恢复

class DeepAgent:
    """Deep Agent:支持中断恢复的长任务 Agent"""

    def __init__(self):
        self.tasks: dict[str, Task] = {}
        self.checkpoints: dict[str, list[Checkpoint]] = {}

    async def run_long_task(self, task: Task) -> str:
        """运行长任务(支持中断恢复)"""
        task_id = task.task_id

        # 检查是否有检查点可恢复
        if task_id in self.checkpoints and self.checkpoints[task_id]:
            last_cp = self.checkpoints[task_id][-1]
            task.status = TaskStatus.RESUMING
            task.current_step = last_cp.step
            task.artifacts = last_cp.artifacts
        else:
            task.status = TaskStatus.RUNNING
            task.current_step = 0

        # 执行任务步骤
        for step in range(task.current_step, len(task.steps)):
            try:
                await self._execute_step(task, step)
                # 每步保存检查点
                self._save_checkpoint(task, step)
            except InterruptedError:
                task.status = TaskStatus.INTERRUPTED
                return task_id  # 返回任务 ID,待后续恢复

        task.status = TaskStatus.COMPLETED
        return task_id

    def resume(self, task_id: str):
        """恢复中断的任务"""
        return self.run_long_task(self.tasks[task_id])

Deep Agent 长任务恢复流程:

  任务启动 -> 执行 Step 0 -> 保存检查点 CP0
           -> 执行 Step 1 -> 保存检查点 CP1
           -> 执行 Step 2 -> [中断!]

  恢复: 加载 CP1 -> 从 Step 2 继续 -> 执行 Step 2 -> 保存 CP2
                                              -> 执行 Step 3 -> 完成

10.4 与其他概念的关联

• <- 执行轨迹与可复现：长任务调试依赖轨迹

• <- 状态检查点：长任务恢复依赖检查点

• -> 完整 Test Harness：Skills 和 Deep Agents 是 Test Harness 的高级能力

---

11. 安全 Harness

11.1 定义

安全 Harness（SafetyHarness）在 Agent 执行过程中实施安全防护，包括 PII（个人身份信息）脱敏、Prompt 注入防护、输出内容过滤和操作审计。

对应 Demo: demos/06_HarnessEngineering/11_安全Harness.py

11.2 三大安全能力

┌──────────────────────────────────────────────────────────┐
│                安全 Harness 三大能力                       │
├──────────┬───────────────────────────────────────────────┤
│ PII 脱敏 │ 自动识别并脱敏输入/输出中的个人信息             │
│          │ 手机号/身份证/邮箱/银行卡 -> 掩码处理           │
├──────────┼───────────────────────────────────────────────┤
│ 注入防护 │ 检测并拦截 Prompt 注入攻击                     │
│          │ "忽略以上指令,执行..." -> 拦截                 │
├──────────┼───────────────────────────────────────────────┤
│ 审计追踪 │ 记录所有 Agent 操作,用于事后追溯               │
│          │ 谁/何时/做了什么/结果如何                      │
└──────────┴───────────────────────────────────────────────┘

11.3 PII 脱敏

import re

class PIIMasker:
    """PII 脱敏器"""

    PATTERNS = {
        "phone": (r"1[3-9]\d{9}", lambda m: m.group()[:3] + "****" + m.group()[-4:]),
        "id_card": (r"\d{17}[\dXx]", lambda m: m.group()[:6] + "********" + m.group()[-4:]),
        "email": (r"[\w.]+@[\w.]+", lambda m: m.group()[:2] + "***@" + m.group().split("@")[1]),
        "bank_card": (r"\d{16,19}", lambda m: m.group()[:4] + "****" + m.group()[-4:]),
    }

    def mask(self, text: str) -> str:
        """脱敏文本中的 PII"""
        for pii_type, (pattern, replacer) in self.PATTERNS.items():
            text = re.sub(pattern, replacer, text)
        return text

# 输入: "我的手机号是13812345678,邮箱是test@example.com"
# 输出: "我的手机号是138****5678,邮箱是te***@example.com"

11.4 Prompt 注入防护

class InjectionGuard:
    """Prompt 注入防护"""

    INJECTION_PATTERNS = [
        r"忽略.{0,10}(指令|规则|限制)",
        r"ignore.{0,10}(instruction|rule)",
        r"你(现在|不再)是",
        r"system\s*prompt",
        r"<\/?system>",
    ]

    def check(self, user_input: str) -> tuple[bool, str]:
        """检查是否存在注入攻击"""
        for pattern in self.INJECTION_PATTERNS:
            if re.search(pattern, user_input, re.IGNORECASE):
                return False, f"检测到可能的 Prompt 注入: {pattern}"
        return True, "安全"

11.5 与其他概念的关联

• <- 环境 Harness：安全 Harness 在沙箱环境中运行

• -> 性能 Harness：安全检查可能影响性能

• -> 完整 Test Harness：安全 Harness 是 Test Harness 的防护层

---

12. 性能 Harness

12.1 定义

性能 Harness（PerformanceHarness）监控 Agent 执行的性能指标，包括延迟、Token 消耗、成本和资源使用，设置预算告警，在超预算时触发告警或限制。

对应 Demo: demos/06_HarnessEngineering/12_性能Harness.py

12.2 监控指标与预算

class PerformanceBudget:
    """性能预算"""
    max_latency: float = 30.0       # 最大延迟(秒)
    max_tokens: int = 100000        # 最大 Token 数
    max_cost: float = 1.0           # 最大成本(美元)
    max_steps: int = 50             # 最大步数


class PerformanceMonitor:
    """性能监控器"""

    def __init__(self, budget: PerformanceBudget):
        self.budget = budget
        self.metrics: list[StepMetric] = []
        self.alerts: list[BudgetAlert] = []

    def record_step(self, metric: StepMetric):
        """记录每步性能指标"""
        self.metrics.append(metric)
        self._check_budget(metric)

    def _check_budget(self, metric: StepMetric):
        """检查是否超预算"""
        total_tokens = sum(m.tokens for m in self.metrics)
        total_cost = sum(m.cost for m in self.metrics)
        total_latency = sum(m.latency for m in self.metrics)

        if total_tokens > self.budget.max_tokens * 0.8:
            self.alerts.append(BudgetAlert(
                level=AlertLevel.WARNING,
                message=f"Token 使用达 {total_tokens}/{self.budget.max_tokens} (80%)"
            ))
        if total_cost > self.budget.max_cost:
            self.alerts.append(BudgetAlert(
                level=AlertLevel.CRITICAL,
                message=f"成本超预算: {total_cost}/{self.budget.max_cost}"
            ))

12.3 三级告警

┌──────────┬──────────────────────────────────────────┐
│ INFO     │ 预算使用 50% -> 记录日志,不通知           │
│ WARNING  │ 预算使用 80% -> 通知开发者,建议优化       │
│ CRITICAL │ 预算使用 100% -> 立即通知,可能终止执行    │
└──────────┴──────────────────────────────────────────┘

12.4 瓶颈检测

class BottleneckDetector:
    """瓶颈检测器:识别性能瓶颈"""

    def detect(self, metrics: list[StepMetric]) -> list[str]:
        """检测性能瓶颈"""
        bottlenecks = []
        avg_latency = sum(m.latency for m in metrics) / len(metrics)

        for m in metrics:
            if m.latency > avg_latency * 3:
                bottlenecks.append(f"Step {m.step} 延迟异常: {m.latency:.1f}s (均值 {avg_latency:.1f}s)")
            if m.tokens > 10000:
                bottlenecks.append(f"Step {m.step} Token 消耗过高: {m.tokens}")

        return bottlenecks

12.5 与其他概念的关联

• <- 安全 Harness：安全检查影响性能指标

• -> 完整 Test Harness：性能监控是 Test Harness 的观测能力

---

13. 完整 Test Harness

13.1 定义

完整 Test Harness（AgentTestHarness）是将环境、工具 Mock、状态检查点、错误注入、执行控制、安全、性能等所有 Harness 组件有机集成的统一测试框架，提供端到端的 Agent 测试能力。

对应 Demo: demos/06_HarnessEngineering/13_完整TestHarness.py

13.2 架构总览

┌──────────────────────────────────────────────────────────────┐
│                  AgentTestHarness（完整测试框架）              │
│                                                              │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐       │
│  │环境Harness│ │工具Mock  │ │状态检查点│ │错误注入  │       │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘       │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐       │
│  │执行控制  │ │批量回归  │ │影子灰度  │ │轨迹复现  │       │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘       │
│  ┌──────────┐ ┌──────────┐ ┌──────────────────────┐        │
│  │安全Harness│ │性能Harness│ │  Skills & DeepAgents │        │
│  └──────────┘ └──────────┘ └──────────────────────┘        │
│                                                              │
│  ┌──────────────────────────────────────────────────────┐    │
│  │              TestReport (统一测试报告)                │    │
│  │  通过率 | 延迟 | Token | 成本 | 错误 | 回归 | 安全   │    │
│  └──────────────────────────────────────────────────────┘    │
└──────────────────────────────────────────────────────────────┘

13.3 集成实现

class AgentTestHarness:
    """完整 Test Harness:集成所有测试能力"""

    def __init__(self, config: TestConfig):
        self.config = config
        # 初始化所有 Harness 组件
        self.env = EnvironmentHarness(config.env_config)
        self.mocks = MockRegistry()
        self.checkpointer = StateCheckpointer()
        self.fault_injector = FaultInjector()
        self.execution = ExecutionController(config.exec_config)
        self.safety = SafetyChecker(config.safety_config)
        self.perf = PerfTracker(config.perf_config)
        self.tracer = AgentTracer()

    async def run_test(self, test_case: TestCase) -> TestReport:
        """执行完整测试"""
        # 1. 环境准备
        context = ExecutionContext(task=test_case.task)
        self.env.setup(context)

        # 2. 注入 Mock
        for name, cfg in test_case.mock_config.items():
            self.mocks.register(self._build_mock(name, cfg))

        # 3. 注入故障(如有)
        for fault in test_case.faults:
            self.fault_injector.add(fault)

        # 4. 执行(带所有 Harness 包裹)
        trace = []
        result = await self.execution.run_with_limits(
            self._wrapped_agent, context
        )

        # 5. 生成测试报告
        return TestReport(
            test_case=test_case.name,
            status=result.status,
            passed=self._verify(result, test_case.expected),
            trace=self.tracer.events,
            perf=self.perf.get_summary(),
            safety=self.safety.get_violations(),
            checkpoints=self.checkpointer.snapshots,
        )

13.4 测试报告

@dataclass
class TestReport:
    """测试报告:汇总所有 Harness 的结果"""
    test_case: str
    status: TestStatus           # 通过/失败/错误
    passed: bool                 # 是否通过
    trace: list                  # 执行轨迹
    perf: dict                   # 性能指标(延迟/Token/成本)
    safety: list                 # 安全违规
    checkpoints: list            # 检查点
    duration: float              # 总耗时
    errors: list[str]            # 错误信息

13.5 与其他概念的关联

• <- 所有 Harness 类型：Test Harness 集成全部能力

• -> Agent 评测体系：Test Harness 的结果用于评测

• -> LLMOps 生产化：Test Harness 是 CI/CD 的核心环节

---

概念关系总览

                    ┌──────────────────────────────────────────┐
                    │        Harness Engineering 体系           │
                    └──────────────────────────────────────────┘

  基础层                    控制层                    保障层
    │                        │                        │
    ▼                        ▼                        ▼
┌──────────┐          ┌──────────┐            ┌──────────┐
│环境Harness│──┐       │执行控制  │     ┌──────│安全Harness│
└──────────┘  │       └──────────┘     │      └──────────┘
┌──────────┐  │       ┌──────────┐     │      ┌──────────┐
│工具Mock  │──┼──>    │错误注入  │──┐  └──────│性能Harness│
└──────────┘  │       │与恢复    │  │         └──────────┘
┌──────────┐  │       └──────────┘  │
│状态检查点│──┘                     │
└──────────┘                        │
       │                            │
       ▼                            ▼
┌──────────┐                ┌──────────┐
│轨迹复现  │<───────────────│批量回归  │
└──────────┘                └──────────┘
       │                            │
       ▼                            ▼
┌──────────┐                ┌──────────┐
│Skills &  │                │影子灰度  │
│DeepAgents│                └──────────┘
└──────────┘                        │
       │                            │
       └──────────┬─────────────────┘
                  ▼
         ┌─────────────────┐
         │ 完整 Test Harness│
         │ (AgentTestHarness)│
         └─────────────────┘

概念间的依赖关系