> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-yi-bu-fen-harness-gong-cheng-ji-chu/03_principles/3.2_verifiability.md).

# 3.2 可验证性原则

本节阐述可验证性原则的含义，说明其在生产系统中的重要性，深入讨论可验证性的三个层次及其实现方法。

## 3.2.1 原则的核心

**可验证性** 意味着：系统的每一个行为都应该是可观测的、可审计的、可重放的。当被问“Agent 做了什么？”和“为什么这样做？”时，系统应该能够提供清晰、完整的答案。

这一原则在智能体时代变得尤为关键。传统软件的行为由预先编写的代码决定，而 Agent 的行为路径是 LLM 在运行时动态生成的——这些被称为“暗码(Dark Code)”的运行时行为执行完毕后即消散，无法像源代码一样被事先审查或精确复现。可验证性正是 Harness 对抗暗码的核心武器：通过将“即生即灭”的行为转化为“可观测、可追踪、可重放”的记录，让动态行为获得与静态代码同等的可审计性。

可验证性涵盖三个层面：

1. **可观测性**：看得见智能体在做什么
2. **可追踪性**：追踪来自何处，去往何处
3. **可重放性**：给定相同的输入，能够重现相同的结果

## 3.2.2 为什么可验证性至关重要

本小节通过具体问题场景、法规要求和技术挑战，说明可验证性为什么是系统可靠运行的基础。

### 问题场景

想象以下情况：

* 智能体执行了一个转账操作，但钱到了错误的账户
* Agent 生成了一个不符合规范的法律文件，造成了合规问题
* Agent 多次调用同一个 API，导致重复扣款

在这些情况下，最紧迫的问题是：**发生了什么？以及为什么？**

如果系统不可验证，这些问题就无法被回答。开发者只能：

* 陷入无尽的调试
* 无法向用户解释
* 无法修复根本原因

### 可验证性的价值

**1. 快速诊断** 有完整的审计日志和执行追踪，我们可以立即看到问题发生的位置。

**2. 用户信任** 当用户询问“这笔钱去哪了？”，如果我们能够提供详细的操作日志，用户会更有信心。

**3. 法律合规** 许多行业（金融、医疗、法律）都要求对所有关键操作进行审计。

**4. 系统改进** 通过观察和分析执行数据，我们能够识别出系统的瓶颈和改进空间。

## 3.2.3 可验证性的三个层次

本小节从基础到高级，分别介绍操作日志、执行追踪和可重放性三个层次的可验证性，以及它们的实现方法。

### 第一层：操作日志

最基础的可验证性：记录发生了什么。

```python
@dataclass
class OperationLog:
    """操作日志"""
    timestamp: datetime
    agent_id: str
    operation_type: str  # "tool_call", "permission_check", "decision"
    operation_name: str  # 具体的操作名称
    input: dict          # 输入参数
    output: dict         # 输出结果
    status: str          # "success", "error", "permission_denied"
    duration_ms: float   # 耗时
    error_message: Optional[str] = None

class OperationLogger:
    """操作日志记录器"""

    async def log(self, operation_log: OperationLog) -> None:
        """记录一个操作"""
        # 格式化为可读的形式
        log_entry = {
            "timestamp": operation_log.timestamp.isoformat(),
            "agent": operation_log.agent_id,
            "type": operation_log.operation_type,
            "operation": operation_log.operation_name,
            "status": operation_log.status,
            "duration_ms": operation_log.duration_ms
        }

        # 写入日志存储
        await self.storage.append(log_entry)

        # 如果是错误,也要输出到错误流
        if operation_log.status == "error":
            logger.error(f"Operation failed: {operation_log.operation_name}",
                        extra={"log_entry": log_entry})
```

### 第二层：执行追踪

更高级的可验证性：记录操作之间的因果关系。

```python
@dataclass
class ExecutionTrace:
    """执行追踪"""
    trace_id: str                  # 追踪的全局唯一ID
    agent_id: str
    task_id: str
    steps: List[TraceStep] = field(default_factory=list)
    start_time: datetime = field(default_factory=datetime.now)
    end_time: Optional[datetime] = None

@dataclass
class TraceStep:
    """追踪中的一步"""
    step_id: str                   # 步骤ID
    parent_step_id: Optional[str]  # 父步骤(用于嵌套调用)
    operation: str
    input: dict
    output: dict
    duration_ms: float
    timestamp: datetime
    status: str

class ExecutionTracer:
    """执行追踪器"""

    def __init__(self):
        self.traces: Dict[str, ExecutionTrace] = {}

    def start_trace(
        self,
        agent_id: str,
        task_id: str,
        trace_id: str = None
    ) -> str:
        """开始一个新的执行追踪"""
        trace_id = trace_id or str(uuid.uuid4())
        self.traces[trace_id] = ExecutionTrace(
            trace_id=trace_id,
            agent_id=agent_id,
            task_id=task_id
        )
        return trace_id

    def add_step(
        self,
        trace_id: str,
        operation: str,
        input: dict,
        output: dict,
        duration_ms: float,
        parent_step_id: str = None
    ) -> str:
        """在追踪中添加一步"""
        step_id = str(uuid.uuid4())
        step = TraceStep(
            step_id=step_id,
            parent_step_id=parent_step_id,
            operation=operation,
            input=input,
            output=output,
            duration_ms=duration_ms,
            timestamp=datetime.now(),
            status="success"
        )
        self.traces[trace_id].steps.append(step)
        return step_id

    def get_trace(self, trace_id: str) -> Optional[ExecutionTrace]:
        """获取完整的执行追踪"""
        return self.traces.get(trace_id)

    def visualize_trace(self, trace_id: str) -> str:
        """生成可视化的执行追踪"""
        trace = self.traces.get(trace_id)
        if not trace:
            return ""

        # 生成缩进的树形结构
        lines = [f"Trace: {trace.trace_id}"]
        lines.append(f"Agent: {trace.agent_id}, Task: {trace.task_id}")
        lines.append(f"Duration: {(trace.end_time - trace.start_time).total_seconds():.2f}s")
        lines.append("")

        def add_step_lines(step: TraceStep, indent: int = 0):
            prefix = "  " * indent
            lines.append(f"{prefix}├─ {step.operation} ({step.duration_ms:.0f}ms)")
            lines.append(f"{prefix}│  Input: {step.input}")
            lines.append(f"{prefix}│  Output: {step.output}")

        for step in trace.steps:
            if step.parent_step_id is None:
                add_step_lines(step)

        return "\n".join(lines)
```

### 第三层：可重放性

最高级的可验证性：给定相同的输入，能够重现执行过程。

```python
class ReplayableExecution:
    """可重放的执行"""

    def __init__(self, trace: ExecutionTrace):
        self.trace = trace
        self.step_index = 0

    async def replay(
        self,
        runtime: RuntimeEngine,
        simulation: bool = False
    ) -> ReplayResult:
        """
        重放执行。

        simulation: 如果为True,不实际执行工具,而是返回记录的结果
        """
        results = []

        for step in self.trace.steps:
            if simulation:
                # 模拟模式:直接返回记录的结果
                results.append({
                    "operation": step.operation,
                    "recorded_output": step.output,
                    "simulation": True
                })
            else:
                # 实际重放:重新执行相同的操作
                try:
                    actual_output = await self._execute_step(
                        step.operation,
                        step.input
                    )

                    # 验证输出是否一致
                    if actual_output == step.output:
                        results.append({
                            "operation": step.operation,
                            "status": "consistent",
                            "output": actual_output
                        })
                    else:
                        results.append({
                            "operation": step.operation,
                            "status": "diverged",
                            "recorded_output": step.output,
                            "actual_output": actual_output
                        })

                except Exception as e:
                    results.append({
                        "operation": step.operation,
                        "status": "error",
                        "error": str(e)
                    })

        return ReplayResult(
            trace_id=self.trace.trace_id,
            results=results,
            is_consistent=all(r["status"] == "consistent" for r in results)
        )

    async def _execute_step(self, operation: str, input: dict):
        """执行一个步骤"""
        # 实现取决于具体的操作类型
        pass
```

## 3.2.4 生产系统中的可验证性实践

Claude Code 的可观测性以 OpenTelemetry 配置导出 metrics 和日志事件为基础；分布式 traces 是增强遥测 beta，默认关闭，开启后可与支持 OTLP 的 APM 工具集成：

```typescript
// Claude Code 追踪示例
const span = tracer.startSpan("tool_execution", {
  attributes: {
    "tool.name": "weather_api",
    "tool.params": JSON.stringify(params),
    "agent.id": agentId
  }
});

try {
  const result = await executeWeatherAPI(params);
  span.setStatus({ code: SpanStatusCode.OK });
  span.end();
} catch (error) {
  span.recordException(error);
  span.setStatus({ code: SpanStatusCode.ERROR });
  span.end();
}
```

这种标准化的追踪方式的好处是：

* 可以与 Jaeger、Zipkin 等可视化工具集成
* 支持分布式追踪（跨多个微服务）
* 标准的性能分析

OpenClaw 的 Lobster 引擎特别强调可重放性。每个工作流的执行都被记录在一个确定性的日志中：

```yaml
## Lobster execution log
workflow_id: workflow-123
execution_id: exec-456
steps:
  - step_id: 1
    operation: query_database
    input:
      query: "SELECT * FROM users WHERE id = ?"
      params: [123]
    output:
      - id: 123
        name: "John Doe"
    timestamp: 2024-04-01T10:30:00Z

  - step_id: 2
    operation: send_notification
    input:
      user_id: 123
      message: "Hello John"
    output:
      status: "sent"
      notification_id: "notif-789"
    timestamp: 2024-04-01T10:30:01Z
```

Lobster 的特点：

1. **确定性**：给定相同的 workflow 和输入，执行顺序总是相同
2. **可重放**：可以从任何步骤重新开始执行
3. **可审计**：完整的操作历史，精确到每一步

## 3.2.5 可验证性在实战中的应用

**场景：转账操作**

一个典型的转账操作应该被追踪如下：

**Trace ID:** trace-2024-04-01-001

| 步骤 | 操作                        | 输入                                                 | 输出                                             | 耗时    |
| -- | ------------------------- | -------------------------------------------------- | ---------------------------------------------- | ----- |
| 1  | Parse Request             | {"from": "ACC001", "to": "ACC002", "amount": 1000} | {"from\_account": {...}, "to\_account": {...}} | 2ms   |
| 2  | Check Balance             | {"account\_id": "ACC001"}                          | {"balance": 5000}                              | 15ms  |
| 3  | Validate Transaction      | {"amount": 1000, "available\_balance": 5000}       | {"valid": true}                                | 1ms   |
| 4  | Create Transaction Record | {"from": "ACC001", "to": "ACC002", "amount": 1000} | {"transaction\_id": "TXN12345"}                | 50ms  |
| 5  | Execute Transfer          | {"transaction\_id": "TXN12345"}                    | {"status": "completed"}                        | 100ms |
| 6  | Send Confirmation         | {"transaction\_id": "TXN12345"}                    | {"notification\_sent": true}                   | 30ms  |

**总耗时：** 198ms **最终状态：** Success

当用户问“我的转账怎么样了？”，我们可以立即从这个追踪中回答：

* **什么时候执行的？** 2024-04-01 at 10:30:00
* **执行了哪些步骤？** 6 个步骤，每一步都成功了
* **如果失败了，在哪个步骤失败？**（不适用，这里全部成功）
* **整个过程花了多长时间？** 198 毫秒

## 3.2.6 实现可验证性的最佳实践

本小节介绍实现可验证性的具体方法，包括结构化日志、链接日志和追踪、定期验证和用户可读摘要。

### 1. 使用结构化日志

结构化日志使用 JSON 格式记录关键字段，便于验证和审计：

```python
# 不好:非结构化日志
logger.info("Transfer from ACC001 to ACC002 for amount 1000")

# 好:结构化日志
logger.info("transfer_executed", extra={
    "from_account": "ACC001",
    "to_account": "ACC002",
    "amount": 1000,
    "transaction_id": "TXN12345",
    "status": "success",
    "duration_ms": 198
})
```

### 2. 链接日志和追踪

通过 trace ID 将分布式日志和追踪关联起来，实现全链路可观测性：

```python
# 在日志中包含trace ID,便于关联
logger.info("Step completed", extra={
    "trace_id": trace.trace_id,
    "step_id": step.step_id,
    "operation": "tool_call"
})
```

### 3. 定期验证一致性

通过重放历史执行来验证系统行为的确定性和一致性：

```python
async def periodic_verification():
    """定期验证执行的一致性"""
    # 每小时,随机选择一个旧的执行,尝试重放
    old_traces = await trace_storage.list_old_traces(
        before_hours=1,
        limit=10
    )

    for trace in old_traces:
        replayer = ReplayableExecution(trace)
        result = await replayer.replay(runtime, simulation=False)

        if not result.is_consistent:
            logger.error("Inconsistency detected in replay",
                        extra={"trace_id": trace.trace_id})
            # 触发告警,通知运维
```

### 4. 提供用户可读的摘要

虽然完整的追踪对于调试很有用，但用户通常需要一个简化版本：

```python
def generate_user_summary(trace: ExecutionTrace) -> str:
    """生成用户友好的摘要"""
    total_duration = (trace.end_time - trace.start_time).total_seconds()
    success_count = sum(1 for step in trace.steps if step.status == "success")
    error_count = sum(1 for step in trace.steps if step.status == "error")

    return f"""
    执行摘要:
    - 执行ID:{trace.trace_id[:8]}...
    - 执行时间:{trace.start_time.strftime('%Y-%m-%d %H:%M:%S')}
    - 耗时:{total_duration:.2f}秒
    - 步骤数:{len(trace.steps)}(成功{success_count},失败{error_count})
    - 状态:{'成功' if error_count == 0 else '失败'}

    详细日志:{generate_trace_url(trace.trace_id)}
    """
```

## 3.2.7 总结

可验证性原则的关键要点：

1. **三个层次**：操作日志 → 执行追踪 → 可重放性
2. **结构化数据**：使用统一的格式，便于机器和人类读取
3. **完整的链路**：从请求到响应，每一步都能被追踪
4. **可视化**：提供人类可读的总结和可视化工具
5. **定期验证**：通过重放，验证系统的一致性

实现可验证性看起来增加了复杂性，但它的回报是巨大的：快速诊断、用户信任、法律合规、系统改进。