11.5 实战：为 MiniHarness 添加可靠性保障

本节为生产级 MiniHarness 添加可靠性工程保障。核心设计是“可观测性驱动、容错重试、健康检查”三层递进。

完整代码见 lab/mini_harness/reliability/，本节聚焦设计模式与集成策略。

11.5.1 结构化日志系统

生产系统必须支持日志聚合与分析。设计两个核心能力：

字段化日志格式：每条日志包含标准字段(timestamp、level、trace_id)和应用字段(tool_name、duration_ms)：

import json
from datetime import datetime

class StructuredLogger:
    def log(self, level: str, message: str, **fields):
        """记录结构化日志为 JSON"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'trace_id': get_trace_id(),  # 链接相关日志
            'message': message,
            'level': level,
            **fields  # 工具名、耗时等
        }
        print(json.dumps(log_entry))

    def log_tool_call(self, tool_name: str, duration_ms: float,
                      status: str, error: str = None):
        """专用方法记录工具调用"""
        self.log('INFO', f'Tool call: {tool_name}',
                 tool_name=tool_name, duration_ms=duration_ms,
                 status=status, error=error)

关键设计决策：

• JSON 序列化：便于日志系统解析和搜索（ELK、DataDog 等）

• trace_id 贯穿：关联同一请求的所有日志，便于链路追踪

• 字段聚合：便于建立告警规则（如 error_count > 100 在 5 分钟内）

完整实现参见 lab/mini_harness/reliability/logging.py。

11.5.2 分布式追踪上下文

追踪用于“跟踪请求经过的所有函数调用”，遵循 OpenTelemetry 标准：

class Span:
    """单个追踪跨度"""
    def __init__(self, name: str, trace_id: str, parent_id: str = None):
        self.span_id = uuid.uuid4()  # 唯一标识
        self.name = name
        self.trace_id = trace_id  # 链接到同一请求
        self.parent_span_id = parent_id  # 形成树结构

        self.start_time = None
        self.end_time = None
        self.attributes = {}  # 自定义属性

    def set_attribute(self, key: str, value: Any):
        self.attributes[key] = value

    def to_dict(self):
        """转换为追踪记录"""
        return {
            'trace_id': self.trace_id,
            'span_id': self.span_id,
            'parent_span_id': self.parent_span_id,
            'name': self.name,
            'duration_ms': self._calculate_duration(),
            'attributes': self.attributes
        }

上下文管理器简化使用：

# 使用示例
with trace_span('tool_execution') as span:
    span.set_attribute('tool_name', 'web_search')
    result = await execute_tool()
    span.set_attribute('status', 'success')
# 自动记录耗时、错误等

设计优势：

1. 树形结构：Parent Span 可视化整个请求的调用树

2. 自动计时：span 自动记录开始/结束时间，无需手动

3. 异常自动捕获：上下文管理器捕获异常并标记为错误

完整实现参见 lab/mini_harness/reliability/tracing.py。

11.5.3 重试与超时策略

“客户端重试 + 超时保护”是应对瞬时故障的标准做法。核心设计：

class RetryDecorator:
    def __init__(self, max_attempts: int = 3,
                 initial_delay: float = 1.0):
        self.max_attempts = max_attempts
        self.initial_delay = initial_delay

    def __call__(self, func):
        async def wrapper(*args, **kwargs):
            for attempt in range(1, self.max_attempts + 1):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == self.max_attempts:
                        raise

                    # 指数退避 + 抖动
                    delay = self.initial_delay * (2 ** (attempt - 1))
                    delay += random.uniform(0, delay * 0.1)
                    await asyncio.sleep(delay)

        return wrapper

# 使用:
@RetryDecorator(max_attempts=3, initial_delay=1.0)
async def call_external_api(url: str):
    return await make_http_request(url)

设计细节：

1. 指数退避：等待时间按 1s → 2s → 4s 递增，避免打垮故障服务

2. 抖动(Jitter)：防止多个客户端同时重试导致雷鸣羊群效应

3. 可恢复异常：仅对网络错误重试(ConnectionError、Timeout)，不对业务错误(ValidationError)重试

完整实现参见 lab/mini_harness/reliability/resilience.py。

11.5.4 可靠智能体的集成

将日志、追踪、重试三层叠加在 Agent 执行中：

class ReliableAgent(MiniHarnessAgent):
    async def execute(self, user_input: str) -> Dict:
        """带完整可靠性保障的执行"""
        with trace_span('agent_execution') as span:
            start = time.time()

            try:
                result = await super().execute(user_input)

                duration_ms = (time.time() - start) * 1000
                span.set_attribute('status', 'success')
                span.set_attribute('duration_ms', duration_ms)

                self.logger.log_agent_step(
                    step_id=span.span_id,
                    action='execute',
                    duration_ms=duration_ms
                )

                return result

            except Exception as e:
                span.set_error(str(e))
                self.logger.log('ERROR', f'Execution failed: {e}',
                               user_input=user_input[:100])
                raise

    @RetryDecorator(max_attempts=3)
    async def _execute_single_tool(self, tool_call):
        """工具执行层带重试"""
        with trace_span(f'tool:{tool_call["tool_name"]}'):
            # 工具执行逻辑
            pass

关键层级：

层级	职责	例子
Agent	追踪整体请求	trace_span('agent_execution')
工具执行	单工具重试保护	@RetryDecorator on _execute_single_tool
外部 API	流量限制+重试	@RetryDecorator on api_client

完整实现参见 lab/mini_harness/reliability/resilience.py（重试/断路器）与 lab/mini_harness/reliability/monitoring.py（指标采集）。

11.5.5 健康检查与告警

系统运行过程中持续监控关键指标：

class MonitoringSystem:
    def check_health(self, metrics: Dict) -> List[Alert]:
        """基于指标生成告警"""
        alerts = []

        # 错误率 > 10% -> CRITICAL
        error_rate = 1 - metrics['success_rate']
        if error_rate > 0.1:
            alerts.append(Alert('error_rate_high', error_rate))

        # 平均延迟 > 5s -> WARNING
        if metrics['avg_latency_ms'] > 5000:
            alerts.append(Alert('latency_high', metrics['avg_latency_ms']))

        # 某工具失败率 > 20% -> WARNING
        for tool, stats in metrics['tool_stats'].items():
            if stats['failure_rate'] > 0.2:
                alerts.append(Alert(f'tool_{tool}_unreliable',
                                   stats['failure_rate']))

        return alerts

告警处理的最佳实践：

1. 不同级别的行动：INFO（记录） vs WARNING（通知） vs CRITICAL（页面告警）

2. 告警去重：避免 5 分钟内重复告警（使用告警窗口）

3. 自愈反馈：告警触发后自动尝试恢复（如重启服务）

完整实现参见 lab/mini_harness/reliability/monitoring.py。

11.5.6 部署检查清单

• 日志格式转换为 JSON（便于 ELK/DataDog 解析）

• Trace ID 在所有日志中传播（便于链路追踪）

• 工具执行添加 @RetryDecorator（对瞬时故障自动重试）

• 设置合理的超时时间（API 调用通常 30-60s）

• 监控告警已配置（错误率 > 10%、延迟 > 5s）

• 告警通知渠道已接入（PagerDuty/Slack/钉钉）

• 可靠性报告在 Admin Dashboard 展示（成功率、平均延迟、工具统计）

11.5.7 总结

MiniHarness 可靠性工程的四大柱石：

1. 可观测性：结构化日志 + 分布式追踪，快速定位问题

2. 容错重试：指数退避，应对瞬时故障

3. 健康监控：实时告警，关键指标异常立即发现

4. 指标驱动：追踪成功率、延迟、工具可靠性，数据驱动改进