【DeepSeek 实战】降本增效：DeepSeek V4 推理成本控制与生产环境监控

总成本 = (输入 Tokens × 2 + 输出 Tokens × 8) / 1,000,000

import hashlib
import redis
import json
from typing import Optional
class ResponseCache:
    """
    基于 Redis 的响应缓存
    """
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis_client = redis.from_url(redis_url)
        self.ttl = 3600  # 缓存有效期 1 小时
    def _generate_cache_key(self, messages: list, model: str) -> str:
        """
        生成缓存键
        """
        content = json.dumps(messages, sort_keys=True) + model
        hash_value = hashlib.md5(content.encode()).hexdigest()
        return f"deepseek:{hash_value}"
    def get_cached_response(self, messages: list, model: str) -> Optional[str]:
        """
        获取缓存的响应
        """
        cache_key = self._generate_cache_key(messages, model)
        cached = self.redis_client.get(cache_key)
        if cached:
            print(f"[Cache Hit] {cache_key[:16]}...")
            return cached.decode('utf-8')
        return None
    def cache_response(self, messages: list, model: str, response: str):
        """
        缓存响应结果
        """
        cache_key = self._generate_cache_key(messages, model)
        self.redis_client.setex(cache_key, self.ttl, response)
        print(f"[Cache Set] {cache_key[:16]}...")

from chromadb.utils import embedding_functions
class SemanticCache:
    """
    基于向量相似度的语义缓存
    """
    def __init__(self, similarity_threshold: float = 0.95):
        self.embedding_func = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name="all-MiniLM-L6-v2"
        )
        self.threshold = similarity_threshold
        self.cache_db = {}  # 简化示例，生产环境应使用向量数据库
    def find_similar_query(self, query: str) -> Optional[str]:
        """
        查找语义相似的已缓存查询
        """
        query_embedding = self.embedding_func([query])[0]
        for cached_query, (cached_embedding, response) in self.cache_db.items():
            similarity = self._cosine_similarity(query_embedding, cached_embedding)
            if similarity >= self.threshold:
                return response
        return None
    def _cosine_similarity(self, vec1, vec2):
        """计算余弦相似度"""
        dot_product = sum(a * b for a, b in zip(vec1, vec2))
        norm1 = sum(a ** 2 for a in vec1) ** 0.5
        norm2 = sum(b ** 2 for b in vec2) ** 0.5
        return dot_product / (norm1 * norm2) if norm1 and norm2 else 0

import asyncio
from typing import List
class BatchProcessor:
    """
    批量请求处理器
    """
    def __init__(self, batch_size: int = 10, max_wait_time: float = 2.0):
        self.batch_size = batch_size
        self.max_wait_time = max_wait_time
        self.request_queue = asyncio.Queue()
        self.is_running = False
    async def start(self):
        """
        启动批量处理循环
        """
        self.is_running = True
        while self.is_running:
            batch = []
            # 收集一批请求
            try:
                first_request = await asyncio.wait_for(
                    self.request_queue.get(), 
                    timeout=self.max_wait_time
                )
                batch.append(first_request)
                while len(batch) < self.batch_size:
                    try:
                        request = self.request_queue.get_nowait()
                        batch.append(request)
                    except asyncio.QueueEmpty:
                        break
                # 处理批量请求
                await self._process_batch(batch)
            except asyncio.TimeoutError:
                if batch:
                    await self._process_batch(batch)
    async def submit_request(self, messages: list) -> asyncio.Future:
        """
        提交请求到队列
        """
        future = asyncio.Future()
        await self.request_queue.put((messages, future))
        return await future
    async def _process_batch(self, batch):
        """
        处理批量请求（需根据实际 API 调整）
        """
        # 注意：DeepSeek API 目前不支持原生批量接口
        # 这里展示的是并发处理优化
        tasks = [self._call_api(messages, future) for messages, future in batch]
        await asyncio.gather(*tasks)
    async def _call_api(self, messages, future):
        """
        调用 API 并设置结果
        """
        try:
            # 实际 API 调用逻辑
            result = await deepseek_client.chat.completions.create(
                model="deepseek-chat",
                messages=messages
            )
            future.set_result(result)
        except Exception as e:
            future.set_exception(e)

class TokenBudgetManager:
    """
    Token 预算管理器
    """
    def __init__(self):
        self.daily_budgets = {}  # {user_id: {"used": 0, "limit": 100000}}
    def set_budget(self, user_id: str, daily_limit: int):
        """
        设置用户每日 Token 预算
        """
        self.daily_budgets[user_id] = {
            "used": 0,
            "limit": daily_limit,
            "reset_time": self._get_next_midnight()
        }
    def check_and_consume(self, user_id: str, token_count: int) -> bool:
        """
        检查并消耗 Token 预算
        :return: True 如果允许调用，False 如果超出预算
        """
        if user_id not in self.daily_budgets:
            return False
        budget = self.daily_budgets[user_id]
        # 检查是否需要重置
        if self._is_past_reset_time(budget["reset_time"]):
            budget["used"] = 0
            budget["reset_time"] = self._get_next_midnight()
        # 检查预算
        if budget["used"] + token_count > budget["limit"]:
            print(f"[Budget Exceeded] User {user_id} has used {budget['used']}/{budget['limit']} tokens")
            return False
        # 消耗预算
        budget["used"] += token_count
        return True
    def get_usage_stats(self, user_id: str) -> dict:
        """
        获取用户使用统计
        """
        if user_id not in self.daily_budgets:
            return {}
        budget = self.daily_budgets[user_id]
        return {
            "used": budget["used"],
            "limit": budget["limit"],
            "remaining": budget["limit"] - budget["used"],
            "usage_percentage": (budget["used"] / budget["limit"]) * 100
        }

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
# 定义监控指标
API_CALLS_TOTAL = Counter(
    'deepseek_api_calls_total',
    'Total number of API calls',
    ['model', 'status']
)
API_LATENCY = Histogram(
    'deepseek_api_latency_seconds',
    'API call latency in seconds',
    ['model']
)
TOKEN_USAGE = Counter(
    'deepseek_token_usage_total',
    'Total token usage',
    ['type']  # prompt or completion
)
ACTIVE_REQUESTS = Gauge(
    'deepseek_active_requests',
    'Number of active requests'
)
class MetricsCollector:
    """
    监控指标采集器
    """
    def __init__(self, port: int = 9090):
        start_http_server(port)
        print(f"Metrics server started on port {port}")
    def record_api_call(self, model: str, status: str, latency: float, 
                       prompt_tokens: int, completion_tokens: int):
        """
        记录一次 API 调用
        """
        API_CALLS_TOTAL.labels(model=model, status=status).inc()
        API_LATENCY.labels(model=model).observe(latency)
        TOKEN_USAGE.labels(type='prompt').inc(prompt_tokens)
        TOKEN_USAGE.labels(type='completion').inc(completion_tokens)

# Prometheus 告警规则
groups:
  - name: deepseek_alerts
    rules:
      - alert: HighErrorRate
        expr: rate(deepseek_api_calls_total{status="error"}[5m]) > 0.05
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "DeepSeek API 错误率过高"
          description: "过去 5 分钟错误率超过 5%"
      - alert: HighLatency
        expr: histogram_quantile(0.95, deepseek_api_latency_seconds) > 5
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "DeepSeek API 响应延迟过高"
          description: "P95 延迟超过 5 秒"
      - alert: BudgetExceeded
        expr: deepseek_token_usage_total > 1000000
        for: 0m
        labels:
          severity: critical
        annotations:
          summary: "Token 用量超出预算"
          description: "今日 Token 用量已超过 100 万"

优化前年度成本:
├── API 费用:   ¥150,000 × 12 = ¥1,800,000
├── 服务器费用: 30台 × ¥2,000/月 × 12 = ¥720,000
├── 运维人力:   5人 × ¥20,000/月 × 12 = ¥1,200,000
└── 总计:       ¥3,720,000
全面优化后年度成本:
├── API 费用:   ¥75,000 × 12 = ¥900,000
├── 服务器费用: 8台 × ¥2,000/月 × 12 = ¥192,000
├── 运维人力:   2人 × ¥20,000/月 × 12 = ¥480,000
└── 总计:       ¥1,572,000
🎉 年度节省: ¥2,148,000 (约 215 万元)