DeepSeek 应用开发与商业变现：从 API 调用到产品化全栈实践

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1"  # 官方兼容 OpenAI 接口
)

response = client.chat.completions.create(
    model="deepseek-chat",  # 或 deepseek-reasoner
    messages=[
        {"role": "system", "content": "你是一个专业的金融数据分析助手"},
        {"role": "user", "content": "请总结 2026 年 Q2 新能源车行业趋势"}
    ],
    temperature=0.3,
    max_tokens=4096,
    response_format={"type": "json_object"}  # 强制 JSON 输出
)
print(response.choices[0].message.content)

import asyncio
import aiohttp

async def deepseek_async(prompt, session):
    url = "https://api.deepseek.com/v1/chat/completions"
    headers = {"Authorization": f"Bearer {API_KEY}"}
    payload = {
        "model": "deepseek-chat",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.1
    }
    async with session.post(url, json=payload, headers=headers) as resp:
        return await resp.json()

async def batch_process(prompts):
    async with aiohttp.ClientSession() as session:
        tasks = [deepseek_async(p, session) for p in prompts]
        return await asyncio.gather(*tasks)

用户 Query → Query Rewriting（HyDE） → 向量检索（Milvus/PGVector） 
→ 重排序（BGE-reranker） → 上下文压缩（LLMLingua） → DeepSeek 生成 → 引用溯源

def rag_with_citation(query, top_k=5):
    # 1. 向量检索
    docs = vector_store.similarity_search(query, k=top_k)
    # 2. 构建上下文，保留来源
    context = "\n\n".join([f"[{i+1}] {doc.page_content}" for i, doc in enumerate(docs)])
    sources = [doc.metadata["source"] for doc in docs]
    
    # 3. 构造 prompt
    prompt = f"""基于以下参考资料回答问题，并标注引用编号（如 [1]）：
参考资料：
{context}

问题：{query}
答案："""
    
    response = deepseek_chat(prompt)
    # 4. 后处理：校验引用编号是否存在
    return response, sources

tools = [{
    "type": "function",
    "function": {
        "name": "get_stock_price",
        "description": "获取指定股票代码的实时价格",
        "parameters": {
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "股票代码，如 AAPL"}
            },
            "required": ["symbol"]
        }
    }
}]

messages = [{"role": "user", "content": "现在苹果的股价是多少？"}]
resp = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)
# 解析 tool_calls 并执行本地函数，二次调用生成最终回答

def route_model(query):
    if len(query) < 50 and any(kw in query for kw in ["分类", "提取", "翻译"]):
        return "deepseek-chat"
    else:
        return "deepseek-reasoner"

PDF/图片 → 版面分析（LayoutLMv3） → 文本抽取 + 表格结构化 
→ 合并为 Markdown 格式 → 送入 DeepSeek 进行 QA/摘要

def hierarchical_summary(long_text, chunk_size=10000, overlap=1000):
    chunks = split_text(long_text, chunk_size, overlap)
    summaries = []
    for chunk in chunks:
        prompt = f"请用 200 字以内总结以下段落：\n{chunk}"
        summaries.append(deepseek_chat(prompt))
    # 二次聚合
    final_prompt = "综合以下各段落摘要，生成整体总结：\n" + "\n".join(summaries)
    return deepseek_chat(final_prompt)

# vLLM 部署示例
apiVersion: apps/v1
kind: Deployment
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: deepseek-server
        image: vllm/vllm-openai:latest
        args: ["--model", "/mnt/models/DeepSeek-V3-Lite", "--tensor-parallel-size", "2"]
        resources:
          limits:
            nvidia.com/gpu: 2