作为一名在生产环境跑了两年多 AI Agent 的开发者，我在 2025 年经历了三次重大故障：API Key 泄露被滥用导致账单爆表、单一模型服务宕机引发全线崩溃、以及没有审计日志导致排查问题花了整整两天。这些教训让我意识到，AI API 的生产环境管理绝对不是把 Key 配进 .env 就完事了。

最近我把团队的所有 AI 调用迁移到了 HolySheep AI，花了两个月时间整理出一套完整的生产环境 checklist。今天这篇文章，我会从延迟、成功率、支付便捷性、模型覆盖、控制台体验五个维度给出真实测评分数，同时手把手教你实现 API Key 轮换、模型 fallback、成本封顶与审计四大核心功能。

一、生产环境为什么需要 checklist

很多团队在开发环境调通 AI 功能后就直接上线，结果在生产环境遇到各种问题：

我的团队去年因此造成的损失超过 2 万元，这还不算业务中断带来的隐性成本。下面这套 checklist，是用真金白银换来的经验。

二、HolySheep AI 基础信息与定价实测

在开始技术实践之前，先给你看几个关键数据，这是我在 HolySheep AI 注册后实测的结果：

测试维度测试方法实测结果评分（10分制）
国内延迟（上海）curl 测量 API 响应时间平均 38ms9.5
API 可用性7×24 小时监控 30 天99.7%9.2
支付便捷性实际充值体验微信/支付宝秒到账10
模型覆盖统计支持模型数量20+ 主流模型9.0
成本节省对比官方定价汇率节省 >85%10

2026 主流模型定价对比

模型官方价格 ($/MTok)HolySheep 价格节省比例
GPT-4.1$8.00$8.00（汇率优势）节省 85%（汇率差）
Claude Sonnet 4.5$15.00$15.00（汇率优势）节省 85%（汇率差）
Gemini 2.5 Flash$2.50$2.50（汇率优势）节省 85%（汇率差）
DeepSeek V3.2$0.42$0.42（汇率优势）节省 85%（汇率差）

注意：HolySheep 的核心优势是 ¥1=$1 无损汇率，官方人民币定价是 ¥7.3=$1，用 HolySheep 直接省下超过 85% 的汇率损耗。我上个月调用 GPT-4.1 花了 $23.6，用 HolySheep 节省了约 ¥165。

三、API Key 轮换实现

3.1 为什么需要 Key 轮换

我在去年遇到过一次惨痛教训：单个 API Key 被高频调用，触发了目标平台的 rate limit，整个服务挂了 4 小时。API Key 轮换的核心目的是：

3.2 多 Key 轮换代码实现

import random
import time
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import threading

@dataclass
class APIKeyConfig:
    key: str
    model: str
    weight: int = 1  # 权重，用于负载分配
    rpm_limit: int = 500  # 每分钟请求限制
    tpm_limit: int = 150000  # 每分钟 token 限制

class HolySheepKeyManager:
    """
    HolySheep AI API Key 轮换管理器
    支持多 Key 负载均衡、限流保护、自动切换
    """
    
    def __init__(self):
        self.keys: List[APIKeyConfig] = []
        self.request_counts: Dict[str, List[datetime]] = {}
        self.token_counts: Dict[str, List[tuple]] = {}  # (timestamp, token_count)
        self._lock = threading.Lock()
        
    def add_key(self, key: str, model: str, weight: int = 1,
                rpm_limit: int = 500, tpm_limit: int = 150000):
        """添加一个新的 API Key"""
        config = APIKeyConfig(
            key=key,
            model=model,
            weight=weight,
            rpm_limit=rpm_limit,
            tpm_limit=tpm_limit
        )
        self.keys.append(config)
        self.request_counts[key] = []
        self.token_counts[key] = []
        
    def _cleanup_old_records(self, records: List[datetime], window: int = 60):
        """清理超过时间窗口的记录"""
        cutoff = datetime.now() - timedelta(seconds=window)
        return [r for r in records if r > cutoff]
    
    def _check_rate_limit(self, key_config: APIKeyConfig) -> tuple[bool, str]:
        """检查单个 Key 是否超过限流"""
        now = datetime.now()
        
        # 检查 RPM
        recent_requests = self._cleanup_old_records(
            self.request_counts.get(key_config.key, []), 60
        )
        if len(recent_requests) >= key_config.rpm_limit:
            return False, f"RPM limit exceeded: {len(recent_requests)}/{key_config.rpm_limit}"
        
        # 检查 TPM（简化版，每分钟 token 统计）
        recent_tokens = [
            tc for tc in self.token_counts.get(key_config.key, [])
            if now - tc[0] < timedelta(seconds=60)
        ]
        total_tokens = sum(tc[1] for tc in recent_tokens)
        if total_tokens >= key_config.tpm_limit:
            return False, f"TPM limit exceeded: {total_tokens}/{key_config.tpm_limit}"
            
        return True, "OK"
    
    def get_available_key(self, model: str) -> Optional[APIKeyConfig]:
        """获取一个可用的 Key（带权重随机选择）"""
        with self._lock:
            available = []
            
            for key_config in self.keys:
                if key_config.model != model:
                    continue
                    
                can_use, reason = self._check_rate_limit(key_config)
                if can_use:
                    # 按权重添加到候选列表
                    available.extend([key_config] * key_config.weight)
                    
            if not available:
                return None
                
            return random.choice(available)
    
    def record_request(self, key: str, token_count: int):
        """记录一次请求（用于限流统计）"""
        with self._lock:
            now = datetime.now()
            if key in self.request_counts:
                self.request_counts[key].append(now)
            if key in self.token_counts:
                self.token_counts[key].append((now, token_count))
    
    def rotate_key(self, failed_key: str, model: str) -> Optional[APIKeyConfig]:
        """当 Key 失败时，轮换到备用 Key"""
        with self._lock:
            for key_config in self.keys:
                if key_config.key == failed_key or key_config.model != model:
                    continue
                can_use, _ = self._check_rate_limit(key_config)
                if can_use:
                    return key_config
        return None



使用示例

if __name__ == "__main__":
    manager = HolySheepKeyManager()
    
    # 添加多个 Key（实际使用时替换为你的 HolySheep API Keys）
    manager.add_key(
        key="YOUR_HOLYSHEEP_API_KEY_1",
        model="gpt-4.1",
        weight=3,
        rpm_limit=500,
        tpm_limit=150000
    )
    manager.add_key(
        key="YOUR_HOLYSHEEP_API_KEY_2",
        model="gpt-4.1",
        weight=2,
        rpm_limit=500,
        tpm_limit=150000
    )
    manager.add_key(
        key="YOUR_HOLYSHEEP_API_KEY_3",
        model="claude-sonnet-4.5",
        weight=2,
        rpm_limit=400,
        tpm_limit=120000
    )
    
    # 获取可用 Key
    key = manager.get_available_key("gpt-4.1")
    if key:
        print(f"Using key: {key.key[:10]}...")
        manager.record_request(key.key, 500)  # 记录 500 tokens

3.3 与 HolySheep API 对接

import requests
from typing import Optional, Dict, Any

class HolySheepAIClient:
    """
    HolySheep AI API 客户端
    base_url: https://api.holysheep.ai/v1
    """
    
    def __init__(self, key_manager: HolySheepKeyManager):
        self.base_url = "https://api.holysheep.ai/v1"
        self.key_manager = key_manager
        
    def chat_completions(
        self,
        model: str,
        messages: list,
        temperature: float = 0.7,
        max_tokens: Optional[int] = None
    ) -> Dict[str, Any]:
        """发送聊天请求，自动 Key 轮换和 fallback"""
        
        # Step 1: 获取可用 Key
        key_config = self.key_manager.get_available_key(model)
        if not key_config:
            raise Exception(f"No available key for model {model}, all keys rate limited")
        
        # Step 2: 构建请求
        headers = {
            "Authorization": f"Bearer {key_config.key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature
        }
        if max_tokens:
            payload["max_tokens"] = max_tokens
            
        # Step 3: 发送请求
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            
            if response.status_code == 200:
                # 估算 token 使用量（实际从响应中获取更准确）
                total_tokens = sum(
                    len(str(m)) // 4 for m in messages
                ) + (max_tokens or 1000)
                self.key_manager.record_request(key_config.key, total_tokens)
                return response.json()
                
            elif response.status_code == 429:
                # Rate limit，尝试其他 Key
                print(f"Rate limited on key {key_config.key[:10]}..., trying fallback")
                fallback_key = self.key_manager.rotate_key(key_config.key, model)
                if fallback_key:
                    return self._retry_with_key(model, messages, fallback_key, temperature, max_tokens)
                else:
                    raise Exception("All keys rate limited")
                    
            else:
                raise Exception(f"API error: {response.status_code} - {response.text}")
                
        except requests.exceptions.RequestException as e:
            # 网络错误，尝试 fallback
            fallback_key = self.key_manager.rotate_key(key_config.key, model)
            if fallback_key:
                return self._retry_with_key(model, messages, fallback_key, temperature, max_tokens)
            raise
            
    def _retry_with_key(
        self,
        model: str,
        messages: list,
        key_config: APIKeyConfig,
        temperature: float,
        max_tokens: Optional[int]
    ) -> Dict[str, Any]:
        """使用备用 Key 重试"""
        headers = {
            "Authorization": f"Bearer {key_config.key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature
        }
        if max_tokens:
            payload["max_tokens"] = max_tokens
            
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code == 200:
            self.key_manager.record_request(key_config.key, max_tokens or 1000)
            return response.json()
        else:
            raise Exception(f"Fallback failed: {response.status_code}")



使用示例

if __name__ == "__main__":
    key_manager = HolySheepKeyManager()
    key_manager.add_key("YOUR_HOLYSHEEP_API_KEY", "gpt-4.1", weight=1)
    
    client = HolySheepAIClient(key_manager)
    
    response = client.chat_completions(
        model="gpt-4.1",
        messages=[
            {"role": "system", "content": "你是一个有用的助手"},
            {"role": "user", "content": "解释什么是 API Key 轮换"}
        ],
        temperature=0.7,
        max_tokens=500
    )
    print(response)

四、模型 Fallback 策略

4.1 为什么需要 Fallback

我在 2025 年 8 月经历过一次 Claude API 全球性故障，整整 6 小时无法服务。如果有完善的 fallback 策略，可以自动切换到备用模型，服务中断时间可以控制在秒级。

4.2 多层 Fallback 实现

from typing import List, Optional, Callable
from enum import Enum
import logging

class FallbackLevel(Enum):
    PRIMARY = 1      # 主模型
    SECONDARY = 2    # 二级 fallback
    TERTIARY = 3     # 三级 fallback
    EMERGENCY = 4    # 紧急备用

class ModelFallbackManager:
    """
    多层模型 Fallback 管理器
    当主模型不可用时，自动切换到备用模型
    """
    
    def __init__(self):
        self.fallback_chains: dict[str, List[tuple]] = {}
        self.logger = logging.getLogger(__name__)
        self.failure_counts: dict[str, int] = {}
        self.circuit_breaker: dict[str, dict] = {}
        
    def register_chain(self, model: str, chain: List[tuple]):
        """
        注册 fallback 链
        
        Args:
            model: 主模型名称
            chain: [(model_name, weight, params), ...]
                  例如: [("gpt-4.1", 3, {}), ("claude-sonnet-4.5", 2, {}), 
                        ("gemini-2.5-flash", 2, {"temperature": 0.5}), 
                        ("deepseek-v3.2", 1, {"max_tokens": 1000})]
        """
        self.fallback_chains[model] = chain
        
    def _should_activate_circuit_breaker(self, model: str) -> bool:
        """检查是否应该激活熔断器"""
        if model not in self.circuit_breaker:
            self.circuit_breaker[model] = {
                "failures": 0,
                "last_failure": None,
                "circuit_open": False
            }
            
        cb = self.circuit_breaker[model]
        
        # 如果熔断已打开，检查是否应该关闭
        if cb["circuit_open"]:
            from datetime import datetime, timedelta
            if datetime.now() - cb["last_failure"] > timedelta(minutes=5):
                cb["circuit_open"] = False
                cb["failures"] = 0
                self.logger.info(f"Circuit breaker reset for {model}")
            return True
            
        # 检查失败次数
        if cb["failures"] >= 5:
            cb["circuit_open"] = True
            self.logger.warning(f"Circuit breaker opened for {model}")
            return True
            
        return False
        
    def _record_failure(self, model: str):
        """记录失败"""
        if model not in self.circuit_breaker:
            self._should_activate_circuit_breaker(model)
        self.circuit_breaker[model]["failures"] += 1
        self.circuit_breaker[model]["last_failure"] = datetime.now()
        
    def get_fallback_model(self, original_model: str, 
                           previous_failures: List[str]) -> Optional[tuple]:
        """获取下一个可用的 fallback 模型"""
        
        if original_model not in self.fallback_chains:
            return None
            
        chain = self.fallback_chains[original_model]
        
        for i, (model, weight, params) in enumerate(chain):
            # 跳过已失败的模型
            if model in previous_failures:
                continue
                
            # 检查熔断器
            if self._should_activate_circuit_breaker(model):
                continue
                
            return (model, params)
            
        return None
        
    def execute_with_fallback(
        self,
        client: HolySheepAIClient,
        original_model: str,
        messages: list,
        callback: Optional[Callable] = None,
        max_fallbacks: int = 3
    ) -> dict:
        """
        执行带 fallback 的请求
        
        Args:
            client: HolySheep AI 客户端
            original_model: 原始请求的模型
            messages: 消息列表
            callback: 可选的回调函数，在每次 fallback 时调用
            max_fallbacks: 最大 fallback 次数
        """
        
        current_model = original_model
        current_params = {}
        failed_models = []
        last_error = None
        
        for attempt in range(max_fallbacks + 1):
            try:
                self.logger.info(f"Attempt {attempt + 1}: using model {current_model}")
                
                response = client.chat_completions(
                    model=current_model,
                    messages=messages,
                    **current_params
                )
                
                # 成功，清除失败计数
                if current_model in self.failure_counts:
                    del self.failure_counts[current_model]
                    
                return response
                
            except Exception as e:
                last_error = e
                failed_models.append(current_model)
                self._record_failure(current_model)
                
                self.logger.warning(
                    f"Model {current_model} failed: {str(e)}, trying fallback"
                )
                
                # 获取下一个 fallback 模型
                fallback = self.get_fallback_model(original_model, failed_models)
                
                if fallback:
                    current_model, current_params = fallback
                    if callback:
                        callback(original_model, current_model)
                else:
                    break
                    
        # 所有模型都失败了
        raise Exception(
            f"All fallback models failed. Last error: {last_error}. "
            f"Failed models: {failed_models}"
        )



使用示例

if __name__ == "__main__":
    # 初始化
    key_manager = HolySheepKeyManager()
    key_manager.add_key("YOUR_HOLYSHEEP_API_KEY", "gpt-4.1", weight=1)
    key_manager.add_key("YOUR_HOLYSHEEP_API_KEY", "claude-sonnet-4.5", weight=1)
    key_manager.add_key("YOUR_HOLYSHEEP_API_KEY", "gemini-2.5-flash", weight=1)
    key_manager.add_key("YOUR_HOLYSHEEP_API_KEY", "deepseek-v3.2", weight=1)
    
    client = HolySheepAIClient(key_manager)
    fallback_manager = ModelFallbackManager()
    
    # 注册 fallback 链：从 GPT-4.1 开始，依次 fallback
    fallback_manager.register_chain(
        "gpt-4.1",
        [
            ("gpt-4.1", 3, {}),                    # 主模型
            ("claude-sonnet-4.5", 2, {"temperature": 0.5}),  # 二级
            ("gemini-2.5-flash", 2, {"temperature": 0.7, "max_tokens": 2000}),  # 三级
            ("deepseek-v3.2", 1, {"max_tokens": 1000})  # 紧急备用
        ]
    )
    
    # 定义 fallback 回调
    def on_fallback(original: str, current: str):
        print(f"⚠️ Fallback triggered: {original} → {current}")
        
    # 执行请求（会自动 fallback）
    try:
        response = fallback_manager.execute_with_fallback(
            client=client,
            original_model="gpt-4.1",
            messages=[{"role": "user", "content": "你好，请介绍一下你自己"}],
            callback=on_fallback,
            max_fallbacks=3
        )
        print(f"Success: {response['choices'][0]['message']['content'][:100]}...")
    except Exception as e:
        print(f"All models failed: {e}")

五、成本封顶与预算告警

5.1 成本监控的重要性

我在 2025 年 6 月收到一张 $847 的账单，当时整个人都懵了。后来分析日志发现，是一个 bug 导致某个接口被无限循环调用。这个故事告诉我：没有成本封顶的生产环境就是在裸奔。

5.2 成本封顶实现

from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta
import threading
import time

@dataclass
class BudgetConfig:
    """预算配置"""
    daily_limit: float = 50.0      # 每日预算（美元）
    monthly_limit: float = 500.0    # 每月预算（美元）
    per_request_limit: float = 2.0 # 单次请求上限（美元）
    warning_threshold: float = 0.8 # 告警阈值（80%）

@dataclass
class CostRecord:
    """成本记录"""
    timestamp: datetime
    amount: float
    model: str
    tokens: int
    request_id: str

class CostGuard:
    """
    成本守卫：实时监控、封顶、告警
    
    HolySheep AI 的汇率优势（¥1=$1）让成本计算更简单：
    假设你的预算是 ¥100/月，折算成 $13.7/月（约 $0.46/天）
    """
    
    def __init__(self, config: BudgetConfig):
        self.config = config
        self.daily_spend: float = 0.0
        self.monthly_spend: float = 0.0
        self.records: list[CostRecord] = []
        self.last_reset_daily = datetime.now()
        self.last_reset_monthly = datetime.now().replace(day=1, hour=0, minute=0, second=0)
        self._lock = threading.Lock()
        self._callbacks: list[callable] = []
        
        # 告警状态
        self.warning_sent_daily = False
        self.warning_sent_monthly = False
        self.blocked = False
        
    def add_warning_callback(self, callback: callable):
        """添加告警回调"""
        self._callbacks.append(callback)
        
    def _check_and_reset_periods(self):
        """检查是否需要重置周期计数"""
        now = datetime.now()
        
        # 检查每日重置
        if now.date() > self.last_reset_daily.date():
            self.daily_spend = 0.0
            self.last_reset_daily = now
            self.warning_sent_daily = False
            
        # 检查每月重置
        if now.month != self.last_reset_monthly.month:
            self.monthly_spend = 0.0
            self.last_reset_monthly = now
            self.warning_sent_monthly = False
            
    def _calculate_request_cost(self, model: str, tokens: int) -> float:
        """计算单次请求成本（基于 HolySheep 定价）"""
        # output token 价格表（$/MTok）
        price_map = {
            "gpt-4.1": 8.0,
            "claude-sonnet-4.5": 15.0,
            "claude-sonnet-3.5": 15.0,
            "gemini-2.5-flash": 2.5,
            "deepseek-v3.2": 0.42,
            "deepseek-v3": 0.42,
        }
        
        price = price_map.get(model, 3.0)  # 默认 $3/MTok
        return (tokens / 1_000_000) * price
        
    def check_request(self, model: str, estimated_tokens: int) -> tuple[bool, str]:
        """
        检查请求是否允许执行
        
        Returns:
            (allowed, reason)
        """
        with self._lock:
            self._check_and_reset_periods()
            
            # 计算预估成本
            estimated_cost = self._calculate_request_cost(model, estimated_tokens)
            
            # 检查是否被封顶
            if self.blocked:
                return False, f"Cost guard BLOCKED: monthly budget exceeded"
            
            # 检查单次请求上限
            if estimated_cost > self.config.per_request_limit:
                return False, (
                    f"Request too expensive: ${estimated_cost:.4f} > "
                    f"${self.config.per_request_limit:.4f} limit"
                )
                
            # 检查每日预算
            if self.daily_spend + estimated_cost > self.config.daily_limit:
                return False, (
                    f"Daily budget exceeded: ${self.daily_spend + estimated_cost:.4f} > "
                    f"${self.config.daily_limit:.4f}"
                )
                
            # 检查每月预算
            if self.monthly_spend + estimated_cost > self.config.monthly_limit:
                self.blocked = True
                return False, (
                    f"Monthly budget EXCEEDED: ${self.monthly_spend + estimated_cost:.4f} > "
                    f"${self.config.monthly_limit:.4f}"
                )
                
            # 检查告警阈值
            self._check_warnings(estimated_cost)
            
            return True, "OK"
            
    def _check_warnings(self, estimated_cost: float):
        """检查是否需要发送告警"""
        
        # 每日告警
        if not self.warning_sent_daily:
            if (self.daily_spend + estimated_cost) / self.config.daily_limit >= self.config.warning_threshold:
                self.warning_sent_daily = True
                self._send_warning(
                    "Daily Budget Warning",
                    f"Daily spend: ${self.daily_spend:.2f}/${self.config.daily_limit:.2f}"
                )
                
        # 每月告警
        if not self.warning_sent_monthly:
            if (self.monthly_spend + estimated_cost) / self.config.monthly_limit >= self.config.warning_threshold:
                self.warning_sent_monthly = True
                self._send_warning(
                    "Monthly Budget Warning",
                    f"Monthly spend: ${self.monthly_spend:.2f}/${self.config.monthly_limit:.2f}"
                )
                
    def _send_warning(self, title: str, message: str):
        """发送告警"""
        for callback in self._callbacks:
            try:
                callback(title, message)
            except Exception as e:
                print(f"Warning callback failed: {e}")
                
    def record_request(
        self,
        model: str,
        tokens: int,
        request_id: str,
        actual_cost: Optional[float] = None
    ):
        """记录已完成的请求"""
        with self._lock:
            cost = actual_cost or self._calculate_request_cost(model, tokens)
            
            self.daily_spend += cost
            self.monthly_spend += cost
            
            record = CostRecord(
                timestamp=datetime.now(),
                amount=cost,
                model=model,
                tokens=tokens,
                request_id=request_id
            )
            self.records.append(record)
            
            # 只保留最近 10000 条记录
            if len(self.records) > 10000:
                self.records = self.records[-10000:]
                
    def get_stats(self) -> dict:
        """获取成本统计"""
        with self._lock:
            self._check_and_reset_periods()
            
            return {
                "daily_spend": self.daily_spend,
                "daily_limit": self.config.daily_limit,
                "daily_remaining": self.config.daily_limit - self.daily_spend,
                "monthly_spend": self.monthly_spend,
                "monthly_limit": self.config.monthly_limit,
                "monthly_remaining": self.config.monthly_limit - self.monthly_spend,
                "total_requests": len(self.records),
                "blocked": self.blocked
            }
            
    def reset_block(self):
        """重置封顶状态（用于紧急恢复）"""
        self.blocked = False
        self.warning_sent_daily = False
        self.warning_sent_monthly = False



使用示例

if __name__ == "__main__":
    # 配置预算：每天 $10，每月 $100
    config = BudgetConfig(
        daily_limit=10.0,
        monthly_limit=100.0,
        per_request_limit=1.0,
        warning_threshold=0.8
    )
    
    guard = CostGuard(config)
    
    # 添加告警回调（可以发邮件、Slack 等）
    def on_warning(title: str, message: str):
        print(f"🚨 ALERT: {title} - {message}")
        
    guard.add_warning_callback(on_warning)
    
    # 模拟请求
    test_cases = [
        ("gpt-4.1", 50000),      # ~$0.40
        ("deepseek-v3.2", 100000),  # ~$0.042
        ("claude-sonnet-4.5", 80000),  # ~$1.20（会触发单次限制）
    ]
    
    for model, tokens in test_cases:
        allowed, reason = guard.check_request(model, tokens)
        if allowed:
            guard.record_request(model, tokens, f"req_{time.time()}")
            print(f"✅ {model}: {tokens} tokens allowed")
        else:
            print(f"❌ {model}: {tokens} tokens blocked - {reason}")
            
    print(f"\n📊 Current stats: {guard.get_stats()}")

六、审计日志系统

6.1 为什么需要审计

生产环境的 AI 调用审计，不仅是合规要求，更是问题排查的基础。我的审计日志帮我定位过：某业务线的 Token 消耗异常高、某时间段 API 调用全部失败、某个模型一直超时等问题。

6.2 审计日志实现

import json
import sqlite3
from datetime import datetime
from typing import Optional, List, Dict
from dataclasses import dataclass, asdict
import threading
import os

@dataclass
class AuditEntry:
    """审计日志条目"""
    id: Optional[int]
    timestamp: str
    request_id: str
    api_key_id: str  # 脱敏后的 Key ID
    model: str
    operation: str   # chat/completion/embedding
    input_tokens: int
    output_tokens: int
    total_cost: float
    latency_ms: int
    status: str      # success/failed/timeout
    error_message: Optional[str]
    user_id: Optional[str]
    session_id: Optional[str]
    metadata: Optional[str]  # JSON 字符串

class AuditLogger:
    """
    AI API 审计日志系统
    
    支持 SQLite 本地存储，方便查询和分析
    实际生产环境建议接入 Elasticsearch/Splunk 等
    """
    
    def __init__(self, db_path: str = "ai_audit.db"):
        self.db_path = db_path
        self._lock = threading.Lock()
        self._init_db()
        
    def _init_db(self):
        """初始化数据库表"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS audit_logs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                request_id TEXT NOT NULL,
                api_key_id TEXT NOT NULL,
                model TEXT NOT NULL,
                operation TEXT NOT NULL,
                input_tokens INTEGER DEFAULT 0,
                output_tokens INTEGER DEFAULT 0,
                total_cost REAL DEFAULT 0,
                latency_ms INTEGER DEFAULT 0,
                status TEXT NOT NULL,
                error_message TEXT,
                user_id TEXT,
                session_id TEXT,
                metadata TEXT,
                created_at TEXT DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # 创建索引
        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_timestamp 
            ON audit_logs(timestamp)
        """)
        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_model 
            ON audit_logs(model)
        """)
        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_status 
            ON audit_logs(status)
        """)
        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_user_id 
            ON audit_logs(user_id)
        """)
        
        conn.commit()
        conn.close()
        
    def log_request(
        self,
        request_id: str,
        api_key_id: str,
        model:

相关资源
相关文章