真实案例：错误信息导致的凌晨三点紧急修复
作为一名后端开发工程师，我至今记得那个让人崩溃的夜晚。凌晨三点，我的手机突然响起警报——生产环境的 AI 服务彻底宕机。登录服务器查看日志，满屏都是令人头疼的错误信息：
ConnectionError: timeout after 30 seconds
HTTPSConnectionPool(host='api.openai.com', port=443): Max retries exceeded
httpx.ReadTimeout: HTTP 504 Gateway Timeout更糟糕的是，当时代码中硬编码的 API 密钥突然失效，账单欠费导致账户被冻结。我不得不在半夜联系财务申请紧急额度，一边忍受着客户的催促电话，一边眼睁睁看着系统瘫痪。
这次惨痛的经历让我下定决心彻底重构 AI API 调用架构。如果你也遇到过类似的困境，或者正在为高昂的 API 成本和不确定的稳定性而烦恼，这篇文章将分享我从血泪教训中总结出的完整解决方案。
理解 AI API 资源利用率的本质
AI API 资源利用率并不仅仅是「用多少花多少」这么简单。它涉及四个核心维度：
- 响应延迟优化：每次 API 调用的平均响应时间
- 吞吐量管理：单位时间内能够处理的请求数量
- 成本效率：每美元预算能够产出的有效 token 数量
- 错误恢复：系统对异常情况的容错和处理能力
大多数开发者在集成 AI API 时，往往只关注功能实现，忽略了这些关键指标。等到真正上线运营时，才发现成本远超预期，延迟严重影响用户体验，错误处理更是一片空白。
基础配置：正确的 API 调用方式
首先，我们来看一个标准的 AI API 调用配置。以下示例使用 HolySheep AI 作为服务提供商，它提供低于 50 毫秒的响应延迟，并且采用美元等价计价（¥1=$1），相比其他主流平台可节省超过 85% 的成本。
import httpx
import asyncio
from typing import Optional, Dict, Any
class AIServiceConfig:
"""AI 服务配置类"""
# HolySheep AI 官方端点
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
# 连接池配置
MAX_CONNECTIONS = 100
MAX_KEEPALIVE_CONNECTIONS = 20
# 超时配置（毫秒）
CONNECT_TIMEOUT = 5000
READ_TIMEOUT = 30000
WRITE_TIMEOUT = 10000
POOL_TIMEOUT = 5
# 重试配置
MAX_RETRIES = 3
RETRY_BACKOFF_FACTOR = 0.5
@classmethod
def create_client(cls) -> httpx.AsyncClient:
"""创建配置好的 HTTP 客户端"""
return httpx.AsyncClient(
base_url=cls.BASE_URL,
timeout=httpx.Timeout(
connect=cls.CONNECT_TIMEOUT / 1000,
read=cls.READ_TIMEOUT / 1000,
write=cls.WRITE_TIMEOUT / 1000,
pool=cls.POOL_TIMEOUT
),
limits=httpx.Limits(
max_connections=cls.MAX_CONNECTIONS,
max_keepalive_connections=cls.MAX_KEEPALIVE_CONNECTIONS
),
headers={
"Authorization": f"Bearer {cls.API_KEY}",
"Content-Type": "application/json"
}
)
使用示例
config = AIServiceConfig()
client = config.create_client()
print(f"连接池已创建，最大连接数: {config.MAX_CONNECTIONS}")智能请求管理：提升资源利用效率
仅仅配置好客户端是不够的，我们需要一套完整的请求管理机制来最大化资源利用率。以下是一个经过生产环境验证的高级请求管理类：
import asyncio
import time
import logging
from dataclasses import dataclass, field
from typing import List, Optional, Callable
from collections import deque
import httpx
logger = logging.getLogger(__name__)
@dataclass
class RequestMetrics:
"""请求指标追踪"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_tokens: int = 0
total_cost_usd: float = 0.0
avg_latency_ms: float = 0.0
last_request_time: float = field(default_factory=time.time)
# 滑动窗口延迟追踪
_latency_window: deque = field(default_factory=lambda: deque(maxlen=100))
def record_success(self, latency_ms: float, tokens: int, cost: float):
self.total_requests += 1
self.successful_requests += 1
self.total_tokens += tokens
self.total_cost_usd += cost
self._latency_window.append(latency_ms)
self.avg_latency_ms = sum(self._latency_window) / len(self._latency_window)
self.last_request_time = time.time()
def record_failure(self):
self.total_requests += 1
self.failed_requests += 1
class SmartRequestManager:
"""智能请求管理器"""
def __init__(
self,
client: httpx.AsyncClient,
rate_limit: int = 60, # 每分钟请求数
burst_limit: int = 10, # 突发请求上限
circuit_breaker_threshold: int = 5,
circuit_breaker_timeout: float = 30.0
):
self.client = client
self.rate_limit = rate_limit
self.burst_limit = burst_limit
self.metrics = RequestMetrics()
# 令牌桶算法
self._tokens = burst_limit
self._last_refill = time.time()
self._lock = asyncio.Lock()
# 熔断器
self._failure_count = 0
self._circuit_open = False
self._circuit_open_time = 0
self.circuit_breaker_threshold = circuit_breaker_threshold
self.circuit_breaker_timeout = circuit_breaker_timeout
async def _refill_tokens(self):
"""补充令牌"""
now = time.time()
elapsed = now - self._last_refill
tokens_to_add = elapsed * (self.rate_limit / 60)
self._tokens = min(self.burst_limit, self._tokens + tokens_to_add)
self._last_refill = now
async def _acquire_token(self):
"""获取令牌"""
async with self._lock:
await self._refill_tokens()
while self._tokens < 1:
await asyncio.sleep(0.1)
await self._refill_tokens()
self._tokens -= 1
def _should_open_circuit(self) -> bool:
"""检查是否应该开启熔断器"""
if self._failure_count >= self.circuit_breaker_threshold:
if time.time() - self._circuit_open_time > self.circuit_breaker_timeout:
logger.info("熔断器尝试恢复")
self._failure_count = 0
self._circuit_open = False
return False
return True
return False
async def chat_completion(
self,
messages: List[Dict],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 1000
) -> Optional[Dict]:
"""发送聊天完成请求"""
# 检查熔断器
if self._circuit_open:
if self._should_open_circuit():
raise Exception("Circuit breaker is open - service unavailable")
# 获取令牌
await self._acquire_token()
start_time = time.time()
try:
response = await self.client.post(
"/chat/completions",
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
tokens = usage.get("total_tokens", 0)
cost = self._calculate_cost(model, tokens)
self.metrics.record_success(latency_ms, tokens, cost)
self._failure_count = 0
return data
else:
self._handle_error(response)
except Exception as e:
self._failure_count += 1
if self._failure_count == 1:
self._circuit_open_time = time.time()
self.metrics.record_failure()
raise
def _calculate_cost(self, model: str, tokens: int) -> float:
"""计算请求成本（基于 2026 年定价）"""
pricing = {
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42 # $0.42/MTok
}
rate = pricing.get(model, 8.0)
return (tokens / 1_000_000) * rate
def _handle_error(self, response: httpx.Response):
"""处理错误响应"""
error_messages = {
401: "API 密钥无效或已过期",
429: "请求频率超限，请稍后重试",
500: "服务器内部错误",
502: "网关错误，服务暂时不可用",
503: "服务暂时不可用"
}
raise Exception(error_messages.get(response.status_code, f"未知错误: {response.status_code}"))
def get_metrics_report(self) -> Dict:
"""获取指标报告"""
success_rate = (self.metrics.successful_requests / max(1, self.metrics.total_requests)) * 100
return {
"总请求数": self.metrics.total_requests,
"成功请求": self.metrics.successful_requests,
"失败请求": self.metrics.failed_requests,
"成功率": f"{success_rate:.2f}%",
"总 Token 数": self.metrics.total_tokens,
"总成本": f"${self.metrics.total_cost_usd:.4f}",
"平均延迟": f"{self.metrics.avg_latency_ms:.2f}ms",
"熔断器状态": "开启" if self._circuit_open else "关闭"
}
使用示例
async def main():
client = AIServiceConfig.create_client()
manager = SmartRequestManager(client, rate_limit=60, burst_limit=10)
messages = [{"role": "user", "content": "你好，请介绍一下自己"}]
try:
response = await manager.chat_completion(messages, model="deepseek-v3.2")
print(f"响应: {response['choices'][0]['message']['content']}")
print(f"\n指标报告: {manager.get_metrics_report()}")
except Exception as e:
print(f"请求失败: {e}")
asyncio.run(main())
高级优化策略：批量处理与缓存
对于需要处理大量请求的场景，批量处理和智能缓存可以显著提升资源利用率。以下是一个完整的优化方案：
import hashlib
import json
import asyncio
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
import aioredis
class BatchProcessor:
"""批量请求处理器"""
def __init__(self, request_manager: SmartRequestManager, batch_size: int = 10):
self.manager = request_manager
self.batch_size = batch_size
self._queue: asyncio.Queue = asyncio.Queue()
self._responses: Dict[str, asyncio.Future] = {}
async def start(self):
"""启动批量处理器"""
asyncio.create_task(self._process_loop())
async def _process_loop(self):
"""批量处理循环"""
while True:
batch = []
# 收集批次请求
while len(batch) < self.batch_size:
try:
item = await asyncio.wait_for(
self._queue.get(),
timeout=0.5 # 500ms 超时，强制处理当前批次
)
batch.append(item)
except asyncio.TimeoutError:
break
if batch:
await self._execute_batch(batch)
async def _execute_batch(self, batch: List[Dict]):
"""执行批次请求"""
tasks = []
for item in batch:
task = asyncio.create_task(
self.manager.chat_completion(
item["messages"],
item.get("model", "deepseek-v3.2"),
item.get("temperature", 0.7),
item.get("max_tokens", 1000)
)
)
tasks.append((item["request_id"], task))
# 并发执行所有请求
results = await asyncio.gather(*[t[1] for t in tasks], return_exceptions=True)
# 填充响应
for (request_id, _), result in zip(tasks, results):
if isinstance(result, Exception):
self._responses[request_id].set_exception(result)
else:
self._responses[request_id].set_result(result)
async def submit(
self,
messages: List[Dict],
model: str = "deepseek-v3.2",
timeout: float = 30.0
) -> Optional[Dict]:
"""提交请求"""
request_id = hashlib.md5(
json.dumps(messages, sort_keys=True).encode()
).hexdigest()
future = asyncio.Future()
self._responses[request_id] = future
await self._queue.put({
"request_id": request_id,
"messages": messages,
"model": model
})
try:
return await asyncio.wait_for(future, timeout)
except asyncio.TimeoutError:
raise Exception(f"请求超时（{timeout}s）")
finally:
del self._responses[request_id]
class IntelligentCache:
"""智能语义缓存"""
def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
self.redis_url = redis_url
self.ttl = ttl
self._local_cache: Dict[str, tuple] = {}
self._max_local_items = 1000
def _generate_key(self, messages: List[Dict], model: str, temperature: float) -> str:
"""生成缓存键"""
content = json.dumps({
"messages": messages,
"model": model,
"temperature": temperature
}, sort_keys=True)
return f"ai_cache:{hashlib.sha256(content.encode()).hexdigest()[:32]}"
async def get(
self,
messages: List[Dict],
model: str,
temperature: float
) -> Optional[Dict]:
"""获取缓存"""
key = self._generate_key(messages, model, temperature)
# 先查本地缓存
if key in self._local_cache:
cached_data, expiry = self._local_cache[key]
if datetime.now() < expiry:
return cached_data
del self._local_cache[key]
# 查 Redis
try:
redis = await aioredis.from_url(self.redis_url)
cached = await redis.get(key)
if cached:
data = json.loads(cached)
# 回填本地缓存
self._set_local_cache(key, data)
return data
except Exception:
pass
return None
async def set(
self,
messages: List[Dict],
model: str,
temperature: float,
response: Dict
):
"""设置缓存"""
key = self._generate_key(messages, model, temperature)
# 写入本地缓存
self._set_local_cache(key, response)
# 写入 Redis
try:
redis = await aioredis.from_url(self.redis_url)
await redis.setex(key, self.ttl, json.dumps(response))
except Exception:
pass
def _set_local_cache(self, key: str, data: Dict):
"""设置本地缓存"""
if len(self._local_cache) >= self._max_local_items:
# 删除最旧的条目
oldest_key = min(
self._local_cache.keys(),
key=lambda k: self._local_cache[k][1]
)
del self._local_cache[oldest_key]
expiry = datetime.now() + timedelta(seconds=self.ttl)
self._local_cache[key] = (data, expiry)
使用示例
async def optimized_ai_service():
"""优化后的 AI 服务"""
client = AIServiceConfig.create_client()
manager = SmartRequestManager(client)
batch_processor = BatchProcessor(manager, batch_size=20)
cache = IntelligentCache()
await batch_processor.start()
# 示例请求
messages = [
{"role": "user", "content": "什么是人工智能？"}
]
# 尝试从缓存获取
cached = await cache.get(messages, "deepseek-v3.2", 0.7)
if cached:
print("命中缓存！")
return cached
# 发送请求
response = await batch_processor.submit(messages, "deepseek-v3.2")
# 存入缓存
await cache.set(messages, "deepseek-v3.2", 0.7, response)
return response2026年AI API定价对比与成本优化建议
了解不同模型的价格差异对于优化资源利用率至关重要。根据 2026 年最新定价：
- DeepSeek V3.2：$0.42/MTok — 性价比最高，适合大多数场景
- Gemini 2.5 Flash：$2.50/MTok — 速度快，适合实时应用
- GPT-4.1：$8.00/MTok — 能力最强，适合复杂推理任务
- Claude Sonnet 4.5：$15.00/MTok — 适合长文本处理
HolySheep AI 作为新一代 AI API 提供商，采用美元等价计价（¥1=$1），支持微信和支付宝支付，响应延迟低于 50 毫秒，并且新用户注册即送免费额度，是中小型项目的理想选择。
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
在开发和运维 AI API 集成项目时，我整理了最常见的三大问题及其完整解决方案：
错误一：401 Unauthorized - API 密钥无效或已过期
# 错误日志示例
httpx.HTTPStatusError: 401 Client Error: Unauthorized
{"error": {"message": "Invalid API key provided", "type": "invalid_request_error"}}
import os
from dotenv import load_dotenv
class APIKeyManager:
"""API 密钥管理器"""
def __init__(self):
load_dotenv()
self._api_key = None
self._validate_key()
def _validate_key(self):
"""验证 API 密钥格式"""
self._api_key = os.getenv("HOLYSHEEP_API_KEY")
if not self._api_key:
raise ValueError(
"未设置 API 密钥！请在 .env 文件中设置 HOLYSHEEP_API_KEY\n"
"获取密钥地址：https://www.holysheep.ai/register"
)
if len(self._api_key) < 20:
raise ValueError("API 密钥格式不正确，长度应至少为 20 个字符")
if self._api_key.startswith("sk-"):
# 这是 OpenAI 格式的密钥，HolySheep 使用不同格式
raise ValueError(
"检测到 OpenAI 格式的密钥！\n"
"HolySheep AI 使用专属密钥格式，请在后台重新获取"
)
def get_key(self) -> str:
"""获取验证后的密钥"""
return self._api_key
使用方式
try:
key_manager = APIKeyManager()
valid_key = key_manager.get_key()
print(f"密钥验证通过，长度: {len(valid_key)}")
except ValueError as e:
print(f"密钥配置错误: {e}")错误二：ConnectionError / Timeout - 连接超时
# 错误日志示例
asyncio.exceptions.TimeoutError: Request timed out
httpx.ConnectTimeout: Connection timeout after 5000ms
HTTPSConnectionPool(host='api.holysheep.ai', port=443): Max retries exceeded
import asyncio
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
class ResilientHTTPClient:
"""具备弹性恢复能力的 HTTP 客户端"""
def __init__(
self,
base_url: str = "https://api.holysheep.ai/v1",
api_key: str = None,
max_retries: int = 3,
timeout: float = 30.0
):
self.base_url = base_url
self.api_key = api_key
self.timeout = timeout
# 配置重试策略
self.client = httpx.AsyncClient(
base_url=base_url,
timeout=httpx.Timeout(timeout),
limits=httpx.Limits(
max_connections=50,
max_keepalive_connections=20,
keepalive_expiry=30
),
proxies=None # 不使用代理以减少延迟
)
async def post_with_retry(
self,
endpoint: str,
json_data: dict,
retries: int = 3
) -> httpx.Response:
"""带重试机制的 POST 请求"""
last_exception = None
for attempt in range(retries):
try:
response = await self.client.post(
endpoint,
json=json_data,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return response
except httpx.ConnectTimeout:
last_exception = ConnectionError(
f"连接超时（第 {attempt + 1}/{retries} 次尝试）\n"
"请检查网络连接或尝试更换网络环境"
)
except httpx.ReadTimeout:
last_exception = ConnectionError(
f"读取超时（第 {attempt + 1}/{retries} 次尝试）\n"
"服务器响应过慢，可能是负载过高"
)
except httpx.ConnectError as e:
last_exception = ConnectionError(
f"连接错误（第 {attempt + 1}/{retries} 次尝试）\n"
f"错误详情: {str(e)}"
)
except Exception as e:
last_exception = ConnectionError(
f"未知网络错误（第 {attempt + 1}/{retries} 次尝试）\n"
f"错误类型: {type(e).__name__}"
)
# 指数退避等待
if attempt < retries - 1:
wait_time = 2 ** attempt
print(f"等待 {wait_time} 秒后重试...")
await asyncio.sleep(wait_time)
raise last_exception
async def close(self):
"""关闭客户端连接"""
await self.client.aclose()
使用示例
async def test_connection():
client = ResilientHTTPClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_retries=3,
timeout=30.0
)
try:
response = await client.post_with_retry(
"/chat/completions",
{
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": "测试连接"}]
}
)
print(f"请求成功: {response.status_code}")
print(f"响应内容: {response.json()}")
except ConnectionError as e:
print(f"连接失败: {e}")
finally:
await client.close()错误三：429 Rate Limit Exceeded - 请求频率超限
# 错误日志示例
httpx.HTTPStatusError: 429 Client Error: Too Many Requests
{"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}
import time
import asyncio
from collections import deque
from typing import Optional
class RateLimitHandler:
"""速率限制处理器 - 实现精确的令牌桶算法"""
def __init__(
self,
requests_per_minute: int = 60,
requests_per_second: int = 10,
burst_size: int = 20
):
self.rpm = requests_per_minute
self.rps = requests_per_second
self.burst = burst_size
# 令牌桶状态
self._tokens = burst_size
self._last_update = time.time()
# 请求时间戳记录（用于 RPM 控制）
self._request_times: deque = deque(maxlen=requests_per_minute + 1)
# 锁
self._lock = asyncio.Lock()
# 指数退避状态
self._backoff_until: float = 0
self._backoff_factor: float = 1.0
def _refill_tokens(self):
"""补充令牌"""
now = time.time()
elapsed = now - self._last_update
# 每秒补充 RPS 个令牌
tokens_to_add = elapsed * self.rps
self._tokens = min(self.burst, self._tokens + tokens_to_add)
self._last_update = now
def _is_rate_limited(self) -> tuple[bool, Optional[float]]:
"""检查是否触发速率限制"""
now = time.time()
# 检查是否处于退避期
if now < self._backoff_until:
retry_after = self._backoff_until - now
return True, retry_after
# 检查 RPM 限制
current_time = time.time()
recent_requests = [
t for t in self._request_times
if current_time - t < 60
]
if len(recent_requests) >= self.rpm:
oldest_request = min(recent_requests)
retry_after = 60 - (current_time - oldest_request)
return True, max(0.1, retry_after)
return False, None
async def acquire(self, timeout: float = 60.0) -> bool:
"""获取请求许可"""
start_time = time.time()
while True:
async with self._lock:
self._refill_tokens()
# 检查速率限制
is_limited, retry_after = self._is_rate_limited()
if is_limited and retry_after:
# 如果超时则抛出异常
if time.time() - start_time + retry_after > timeout:
raise TimeoutError(
f"等待速率限制释放超时（已等待 {timeout}s）"
)
# 增加退避因子
self._backoff_factor = min(4.0, self._backoff_factor * 1.5)
wait_time = retry_after * self._backoff_factor
print(f"触发速率限制，等待 {wait_time:.2f}s（退避因子: {self._backoff_factor}）")
await asyncio.sleep(wait_time)
continue
# 获取令牌
if self._tokens >= 1:
self._tokens -= 1
self._request_times.append(time.time())
# 重置退避因子
self._backoff_factor = 1.0
return True
# 未获取到令牌，稍后重试
await asyncio.sleep(0.05)
def handle_rate_limit_response(self, retry_after: Optional[int] = None):
"""处理 429 响应"""
wait_time = retry_after if retry_after else 60 // self.rpm
self._backoff_until = time.time() + wait_time * 2
print(f"服务器返回 429，将在 {wait_time * 2}s 后重试")
装饰器方式使用
def rate_limited(rpm: int = 60, rps: int = 10):
"""速率限制装饰器"""
handler = RateLimitHandler(rpm, rps)
def decorator(func):
async def wrapper(*args, **kwargs):
await handler.acquire(timeout=120)
return await func(*args, **kwargs)
return wrapper
return decorator
使用示例
async def main():
handler = RateLimitHandler(rpm=60, rps=10)
for i in range(65): # 超过 RPM 限制
try:
await handler.acquire(timeout=10)
print(f"请求 {i + 1}: 成功发送")
except TimeoutError as e:
print(f"请求 {i + 1}: {e}")
break
await asyncio.sleep(0.1) # 模拟请求间隔
asyncio.run(main())
监控与告警：生产环境的必备保障
即使代码逻辑完美，生产环境仍然可能遇到各种突发状况。建立完善的监控和告警系统是保障服务稳定性的最后一道防线。
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Callable
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AIMonitor:
"""AI API 监控系统"""
def __init__(self, metrics: RequestMetrics, alert_threshold: Dict):
self.metrics = metrics
self.alert_threshold = alert_threshold
self._alert_history: List[Dict] = []
self._alert_callbacks: List[Callable] = []
def add_alert_callback(self, callback: Callable):
"""添加告警回调函数"""
self._alert_callbacks.append(callback)
def check_and_alert(self) -> List[str]:
"""检查指标并触发告警"""
alerts = []
now = datetime.now()
# 检查错误率
if self.metrics.total_requests > 0:
error_rate = self.metrics.failed_requests / self.metrics.total_requests
if error_rate > self.alert_threshold.get("error_rate", 0.1):
alert_msg = (
f"【严重】错误率过高: {error_rate*100:.2f}%\n"
f"失败请求数: {self.metrics.failed_requests}/{self.metrics.total_requests}"
)
alerts.append(alert_msg)
# 检查平均延迟
if self.metrics.avg_latency_ms > self.alert_threshold.get("latency_ms", 5000):
alert_msg = (
f"【警告】平均延迟过高: {self.metrics.avg_latency_ms:.2f}ms\n"
f"阈值: {self.alert_threshold['latency_ms']}ms"
)
alerts.append(alert_msg)
# 检查成本
if self.metrics.total_cost_usd > self.alert_threshold.get("cost_usd", 100):
alert_msg = (
f"【通知】API 成本超出预期: ${self.metrics.total_cost_usd:.2f}\n"
f"Token 使用量: {self.metrics.total_tokens:,}"
)
alerts.append(alert_msg)
# 检查最后请求时间
time_since_last = (now - datetime.fromtimestamp(self.metrics.last_request_time)).total_seconds()
if time_since_last > 300: # 5 分钟无请求
alert_msg = f"【信息】系统已 {time_since_last/60:.1f} 分钟无活动请求"
alerts.append(alert_msg)
# 触发告警回调
for alert in alerts:
logger.warning(alert)
for callback in self._alert_callbacks:
try:
callback(alert)
except Exception as e:
logger.error(f"告警回调执行失败: {e}")
return alerts
告警回调示例：发送邮件
def email_alert(recipients: List[str]):
"""创建邮件告警回调"""
def callback(message: str):
msg = MIMEMultipart()
msg['From'] = "[email protected]"
msg['To'] = ", ".join(recipients)
msg['Subject'] = "AI API 监控系统告警"
msg.attach(MIMEText(message, 'plain'))
try:
with smtplib.SMTP('smtp.company.com', 587) as server:
server.starttls()
server.login("[email protected]", "password")
server.send_message(msg)
except Exception as e:
logger.error(f"邮件发送失败: {e}")
return callback
使用示例
monitor = AIMonitor(
metrics=RequestMetrics(),
alert_threshold={
"error_rate": 0.05, # 5% 错误率告警
"