DeepSeek应用开发与商业变现实战：从API调用到多租户SaaS系统全栈落地

# core/deepseek_client.py
import asyncio
import time
from typing import AsyncGenerator, Optional
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import logging

logger = logging.getLogger(__name__)

class DeepSeekAsyncClient:
    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com/v1"):
        self.client = AsyncOpenAI(
            api_key=api_key,
            base_url=base_url,
            timeout=60.0,
            max_retries=0  # 由tenacity接管
        )
        self.default_model = "deepseek-chat"  # 或 deepseek-reasoner

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((ConnectionError, TimeoutError))
    )
    async def chat_completion(
        self,
        messages: list[dict],
        model: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 4096,
        stream: bool = False,
        **kwargs
    ) -> dict:
        """核心对话接口，返回响应及token消耗"""
        try:
            response = await self.client.chat.completions.create(
                model=model or self.default_model,
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens,
                stream=stream,
                **kwargs
            )
            # 返回标准化格式
            return {
                "content": response.choices[0].message.content,
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens,
                "finish_reason": response.choices[0].finish_reason
            }
        except Exception as e:
            logger.error(f"DeepSeek API调用失败: {str(e)}")
            raise

    async def stream_chat(self, messages: list[dict], **kwargs) -> AsyncGenerator[str, None]:
        """流式输出，降低用户端首字延迟"""
        try:
            stream = await self.client.chat.completions.create(
                model=self.default_model,
                messages=messages,
                stream=True,
                **kwargs
            )
            async for chunk in stream:
                if chunk.choices[0].delta.content:
                    yield chunk.choices[0].delta.content
        except Exception as e:
            logger.error(f"流式响应中断: {e}")
            yield f"[ERROR] {str(e)}"

# rag/document_processor.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
from PyPDF2 import PdfReader
import io

def parse_pdf(file_bytes: bytes) -> str:
    reader = PdfReader(io.BytesIO(file_bytes))
    text = ""
    for page in reader.pages:
        text += page.extract_text() + "\n"
    return text

def chunk_document(text: str, chunk_size: int = 512, overlap: int = 50) -> list[str]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
        length_function=len
    )
    return splitter.split_text(text)

# rag/vector_store.py
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
import numpy as np
from sentence_transformers import SentenceTransformer

# 加载Embedding模型（BGE-large-zh-v1.5）
embedder = SentenceTransformer('BAAI/bge-large-zh-v1.5', device='cuda')  # 若GPU受限可换CPU

class MilvusClient:
    def __init__(self, host='localhost', port='19530'):
        connections.connect(alias="default", host=host, port=port)
        self.collection_name = "doc_chunks"

    def create_collection(self):
        fields = [
            FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
            FieldSchema(name="tenant_id", dtype=DataType.VARCHAR, max_length=64, is_partition_key=True),
            FieldSchema(name="chunk_text", dtype=DataType.VARCHAR, max_length=2048),
            FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024)
        ]
        schema = CollectionSchema(fields=fields, description="RAG chunks")
        collection = Collection(self.collection_name, schema)
        # 创建索引（IVF_FLAT平衡速度和精度）
        index_params = {"metric_type": "IP", "index_type": "IVF_FLAT", "params": {"nlist": 128}}
        collection.create_index(field_name="embedding", index_params=index_params)
        collection.load()
        return collection

    def insert_chunks(self, tenant_id: str, chunks: list[str]):
        collection = Collection(self.collection_name)
        embeddings = embedder.encode(chunks, normalize_embeddings=True).tolist()
        data = [[tenant_id] * len(chunks), chunks, embeddings]
        collection.insert(data)
        collection.flush()

    def search(self, tenant_id: str, query: str, top_k: int = 5) -> list[str]:
        collection = Collection(self.collection_name)
        query_emb = embedder.encode([query], normalize_embeddings=True).tolist()
        # 多租户过滤
        search_params = {"metric_type": "IP", "params": {"nprobe": 10}}
        results = collection.search(
            data=query_emb,
            anns_field="embedding",
            param=search_params,
            limit=top_k,
            expr=f'tenant_id == "{tenant_id}"'
        )
        return [hit.entity.get('chunk_text') for hit in results[0]]

def build_rag_prompt(query: str, context_chunks: list[str]) -> list[dict]:
    context = "\n---\n".join(context_chunks)
    system_prompt = """你是一个严谨的企业文档助手。请仅根据以下【参考文档】回答问题。若文档中无相关信息，请明确告知“当前知识库未覆盖该问题”，严禁编造。参考文档：\n""" + context
    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": query}
    ]

# tools/crm_tools.py
tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "get_customer_balance",
            "description": "根据客户手机号或企业ID查询当前账户余额（单位：元）",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string", "description": "客户唯一标识"}
                },
                "required": ["customer_id"]
            }
        }
    }
]

async def execute_tool(tool_name: str, arguments: dict) -> str:
    if tool_name == "get_customer_balance":
        # 模拟查询数据库
        return f"客户 {arguments['customer_id']} 当前余额为 12450.00 元"
    return "未知工具"

async def chat_with_tools(client: DeepSeekAsyncClient, messages: list[dict]):
    # 第一轮：让模型判断是否需要调用工具
    response = await client.chat_completion(messages=messages, tools=tools_schema, tool_choice="auto")
    msg = response["content"]
    tool_calls = response.get("tool_calls", [])

    if not tool_calls:
        return msg

    # 执行工具并追加结果
    messages.append({"role": "assistant", "content": msg, "tool_calls": tool_calls})
    for tc in tool_calls:
        result = await execute_tool(tc.function.name, json.loads(tc.function.arguments))
        messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})

    # 第二轮：生成最终回复
    final_response = await client.chat_completion(messages=messages)
    return final_response["content"]

-- 租户表
CREATE TABLE tenants (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(100) NOT NULL,
    api_key VARCHAR(64) UNIQUE NOT NULL,  -- 用于SDK鉴权
    balance DECIMAL(10,2) DEFAULT 0.00,   -- 预充值余额
    tier VARCHAR(20) DEFAULT 'standard',  -- standard/premium
    created_at TIMESTAMP DEFAULT NOW()
);

-- 调用日志（计费依据）
CREATE TABLE usage_logs (
    id BIGSERIAL PRIMARY KEY,
    tenant_id UUID REFERENCES tenants(id),
    request_id VARCHAR(64),
    prompt_tokens INT,
    completion_tokens INT,
    total_tokens INT,
    cost DECIMAL(10,6),  -- 实际消耗金额
    created_at TIMESTAMP DEFAULT NOW()
);

# middleware/billing.py
from fastapi import Request, HTTPException
import redis.asyncio as redis
import json

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
TOKEN_PRICE_PROMPT = 0.000001  # 1元/百万token -> 每token 0.000001元
TOKEN_PRICE_COMPLETION = 0.000002

async def billing_dependency(request: Request):
    tenant_id = request.headers.get("X-Tenant-ID")
    if not tenant_id:
        raise HTTPException(401, "Missing Tenant ID")
    
    # 检查缓存余额（提升性能）
    cached_balance = await redis_client.get(f"balance:{tenant_id}")
    if cached_balance is None:
        # 从DB查询并缓存5分钟
        balance = await get_balance_from_db(tenant_id)
        await redis_client.setex(f"balance:{tenant_id}", 300, balance)
    else:
        balance = float(cached_balance)
    
    if balance <= 0:
        raise HTTPException(402, "Insufficient balance. Please recharge.")
    
    # 将余额存入request.state供后续使用
    request.state.tenant_id = tenant_id
    request.state.balance = balance
    return True

# 后置扣费Hook（在响应后执行）
async def deduct_cost(request: Request, prompt_tokens: int, completion_tokens: int):
    cost = (prompt_tokens * TOKEN_PRICE_PROMPT) + (completion_tokens * TOKEN_PRICE_COMPLETION)
    tenant_id = request.state.tenant_id
    
    # 原子扣减（使用Redis Lua脚本防止并发超扣）
    lua_script = """
        local current = redis.call('GET', KEYS[1])
        if current and tonumber(current) >= tonumber(ARGV[1]) then
            redis.call('DECRBY', KEYS[1], ARGV[1] * 1000000)  -- 存储为整数分避免浮点
            return 1
        end
        return 0
    """
    # 实际开发中需处理Redis与DB最终一致性（异步落库）
    await redis_client.decrbyfloat(f"balance:{tenant_id}", cost)
    # 异步记录usage_log（推送至Celery）
    add_usage_log.delay(tenant_id, prompt_tokens, completion_tokens, cost)

@app.post("/api/webhook/recharge")
async def recharge_callback(payload: dict):
    tenant_id = payload.get("tenant_id")
    amount = payload.get("amount")
    signature = payload.get("sign")
    # 验证签名（略）
    await redis_client.incrbyfloat(f"balance:{tenant_id}", amount)
    # 更新DB
    await update_balance_db(tenant_id, amount)
    return {"code": 0, "msg": "success"}

# 在DeepSeekAsyncClient中注入自定义httpx client
import httpx
http_client = httpx.AsyncClient(
    limits=httpx.Limits(max_keepalive_connections=20, max_connections=100),
    timeout=60.0
)
self.client = AsyncOpenAI(http_client=http_client, ...)

async def get_cache_key(tenant_id: str, query: str):
    emb = embedder.encode(query).tobytes()
    # 使用向量索引快速检索缓存库（此处略）

version: '3.8'
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: ai_saas
      POSTGRES_USER: admin
      POSTGRES_PASSWORD: your_password
    volumes:
      - pg_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  milvus:
    image: milvusdb/milvus:2.3.3
    ports:
      - "19530:19530"
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
    depends_on:
      - etcd
      - minio

  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql://admin:your_password@postgres/ai_saas
      REDIS_URL: redis://redis:6379
      DEEPSEEK_API_KEY: ${DEEPSEEK_API_KEY}
    depends_on:
      - postgres
      - redis
      - milvus
    command: uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

volumes:
  pg_data:

docker-compose up -d --build

@app.get("/admin/dashboard")
async def get_dashboard():
    # 今日调用次数
    today_count = await db.fetchval("SELECT COUNT(*) FROM usage_logs WHERE created_at > NOW() - INTERVAL '1 day'")
    # 今日总收入
    today_revenue = await db.fetchval("SELECT SUM(cost) FROM usage_logs WHERE created_at > NOW() - INTERVAL '1 day'")
    # Token消耗趋势
    trend = await db.fetch("SELECT DATE_TRUNC('hour', created_at) as hour, SUM(total_tokens) FROM usage_logs GROUP BY hour ORDER BY hour DESC LIMIT 24")
    return {
        "total_requests_today": today_count,
        "revenue_today": round(today_revenue or 0, 2),
        "trend": [dict(r) for r in trend]
    }