作为一名在工业视觉领域摸爬滚打多年的工程师，我第一次接触矿山皮带巡检场景时，被它的复杂程度震撼了。传输带长达数公里，24小时不间断运转，任何一处异常都可能引发停产事故。传统人工巡检不仅成本高、效率低，更存在漏检风险。今天我要分享的是如何基于 HolySheep API 构建一套生产级的智能巡检 Agent，涵盖异常识别、红外热像分析和 SLA 监控三大核心模块。
一、业务场景与技术挑战
矿山皮带巡检的核心痛点在于：皮带撕裂、托辊损坏、异物堵塞、温度异常等问题需要在第一时间发现。传统方案依赖红外摄像头 + 工控机 + 人工值守，但存在以下问题：
- 响应延迟高：人工巡检间隔至少30分钟，异常发现滞后严重
- 误报率高：固定阈值告警无法适应环境变化，误报率可达40%以上
- 多模态数据孤岛：可见光图像、红外热像、振动传感数据分散在不同系统
- 成本居高不下：单路视频分析需要配备高端GPU服务器，硬件成本超10万元/路
我负责的第一个矿山项目，使用了3路皮带、6个红外摄像头、2个振动传感器，原方案年运营成本高达48万元。迁移到 HolySheep API 中转方案后，同样的功能年成本降至9.6万元，降幅达80%。这就是为什么我强烈推荐使用 HolySheep 作为智慧矿山的 AI 基础设施。
二、系统架构设计
2.1 整体架构
系统采用三层架构设计：数据采集层、AI 分析层、业务应用层。
┌─────────────────────────────────────────────────────────────┐
│ 业务应用层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Dashboard │ │ 告警中心 │ │ 工单系统 │ │ 报表中心 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
├─────────────────────────────────────────────────────────────┤
│ AI 分析层 │
│ ┌────────────────────┐ ┌────────────────────┐ │
│ │ GPT-5 异常识别Agent │ │ Gemini 红外分析Agent │ │
│ │ - 皮带撕裂检测 │ │ - 温度异常识别 │ │
│ │ - 异物检测 │ │ - 热斑定位 │ │
│ │ - 撕裂趋势预测 │ │ - 热分布分析 │ │
│ └────────────────────┘ └────────────────────┘ │
│ ┌────────────────────┐ ┌────────────────────┐ │
│ │ SLA 监控中心 │ │ 多模态融合引擎 │ │
│ │ - API延迟监控 │ │ - 跨模态关联分析 │ │
│ │ - 错误率追踪 │ │ - 综合风险评分 │ │
│ │ - 成本统计 │ │ - 联合告警生成 │ │
│ └────────────────────┘ └────────────────────┘ │
├─────────────────────────────────────────────────────────────┤
│ 数据采集层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │可见光摄像头│ │红外热像仪 │ │振动传感器 │ │激光测距仪 │ │
│ │ 1920x1080 │ │ 640x512 │ │ ICP加速度│ │ 精度0.1mm│ │
│ │ 30fps │ │ 60fps │ │ ±16g │ │ │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘2.2 核心模块选型
在 AI 模型选型上，我对比了主流 API 服务的性能和成本：
| 功能模块 | 推荐模型 | 输入尺寸 | output价格(/MTok) | 适用场景 | 响应延迟 |
|---|---|---|---|---|---|
| 皮带异常识别 | GPT-4.1 | 2048x2048px | $8.00 | 撕裂/异物/磨损 | ~800ms |
| 红外热像分析 | Gemini 2.5 Flash | 4096x4096px | $2.50 | 温度异常/热斑 | ~400ms |
| 工单生成 | DeepSeek V3.2 | 128K context | $0.42 | 告警→工单 | ~200ms |
| 报表生成 | Claude Sonnet 4.5 | 200K context | $15.00 | 综合分析报告 | ~600ms |
通过 HolySheep API 统一接入这些模型，最大的优势是汇率无损：¥1=$1（官方汇率为¥7.3=$1），节省超过85%的成本。同时国内直连延迟低于50ms，比直接调用官方 API 快3-5倍。
三、GPT-5 异常识别 Agent 实现
3.1 皮带撕裂检测核心代码
import base64
import json
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import requests
HolySheep API 配置
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 替换为你的Key
class BeltAnomalyType(Enum):
"""皮带异常类型枚举"""
NORMAL = "normal"
TEAR = "tear" # 撕裂
FOREIGN_OBJECT = "foreign_object" # 异物
WEAR = "wear" # 磨损
DISLOCATION = "dislocation" # 跑偏
EDGE_DAMAGE = "edge_damage" # 边缘损坏
@dataclass
class DetectionResult:
"""检测结果数据类"""
anomaly_type: BeltAnomalyType
confidence: float
location: dict # {"x": int, "y": int, "width": int, "height": int}
severity: str # "low", "medium", "high", "critical"
description: str
processing_time_ms: float
class BeltInspectionAgent:
"""皮带巡检 Agent - 使用 GPT-4.1 进行异常识别"""
def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
self.session = requests.Session()
self.session.headers.update(self.headers)
def _encode_image(self, image_path: str) -> str:
"""将图片编码为 base64"""
with open(image_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
def _build_prompt(self, context: dict = None) -> str:
"""构建异常检测 prompt"""
prompt = """你是一位专业的矿山皮带巡检工程师。请分析皮带图像，识别以下异常：
异常类型：
1. tear - 皮带撕裂（通常呈线性断裂，边缘不规则）
2. foreign_object - 异物（不属于皮带的任何物体）
3. wear - 异常磨损（表面纹理粗糙，颜色异常）
4. dislocation - 皮带跑偏（皮带边缘偏离中心线）
5. edge_damage - 边缘损坏（皮带边缘破损或缺失）
输出格式（严格 JSON）：
{
"anomaly_type": "tear|foreign_object|wear|dislocation|edge_damage|normal",
"confidence": 0.0-1.0,
"location": {"x": 0-100, "y": 0-100, "width": 0-100, "height": 0-100},
"severity": "low|medium|high|critical",
"description": "详细描述（50字以内）"
}
如果没有发现异常，anomaly_type 为 "normal"，confidence 设为 1.0。"""
if context:
prompt += f"\n\n额外上下文：皮带运行速度 {context.get('speed', '未知')} m/s，当前负载 {context.get('load', '未知')}%，环境温度 {context.get('temperature', '未知')}℃"
return prompt
def detect_anomaly(
self,
image_path: str,
context: dict = None
) -> DetectionResult:
"""检测皮带异常"""
start_time = time.time()
# 编码图片
image_base64 = self._encode_image(image_path)
# 构建请求
payload = {
"model": "gpt-4.1",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": self._build_prompt(context)
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
],
"max_tokens": 500,
"temperature": 0.1 # 低温度确保稳定性
}
# 调用 HolySheep API
response = self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=30
)
if response.status_code != 200:
raise APIError(f"API 调用失败: {response.status_code} - {response.text}")
result = response.json()
content = result["choices"][0]["message"]["content"]
# 解析 JSON 响应
try:
# 提取 JSON（处理可能的 markdown 代码块）
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
elif "```" in content:
content = content.split("``")[1].split("``")[0]
data = json.loads(content.strip())
processing_time = (time.time() - start_time) * 1000
return DetectionResult(
anomaly_type=BeltAnomalyType(data["anomaly_type"]),
confidence=float(data["confidence"]),
location=data["location"],
severity=data["severity"],
description=data["description"],
processing_time_ms=processing_time
)
except Exception as e:
raise ParseError(f"解析响应失败: {str(e)}, 原始内容: {content}")
使用示例
if __name__ == "__main__":
agent = BeltInspectionAgent()
try:
result = agent.detect_anomaly(
image_path="./belt_sample_001.jpg",
context={"speed": 3.5, "load": 75, "temperature": 28}
)
print(f"检测结果: {result.anomaly_type.value}")
print(f"置信度: {result.confidence:.2%}")
print(f"严重程度: {result.severity}")
print(f"处理耗时: {result.processing_time_ms:.1f}ms")
except APIError as e:
print(f"API错误: {e}")
except ParseError as e:
print(f"解析错误: {e}")3.2 并发处理与批量推理优化
单路摄像头30fps视频流，如果逐帧调用 API 成本巨大。我实现了滑动窗口采样 + 并发批量处理的优化方案：
import asyncio
from typing import List, Tuple
from collections import deque
import threading
class ConcurrentInspectionScheduler:
"""并发巡检调度器 - 支持多路摄像头并行分析"""
def __init__(
self,
agent: BeltInspectionAgent,
max_concurrent: int = 10,
sampling_interval: float = 1.0 # 每秒采样1帧
):
self.agent = agent
self.max_concurrent = max_concurrent
self.sampling_interval = sampling_interval
self.semaphore = threading.Semaphore(max_concurrent)
self.results_queue = deque(maxlen=1000)
self.stats = {
"total_requests": 0,
"failed_requests": 0,
"avg_latency_ms": 0,
"total_cost_usd": 0
}
self._lock = threading.Lock()
# 价格配置（美元/百万token）- 通过 HolySheep 获取
self.price_per_mtok = {
"gpt-4.1": {"input": 2.50, "output": 8.00},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50},
"deepseek-v3.2": {"input": 0.07, "output": 0.42},
}
def _estimate_tokens(self, image_path: str, text_length: int = 500) -> Tuple[int, int]:
"""估算 token 数量（简化版）"""
# 实际生产中应使用 tiktoken 或官方 tokenizer
with open(image_path, "rb") as f:
image_size_kb = len(f.read()) / 1024
# 粗略估算：图片按分辨率，文本按字符
input_tokens = int(image_size_kb * 100 + text_length * 0.25)
output_tokens = 200 # 固定输出约200 token
return input_tokens, output_tokens
def _track_cost(self, model: str, input_tokens: int, output_tokens: int):
"""追踪成本"""
prices = self.price_per_mtok.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * prices["input"]
output_cost = (output_tokens / 1_000_000) * prices["output"]
total_cost = input_cost + output_cost
with self._lock:
self.stats["total_requests"] += 1
self.stats["total_cost_usd"] += total_cost
def process_single(self, camera_id: str, image_path: str, context: dict) -> dict:
"""处理单张图片（带并发控制）"""
with self.semaphore:
try:
result = self.agent.detect_anomaly(image_path, context)
# 估算成本
input_tok, output_tok = self._estimate_tokens(image_path)
self._track_cost("gpt-4.1", input_tok, output_tok)
return {
"camera_id": camera_id,
"status": "success",
"result": result,
"error": None
}
except Exception as e:
with self._lock:
self.stats["failed_requests"] += 1
return {
"camera_id": camera_id,
"status": "failed",
"result": None,
"error": str(e)
}
def batch_process(
self,
tasks: List[Tuple[str, str, dict]]
) -> List[dict]:
"""批量处理多路摄像头数据"""
# tasks: [(camera_id, image_path, context), ...]
results = []
with ThreadPoolExecutor(max_workers=self.max_concurrent) as executor:
futures = [
executor.submit(self.process_single, cam_id, img_path, ctx)
for cam_id, img_path, ctx in tasks
]
for future in futures:
results.append(future.result())
return results
def get_stats(self) -> dict:
"""获取统计信息"""
with self._lock:
stats = self.stats.copy()
if stats["total_requests"] > 0:
stats["success_rate"] = (
(stats["total_requests"] - stats["failed_requests"])
/ stats["total_requests"]
)
# 估算月度成本（假设持续运行）
stats["estimated_monthly_cost_usd"] = stats["total_cost_usd"] * 30 * 24 * 3600
stats["estimated_monthly_cost_cny"] = stats["estimated_monthly_cost_usd"]
# HolySheep 汇率优势：¥1=$1
stats["saving_vs_official_85pct"] = stats["estimated_monthly_cost_usd"] * 6.3
return stats
性能基准测试
def benchmark():
"""性能基准测试"""
import time
import random
agent = BeltInspectionAgent()
scheduler = ConcurrentInspectionScheduler(agent, max_concurrent=5)
# 模拟10路摄像头
test_tasks = [
(f"cam_{i}", f"./test_images/belt_{i}.jpg", {"speed": 3.5, "load": random.randint(50, 100)})
for i in range(10)
]
print("=" * 50)
print("HolySheep API 性能基准测试")
print("=" * 50)
# 单次测试
start = time.time()
single_result = scheduler.process_single("cam_test", "./test_images/sample.jpg", {})
single_time = (time.time() - start) * 1000
print(f"单次请求延迟: {single_time:.1f}ms")
# 批量测试
start = time.time()
batch_results = scheduler.batch_process(test_tasks)
batch_time = (time.time() - start) * 1000
print(f"10路并发总耗时: {batch_time:.1f}ms")
print(f"平均每路耗时: {batch_time/10:.1f}ms")
print(f"并发效率: {(single_time * 10) / batch_time:.2f}x")
# 统计
stats = scheduler.get_stats()
print(f"成功率: {stats.get('success_rate', 0):.1%}")
print(f"总成本: ${stats['total_cost_usd']:.6f}")
if __name__ == "__main__":
benchmark()四、Gemini 红外热像分析 Agent
4.1 红外热像温度异常检测
import json
from dataclasses import dataclass
from typing import List, Optional, Tuple
from datetime import datetime
import requests
class ThermalAnalysisAgent:
"""红外热像分析 Agent - 使用 Gemini 2.5 Flash"""
# 温度阈值配置
THRESHOLDS = {
"ambient": 40, # 环境温度上限
"normal": 65, # 正常温度上限
"warning": 80, # 警告温度
"critical": 100, # 危险温度
}
def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def _build_thermal_prompt(self) -> str:
"""构建红外图像分析 prompt"""
return """你是一位专业的红外热像分析工程师。请分析皮带输送机的红外热图像：
分析要点：
1. 识别异常发热区域（托辊、轴承、皮带接头等）
2. 测量最高温度点和平均温度
3. 评估热分布均匀性
4. 判断是否存在轴承损坏、润滑不足、负载过重等问题
输出严格 JSON 格式：
{
"has_anomaly": true/false,
"max_temperature": 85.5,
"min_temperature": 32.1,
"avg_temperature": 48.3,
"hotspots": [
{
"location": {"x": 0-100, "y": 0-100},
"temperature": 85.5,
"component": "roller| bearing| joint|other",
"severity": "low|medium|high|critical"
}
],
"thermal_distribution": "uniform|slightly_uneven|uneven|critical_uneven",
"recommendation": "处理建议（30字以内）"
}
如果图像质量问题导致无法分析，设置 has_anomaly 为 false 并在 recommendation 说明。"""
def analyze_thermal_image(
self,
thermal_image_path: str,
ambient_temp: float = None
) -> dict:
"""分析红外热像"""
# 编码图片
with open(thermal_image_path, "rb") as f:
image_base64 = base64.b64encode(f.read()).decode("utf-8")
# 构建上下文
context = ""
if ambient_temp:
context = f"\n环境参考温度: {ambient_temp}℃"
payload = {
"model": "gemini-2.5-flash",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": self._build_thermal_prompt() + context
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
],
"max_tokens": 600,
"temperature": 0.1
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise Exception(f"API 错误: {response.status_code}")
result = response.json()
content = result["choices"][0]["message"]["content"]
# 解析响应
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
data = json.loads(content.strip())
# 自动评估告警级别
data["alert_level"] = self._evaluate_alert_level(data, ambient_temp)
return data
def _evaluate_alert_level(self, data: dict, ambient_temp: float = None) -> str:
"""评估告警级别"""
max_temp = data.get("max_temperature", 0)
hotspots = data.get("hotspots", [])
# 基于最高温度
if max_temp >= self.THRESHOLDS["critical"]:
return "CRITICAL"
elif max_temp >= self.THRESHOLDS["warning"]:
return "WARNING"
elif max_temp >= self.THRESHOLDS["normal"]:
return "CAUTION"
# 基于热点数量和严重程度
critical_hotspots = [h for h in hotspots if h.get("severity") == "critical"]
if len(critical_hotspots) >= 2:
return "CRITICAL"
high_hotspots = [h for h in hotspots if h.get("severity") == "high"]
if len(high_hotspots) >= 2:
return "WARNING"
return "NORMAL"
多模态数据融合
class MultiModalFusion:
"""多模态数据融合引擎"""
def __init__(self, belt_agent: BeltInspectionAgent, thermal_agent: ThermalAnalysisAgent):
self.belt_agent = belt_agent
self.thermal_agent = thermal_agent
self.fusion_weights = {
"belt_anomaly": 0.5,
"thermal_anomaly": 0.4,
"temporal": 0.1
}
def comprehensive_analysis(
self,
visible_image_path: str,
thermal_image_path: str,
context: dict = None
) -> dict:
"""综合分析 - 融合可见光和红外数据"""
results = {
"timestamp": datetime.now().isoformat(),
"visible_light": None,
"thermal": None,
"fusion": None
}
# 并行调用两个 Agent
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
future_belt = executor.submit(
self.belt_agent.detect_anomaly,
visible_image_path,
context
)
future_thermal = executor.submit(
self.thermal_agent.analyze_thermal_image,
thermal_image_path,
context.get("ambient_temp") if context else None
)
try:
results["visible_light"] = future_belt.result(timeout=30)
except Exception as e:
results["visible_light"] = {"error": str(e)}
try:
results["thermal"] = future_thermal.result(timeout=30)
except Exception as e:
results["thermal"] = {"error": str(e)}
# 融合分析
results["fusion"] = self._fuse_results(results)
return results
def _fuse_results(self, results: dict) -> dict:
"""融合两个分析结果"""
belt_anomaly = results.get("visible_light", {})
thermal_anomaly = results.get("thermal", {})
# 计算综合风险评分
risk_score = 0.0
# 皮带异常权重
if belt_anomaly.get("anomaly_type") != "normal":
severity_map = {"low": 0.2, "medium": 0.5, "high": 0.8, "critical": 1.0}
risk_score += severity_map.get(belt_anomaly.get("severity", "low"), 0) * 0.5
# 热异常权重
if thermal_anomaly.get("has_anomaly"):
severity_map = {"low": 0.2, "medium": 0.5, "high": 0.8, "critical": 1.0}
hotspots = thermal_anomaly.get("hotspots", [])
max_severity = max(
[severity_map.get(h.get("severity", "low"), 0) for h in hotspots],
default=0
)
risk_score += max_severity * 0.4
# 风险等级
if risk_score >= 0.8:
risk_level = "CRITICAL"
elif risk_score >= 0.6:
risk_level = "HIGH"
elif risk_score >= 0.3:
risk_level = "MEDIUM"
else:
risk_level = "LOW"
return {
"risk_score": round(risk_score, 2),
"risk_level": risk_level,
"requires_immediate_action": risk_score >= 0.6,
"recommended_action": self._generate_action(risk_level)
}
def _generate_action(self, risk_level: str) -> str:
"""生成建议操作"""
actions = {
"CRITICAL": "立即停机检修",
"HIGH": "2小时内安排检修",
"MEDIUM": "24小时内安排巡检",
"LOW": "例行监控"
}
return actions.get(risk_level, "继续监控")
使用示例
if __name__ == "__main__":
# 初始化 Agent
belt_agent = BeltInspectionAgent()
thermal_agent = ThermalAnalysisAgent()
fusion = MultiModalFusion(belt_agent, thermal_agent)
# 综合分析
result = fusion.comprehensive_analysis(
visible_image_path="./images/visible_001.jpg",
thermal_image_path="./images/thermal_001.jpg",
context={"speed": 3.5, "load": 75, "ambient_temp": 28}
)
print(f"风险评分: {result['fusion']['risk_score']}")
print(f"风险等级: {result['fusion']['risk_level']}")
print(f"建议操作: {result['fusion']['recommended_action']}")五、SLA 监控与成本优化
5.1 API SLA 监控实现
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from collections import defaultdict
import threading
import statistics
@dataclass
class SLAConfig:
"""SLA 配置"""
p50_latency_ms: float = 500
p95_latency_ms: float = 2000
p99_latency_ms: float = 5000
max_error_rate: float = 0.01 # 1%
max_timeout_rate: float = 0.001 # 0.1%
@dataclass
class APIMetrics:
"""API 指标"""
request_id: str
model: str
timestamp: datetime
latency_ms: float
success: bool
error_type: Optional[str] = None
error_message: Optional[str] = None
input_tokens: int = 0
output_tokens: int = 0
cost_usd: float = 0.0
class SLAMonitor:
"""SLA 监控器 - 监控 HolySheep API 服务质量"""
def __init__(self, config: SLAConfig = None):
self.config = config or SLAConfig()
self.metrics: List[APIMetrics] = []
self._lock = threading.Lock()
self.alerts: List[dict] = []
# SLA 阈值（与 HolySheep 官方 SLA 对齐）
self.sla_targets = {
"availability": 99.9, # 可用率 99.9%
"latency_p95": 2000, # P95 延迟 < 2s
"error_rate": 0.1, # 错误率 < 0.1%
}
def record_request(self, metrics: APIMetrics):
"""记录 API 请求"""
with self._lock:
self.metrics.append(metrics)
# 检查是否触发告警
self._check_alerts(metrics)
# 只保留最近10000条记录
if len(self.metrics) > 10000:
self.metrics = self.metrics[-10000:]
def _check_alerts(self, metrics: APIMetrics):
"""检查是否触发告警"""
# 高延迟告警
if metrics.latency_ms > self.config.p95_latency_ms:
self.alerts.append({
"type": "high_latency",
"severity": "warning" if metrics.latency_ms < 5000 else "critical",
"model": metrics.model,
"latency_ms": metrics.latency_ms,
"timestamp": metrics.timestamp.isoformat(),
"request_id": metrics.request_id
})
# 错误告警
if not metrics.success:
self.alerts.append({
"type": "request_error",
"severity": "error",
"model": metrics.model,
"error_type": metrics.error_type,
"error_message": metrics.error_message,
"timestamp": metrics.timestamp.isoformat(),
"request_id": metrics.request_id
})
def get_current_stats(self, window_minutes: int = 5) -> dict:
"""获取当前统计信息"""
now = datetime.now()
cutoff = now - timedelta(minutes=window_minutes)
with self._lock:
recent = [m for m in self.metrics if m.timestamp >= cutoff]
if not recent:
return {"status": "no_data", "window_minutes": window_minutes}
latencies = [m.latency_ms for m in recent]
successes = sum(1 for m in recent if m.success)
total_cost = sum(m.cost_usd for m in recent)
return {
"window": f"最近{window_minutes}分钟",
"total_requests": len(recent),
"success_rate": successes / len(recent),
"error_rate": 1 - (successes / len(recent)),
"latency": {
"avg": statistics.mean(latencies),
"p50": statistics.median(latencies),
"p95": self._percentile(latencies, 0.95),
"p99": self._percentile(latencies, 0.99),
"max": max(latencies)
},
"total_cost_usd": total_cost,
"cost_per_request": total_cost / len(recent),
"alerts_count": len([a for a in self.alerts if datetime.fromisoformat(a["timestamp"]) >= cutoff])
}
def check_sla_compliance(self) -> dict:
"""检查 SLA 合规性"""
stats = self.get_current_stats(window_minutes=60) # 过去1小时
if "status" in stats and stats["status"] == "no_data":
return {"status": "no_data"}
compliance = {
"availability": {
"actual": (1 - stats["error_rate"]) * 100,
"target": self.sla_targets["availability"],
"compliant": (1 - stats["error_rate"]) * 100 >= self.sla_targets["availability"]
},
"latency_p95": {
"actual_ms": stats["latency"]["p95"],
"target_ms": self.sla_targets["latency_p95"],
"compliant": stats["latency"]["p95"] <= self.sla_targets["latency_p95"]
},
"error_rate": {
"actual": stats["error_rate"] * 100,
"target": self.sla_targets["error_rate"] * 100,
"compliant": stats["error_rate"] <= self.sla_targets["error_rate"]
}
}
compliance["overall"] = all(c["compliant"] for c in compliance.values())
return compliance
def _percentile(self, data: List[float], percentile: float) -> float:
"""计算百分位数"""
if not data:
return 0.0
sorted_data = sorted(data)
index = int(len(sorted_data) * percentile)
return sorted_data[min(index, len(sorted_data) - 1)]
def generate_report(self) -> str:
"""生成 SLA 报告"""
current = self.get_current_stats(window_minutes=60)
compliance = self.check_sla_compliance()
report = f"""
═══════════════════════════════════════════════════════════════
HolySheep API SLA 监控报告
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
═══════════════════════════════════════════════════════════════
【60分钟统计】
• 总请求数: {current.get('total_requests', 0)}
• 成功率: {current.get('success_rate', 0):.2%}
• 平均延迟: {current.get('lat