DeepSeek API 最佳实践与性能优化指南

在生产环境中使用 DeepSeek API 时，合理的架构设计和性能优化策略至关重要。本文将深入讲解从基础调用到高级优化的完整实践方案，帮助你构建高效、稳定、低成本的 AI 应用。

一、DeepSeek API 概述

DeepSeek 提供多个模型供不同场景使用，均通过兼容 OpenAI 格式的 API 进行调用：

所有模型共享统一的 API 端点，切换模型只需修改 model 参数即可。

二、API 基础使用

2.1 认证配置

DeepSeek API 使用 Bearer Token 认证，完全兼容 OpenAI SDK 格式：

from openai import OpenAI

# 初始化客户端，指向 DeepSeek API 端点
client = OpenAI(
    api_key="sk-your-api-key",
    base_url="https://api.deepseek.com"  # DeepSeek API 基础地址
)

import OpenAI from 'openai';

// 初始化客户端，配置 DeepSeek 端点
const client = new OpenAI({
  apiKey: 'sk-your-api-key',
  baseURL: 'https://api.deepseek.com',  // DeepSeek API 基础地址
});

2.2 基础调用示例

# 基础对话调用
response = client.chat.completions.create(
    model="deepseek-chat",        # 使用 V3 模型
    messages=[
        {"role": "system", "content": "你是一个专业的技术助手。"},
        {"role": "user", "content": "解释 Python 的 GIL 机制"}
    ],
    temperature=0.7,               # 控制输出随机性
    max_tokens=2048,               # 最大输出 token 数
    top_p=0.95                     # 核采样参数
)

print(response.choices[0].message.content)

使用 curl 的等效调用：

curl https://api.deepseek.com/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer sk-your-api-key" \
  -d '{
    "model": "deepseek-chat",
    "messages": [
      {"role": "system", "content": "你是一个专业的技术助手。"},
      {"role": "user", "content": "解释 Python 的 GIL 机制"}
    ],
    "temperature": 0.7,
    "max_tokens": 2048
  }'

三、提示工程最佳实践（Prompt Engineering）

3.1 系统提示词设计原则

# 结构化系统提示词示例
system_prompt = """你是一个专业的数据分析师，请遵循以下规则：

## 角色定义
- 专注于数据分析和可视化建议
- 使用专业但易懂的语言

## 输出格式
- 使用 Markdown 格式
- 包含具体的代码示例
- 关键数据用表格展示

## 约束条件
- 不编造数据
- 对不确定的内容明确标注
- 回答长度控制在 500 字以内
"""

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": "分析这份销售数据的趋势"}
    ]
)

3.2 Few-shot 示例优化

# 使用 few-shot 提升输出一致性
messages = [
    {"role": "system", "content": "你是一个 JSON 格式化助手，将自然语言转为结构化数据。"},
    # 示例 1
    {"role": "user", "content": "张三，男，28岁，北京"},
    {"role": "assistant", "content": '{"name": "张三", "gender": "男", "age": 28, "city": "北京"}'},
    # 示例 2
    {"role": "user", "content": "李四，女，35岁，上海"},
    {"role": "assistant", "content": '{"name": "李四", "gender": "女", "age": 35, "city": "上海"}'},
    # 实际查询
    {"role": "user", "content": "王五，男，42岁，深圳"}
]

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    temperature=0       # 结构化输出建议使用低温度
)

3.3 思维链（Chain-of-Thought）提示

# 使用 R1 模型进行深度推理
response = client.chat.completions.create(
    model="deepseek-reasoner",   # R1 推理模型
    messages=[
        {
            "role": "user",
            "content": """请逐步分析以下问题：

一个水池有两个进水管和一个出水管。
进水管A每小时注入 3 立方米，进水管B每小时注入 5 立方米。
出水管每小时排出 2 立方米。
水池容量为 120 立方米。
问：从空池开始，多少小时能注满？

请展示完整的推理过程。"""
        }
    ]
)

四、流式输出（Streaming）实现

4.1 Python 流式输出

# 流式输出 - 逐 token 返回，降低首字延迟
stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "user", "content": "写一篇关于人工智能的短文"}
    ],
    stream=True          # 启用流式输出
)

# 逐块处理响应
full_response = ""
for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        content = chunk.choices[0].delta.content
        full_response += content
        print(content, end="", flush=True)  # 实时打印

4.2 Node.js 流式输出

// 使用 async iterator 处理流式响应
async function streamChat(prompt) {
  const stream = await client.chat.completions.create({
    model: 'deepseek-chat',
    messages: [{ role: 'user', content: prompt }],
    stream: true,  // 启用流式输出
  });

  let fullResponse = '';
  for await (const chunk of stream) {
    const content = chunk.choices[0]?.delta?.content || '';
    fullResponse += content;
    process.stdout.write(content);  // 实时输出到控制台
  }
  return fullResponse;
}

// 调用示例
streamChat('用 JavaScript 实现一个快速排序算法');

4.3 SSE（Server-Sent Events）Web 集成

from flask import Flask, Response
import json

app = Flask(__name__)

@app.route('/api/chat', methods=['POST'])
def chat_stream():
    """SSE 流式接口，适用于前端实时展示"""
    def generate():
        stream = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": "你好"}],
            stream=True
        )
        for chunk in stream:
            content = chunk.choices[0].delta.content
            if content:
                # 按 SSE 格式推送数据
                yield f"data: {json.dumps({'content': content})}\n\n"
        yield "data: [DONE]\n\n"  # 结束标识

    return Response(generate(), mimetype='text/event-stream')

五、批处理（Batch API）优化

5.1 批量请求处理

import asyncio
from openai import AsyncOpenAI

# 使用异步客户端实现批量请求
async_client = AsyncOpenAI(
    api_key="sk-your-api-key",
    base_url="https://api.deepseek.com"
)

async def process_batch(prompts: list[str], max_concurrent: int = 5):
    """批量处理多个请求，使用信号量控制并发数"""
    semaphore = asyncio.Semaphore(max_concurrent)  # 限制并发数

    async def single_request(prompt):
        async with semaphore:
            response = await async_client.chat.completions.create(
                model="deepseek-chat",
                messages=[{"role": "user", "content": prompt}],
                max_tokens=1024
            )
            return response.choices[0].message.content

    # 并发执行所有请求
    tasks = [single_request(p) for p in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# 使用示例
prompts = [
    "总结量子计算的核心原理",
    "解释区块链的共识机制",
    "描述神经网络的反向传播算法",
    "介绍强化学习的基本概念",
    "解释 Transformer 架构的注意力机制"
]

results = asyncio.run(process_batch(prompts, max_concurrent=3))
for i, result in enumerate(results):
    print(f"--- 问题 {i+1} ---")
    print(result[:200])  # 打印前200字符

5.2 JSONL 批处理文件格式

import json

def create_batch_file(requests: list[dict], output_path: str):
    """创建 JSONL 格式的批处理文件"""
    with open(output_path, 'w', encoding='utf-8') as f:
        for i, req in enumerate(requests):
            batch_item = {
                "custom_id": f"request-{i}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": "deepseek-chat",
                    "messages": req["messages"],
                    "max_tokens": req.get("max_tokens", 1024)
                }
            }
            f.write(json.dumps(batch_item, ensure_ascii=False) + "\n")

# 构造批处理请求列表
batch_requests = [
    {"messages": [{"role": "user", "content": f"翻译为英文：{text}"}]}
    for text in ["你好世界", "人工智能", "深度学习", "自然语言处理"]
]

create_batch_file(batch_requests, "batch_input.jsonl")

六、Function Calling / Tool Use

6.1 定义工具函数

# 定义可供模型调用的工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如：北京、上海"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "在数据库中搜索产品信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    },
                    "category": {
                        "type": "string",
                        "description": "产品类别"
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "最大返回结果数"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

6.2 完整的 Tool Use 工作流

import json

def get_weather(city: str, unit: str = "celsius") -> dict:
    """模拟天气查询接口"""
    # 实际项目中应调用真实的天气 API
    return {"city": city, "temperature": 22, "unit": unit, "condition": "晴"}

def search_database(query: str, category: str = None, max_results: int = 5) -> list:
    """模拟数据库查询"""
    return [{"name": f"{query}相关产品", "price": 99.9, "category": category}]

# 工具函数映射表
tool_functions = {
    "get_weather": get_weather,
    "search_database": search_database,
}

def run_with_tools(user_message: str):
    """带工具调用的完整对话流程"""
    messages = [{"role": "user", "content": user_message}]

    # 第一次调用：让模型决定是否使用工具
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=messages,
        tools=tools,
        tool_choice="auto"   # 自动决定是否调用工具
    )

    assistant_message = response.choices[0].message

    # 检查是否有工具调用
    if assistant_message.tool_calls:
        messages.append(assistant_message)

        # 执行每个工具调用
        for tool_call in assistant_message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            # 调用对应的工具函数
            result = tool_functions[func_name](**func_args)

            # 将工具结果加入消息列表
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })

        # 第二次调用：模型根据工具结果生成最终回答
        final_response = client.chat.completions.create(
            model="deepseek-chat",
            messages=messages,
            tools=tools
        )
        return final_response.choices[0].message.content

    return assistant_message.content

# 使用示例
print(run_with_tools("北京今天天气怎么样？帮我查一下防晒霜产品"))

6.3 Node.js Tool Use 实现

// 工具定义
const tools = [
  {
    type: 'function',
    function: {
      name: 'calculate',
      description: '执行数学计算',
      parameters: {
        type: 'object',
        properties: {
          expression: { type: 'string', description: '数学表达式' }
        },
        required: ['expression']
      }
    }
  }
];

// 工具函数实现
const toolFunctions = {
  calculate: ({ expression }) => {
    try {
      // 安全的数学表达式计算
      const result = Function(`"use strict"; return (${expression})`)();
      return { result, expression };
    } catch (e) {
      return { error: '计算失败', expression };
    }
  }
};

// 带工具调用的对话函数
async function chatWithTools(userMessage) {
  const messages = [{ role: 'user', content: userMessage }];

  const response = await client.chat.completions.create({
    model: 'deepseek-chat',
    messages,
    tools,
    tool_choice: 'auto',  // 自动决定是否调用工具
  });

  const assistantMsg = response.choices[0].message;

  if (assistantMsg.tool_calls) {
    messages.push(assistantMsg);

    // 逐个执行工具调用
    for (const toolCall of assistantMsg.tool_calls) {
      const args = JSON.parse(toolCall.function.arguments);
      const result = toolFunctions[toolCall.function.name](args);
      messages.push({
        role: 'tool',
        tool_call_id: toolCall.id,
        content: JSON.stringify(result),
      });
    }

    // 将工具结果发回模型获取最终回答
    const finalResponse = await client.chat.completions.create({
      model: 'deepseek-chat',
      messages,
      tools,
    });
    return finalResponse.choices[0].message.content;
  }

  return assistantMsg.content;
}

七、速率限制与重试策略

7.1 指数退避重试

import time
import random
from openai import RateLimitError, APITimeoutError, APIConnectionError

def call_with_retry(
    func,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0
):
    """带指数退避的重试装饰器"""
    for attempt in range(max_retries):
        try:
            return func()
        except RateLimitError as e:
            # 速率限制：使用较长的退避时间
            delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
            print(f"触发速率限制，等待 {delay:.1f} 秒后重试 (第 {attempt+1}/{max_retries} 次)")
            time.sleep(delay)
        except APITimeoutError:
            # 超时：较短的退避
            delay = min(base_delay * (1.5 ** attempt), max_delay)
            print(f"请求超时，等待 {delay:.1f} 秒后重试")
            time.sleep(delay)
        except APIConnectionError:
            # 连接错误：中等退避
            delay = min(base_delay * (2 ** attempt), max_delay)
            print(f"连接错误，等待 {delay:.1f} 秒后重试")
            time.sleep(delay)

    raise Exception(f"重试 {max_retries} 次后仍然失败")

# 使用示例
result = call_with_retry(
    lambda: client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": "你好"}],
        timeout=30  # 30秒超时
    )
)

7.2 使用 tenacity 库实现高级重试

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
from openai import RateLimitError, APITimeoutError

@retry(
    stop=stop_after_attempt(5),                      # 最多重试5次
    wait=wait_exponential(multiplier=1, max=60),     # 指数退避，最大60秒
    retry=retry_if_exception_type(                   # 仅对特定异常重试
        (RateLimitError, APITimeoutError)
    ),
    before_sleep=lambda info: print(               # 重试前打印日志
        f"重试第 {info.idle_for:.1f} 秒后执行..."
    )
)
def reliable_api_call(messages: list, model: str = "deepseek-chat"):
    """带自动重试的可靠 API 调用"""
    return client.chat.completions.create(
        model=model,
        messages=messages,
        timeout=30
    )

7.3 令牌桶速率限制器

import time
import threading

class TokenBucketRateLimiter:
    """令牌桶算法限速器，控制 API 请求频率"""

    def __init__(self, rate: float, capacity: int):
        self.rate = rate              # 每秒补充的令牌数
        self.capacity = capacity      # 桶的最大容量
        self.tokens = capacity        # 当前令牌数
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self):
        """获取一个令牌，如果没有可用令牌则等待"""
        while True:
            with self.lock:
                now = time.monotonic()
                # 补充令牌
                elapsed = now - self.last_refill
                self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
                self.last_refill = now

                if self.tokens >= 1:
                    self.tokens -= 1
                    return
            time.sleep(0.1)  # 短暂等待后再试

# 使用示例：每秒最多 10 个请求
limiter = TokenBucketRateLimiter(rate=10, capacity=10)

def rate_limited_call(messages):
    """受速率限制的 API 调用"""
    limiter.acquire()
    return client.chat.completions.create(
        model="deepseek-chat",
        messages=messages
    )

八、成本优化技巧

8.1 Prompt 缓存

# 利用前缀缓存降低重复请求成本
# 固定的系统提示词作为缓存前缀
CACHED_SYSTEM_PROMPT = """你是一个专业的客服助手，负责处理以下类型的问题：
1. 产品咨询
2. 订单查询
3. 售后服务
4. 投诉建议

请始终保持礼貌和专业。对于无法处理的问题，请引导用户联系人工客服。
"""

def customer_service_chat(user_message: str, conversation_history: list = None):
    """客服对话 - 利用固定前缀实现缓存复用"""
    messages = [{"role": "system", "content": CACHED_SYSTEM_PROMPT}]

    if conversation_history:
        messages.extend(conversation_history)

    messages.append({"role": "user", "content": user_message})

    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=messages,
        max_tokens=512      # 限制输出长度控制成本
    )

    # 检查缓存命中情况
    usage = response.usage
    print(f"输入 tokens: {usage.prompt_tokens}")
    print(f"输出 tokens: {usage.completion_tokens}")
    if hasattr(usage, 'prompt_cache_hit_tokens'):
        print(f"缓存命中 tokens: {usage.prompt_cache_hit_tokens}")

    return response.choices[0].message.content

8.2 Prompt 压缩策略

def compress_prompt(text: str, max_length: int = 2000) -> str:
    """压缩长文本，减少 token 消耗"""
    if len(text) <= max_length:
        return text

    # 策略1：使用模型自身进行摘要压缩
    summary_response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "请将以下内容压缩为简洁摘要，保留关键信息："},
            {"role": "user", "content": text}
        ],
        max_tokens=500,
        temperature=0
    )
    return summary_response.choices[0].message.content

def smart_context_window(messages: list, max_tokens: int = 4000) -> list:
    """智能上下文窗口管理，保留最重要的消息"""
    if not messages:
        return messages

    # 始终保留系统消息和最近的消息
    system_msgs = [m for m in messages if m["role"] == "system"]
    non_system = [m for m in messages if m["role"] != "system"]

    # 如果消息过多，保留首尾部分
    if len(non_system) > 10:
        # 保留前2条（建立上下文）和后6条（最新对话）
        trimmed = non_system[:2] + [
            {"role": "system", "content": "[中间对话已省略]"}
        ] + non_system[-6:]
        return system_msgs + trimmed

    return messages

8.3 模型选择策略

def smart_model_selection(query: str) -> str:
    """根据任务复杂度自动选择最优模型，平衡性能与成本"""

    # 简单任务关键词 - 使用较小模型
    simple_keywords = ["翻译", "总结", "改写", "格式化", "提取"]
    # 复杂任务关键词 - 使用推理模型
    complex_keywords = ["证明", "推导", "分析", "设计架构", "数学"]
    # 代码任务 - 使用代码模型
    code_keywords = ["代码", "编程", "debug", "重构", "实现函数"]

    query_lower = query.lower()

    if any(kw in query_lower for kw in code_keywords):
        return "deepseek-coder"           # 代码任务用 Coder
    elif any(kw in query_lower for kw in complex_keywords):
        return "deepseek-reasoner"        # 复杂推理用 R1
    else:
        return "deepseek-chat"            # 通用任务用 V3

# 模型定价对比表
MODEL_PRICING = {
    "deepseek-chat": {
        "input": 0.27,       # 每百万 token 输入价格（元）
        "output": 1.10,      # 每百万 token 输出价格（元）
        "cache_hit": 0.07    # 缓存命中价格
    },
    "deepseek-reasoner": {
        "input": 0.55,
        "output": 2.19,
        "cache_hit": 0.14
    }
}

def estimate_cost(input_tokens: int, output_tokens: int, model: str) -> float:
    """估算单次调用成本（元）"""
    pricing = MODEL_PRICING.get(model, MODEL_PRICING["deepseek-chat"])
    cost = (input_tokens / 1_000_000 * pricing["input"] +
            output_tokens / 1_000_000 * pricing["output"])
    return round(cost, 6)

九、错误处理和监控

9.1 全面的错误处理

from openai import (
    APIError,
    AuthenticationError,
    RateLimitError,
    APITimeoutError,
    BadRequestError,
    APIConnectionError
)
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("deepseek_api")

def robust_api_call(messages: list, **kwargs):
    """健壮的 API 调用，包含完整错误处理"""
    try:
        response = client.chat.completions.create(
            model=kwargs.get("model", "deepseek-chat"),
            messages=messages,
            **{k: v for k, v in kwargs.items() if k != "model"}
        )
        # 记录 token 使用量
        logger.info(
            f"API 调用成功 | 输入: {response.usage.prompt_tokens} tokens "
            f"| 输出: {response.usage.completion_tokens} tokens"
        )
        return response

    except AuthenticationError:
        logger.error("认证失败：请检查 API Key 是否正确")
        raise
    except RateLimitError as e:
        logger.warning(f"触发速率限制：{e.message}")
        raise
    except BadRequestError as e:
        logger.error(f"请求参数错误：{e.message}")
        # 常见原因：上下文过长、参数格式错误
        raise
    except APITimeoutError:
        logger.warning("请求超时，建议减少输入长度或增加超时时间")
        raise
    except APIConnectionError:
        logger.error("网络连接失败，请检查网络和 API 端点配置")
        raise
    except APIError as e:
        logger.error(f"API 内部错误 (状态码 {e.status_code}): {e.message}")
        raise

9.2 使用监控和指标收集

import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class APIMetrics:
    """API 调用指标收集器"""
    total_calls: int = 0
    successful_calls: int = 0
    failed_calls: int = 0
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    total_latency: float = 0.0
    errors: dict = field(default_factory=dict)

    @property
    def avg_latency(self) -> float:
        """平均延迟（秒）"""
        return self.total_latency / max(self.total_calls, 1)

    @property
    def success_rate(self) -> float:
        """成功率"""
        return self.successful_calls / max(self.total_calls, 1)

    def report(self) -> str:
        """生成监控报告"""
        return f"""
=== DeepSeek API 监控报告 ===
总调用次数: {self.total_calls}
成功率: {self.success_rate:.1%}
平均延迟: {self.avg_latency:.2f}s
总输入 Tokens: {self.total_input_tokens:,}
总输出 Tokens: {self.total_output_tokens:,}
错误分布: {self.errors}
"""

metrics = APIMetrics()

def monitored_call(messages: list, **kwargs):
    """带监控的 API 调用"""
    metrics.total_calls += 1
    start = time.time()

    try:
        response = client.chat.completions.create(
            model=kwargs.get("model", "deepseek-chat"),
            messages=messages,
            **{k: v for k, v in kwargs.items() if k != "model"}
        )
        metrics.successful_calls += 1
        metrics.total_input_tokens += response.usage.prompt_tokens
        metrics.total_output_tokens += response.usage.completion_tokens
        return response

    except Exception as e:
        metrics.failed_calls += 1
        error_type = type(e).__name__
        metrics.errors[error_type] = metrics.errors.get(error_type, 0) + 1
        raise
    finally:
        metrics.total_latency += time.time() - start

十、LangChain 集成

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate

# 初始化 DeepSeek 模型（兼容 LangChain 的 OpenAI 接口）
llm = ChatOpenAI(
    model="deepseek-chat",
    openai_api_key="sk-your-api-key",
    openai_api_base="https://api.deepseek.com",
    temperature=0.7,
    max_tokens=2048,
    streaming=True            # 启用流式输出
)

# 使用 Prompt Template
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}，请用{style}的风格回答问题。"),
    ("human", "{question}")
])

# 创建调用链
chain = prompt | llm

# 同步调用
result = chain.invoke({
    "role": "技术专家",
    "style": "简洁专业",
    "question": "微服务架构的优缺点是什么？"
})
print(result.content)

# 流式调用
async def stream_langchain():
    async for chunk in chain.astream({
        "role": "技术专家",
        "style": "简洁专业",
        "question": "微服务架构的优缺点是什么？"
    }):
        print(chunk.content, end="", flush=True)

十一、LlamaIndex 集成

from llama_index.llms.openai_like import OpenAILike
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings

# 配置 DeepSeek 作为 LlamaIndex 的 LLM
llm = OpenAILike(
    model="deepseek-chat",
    api_base="https://api.deepseek.com",
    api_key="sk-your-api-key",
    is_chat_model=True,
    temperature=0.7,
    max_tokens=2048
)

# 设置全局默认 LLM
Settings.llm = llm

# 构建 RAG 管道
documents = SimpleDirectoryReader("./data").load_data()  # 加载文档
index = VectorStoreIndex.from_documents(documents)       # 构建索引
query_engine = index.as_query_engine(                    # 创建查询引擎
    similarity_top_k=3,            # 检索前3个相关片段
    streaming=True                 # 启用流式输出
)

# 执行查询
response = query_engine.query("DeepSeek V3 的 MoE 架构有什么特点？")
print(response)

十二、延迟优化

12.1 上下文长度控制

import tiktoken

def count_tokens(text: str, model: str = "deepseek-chat") -> int:
    """估算文本的 token 数量"""
    # DeepSeek 使用自定义分词器，这里使用 cl100k_base 近似估算
    encoder = tiktoken.get_encoding("cl100k_base")
    return len(encoder.encode(text))

def optimize_context(messages: list, max_context_tokens: int = 8000) -> list:
    """优化上下文长度，减少延迟"""
    total_tokens = sum(count_tokens(m["content"]) for m in messages)

    if total_tokens <= max_context_tokens:
        return messages

    # 策略：压缩历史消息，保留最近对话
    optimized = []
    system_msg = None
    remaining_tokens = max_context_tokens

    # 保留系统消息
    for msg in messages:
        if msg["role"] == "system":
            system_msg = msg
            remaining_tokens -= count_tokens(msg["content"])
            break

    if system_msg:
        optimized.append(system_msg)

    # 从最新消息开始，逆序添加直到达到限制
    non_system = [m for m in messages if m["role"] != "system"]
    for msg in reversed(non_system):
        msg_tokens = count_tokens(msg["content"])
        if remaining_tokens >= msg_tokens:
            optimized.insert(len(optimized), msg)  # 保持顺序
            remaining_tokens -= msg_tokens
        else:
            break

    return optimized

12.2 并发请求优化

import asyncio
import aiohttp
from typing import Any

class DeepSeekBatchClient:
    """高性能批量请求客户端"""

    def __init__(self, api_key: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.base_url = "https://api.deepseek.com"
        self.semaphore = asyncio.Semaphore(max_concurrent)  # 并发控制
        self.session = None

    async def __aenter__(self):
        # 使用连接池复用 TCP 连接
        connector = aiohttp.TCPConnector(
            limit=20,               # 最大连接数
            keepalive_timeout=30     # 连接保活时间
        )
        self.session = aiohttp.ClientSession(
            connector=connector,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
        )
        return self

    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()

    async def _single_request(self, payload: dict) -> dict:
        """执行单个请求，使用信号量控制并发"""
        async with self.semaphore:
            async with self.session.post(
                f"{self.base_url}/chat/completions",
                json=payload,
                timeout=aiohttp.ClientTimeout(total=60)
            ) as response:
                return await response.json()

    async def batch_complete(self, prompts: list[str], model: str = "deepseek-chat") -> list:
        """批量完成请求"""
        payloads = [
            {
                "model": model,
                "messages": [{"role": "user", "content": p}],
                "max_tokens": 1024
            }
            for p in prompts
        ]

        tasks = [self._single_request(p) for p in payloads]
        return await asyncio.gather(*tasks, return_exceptions=True)

# 使用示例
async def main():
    async with DeepSeekBatchClient("sk-your-api-key", max_concurrent=5) as client:
        prompts = [f"简要解释概念：{topic}" for topic in [
            "机器学习", "深度学习", "强化学习", "迁移学习", "联邦学习"
        ]]
        results = await client.batch_complete(prompts)
        for r in results:
            if isinstance(r, dict) and "choices" in r:
                print(r["choices"][0]["message"]["content"][:100])

asyncio.run(main())

12.3 响应延迟对比与优化建议

总结

DeepSeek API 在生产环境中的高效使用需要综合考虑以下几个维度：

1. 模型选择：根据任务复杂度匹配合适的模型（V3 / R1 / Coder）
2. 提示工程：结构化 prompt 设计、few-shot 示例、思维链推理
3. 性能优化：流式输出、并发控制、上下文管理
4. 成本控制：缓存利用、prompt 压缩、输出长度限制
5. 稳定性保障：重试策略、错误处理、监控告警

通过合理应用这些最佳实践，你可以构建出高效、稳定、经济的 AI 应用。建议从小规模验证开始，逐步应用各项优化策略，持续监控关键指标以确保系统始终运行在最佳状态。