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 可验证性的三个层次

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

第一层：操作日志

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

@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})

第二层：执行追踪

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

@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)

第三层：可重放性

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

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 工具集成：

// 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引擎特别强调可重放性。每个工作流的执行都被记录在一个确定性的日志中：

## 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格式记录关键字段，便于验证和审计：

# 不好:非结构化日志
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将分布式日志和追踪关联起来，实现全链路可观测性：

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

3. 定期验证一致性

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

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. 提供用户可读的摘要

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

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. 定期验证：通过重放，验证系统的一致性

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