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

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

```python
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
        }
```

上下文管理器简化使用：

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


---

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