# 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 标准进行追踪，这使得可以与许多现成的 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. **定期验证**：通过重放，验证系统的一致性

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


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://yeasy.gitbook.io/harness_engineering_guide/di-yi-bu-fen-harness-gong-cheng-ji-chu/03_principles/3.2_verifiability.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.