作为在生产环境处理过日均千万级 API 调用的工程师，我深知日志分析与异常检测对于保障服务稳定性的关键价值。本文将从零构建一套完整的日志分析系统，涵盖架构设计、代码实现、性能调优与成本优化，代码可直接部署至生产环境。
如果你还没有 立即注册 HolySheep AI，建议先注册获取免费额度，我们的测试将在这个平台上进行。
一、为什么需要 API 日志分析与异常检测
在 AI API 调用场景中，日志分析的价值体现在三个维度：
- 成本控制：Token 消耗是按量计费，日志分析能精准定位异常调用，防止预算超支
- 性能优化：通过延迟分布分析，找出 P99 延迟异常点，针对性优化
- 稳定性保障：异常检测能在服务中断前预警，将 MTTR（平均恢复时间）从分钟级降至秒级
二、整体架构设计
我设计的架构包含四个核心组件：
- 日志采集层：Python SDK 自动拦截 + 结构化输出
- 传输缓冲层：本地 Redis 队列，异步批量上传
- 存储分析层：Elasticsearch + Kibana 可视化
- 告警层：基于规则 + 机器学习的双层异常检测
三、核心代码实现
3.1 日志采集客户端（生产级代码）
import hashlib
import json
import time
import asyncio
from datetime import datetime
from typing import Optional, Dict, Any
from dataclasses import dataclass, asdict
import redis
import httpx
@dataclass
class APIRequestLog:
"""API 请求日志结构"""
request_id: str
timestamp: float
model: str
input_tokens: int
output_tokens: int
latency_ms: float
status_code: int
error_message: Optional[str] = None
user_id: Optional[str] = None
session_id: Optional[str] = None
def to_json(self) -> str:
return json.dumps(asdict(self), ensure_ascii=False)
@property
def total_cost(self) -> float:
"""计算单次请求成本（基于 HolySheep 2026 价格）"""
model_prices = {
"gpt-4.1": (3.0, 8.0), # input/output $/MTok
"claude-sonnet-4.5": (3.0, 15.0),
"gemini-2.5-flash": (0.35, 2.50),
"deepseek-v3.2": (0.27, 0.42)
}
if self.model not in model_prices:
return 0.0
input_price, output_price = model_prices[self.model]
return (self.input_tokens / 1_000_000) * input_price + \
(self.output_tokens / 1_000_000) * output_price
class HolySheepAPIClient:
"""带日志分析的 HolySheep API 客户端"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, redis_host: str = "localhost", redis_port: int = 6379):
self.api_key = api_key
self.redis_client = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
self.queue_name = "api_logs:pending"
self._request_count = 0
self._total_latency = 0.0
def _generate_request_id(self) -> str:
"""生成唯一请求 ID"""
raw = f"{time.time()}-{self._request_count}-{id(self)}"
return hashlib.md5(raw.encode()).hexdigest()[:16]
async def chat_completions(
self,
messages: list,
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""调用 HolySheep Chat Completions API"""
request_id = self._generate_request_id()
start_time = time.perf_counter()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
latency_ms = (time.perf_counter() - start_time) * 1000
response_data = response.json()
# 提取 token 使用量
usage = response_data.get("usage", {})
log = APIRequestLog(
request_id=request_id,
timestamp=datetime.now().timestamp(),
model=model,
input_tokens=usage.get("prompt_tokens", 0),
output_tokens=usage.get("completion_tokens", 0),
latency_ms=latency_ms,
status_code=response.status_code,
error_message=None
)
# 异步写入 Redis 队列
asyncio.create_task(self._enqueue_log(log))
self._request_count += 1
self._total_latency += latency_ms
return response_data
except httpx.HTTPError as e:
latency_ms = (time.perf_counter() - start_time) * 1000
log = APIRequestLog(
request_id=request_id,
timestamp=datetime.now().timestamp(),
model=model,
input_tokens=0,
output_tokens=0,
latency_ms=latency_ms,
status_code=0,
error_message=str(e)
)
asyncio.create_task(self._enqueue_log(log))
raise
async def _enqueue_log(self, log: APIRequestLog):
"""写入 Redis 队列（异步批量上传）"""
try:
self.redis_client.rpush(self.queue_name, log.to_json())
except redis.RedisError:
# Redis 不可用时降级到本地文件
with open("/var/log/api_requests.log", "