DeepSeek 应用开发与商业变现实战：从API调用到百万级营收架构

# src/infrastructure/deepseek_client.py
import os
import time
import asyncio
from typing import AsyncGenerator, Optional, Callable, Dict, Any
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import tiktoken

class DeepSeekClient:
    """企业级DeepSeek客户端，集成成本计量与可观测性"""
    
    def __init__(self, api_key: str = None, base_url: str = None, model: str = "deepseek-chat"):
        self.api_key = api_key or os.getenv("DEEPSEEK_API_KEY")
        self.base_url = base_url or os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1")
        self.model = model
        self.client = AsyncOpenAI(
            api_key=self.api_key,
            base_url=self.base_url,
            timeout=60.0,
            max_retries=0,  # 由tenacity接管
        )
        # Tokenizer用于精确计量（与DeepSeek一致）
        self.encoding = tiktoken.get_encoding("cl100k_base")  # DeepSeek兼容
        self.cost_hooks: list[Callable] = []  # 成本回调链
        
    def register_cost_hook(self, hook: Callable[[str, int, int, float], None]):
        """注册成本回调，用于计量模块"""
        self.cost_hooks.append(hook)
    
    async def _notify_cost(self, operation: str, prompt_tokens: int, completion_tokens: int):
        """计算并通知成本"""
        # DeepSeek 价格 (元/百万tokens) - 以deepseek-chat为例
        INPUT_PRICE_PER_M = 1.0   # 输入
        OUTPUT_PRICE_PER_M = 2.0  # 输出
        cost = (prompt_tokens / 1_000_000 * INPUT_PRICE_PER_M) + \
               (completion_tokens / 1_000_000 * OUTPUT_PRICE_PER_M)
        for hook in self.cost_hooks:
            await hook(operation, prompt_tokens, completion_tokens, cost)
        return cost

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        retry=retry_if_exception_type((ConnectionError, TimeoutError)),
        reraise=True
    )
    async def chat_completion(
        self, 
        messages: list[dict], 
        temperature: float = 0.7,
        max_tokens: int = 4096,
        stream: bool = False,
        operation_name: str = "default",
        **kwargs
    ) -> dict | AsyncGenerator:
        """非流式/流式统一入口"""
        start_time = time.perf_counter()
        
        if stream:
            return self._stream_chat(messages, temperature, max_tokens, operation_name, **kwargs)
        
        # 非流式
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=False,
            **kwargs
        )
        
        # 精确计量
        usage = response.usage
        prompt_tokens = usage.prompt_tokens
        completion_tokens = usage.completion_tokens
        cost = await self._notify_cost(operation_name, prompt_tokens, completion_tokens)
        
        # 可观测性
        elapsed = time.perf_counter() - start_time
        print(f"[Metric] {operation_name} | tokens: {prompt_tokens}/{completion_tokens} | cost: ¥{cost:.6f} | latency: {elapsed:.2f}s")
        
        return {
            "content": response.choices[0].message.content,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "cost": cost,
            "model": self.model,
        }

    async def _stream_chat(self, messages, temperature, max_tokens, operation_name, **kwargs):
        """流式处理（带逐字计量）"""
        # 注意：流式无法提前获取token数，需要预估或事后统计
        # 我们使用tiktoken粗略预估输入，输出按流式累加
        input_text = " ".join([m["content"] for m in messages])
        estimated_prompt_tokens = len(self.encoding.encode(input_text))
        
        stream = await self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=True,
            **kwargs
        )
        
        full_content = ""
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                delta = chunk.choices[0].delta.content
                full_content += delta
                yield delta
        
        # 流结束后精准计量（实际API返回usage在流结束时才有，但DeepSeek支持在末尾返回）
        # 此处简化：使用tiktoken估算输出
        estimated_completion_tokens = len(self.encoding.encode(full_content))
        await self._notify_cost(operation_name, estimated_prompt_tokens, estimated_completion_tokens)

# src/prompts/templates.py
from jinja2 import Template
import json
from typing import Dict, Any, Optional

class PromptTemplate:
    def __init__(self, name: str, system_template: str, user_template: str, 
                 variables: list[str], industry: str, price_extra: float = 0.0):
        self.name = name
        self.system = Template(system_template)
        self.user = Template(user_template)
        self.variables = variables
        self.industry = industry
        self.price_extra = price_extra  # 高级模板溢价

# 预置模板库
TEMPLATE_LIBRARY = {
    "ecommerce_product_desc": PromptTemplate(
        name="电商卖点提炼",
        system_template="""你是一位顶尖的电商文案专家，熟悉{{ brand_tone }}风格。
        请根据产品参数{{ product_params }}，生成5个不同角度的卖点标题（每个15字内）和一段200字详情描述。
        要求：包含{{ seo_keywords }}关键词，遵循{{ platform }}平台规则。""",
        user_template="产品名称：{{ product_name }}，核心功能：{{ features }}",
        variables=["brand_tone", "product_params", "seo_keywords", "platform", "product_name", "features"],
        industry="ecommerce",
        price_extra=0.3
    ),
    "wechat_article_hook": PromptTemplate(
        name="公众号爆款开头",
        system_template="""你熟悉公众号文章的开头钩子写法，包含痛点共鸣、数据冲击、故事代入三种手法。
        文章主题：{{ topic }}，目标人群：{{ audience }}。""",
        user_template="请生成3个不同钩子，每个100字以内。",
        variables=["topic", "audience"],
        industry="content",
        price_extra=0.5
    )
}

class PromptEngine:
    @staticmethod
    def render(template_name: str, variables: Dict[str, Any]) -> tuple[str, str]:
        tmpl = TEMPLATE_LIBRARY.get(template_name)
        if not tmpl:
            raise ValueError(f"Template {template_name} not found")
        
        # 校验变量
        missing = set(tmpl.variables) - set(variables.keys())
        if missing:
            raise ValueError(f"Missing variables: {missing}")
        
        system_prompt = tmpl.system.render(**variables)
        user_prompt = tmpl.user.render(**variables)
        return system_prompt, user_prompt

-- scripts/consume_quota.lua
-- KEYS[1]: tenant_quota_key, KEYS[2]: tenant_usage_key
-- ARGV[1]: requested_tokens, ARGV[2]: current_timestamp
local quota_key = KEYS[1]
local usage_key = KEYS[2]
local requested = tonumber(ARGV[1])

-- 获取剩余配额（结构：{total: 1000000, used: 200000}）
local quota_json = redis.call('GET', quota_key)
if not quota_json then
    return {-1, "Quota not found"}  -- 无套餐
end
local quota = cjson.decode(quota_json)
local remaining = quota.total - quota.used

if remaining < requested then
    return {0, remaining}  -- 配额不足，返回剩余量
end

-- 原子扣减
quota.used = quota.used + requested
redis.call('SET', quota_key, cjson.encode(quota))

-- 记录使用明细到Sorted Set（用于对账）
local usage_item = cjson.encode({amount=requested, time=ARGV[2]})
redis.call('ZADD', usage_key, ARGV[2], usage_item)

return {1, remaining - requested}

# src/billing/quota_manager.py
import redis.asyncio as redis
import json
import time

class QuotaManager:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.consume_lua = await self.redis.register_script(open("scripts/consume_quota.lua").read())
    
    async def consume(self, tenant_id: str, requested_tokens: int) -> tuple[bool, int, str]:
        """返回 (是否成功, 剩余配额, 错误信息)"""
        quota_key = f"quota:{tenant_id}"
        usage_key = f"usage:{tenant_id}:{time.strftime('%Y%m')}"  # 按月
        result = await self.consume_lua(
            keys=[quota_key, usage_key],
            args=[requested_tokens, int(time.time())]
        )
        code = result[0]
        if code == -1:
            return False, 0, "No active subscription"
        elif code == 0:
            return False, result[1], f"Insufficient quota, remaining: {result[1]}"
        else:
            return True, result[1], "Success"
    
    async def refill_quota(self, tenant_id: str, total_tokens: int):
        """月初/购买后重置配额"""
        quota_key = f"quota:{tenant_id}"
        # 保留已用量（支持按量付费）
        current = await self.redis.get(quota_key)
        if current:
            data = json.loads(current)
            data["total"] = total_tokens
        else:
            data = {"total": total_tokens, "used": 0}
        await self.redis.set(quota_key, json.dumps(data))

# src/core/orchestrator.py
from typing import AsyncGenerator
from src.infrastructure.deepseek_client import DeepSeekClient
from src.prompts.templates import PromptEngine
from src.billing.quota_manager import QuotaManager
from src.rag.vector_store import VectorStore  # 对接腾讯云向量数据库

class MarketingOrchestrator:
    def __init__(self, deepseek: DeepSeekClient, quota: QuotaManager, vector_store: VectorStore):
        self.llm = deepseek
        self.quota = quota
        self.vector_store = vector_store
        # 注册成本钩子 -> 实时扣费
        self.llm.register_cost_hook(self._billing_hook)
    
    async def _billing_hook(self, operation: str, prompt_tokens: int, completion_tokens: int, cost: float):
        """计费钩子：从请求上下文中获取租户ID"""
        # 通过ContextVar传递租户ID (FastAPI依赖注入)
        from src.api.dependencies import get_current_tenant
        tenant_id = get_current_tenant()
        total_tokens = prompt_tokens + completion_tokens
        # 转换为计费单位（1 token = 1 单位，业务可自定义）
        success, remaining, msg = await self.quota.consume(tenant_id, total_tokens)
        if not success:
            raise RuntimeError(f"Quota exhausted: {msg}")
        # 记录详细账单到TiDB（异步）
        await self._save_billing_record(tenant_id, operation, prompt_tokens, completion_tokens, cost)
        return cost
    
    async def generate_with_rag(
        self, 
        template_name: str, 
        variables: dict, 
        user_query: str,
        tenant_id: str
    ) -> dict:
        """RAG增强生成：从向量库检索企业专属资料"""
        # 1. 检索相关文档
        docs = await self.vector_store.search(
            collection=f"tenant_{tenant_id}_docs",
            query=user_query,
            top_k=3
        )
        context = "\n".join([doc["text"] for doc in docs])
        
        # 2. 渲染模板
        system, user = PromptEngine.render(template_name, variables)
        # 注入RAG上下文
        system += f"\n\n【企业专属知识库参考】\n{context}\n请优先引用上述资料，确保文案符合企业规范。"
        
        messages = [
            {"role": "system", "content": system},
            {"role": "user", "content": user}
        ]
        
        # 3. 调用DeepSeek
        result = await self.llm.chat_completion(
            messages=messages,
            temperature=0.8,
            operation_name=f"rag_{template_name}"
        )
        
        return {
            "content": result["content"],
            "usage": {"prompt": result["prompt_tokens"], "completion": result["completion_tokens"]},
            "cost": result["cost"],
            "rag_docs": docs
        }

# src/api/routes_billing.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from src.billing.quota_manager import QuotaManager
from src.billing.order_service import OrderService

router = APIRouter(prefix="/billing", tags=["billing"])

class PurchaseRequest(BaseModel):
    plan: str  # "starter", "pro", "enterprise"
    quantity: int = 1

class UsageReport(BaseModel):
    tenant_id: str
    month: str  # "2026-08"
    total_tokens: int
    total_cost: float
    overage_cost: float

@router.post("/purchase")
async def purchase_plan(req: PurchaseRequest, tenant_id: str = Depends(get_tenant_from_api_key)):
    """购买套餐/充值"""
    # plan配置从配置中心读取
    plan_config = {
        "starter": {"tokens": 5_000_000, "price": 99},
        "pro": {"tokens": 20_000_000, "price": 299},
        "enterprise": {"tokens": 100_000_000, "price": 999},
    }
    cfg = plan_config.get(req.plan)
    if not cfg:
        raise HTTPException(400, "Invalid plan")
    
    total_tokens = cfg["tokens"] * req.quantity
    total_price = cfg["price"] * req.quantity
    
    # 1. 调用支付网关（腾讯云支付/微信支付）
    payment_url = await OrderService.create_order(tenant_id, total_price, total_tokens)
    
    # 2. 预置配额（支付成功后回调激活，此处仅返回支付链接）
    return {"payment_url": payment_url, "order_id": OrderService.order_id}

@router.get("/usage/{tenant_id}")
async def get_usage(tenant_id: str, month: str, quota: QuotaManager = Depends()):
    """获取月度使用明细"""
    usage_key = f"usage:{tenant_id}:{month}"
    items = await quota.redis.zrange(usage_key, 0, -1, withscores=True)
    total_tokens = sum([json.loads(item[0])["amount"] for item in items])
    # 计算超额费用（假设基础套餐已包含500w）
    base_tokens = 5_000_000
    overage = max(0, total_tokens - base_tokens)
    overage_cost = overage / 1_000_000 * 0.8  # 超额单价0.8元/百万
    return UsageReport(
        tenant_id=tenant_id,
        month=month,
        total_tokens=total_tokens,
        total_cost=overage_cost,
        overage_cost=overage_cost
    )

# src/api/webhooks.py
from fastapi import APIRouter, Request
from src.billing.quota_manager import QuotaManager

router = APIRouter(prefix="/webhook")

@router.post("/payment/callback")
async def payment_callback(request: Request, quota: QuotaManager):
    """腾讯云支付/微信支付回调"""
    payload = await request.json()
    # 验签（省略）
    if payload["status"] == "SUCCESS":
        tenant_id = payload["out_trade_no"].split("_")[0]  # 业务订单号解析
        total_tokens = int(payload["attach"])  # 附加字段
        await quota.refill_quota(tenant_id, total_tokens)
        return {"code": 0, "msg": "OK"}
    return {"code": 1, "msg": "FAIL"}

# src/rag/tencent_vector_db.py
import tcvectordb
from tcvectordb.model.enum import FieldType, IndexType, MetricType
from tcvectordb.model.index import Index, VectorIndex, FilterIndex

class TencentVectorStore:
    def __init__(self, url: str, key: str, username: str = "root"):
        self.client = tcvectordb.VectorDBClient(url=url, username=username, key=key)
    
    async def create_tenant_collection(self, tenant_id: str, dimension: int = 768):
        """为租户创建独立Collection（实现物理隔离）"""
        collection_name = f"tenant_{tenant_id}_docs"
        if not self.client.exists_collection(collection_name):
            idx = Index(
                VectorIndex("vector", dimension, IndexType.FLAT, MetricType.COSINE),
                FilterIndex("doc_id", FieldType.String, IndexType.PRIMARY_KEY),
                FilterIndex("category", FieldType.String, IndexType.FILTER),
            )
            self.client.create_collection(collection_name, index=idx)
        return collection_name

# k8s/hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: deepseek-app-hpa
  namespace: production
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: deepseek-app
  minReplicas: 2
  maxReplicas: 20
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 60
    - type: Pods
      pods:
        metric:
          name: rabbitmq_queue_length
        target:
          type: AverageValue
          averageValue: "50"
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 300   # 缩容冷静期5分钟，避免抖动

# src/cache/semantic_cache.py
from sentence_transformers import SentenceTransformer
import redis.asyncio as redis
import numpy as np
from scipy.spatial.distance import cosine

class SemanticCache:
    def __init__(self, redis_client: redis.Redis, threshold: float = 0.9):
        self.redis = redis_client
        self.model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')  # 多语言
        self.threshold = threshold
    
    async def get(self, query: str) -> str | None:
        q_emb = self.model.encode(query).tobytes()
        # 使用Redis的向量相似度搜索（RediSearch）
        results = await self.redis.ft("idx:embeddings").search(
            Query("*=>[KNN 1 @vector $vec AS score]")
            .return_fields("text", "score")
            .dialect(2),
            query_params={"vec": q_emb}
        )
        if results.total > 0 and results.docs[0].score >= self.threshold:
            return results.docs[0].text
        return None
    
    async def set(self, query: str, response: str):
        emb = self.model.encode(query).tobytes()
        await self.redis.ft("idx:embeddings").add(
            f"cache:{hash(query)}", 
            {"vector": emb, "text": response},
            nx=True
        )

# tests/stress_test.py (使用locust)
from locust import HttpUser, task, between
import random

class DeepSeekSaaSUser(HttpUser):
    wait_time = between(0.5, 1.5)
    
    @task(3)
    def generate_copy(self):
        self.client.post("/v1/generate", json={
            "template": "ecommerce_product_desc",
            "variables": {
                "brand_tone": "年轻潮流",
                "product_name": f"智能手环_{random.randint(1,100)}",
                "features": "心率监测、血氧检测、50米防水",
                "platform": "抖音",
                "seo_keywords": "运动,健康",
                "product_params": "型号X5, 续航7天"
            }
        })
    
    @task(1)
    def check_quota(self):
        self.client.get("/billing/usage/current")