DeepSeek 应用开发与商业变现实战：从 API 调用到高并发 SaaS 架构

# 项目初始化
mkdir deepseek-saas-backend && cd deepseek-saas-backend
python -m venv venv && source venv/bin/activate
pip install fastapi uvicorn redis psycopg2-binary sqlalchemy asyncpg python-dotenv
pip install openai  # DeepSeek 兼容 OpenAI SDK
pip install pydantic-settings stripe  # Stripe 用于国际支付，国内可替换为支付宝

src/
├── core/
│   ├── config.py          # 环境变量与模型配置
│   └── security.py        # JWT 鉴权与租户隔离
├── models/
│   ├── db.py              # SQLAlchemy ORM（User, Quota, ChatHistory）
│   └── schemas.py         # Pydantic 请求/响应模型
├── services/
│   ├── deepseek_client.py # 深度封装的 DeepSeek 客户端（含重试/降级）
│   ├── context_cache.py   # Redis 语义缓存（降低延迟与成本）
│   └── billing.py         # Token 计费与配额管理
├── tools/
│   └── tool_definitions.py # Function Calling 定义（查天气/订单/库存）
├── api/
│   ├── v1/
│   │   ├── chat.py        # 流式聊天接口
│   │   └── webhook.py     # 支付回调
└── main.py                # 应用入口

import asyncio
import json
from typing import AsyncGenerator, Dict, List, Any, Optional
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from src.core.config import settings

class DeepSeekClient:
    def __init__(self):
        self.client = AsyncOpenAI(
            api_key=settings.DEEPSEEK_API_KEY,
            base_url=settings.DEEPSEEK_BASE_URL,  # https://api.deepseek.com/v1
            timeout=60.0
        )
        self.model = settings.DEEPSEEK_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_stream(
        self,
        messages: List[Dict[str, str]],
        tools: Optional[List[Dict]] = None,
        tool_choice: Optional[str] = "auto",
        temperature: float = 0.6,
        max_tokens: int = 4096,
        response_format: Optional[str] = None,  # "json_object" 强制 JSON
        user_id: Optional[str] = None
    ) -> AsyncGenerator[Dict[str, Any], None]:
        """流式请求，支持工具调用与结构化输出"""
        kwargs = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": True,
            "extra_body": {
                "stop": [],  # 可配置终止词
                "include_usage": True,  # 返回 token 消耗，用于计费
            }
        }
        if response_format == "json_object":
            kwargs["response_format"] = { "type": "json_object" }
        if tools:
            kwargs["tools"] = tools
            kwargs["tool_choice"] = tool_choice

        try:
            async with self.client as client:
                stream = await client.chat.completions.create(**kwargs)
                async for chunk in stream:
                    # 处理标准增量
                    if chunk.choices and chunk.choices[0].delta:
                        delta = chunk.choices[0].delta
                        # 检测工具调用指令
                        if delta.tool_calls:
                            yield {"type": "tool_call", "data": delta.tool_calls}
                        elif delta.content:
                            yield {"type": "text", "data": delta.content}
                    # 捕获 usage（DeepSeek 在最后一个 chunk 返回）
                    if chunk.usage:
                        yield {
                            "type": "usage",
                            "data": {
                                "prompt_tokens": chunk.usage.prompt_tokens,
                                "completion_tokens": chunk.usage.completion_tokens,
                                "total_tokens": chunk.usage.total_tokens
                            }
                        }
        except Exception as e:
            # 降级方案：若主模型报错，可切换至备用模型（如 deepseek-reasoner）
            yield {"type": "error", "data": f"主模型调用失败，尝试降级: {str(e)}"}
            # 实际生产中可触发 fallback

    # 非流式调用（用于工具执行后的汇总）
    async def chat_completion_sync(
        self,
        messages: List[Dict[str, str]],
        tools: Optional[List[Dict]] = None,
        **kwargs
    ) -> Dict[str, Any]:
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=tools,
            temperature=kwargs.get("temperature", 0.3),
            max_tokens=kwargs.get("max_tokens", 2048),
            stream=False
        )
        return response.model_dump()

import hashlib
import json
import redis.asyncio as redis
from src.core.config import settings

class SemanticCache:
    def __init__(self):
        self.redis = redis.Redis(
            host=settings.REDIS_HOST,
            port=settings.REDIS_PORT,
            decode_responses=True,
            max_connections=50
        )
        self.ttl = 3600 * 24  # 缓存 24 小时

    def _generate_key(self, messages: list, tools_signature: str = "") -> str:
        """生成缓存键：取系统消息 + 最后用户消息的哈希，减少碰撞"""
        # 仅取 system 和最后一条 user 消息作为缓存依据（业务中可按需调整）
        relevant = [m["content"] for m in messages if m["role"] in ["system", "user"]]
        if not relevant:
            relevant = [messages[-1]["content"]]
        combined = "|".join(relevant) + "|" + tools_signature
        return f"cache:semantic:{hashlib.sha256(combined.encode()).hexdigest()}"

    async def get(self, messages: list, tools: list = None) -> Optional[dict]:
        tools_sig = json.dumps(tools) if tools else ""
        key = self._generate_key(messages, tools_sig)
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None

    async def set(self, messages: list, response: dict, tools: list = None):
        tools_sig = json.dumps(tools) if tools else ""
        key = self._generate_key(messages, tools_sig)
        await self.redis.setex(key, self.ttl, json.dumps(response))

    async def invalidate(self, user_id: str, session_id: str):
        """按会话粒度清理缓存（当业务数据变更时）"""
        pattern = f"cache:semantic:*{user_id}*{session_id}*"
        async for key in self.redis.scan_iter(match=pattern, count=100):
            await self.redis.delete(key)

# api/v1/chat.py 核心逻辑片段
cached = await semantic_cache.get(messages, tools)
if cached:
    # 直接返回缓存，且不计费（注意：缓存需包含完整回复）
    return StreamingResponse(generate_from_cache(cached), media_type="text/event-stream")
# 未命中则调用 DeepSeek，并在流结束后存入缓存

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "query_order_status",
            "description": "根据订单号查询当前物流状态、预计到达时间",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单编号，格式如 ORD-2026-XXXX"
                    },
                    "fields": {
                        "type": "array",
                        "items": {"type": "string", "enum": ["status", "location", "estimated_delivery"]},
                        "description": "需要返回的字段，默认全部"
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取指定城市的实时天气",
            "parameters": { ... }  # 标准天气工具
        }
    }
]

# 模拟业务执行器
async def execute_tool_call(tool_name: str, arguments: dict) -> str:
    if tool_name == "query_order_status":
        order_id = arguments.get("order_id")
        # 这里模拟查 DB 或调用第三方物流 API
        if "2026" in order_id:
            return f"订单 {order_id} 当前状态: 派送中，预计 2026-08-18 送达"
        return f"订单 {order_id} 未找到，请检查编号"
    elif tool_name == "get_current_weather":
        city = arguments.get("city", "北京")
        return f"{city} 当前温度 28°C，晴"
    return "未知工具"

# services/agent_runner.py
async def run_agent_with_tools(
    client: DeepSeekClient,
    user_messages: List[Dict],
    max_iterations: int = 3
) -> AsyncGenerator:
    messages = user_messages.copy()
    for _ in range(max_iterations):
        # 1. 发起流式请求，获取工具调用指令
        collected_tool_calls = []
        async for chunk in client.chat_completion_stream(messages, tools=TOOLS):
            if chunk["type"] == "tool_call":
                collected_tool_calls.extend(chunk["data"])
            elif chunk["type"] == "text":
                # 如果是直接文本，直接透出，不经过工具
                yield chunk["data"]
            elif chunk["type"] == "usage":
                yield {"type": "usage", "data": chunk["data"]}
        
        # 2. 如果存在工具调用，执行并回填结果
        if collected_tool_calls:
            tool_responses = []
            for tool_call in collected_tool_calls:
                func_name = tool_call.function.name
                args = json.loads(tool_call.function.arguments)
                result = await execute_tool_call(func_name, args)
                tool_responses.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
            # 将工具结果追加到消息列表
            messages.extend(tool_responses)
            # 继续循环，让模型基于工具结果生成最终回答
            continue
        else:
            break  # 无工具调用，结束循环

from sqlalchemy import Column, String, Integer, BigInteger, DateTime, Numeric, ForeignKey
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(String(36), primary_key=True)  # UUID
    email = Column(String(128), unique=True, index=True)
    plan = Column(String(20), default="free")  # free, pro, enterprise
    balance_tokens = Column(BigInteger, default=10000)  # 免费额度 1万 tokens

class TokenUsage(Base):
    __tablename__ = "token_usage"
    id = Column(BigInteger, primary_key=True, autoincrement=True)
    user_id = Column(String(36), ForeignKey("users.id"))
    session_id = Column(String(64))
    prompt_tokens = Column(Integer)
    completion_tokens = Column(Integer)
    total_tokens = Column(Integer)
    cost_usd = Column(Numeric(10, 6))  # 精确到微美元
    created_at = Column(DateTime, server_default="now()")

import decimal
from src.core.config import settings

# DeepSeek 官方定价（2026）
PRICING = {
    "deepseek-chat": {"input": 0.002, "output": 0.008}  # 单位：美元 / 1K tokens
}

class BillingService:
    @staticmethod
    async def deduct_quota(user_id: str, usage: dict) -> bool:
        """扣减配额，返回是否成功（余额不足返回 False）"""
        input_tokens = usage.get("prompt_tokens", 0)
        output_tokens = usage.get("completion_tokens", 0)
        cost = (input_tokens / 1000) * PRICING["deepseek-chat"]["input"] + \
               (output_tokens / 1000) * PRICING["deepseek-chat"]["output"]
        cost_usd = decimal.Decimal(str(cost))

        # 异步更新余额（使用乐观锁避免并发超扣）
        async with async_session() as session:
            stmt = select(User).where(User.id == user_id).with_for_update()
            result = await session.execute(stmt)
            user = result.scalar_one()
            if user.balance_tokens < output_tokens * 2:  # 按输出 token 的 2 倍粗略校验
                return False  # 余额不足
            user.balance_tokens -= (input_tokens + output_tokens)
            # 记录明细
            usage_record = TokenUsage(
                user_id=user_id,
                prompt_tokens=input_tokens,
                completion_tokens=output_tokens,
                total_tokens=input_tokens + output_tokens,
                cost_usd=cost_usd
            )
            session.add(usage_record)
            await session.commit()
        return True

    @staticmethod
    async def recharge(user_id: str, amount_usd: float):
        """充值接口（由支付回调触发）"""
        # 1 USD = 1,000,000 tokens (按需换算)
        tokens_to_add = int(amount_usd * 1000000 / PRICING["deepseek-chat"]["input"])
        # 更新余额 ...

# api/v1/webhook.py
from fastapi import Request, HTTPException

@router.post("/webhook/stripe")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")
    try:
        event = stripe.Webhook.construct_event(payload, sig_header, settings.STRIPE_WEBHOOK_SECRET)
    except ValueError:
        raise HTTPException(400, "Invalid payload")
    if event["type"] == "checkout.session.completed":
        session = event["data"]["object"]
        user_id = session["client_reference_id"]  # 在创建 checkout 时传入
        amount = session["amount_total"] / 100  # 美元分转美元
        await BillingService.recharge(user_id, amount)
    return {"status": "success"}

from fastapi import APIRouter, Depends
from sse_starlette.sse import EventSourceResponse
from src.services.agent_runner import run_agent_with_tools

router = APIRouter()

@router.post("/stream")
async def chat_stream(request: ChatRequest, user=Depends(get_current_user)):
    # 1. 鉴权与配额预检
    if user.balance_tokens < 100:
        return JSONResponse({"error": "余额不足，请充值"}, status_code=402)

    # 2. 构建消息列表
    messages = [{"role": "system", "content": "你是一个专业的客服助手..."}] + request.messages

    # 3. 执行 Agent 并流式返回
    async def event_generator():
        total_usage = {}
        async for chunk in run_agent_with_tools(deepseek_client, messages):
            if chunk.get("type") == "usage":
                total_usage = chunk["data"]
                # 流结束后才扣费
            elif chunk.get("type") == "error":
                yield {"event": "error", "data": chunk["data"]}
            else:
                # 标准文本增量
                yield {"event": "message", "data": chunk.get("data", "")}

        # 4. 流结束后进行扣费
        if total_usage:
            success = await BillingService.deduct_quota(user.id, total_usage)
            if not success:
                yield {"event": "error", "data": "扣费失败，请检查账户"}

    return EventSourceResponse(event_generator(), media_type="text/event-stream")

# prometheus 配置示例
- job_name: 'deepseek-api'
  static_configs:
    - targets: ['deepseek-saas-service:8000']