> 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-san-bu-fen-xi-tong-ji-cheng-yu-gong-cheng-shi-jian/11_reliability/11.5_miniharness_reliability.md).

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

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

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

## 11.5.1 结构化日志系统

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

**字段化日志格式**：每条日志包含标准字段(timestamp、level、trace\_id)和应用字段(tool\_name、duration\_ms)：

```python
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 分布式追踪上下文

追踪用于“跟踪请求经过的所有函数调用”。MiniHarness 示例使用 OpenTelemetry/W3C 形态的 trace/span ID 和 JSON 追踪记录；若要接入完整 OpenTelemetry 生态，还需要补充 SpanContext、上下文传播和 OTLP exporter：

```python
import secrets

class Span:
    """单个追踪跨度"""
    def __init__(self, name: str, trace_id: str, parent_id: str = None):
        self.span_id = secrets.token_hex(8)  # 8字节 SpanId，16位十六进制
        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
        }
```

上下文管理器简化使用：

```python
# 使用示例
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 重试与超时策略

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

```python
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 执行中：

```python
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 健康检查与告警

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

```python
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. **指标驱动**：追踪成功率、延迟、工具可靠性，数据驱动改进