教程

DeepSeek 推理模型完全教程

2026年05月29日 11:00 41 次阅读
教程简介

零基础DeepSeek推理模型完全教程，涵盖DeepSeek-V3与R1架构解析、MoE混合专家机制、推理链思维能力、API接入实战、本地部署方案、与OpenAI o1/o3对比、蒸馏模型系列、数学与代码推理、RAG集成、Agent应用等核心技能，配有数学推理助手与代码自动修复系统两大实战项目，适合AI开发者和研究者系统学习。

DeepSeek 推理模型完全教程

零基础入门到实战 — 涵盖架构解析、API开发、本地部署、Agent应用全流程

📚 完整目录

第一章：DeepSeek 模型家族概览

1.1 DeepSeek 发展历程

DeepSeek（深度求索）是由中国量化投资公司幻方量化创立的人工智能研究实验室，自2023年成立以来，迅速在大模型领域崭露头角。其核心理念是通过技术创新降低大模型的训练和推理成本，同时保持甚至超越前沿模型的性能。

关键里程碑：

时间 事件 意义
2023年11月 DeepSeek-Coder 发布 首个代码专用模型，性能媲美 Codex
2024年1月 DeepSeek-V1 发布 通用大语言模型，67B参数
2024年5月 DeepSeek-V2 发布 引入 MLA 注意力机制，大幅降低推理成本
2024年12月 DeepSeek-V3 发布 MoE架构，671B总参数，37B激活参数
2025年1月 DeepSeek-R1 发布 推理专用模型，性能对标 OpenAI o1

1.2 模型家族一览

DeepSeek 的模型家族可大致分为以下几个系列：

  • DeepSeek-V 系列：通用基础模型，代表产品为 DeepSeek-V3
  • DeepSeek-R 系列：推理增强模型，代表产品为 DeepSeek-R1
  • DeepSeek-Coder 系列：代码专用模型
  • DeepSeek-Math 系列：数学专用模型
  • 蒸馏模型系列：基于 R1 蒸馏的轻量级模型（1.5B/7B/8B/14B/32B/70B）

1.3 核心技术特色

DeepSeek 模型的几个核心技术亮点：

  1. Multi-head Latent Attention (MLA)：创新的注意力机制，显著降低 KV Cache 内存占用
  2. Mixture of Experts (MoE)：稀疏激活的混合专家架构，以更低成本实现更大容量
  3. 辅助损失自由负载均衡：在不引入额外损失函数的情况下实现专家负载均衡
  4. 多 Token 预测：训练时同时预测多个后续 Token，提升训练效率

第二章：DeepSeek-V3 架构深度解析

2.1 整体架构

DeepSeek-V3 采用 Transformer 架构，共 61 层，其中包含 1 个 Embedding 层、61 个 Transformer 块和 1 个输出层。关键参数如下：

  • 总参数量：671B（6710亿）
  • 激活参数量：37B（每次推理仅激活约5.5%的参数）
  • 隐藏维度：7168
  • 注意力头数：128（KV头数压缩至128，使用 GQA）
  • 专家数量：256个路由专家 + 1个共享专家
  • 上下文长度：128K tokens

2.2 Multi-head Latent Attention (MLA)

MLA 是 DeepSeek-V2 引入的核心创新，其关键思想是将 Key-Value 对进行低秩压缩：

# MLA 核心原理（简化示意）
import torch
import torch.nn as nn

class MultiHeadLatentAttention(nn.Module):
    def __init__(self, d_model, n_heads, kv_lora_rank, q_lora_rank):
        super().__init__()
        self.n_heads = n_heads
        self.d_head = d_model // n_heads
        self.kv_lora_rank = kv_lora_rank  # KV 压缩维度，远小于原始维度
        
        # KV 压缩投影
        self.kv_compress = nn.Linear(d_model, kv_lora_rank, bias=False)
        # KV 解压投影
        self.kv_decompress_k = nn.Linear(kv_lora_rank, n_heads * self.d_head, bias=False)
        self.kv_decompress_v = nn.Linear(kv_lora_rank, n_heads * self.d_head, bias=False)
        # Query 压缩
        self.q_compress = nn.Linear(d_model, q_lora_rank, bias=False)
        self.q_decompress = nn.Linear(q_lora_rank, n_heads * self.d_head, bias=False)
        
    def forward(self, x, kv_cache=None):
        batch_size, seq_len, _ = x.shape
        
        # 压缩 KV 到低维空间
        kv_compressed = self.kv_compress(x)  # [B, S, kv_lora_rank]
        
        # 从压缩表示恢复 K 和 V
        k = self.kv_decompress_k(kv_compressed)  # [B, S, n_heads * d_head]
        v = self.kv_decompress_v(kv_compressed)  # [B, S, n_heads * d_head]
        
        # Query 处理
        q_compressed = self.q_compress(x)
        q = self.q_decompress(q_compressed)
        
        # 重塑为多头格式
        q = q.view(batch_size, seq_len, self.n_heads, self.d_head).transpose(1, 2)
        k = k.view(batch_size, seq_len, self.n_heads, self.d_head).transpose(1, 2)
        v = v.view(batch_size, seq_len, self.n_heads, self.d_head).transpose(1, 2)
        
        # 标准注意力计算
        scale = self.d_head ** -0.5
        attn_weights = torch.matmul(q, k.transpose(-2, -1)) * scale
        attn_weights = torch.softmax(attn_weights, dim=-1)
        output = torch.matmul(attn_weights, v)
        
        return output

MLA 的优势：传统多头注意力的 KV Cache 随序列长度线性增长，在长上下文场景中成为瓶颈。MLA 将 KV 压缩到低秩空间，推理时只需缓存压缩后的表示，大幅降低内存占用。实测中，MLA 使 DeepSeek-V3 的 KV Cache 减少约 93.3%。

2.3 DeepSeekMoE 架构

DeepSeek-V3 采用的 MoE 架构有两大创新：

1. 细粒度专家分割：将传统的少数大专家分割为更多小专家（256个），提高专家的专业化程度。

2. 共享专家隔离：设置 1 个"共享专家"始终参与计算，捕获通用知识；其余路由专家则根据输入动态激活。

# DeepSeekMoE 架构示意
class DeepSeekMoE(nn.Module):
    def __init__(self, d_model, n_routed_experts=256, n_shared_experts=1, top_k=8):
        super().__init__()
        self.n_routed_experts = n_routed_experts
        self.top_k = top_k
        
        # 共享专家 - 始终激活
        self.shared_experts = nn.ModuleList([
            nn.Sequential(
                nn.Linear(d_model, d_model * 4),
                nn.SiLU(),
                nn.Linear(d_model * 4, d_model)
            ) for _ in range(n_shared_experts)
        ])
        
        # 路由专家 - 按需激活
        self.routed_experts = nn.ModuleList([
            nn.Sequential(
                nn.Linear(d_model, d_model * 4),
                nn.SiLU(),
                nn.Linear(d_model * 4, d_model)
            ) for _ in range(n_routed_experts)
        ])
        
        # 路由器（门控网络）
        self.gate = nn.Linear(d_model, n_routed_experts, bias=False)
    
    def forward(self, x):
        batch_size, seq_len, d_model = x.shape
        
        # 计算路由权重
        router_logits = self.gate(x)  # [B, S, n_routed_experts]
        router_probs = torch.softmax(router_logits, dim=-1)
        
        # 选择 Top-K 专家
        top_k_probs, top_k_indices = torch.topk(router_probs, self.top_k, dim=-1)
        top_k_probs = top_k_probs / top_k_probs.sum(dim=-1, keepdim=True)  # 归一化
        
        # 路由专家计算
        routed_output = torch.zeros_like(x)
        for k in range(self.top_k):
            expert_idx = top_k_indices[:, :, k]  # [B, S]
            expert_prob = top_k_probs[:, :, k:k+1]  # [B, S, 1]
            
            for i in range(self.n_routed_experts):
                mask = (expert_idx == i)
                if mask.any():
                    expert_input = x[mask]
                    expert_out = self.routed_experts[i](expert_input)
                    routed_output[mask] += expert_prob[mask].squeeze(-1).unsqueeze(-1) * expert_out
        
        # 共享专家计算
        shared_output = sum(expert(x) for expert in self.shared_experts)
        
        return routed_output + shared_output

2.4 辅助损失自由负载均衡

传统 MoE 模型使用辅助损失函数（Auxiliary Loss）来鼓励专家负载均衡，但辅助损失会干扰主损失函数。DeepSeek-V3 引入了**偏置项（Bias Term）**方法：

# 辅助损失自由负载均衡
class BiasBasedLoadBalancing(nn.Module):
    def __init__(self, n_experts):
        super().__init__()
        # 可学习的偏置项（不参与梯度计算）
        self.bias = nn.Parameter(torch.zeros(n_experts), requires_grad=False)
    
    def compute_routing(self, router_logits, top_k):
        # 路由分数 = 原始分数 + 偏置项
        biased_logits = router_logits + self.bias
        top_k_probs, top_k_indices = torch.topk(biased_logits, top_k, dim=-1)
        return top_k_probs, top_k_indices
    
    def update_bias(self, expert_loads, target_load):
        """
        根据实际负载情况调整偏置项
        负载过高的专家：降低偏置（减少被选中概率）
        负载过低的专家：提高偏置（增加被选中概率）
        """
        with torch.no_grad():
            load_diff = expert_loads - target_load
            self.bias -= 0.001 * load_diff  # 小步调整

第三章：DeepSeek-R1 推理模型详解

3.1 R1 的核心设计理念

DeepSeek-R1 是一个专注于复杂推理的模型，其设计灵感来源于 OpenAI o1 的推理链（Chain-of-Thought）思路，但在实现路径上有显著不同。

R1 的核心创新：

  1. 纯强化学习训练：R1-Zero 版本完全通过 RL 训练（无 SFT 阶段），自发涌现出推理行为
  2. 推理链蒸馏：将长推理链的能力蒸馏到更小的模型中
  3. 多阶段训练流程：R1 正式版采用 Cold Start → RL → Rejection Sampling → SFT → RL 的多阶段流程

3.2 R1-Zero：纯强化学习的突破

R1-Zero 是 DeepSeek 最具创新性的实验。它直接在 DeepSeek-V3-Base 上应用 GRPO（Group Relative Policy Optimization）强化学习算法，不经过任何监督微调（SFT），模型就自发发展出了以下能力：

  • 自我验证：模型会主动检查自己的推理步骤
  • 反思与纠错：发现错误后能够回溯并修正
  • 推理链扩展：遇到复杂问题时自动延长思考过程
# R1-Zero 的推理行为示例
用户问题：计算 17 × 23

<thinking>
让我计算 17 × 23。

方法一：分解计算
17 × 23 = 17 × 20 + 17 × 3
= 340 + 51
= 391

让我验证一下：
23 × 17 = 23 × 10 + 23 × 7
= 230 + 161
= 391

两种方法结果一致，答案是 391。
</thinking>

答案是 391。

3.3 R1 完整训练流程

DeepSeek-R1 的正式版采用了精心设计的四阶段训练流程：

阶段 1: Cold Start（冷启动）
├── 收集少量高质量长推理链数据
├── 对 DeepSeek-V3-Base 进行 SFT
└── 建立基本的推理格式和模式

阶段 2: 推理导向 RL
├── 使用 GRPO 算法进行强化学习
├── 奖励信号：数学/代码等可验证任务的正确性
├── 格式奖励：确保推理链在 <think> 标签内
└── 模型学习更长、更系统的推理链

阶段 3: Rejection Sampling + SFT
├── 使用阶段2的模型生成大量推理轨迹
├── 通过拒绝采样筛选高质量样本
├── 混合通用 SFT 数据（写作、问答等）
└── 重新训练基础模型，平衡推理与通用能力

阶段 4: 全场景 RL
├── 在推理任务上继续 RL 优化
├── 引入 Helpfulness 和 Harmlessness 奖励
└── 最终模型兼顾推理能力和用户体验

3.4 GRPO 算法详解

GRPO（Group Relative Policy Optimization）是 DeepSeek 提出的 RL 算法，是 PPO 的高效替代方案：

# GRPO 算法核心逻辑
import torch
import torch.nn.functional as F

class GRPOTrainer:
    def __init__(self, model, ref_model, kl_coeff=0.01):
        self.model = model
        self.ref_model = ref_model
        self.kl_coeff = kl_coeff
    
    def compute_grpo_loss(self, prompt, responses, rewards):
        """
        GRPO 核心：组内相对优势估计
        - 不需要单独的 Critic/Value 模型
        - 通过对同一 prompt 的多个采样计算相对优势
        """
        # 1. 对同一 prompt 生成 G 个不同响应
        # responses: list of G responses
        # rewards: tensor of G rewards
        
        # 2. 计算组内标准化优势
        mean_reward = rewards.mean()
        std_reward = rewards.std()
        advantages = (rewards - mean_reward) / (std_reward + 1e-8)
        
        # 3. 计算策略梯度损失
        policy_losses = []
        for response, advantage in zip(responses, advantages):
            # 当前策略的 log 概率
            log_prob = self.model.compute_log_prob(prompt, response)
            # 参考策略的 log 概率
            ref_log_prob = self.ref_model.compute_log_prob(prompt, response)
            
            # KL 惩罚
            kl_div = log_prob - ref_log_prob
            
            # PPO-Clip 目标
            ratio = torch.exp(log_prob - ref_log_prob)
            clipped_ratio = torch.clamp(ratio, 1 - 0.2, 1 + 0.2)
            policy_loss = -torch.min(ratio * advantage, clipped_ratio * advantage)
            
            # 加入 KL 散度约束
            total_loss = policy_loss + self.kl_coeff * kl_div
            policy_losses.append(total_loss)
        
        return torch.stack(policy_losses).mean()

GRPO 的优势：

  • 无需训练额外的 Value 网络，节省约 50% 的计算资源
  • 组内相对比较消除了绝对奖励的偏差
  • 训练更稳定，收敛更快

第四章：MoE 混合专家机制

4.1 MoE 原理回顾

混合专家模型（Mixture of Experts）的核心思想是"术业有专攻"——不同的输入由不同的专家子网络处理：

传统 Dense 模型：
输入 → [所有参数都参与计算] → 输出
参数量 = 激活量

MoE 模型：
输入 → [路由器] → 选择 Top-K 专家 → [仅被选中的专家计算] → 输出
参数量 >> 激活量

4.2 DeepSeek MoE 的独特设计

与 GShard、Switch Transformer 等早期 MoE 方案相比，DeepSeek MoE 有三个核心区别：

1. 细粒度专家分割

传统 MoE：8 个大专家，每个专家参数量 = d_model × 4 × d_model
DeepSeek MoE：256 个小专家 + 1 共享专家，每个专家参数量更小

优势：
- 更高的专家专业化程度
- 更灵活的专家组合
- 更均匀的知识分布

2. 共享专家 + 路由专家

输入 → ┌─ 共享专家（始终激活，捕获通用知识）
       └─ 路由专家（Top-K 动态选择，捕获专业知识）
       
两路输出相加 → 最终输出

3. 无辅助损失负载均衡

传统方法需要额外的辅助损失函数来防止"专家坍缩"（所有 token 都路由到少数专家），但辅助损失会干扰主训练目标。DeepSeek 通过可调偏置项实现负载均衡，不引入额外损失。

4.3 MoE 的工程挑战与解决方案

挑战 解决方案
专家负载不均衡 偏置项调节 + 细粒度专家
通信开销大 专家并行 + 流水线优化
内存占用高 模型并行 + 参数卸载
路由抖动 共享专家提供稳定基线

第五章：API 接入实战

5.1 获取 API 访问

DeepSeek 提供了兼容 OpenAI 格式的 API 服务：

步骤 1：访问 platform.deepseek.com 注册账号

步骤 2：在控制台获取 API Key

步骤 3：安装 OpenAI Python SDK（DeepSeek API 兼容 OpenAI 格式）

pip install openai

5.2 基础 API 调用

from openai import OpenAI

# 初始化客户端（兼容 OpenAI SDK）
client = OpenAI(
    api_key="your-deepseek-api-key",
    base_url="https://api.deepseek.com"
)

# 基础对话
response = client.chat.completions.create(
    model="deepseek-chat",  # DeepSeek-V3
    messages=[
        {"role": "system", "content": "你是一个有帮助的AI助手。"},
        {"role": "user", "content": "请解释什么是量子计算？"}
    ],
    temperature=0.7,
    max_tokens=2048
)

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

5.3 推理模型 API 调用

# 使用 DeepSeek-R1 进行复杂推理
response = client.chat.completions.create(
    model="deepseek-reasoner",  # DeepSeek-R1
    messages=[
        {"role": "user", "content": """
证明：对于任意正整数 n，n³ + 2n 能被 3 整除。
"""}
    ],
    max_tokens=8192
)

# R1 模型返回包含推理过程
message = response.choices[0].message
print("推理过程：")
print(message.reasoning_content)  # 思考过程
print("\n最终答案：")
print(message.content)  # 最终回答

5.4 流式输出

# 流式调用 R1 模型
stream = client.chat.completions.create(
    model="deepseek-reasoner",
    messages=[
        {"role": "user", "content": "用 Python 实现一个红黑树，包含插入和删除操作。"}
    ],
    stream=True,
    max_tokens=16384
)

for chunk in stream:
    if chunk.choices[0].delta.reasoning_content:
        print(chunk.choices[0].delta.reasoning_content, end="", flush=True)
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

5.5 Function Calling（工具调用）

import json

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["city"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=tools,
    tool_choice="auto"
)

# 处理工具调用
message = response.choices[0].message
if message.tool_calls:
    tool_call = message.tool_calls[0]
    function_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    print(f"调用工具: {function_name}")
    print(f"参数: {arguments}")

5.6 API 参数调优指南

参数 推荐值 说明
temperature 0.0-0.3（推理任务）/ 0.7-1.0（创作任务） 越低越确定性
top_p 0.9-0.95 核采样阈值
max_tokens 根据任务调整 R1 推理可能需要 8K-32K
frequency_penalty 0.0-0.5 减少重复
presence_penalty 0.0-0.3 鼓励多样性

第六章：本地部署方案

6.1 硬件需求

模型 最低显存 推荐配置 量化方案
R1-Distill-1.5B 4GB 单卡 RTX 3060 FP16/Q4
R1-Distill-7B 16GB 单卡 RTX 4090 FP16/Q4
R1-Distill-14B 32GB A100-40G FP16/Q4
R1-Distill-32B 64GB 2×A100-40G FP16/Q8
R1-Distill-70B 140GB 2×A100-80G FP16/Q4
R1 完整版(671B) 1.3TB+ 多节点集群 FP8

6.2 使用 Ollama 本地部署（推荐入门）

# 安装 Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# 拉取 DeepSeek-R1 蒸馏模型（根据显存选择）
ollama pull deepseek-r1:7b       # 7B 版本，约 4.7GB
ollama pull deepseek-r1:14b      # 14B 版本，约 9GB
ollama pull deepseek-r1:32b      # 32B 版本，约 19GB
ollama pull deepseek-r1:70b      # 70B 版本，约 42GB

# 运行模型
ollama run deepseek-r1:7b

# API 模式（兼容 OpenAI 格式）
# 默认监听 http://localhost:11434

# 使用 OpenAI SDK 连接本地 Ollama
from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama"  # Ollama 不需要真实 key
)

response = client.chat.completions.create(
    model="deepseek-r1:7b",
    messages=[
        {"role": "user", "content": "解释相对论的基本原理"}
    ]
)
print(response.choices[0].message.content)

6.3 使用 vLLM 部署（高性能推理）

# 安装 vLLM
pip install vllm

# 启动 API 服务
python -m vllm.entrypoints.openai.api_server \
    --model deepseek-ai/DeepSeek-R1-Distill-Qwen-32B \
    --tensor-parallel-size 2 \
    --max-model-len 32768 \
    --gpu-memory-utilization 0.9 \
    --host 0.0.0.0 \
    --port 8000

# 连接 vLLM 服务
from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="not-needed"
)

response = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-R1-Distill-Qwen-32B",
    messages=[
        {"role": "user", "content": "证明勾股定理"}
    ],
    temperature=0.6,
    max_tokens=8192
)

6.4 使用 llama.cpp 部署（CPU/低资源方案）

# 编译 llama.cpp
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp
make -j$(nproc)

# 下载 GGUF 格式量化模型
# 推荐从 HuggingFace 获取：bartowski/DeepSeek-R1-Distill-Qwen-7B-GGUF

# 启动服务器
./llama-server \
    -m DeepSeek-R1-Distill-Qwen-7B-Q4_K_M.gguf \
    --host 0.0.0.0 \
    --port 8080 \
    -c 4096 \
    -ngl 99  # GPU 层数，0 表示纯 CPU

6.5 Docker 一键部署

# docker-compose.yml
version: '3.8'
services:
  deepseek-api:
    image: vllm/vllm-openai:latest
    container_name: deepseek-r1
    ports:
      - "8000:8000"
    volumes:
      - ./models:/models
    command: >
      --model deepseek-ai/DeepSeek-R1-Distill-Qwen-32B
      --tensor-parallel-size 2
      --max-model-len 32768
      --gpu-memory-utilization 0.9
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: all
              capabilities: [gpu]

第七章：与 OpenAI o1/o3 对比分析

7.1 性能对比

基准测试 DeepSeek-R1 OpenAI o1 OpenAI o3-mini
AIME 2024（数学竞赛） 79.8% 79.2% 87.3%
MATH-500 97.3% 96.4% 97.9%
Codeforces（编程竞赛） 2029 Elo 1891 Elo 2073 Elo
GPQA Diamond（研究生级问答） 71.5% 75.7% 79.7%
LiveCodeBench 65.9% 63.4% 79.2%

7.2 核心差异对比

维度 DeepSeek-R1 OpenAI o1/o3
训练方法 纯 RL（R1-Zero）+ 多阶段流程 级 RL + 过程奖励模型
推理链可见性 完全可见（可输出思考过程） 部分可见（o1 仅输出摘要）
开源程度 完全开源（模型权重+训练细节） 闭源
API 价格 极低（约 $0.55/M input, $2.19/M output） 较高（o1: $15/M input, $60/M output）
本地部署 支持（提供蒸馏版本） 不支持
蒸馏模型 提供 1.5B-70B 全系列蒸馏模型 不提供
响应速度 中等 较快（o3-mini）

7.3 价格优势

# 价格对比计算
# DeepSeek-R1 API 价格
deepseek_input_price = 0.55   # $/M tokens (缓存命中: $0.14)
deepseek_output_price = 2.19  # $/M tokens

# OpenAI o1 API 价格
o1_input_price = 15.0   # $/M tokens
o1_output_price = 60.0  # $/M tokens

# 假设每月处理 100M input tokens + 50M output tokens
monthly_input = 100  # M tokens
monthly_output = 50   # M tokens

deepseek_cost = (monthly_input * deepseek_input_price + 
                 monthly_output * deepseek_output_price)
o1_cost = (monthly_input * o1_input_price + 
           monthly_output * o1_output_price)

print(f"DeepSeek-R1 月成本: ${deepseek_cost:,.2f}")
print(f"OpenAI o1 月成本: ${o1_cost:,.2f}")
print(f"成本节省: {(1 - deepseek_cost/o1_cost)*100:.1f}%")
# 输出：成本节省约 96%

7.4 选型建议

  • 选 DeepSeek-R1：需要开源可控、本地部署、成本敏感、需要完整推理链、学术研究
  • 选 OpenAI o1/o3：追求极致性能、需要多模态支持、企业级 SLA、已有 OpenAI 生态

第八章：蒸馏模型系列

8.1 蒸馏原理

DeepSeek-R1 的蒸馏过程是将大模型的推理能力转移到小模型中：

教师模型（DeepSeek-R1, 671B）
    ↓ 生成 80 万条高质量推理样本
    ↓ 包含数学、代码、科学、推理等任务
    ↓ 每条样本包含完整推理链
学生模型（1.5B / 7B / 8B / 14B / 32B / 70B）
    ↓ 使用推理样本进行 SFT
    ↓ 学习教师的推理模式和思维链
蒸馏模型（保留大部分推理能力，参数量小 10-100 倍）

8.2 蒸馏模型系列

模型 基座模型 参数量 AIME 2024 MATH-500 适用场景
R1-Distill-Qwen-1.5B Qwen2.5-1.5B 1.5B 28.9% 83.9% 移动端/边缘设备
R1-Distill-Qwen-7B Qwen2.5-7B 7B 55.5% 92.8% 个人电脑
R1-Distill-Llama-8B Llama-3.1-8B 8B 50.4% 89.1% 生态兼容
R1-Distill-Qwen-14B Qwen2.5-14B 14B 69.7% 93.9% 中等负载服务
R1-Distill-Qwen-32B Qwen2.5-32B 32B 72.6% 94.3% 高质量推理
R1-Distill-Llama-70B Llama-3.3-70B 70B 70.0% 94.5% 最强蒸馏模型

8.3 蒸馏模型使用示例

from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

# 加载 7B 蒸馏模型
model_name = "deepseek-ai/DeepSeek-R1-Distill-Qwen-7B"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map="auto"
)

# 推理
prompt = """请一步步思考并解决这个问题：
一个水池有两个进水管，A管单独注满需要6小时，B管单独注满需要4小时。
如果先开A管2小时，再同时开A、B两管，还需要多少小时注满？"""

messages = [
    {"role": "user", "content": prompt}
]

input_text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tokenizer(input_text, return_tensors="pt").to(model.device)

outputs = model.generate(
    **inputs,
    max_new_tokens=4096,
    temperature=0.6,
    top_p=0.95,
    do_sample=True
)

response = tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
print(response)

第九章：数学推理能力实战

9.1 数学推理 Prompt 工程

针对 DeepSeek-R1 的数学推理，推荐以下 Prompt 策略：

# 数学推理系统提示词
MATH_SYSTEM_PROMPT = """你是一个专业的数学推理助手。请遵循以下原则：

1. **逐步推理**：将复杂问题分解为清晰的步骤
2. **验证每步**：对关键步骤进行验算
3. **多种方法**：尝试用不同方法求解，交叉验证
4. **清晰表达**：使用标准数学符号和格式
5. **最终答案**：在推理完成后明确给出最终答案

对于证明题，请使用严格的数学论证格式。
对于计算题，请展示完整的计算过程。"""

# 复杂数学问题示例
def solve_math_problem(problem: str) -> str:
    response = client.chat.completions.create(
        model="deepseek-reasoner",
        messages=[
            {"role": "system", "content": MATH_SYSTEM_PROMPT},
            {"role": "user", "content": problem}
        ],
        max_tokens=8192,
        temperature=0.0  # 数学推理用低温度
    )
    return response.choices[0].message

# 测试
result = solve_math_problem("""
求以下不定积分：
∫ (x² + 1) / (x⁴ + 1) dx
""")
print("推理过程：", result.reasoning_content)
print("答案：", result.content)

9.2 数学推理评估框架

import re
import json
from typing import List, Dict

class MathEvaluator:
    """数学推理评估器"""
    
    def __init__(self, client):
        self.client = client
        self.results = []
    
    def extract_answer(self, response: str) -> str:
        """从回答中提取最终数值答案"""
        # 尝试从 \boxed{} 中提取
        boxed_match = re.search(r'\\boxed\{([^}]+)\}', response)
        if boxed_match:
            return boxed_match.group(1)
        
        # 尝试提取 "答案是" 后的内容
        answer_match = re.search(r'答案[是为：:]\s*(.+?)(?:\n|$)', response)
        if answer_match:
            return answer_match.group(1).strip()
        
        return response.strip()
    
    def evaluate_problem(self, problem: str, expected_answer: str) -> Dict:
        """评估单个问题"""
        response = self.client.chat.completions.create(
            model="deepseek-reasoner",
            messages=[{"role": "user", "content": problem}],
            max_tokens=4096,
            temperature=0.0
        )
        
        model_answer = self.extract_answer(response.choices[0].message.content)
        is_correct = self.compare_answers(model_answer, expected_answer)
        
        result = {
            "problem": problem,
            "expected": expected_answer,
            "model_answer": model_answer,
            "correct": is_correct,
            "reasoning_length": len(response.choices[0].message.reasoning_content or "")
        }
        self.results.append(result)
        return result
    
    def compare_answers(self, answer1: str, answer2: str) -> bool:
        """比较两个答案是否等价"""
        # 清理答案
        a1 = re.sub(r'[^\d.\-/]', '', answer1)
        a2 = re.sub(r'[^\d.\-/]', '', answer2)
        
        try:
            return abs(float(a1) - float(a2)) < 1e-6
        except:
            return a1.strip() == a2.strip()
    
    def summary(self) -> Dict:
        """生成评估摘要"""
        total = len(self.results)
        correct = sum(1 for r in self.results if r["correct"])
        avg_reasoning = sum(r["reasoning_length"] for r in self.results) / total
        
        return {
            "total": total,
            "correct": correct,
            "accuracy": f"{correct/total*100:.1f}%",
            "avg_reasoning_chars": int(avg_reasoning)
        }

# 使用示例
evaluator = MathEvaluator(client)
test_problems = [
    ("计算 ∫₀¹ x² dx", "1/3"),
    ("求解方程 x² - 5x + 6 = 0", "2, 3"),
    ("求 lim(x→0) sin(x)/x", "1"),
]

for problem, answer in test_problems:
    result = evaluator.evaluate_problem(problem, answer)
    print(f"{'✓' if result['correct'] else '✗'} {problem}")

print("\n评估结果：", json.dumps(evaluator.summary(), ensure_ascii=False))

第十章：代码推理能力实战

10.1 代码生成与推理

DeepSeek-R1 在代码推理方面表现优异，能够理解复杂需求并生成高质量代码：

# 代码生成系统提示词
CODE_SYSTEM_PROMPT = """你是一个高级软件工程师。在编写代码时：

1. 先理解需求，梳理核心逻辑
2. 选择合适的数据结构和算法
3. 编写清晰、可维护的代码
4. 添加必要的注释和文档字符串
5. 考虑边界情况和错误处理
6. 分析时间和空间复杂度

使用 Python 编写代码，遵循 PEP 8 规范。"""

def generate_code(task: str) -> str:
    response = client.chat.completions.create(
        model="deepseek-reasoner",
        messages=[
            {"role": "system", "content": CODE_SYSTEM_PROMPT},
            {"role": "user", "content": task}
        ],
        max_tokens=8192,
        temperature=0.2
    )
    return response.choices[0].message

# 测试：生成一个 LRU Cache 实现
result = generate_code("""
实现一个线程安全的 LRU Cache，支持以下操作：
1. get(key) - 获取值，不存在返回 -1
2. put(key, value) - 插入/更新键值对
3. 容量满时淘汰最久未使用的键
4. 所有操作 O(1) 时间复杂度
5. 支持上下文管理器，用于批量操作
""")

10.2 代码审查与 Debug

def debug_code(code: str, error: str) -> str:
    """使用 R1 进行代码 Debug"""
    prompt = f"""请分析以下代码的错误并提供修复方案。

**代码：**
```python
{code}

错误信息：

请：

  1. 识别错误的根本原因

  2. 解释为什么会出现这个错误

  3. 提供修复后的完整代码

  4. 说明如何避免类似错误"""

    response = client.chat.completions.create( model="deepseek-reasoner", messages=[{"role": "user", "content": prompt}], max_tokens=4096 ) return response.choices[0].message

示例

buggy_code = """ def merge_sort(arr): if len(arr) ⇐ 1: return arr mid = len(arr) // 2 left = merge_sort(arr[:mid]) right = merge_sort(arr[mid:]) result = [] i = j = 0 while i < len(left) and j < len(right): if left[i] ⇐ right[j]: result.append(left[i]) i += 1 else: result.append(right[j]) j += 1 result.extend(left[i:]) result.extend(right[j:]) return result

测试

data = [3, 6, 8, 10, 1, 2, 1] print(merge_sort(data))

期望输出: [1, 1, 2, 3, 6, 8, 10]

实际输出: [1, 1, 2, 3, 6, 8, 10] ✓

但当输入包含 None 值时崩溃

data2 = [3, None, 8, 10, 1, 2, 1] print(merge_sort(data2)) # TypeError """

result = debug_code(buggy_code, "TypeError: '<=' not supported between instances of 'int' and 'NoneType'")


---

## 第十一章：RAG 集成方案

### 11.1 RAG 架构设计

将 DeepSeek-R1 与 RAG（检索增强生成）结合，构建知识问答系统：

```python
from typing import List
import numpy as np

class SimpleRAG:
    """简易 RAG 系统"""
    
    def __init__(self, client, embedding_model="text-embedding-v3"):
        self.client = client
        self.embedding_model = embedding_model
        self.documents = []
        self.embeddings = []
    
    def add_documents(self, docs: List[str]):
        """添加文档到知识库"""
        for doc in docs:
            # 生成嵌入向量
            response = self.client.embeddings.create(
                model=self.embedding_model,
                input=doc
            )
            embedding = response.data[0].embedding
            self.documents.append(doc)
            self.embeddings.append(embedding)
    
    def search(self, query: str, top_k: int = 3) -> List[str]:
        """语义检索最相关文档"""
        query_response = self.client.embeddings.create(
            model=self.embedding_model,
            input=query
        )
        query_embedding = np.array(query_response.data[0].embedding)
        
        # 计算余弦相似度
        similarities = []
        for doc_embedding in self.embeddings:
            doc_embedding = np.array(doc_embedding)
            similarity = np.dot(query_embedding, doc_embedding) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(doc_embedding)
            )
            similarities.append(similarity)
        
        # 返回 Top-K 相关文档
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.documents[i] for i in top_indices]
    
    def query(self, question: str) -> str:
        """RAG 问答"""
        # 1. 检索相关文档
        relevant_docs = self.search(question, top_k=3)
        context = "\n\n".join(relevant_docs)
        
        # 2. 构建提示词
        prompt = f"""基于以下参考资料回答问题。如果参考资料中没有相关信息，请说明。

**参考资料：**
{context}

**问题：**{question}

请给出准确、详细的回答，并引用相关参考资料。"""
        
        # 3. 使用 DeepSeek-R1 生成回答
        response = self.client.chat.completions.create(
            model="deepseek-reasoner",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=4096
        )
        return response.choices[0].message

# 使用示例
rag = SimpleRAG(client)
rag.add_documents([
    "DeepSeek-R1 使用 GRPO 强化学习算法进行训练，该算法不需要额外的 Value 网络。",
    "DeepSeek-V3 采用 MLA 注意力机制，将 KV Cache 压缩到低维空间，减少约 93% 的内存占用。",
    "DeepSeek 的 MoE 架构包含 256 个路由专家和 1 个共享专家，每次推理激活 8 个路由专家。",
])

answer = rag.query("DeepSeek-R1 使用什么训练算法？")
print(answer.content)

11.2 高级 RAG 优化策略

class AdvancedRAG(SimpleRAG):
    """增强版 RAG，支持多种优化策略"""
    
    def __init__(self, client, embedding_model="text-embedding-v3"):
        super().__init__(client, embedding_model)
        self.query_cache = {}
    
    def rewrite_query(self, original_query: str) -> str:
        """查询重写：优化检索效果"""
        response = self.client.chat.completions.create(
            model="deepseek-chat",  # 使用 V3 进行快速查询重写
            messages=[{
                "role": "user",
                "content": f"请将以下问题重写为更适合信息检索的形式，保持核心含义不变：\n{original_query}"
            }],
            max_tokens=256,
            temperature=0.3
        )
        return response.choices[0].message.content
    
    def rerank_documents(self, query: str, docs: List[str]) -> List[str]:
        """重排序：使用 R1 对检索结果进行精排"""
        docs_text = "\n".join([f"[{i+1}] {doc}" for i, doc in enumerate(docs)])
        
        response = self.client.chat.completions.create(
            model="deepseek-chat",
            messages=[{
                "role": "user",
                "content": f"""请根据与问题的相关性，对以下文档进行排序（从最相关到最不相关）。
只需返回文档编号，用逗号分隔。

问题：{query}

文档：
{docs_text}"""
            }],
            max_tokens=128,
            temperature=0.0
        )
        
        # 解析排序结果
        try:
            order = [int(x.strip()) - 1 for x in response.choices[0].message.content.split(",")]
            return [docs[i] for i in order if i < len(docs)]
        except:
            return docs
    
    def query_with_cot(self, question: str) -> str:
        """带推理链的 RAG 查询"""
        relevant_docs = self.search(question, top_k=5)
        reranked_docs = self.rerank_documents(question, relevant_docs)[:3]
        context = "\n\n".join(reranked_docs)
        
        prompt = f"""基于参考资料进行推理并回答问题。

**参考资料：**
{context}

**问题：**{question}

请先分析参考资料中的相关信息，然后进行推理，最后给出答案。"""
        
        response = self.client.chat.completions.create(
            model="deepseek-reasoner",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=8192
        )
        return response.choices[0].message

第十二章：Agent 应用开发

12.1 基础 Agent 框架

import json
from typing import Callable, Dict, Any
from datetime import datetime

class DeepSeekAgent:
    """基于 DeepSeek 的 Agent 框架"""
    
    def __init__(self, client, model="deepseek-reasoner"):
        self.client = client
        self.model = model
        self.tools: Dict[str, Callable] = {}
        self.tool_descriptions: list = []
        self.memory: list = []
    
    def register_tool(self, name: str, description: str, parameters: dict, func: Callable):
        """注册工具"""
        self.tools[name] = func
        self.tool_descriptions.append({
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": parameters
            }
        })
    
    def think(self, task: str) -> str:
        """Agent 思考并执行任务"""
        self.memory.append({"role": "user", "content": task})
        
        max_iterations = 10
        for _ in range(max_iterations):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": self._get_system_prompt()},
                    *self.memory
                ],
                tools=self.tool_descriptions if self.tool_descriptions else None,
                max_tokens=4096
            )
            
            message = response.choices[0].message
            self.memory.append(message)
            
            # 如果有工具调用
            if message.tool_calls:
                for tool_call in message.tool_calls:
                    func_name = tool_call.function.name
                    args = json.loads(tool_call.function.arguments)
                    
                    print(f"[Agent] 调用工具: {func_name}({args})")
                    
                    # 执行工具
                    if func_name in self.tools:
                        result = self.tools[func_name](**args)
                        self.memory.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "content": json.dumps(result, ensure_ascii=False)
                        })
                    else:
                        self.memory.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "content": f"错误：工具 {func_name} 不存在"
                        })
                continue
            
            # 无工具调用，返回最终回答
            return message.content
        
        return "达到最大迭代次数，任务未完成。"
    
    def _get_system_prompt(self) -> str:
        return f"""你是一个智能 Agent，能够使用工具完成复杂任务。

当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

请遵循以下原则：
1. 分析任务，制定执行计划
2. 合理使用可用工具
3. 根据工具返回结果调整策略
4. 给出清晰的最终回答

请用中文回答。"""

# 使用示例
agent = DeepSeekAgent(client)

# 注册工具
def search_web(query: str) -> str:
    """模拟网络搜索"""
    return {"results": [f"关于 '{query}' 的搜索结果..."]}

def calculate(expression: str) -> str:
    """安全计算数学表达式"""
    try:
        allowed_chars = set("0123456789+-*/.() ")
        if all(c in allowed_chars for c in expression):
            result = eval(expression)
            return {"result": result}
        return {"error": "不允许的字符"}
    except Exception as e:
        return {"error": str(e)}

agent.register_tool("search", "搜索网络信息", {
    "type": "object",
    "properties": {"query": {"type": "string", "description": "搜索关键词"}},
    "required": ["query"]
}, search_web)

agent.register_tool("calculate", "计算数学表达式", {
    "type": "object",
    "properties": {"expression": {"type": "string", "description": "数学表达式"}},
    "required": ["expression"]
}, calculate)

# 执行任务
result = agent.think("计算 (15 * 23 + 47) / 8 的值，并解释计算过程")
print(result)

12.2 Multi-Agent 协作

class MultiAgentSystem:
    """多 Agent 协作系统"""
    
    def __init__(self, client):
        self.client = client
        self.agents = {}
    
    def add_agent(self, name: str, role: str, capabilities: list):
        """添加 Agent"""
        self.agents[name] = {
            "role": role,
            "capabilities": capabilities,
            "agent": DeepSeekAgent(self.client)
        }
    
    def delegate_task(self, task: str) -> str:
        """将任务分配给最合适的 Agent"""
        # 选择合适的 Agent
        agent_descriptions = "\n".join([
            f"- {name}: {info['role']}，擅长 {', '.join(info['capabilities'])}"
            for name, info in self.agents.items()
        ])
        
        response = self.client.chat.completions.create(
            model="deepseek-chat",
            messages=[{
                "role": "user",
                "content": f"""根据任务选择最合适的 Agent。

任务：{task}

可用 Agent：
{agent_descriptions}

只返回 Agent 名称。"""
            }],
            max_tokens=50,
            temperature=0.0
        )
        
        chosen_agent = response.choices[0].message.content.strip()
        
        if chosen_agent in self.agents:
            print(f"[系统] 将任务分配给: {chosen_agent}")
            return self.agents[chosen_agent]["agent"].think(task)
        
        return "无法确定合适的 Agent"

# 使用示例
system = MultiAgentSystem(client)
system.add_agent("coder", "代码专家", ["编程", "Debug", "代码审查"])
system.add_agent("mathematician", "数学专家", ["数学推理", "证明", "计算"])
system.add_agent("researcher", "研究专家", ["信息检索", "知识问答", "分析"])

result = system.delegate_task("请证明：对于任意正整数 n，1+2+...+n = n(n+1)/2")

实战项目一：数学推理助手

项目概述

构建一个功能完整的数学推理助手，支持多种数学问题类型，具备推理过程可视化、步骤验证、多种解法对比等功能。

完整代码实现

"""
数学推理助手 - 基于 DeepSeek-R1
支持：代数、微积分、概率统计、线性代数、数论等
"""

import json
import re
import time
from typing import Optional, List, Dict
from dataclasses import dataclass, field
from enum import Enum
from openai import OpenAI


class MathDomain(Enum):
    ALGEBRA = "代数"
    CALCULUS = "微积分"
    PROBABILITY = "概率统计"
    LINEAR_ALGEBRA = "线性代数"
    NUMBER_THEORY = "数论"
    GEOMETRY = "几何"
    COMBINATORICS = "组合数学"


@dataclass
class MathSolution:
    problem: str
    domain: MathDomain
    steps: List[str]
    answer: str
    verification: str
    alternative_methods: List[str] = field(default_factory=list)
    confidence: float = 0.0
    time_taken: float = 0.0


class MathReasoningAssistant:
    """数学推理助手"""
    
    SYSTEM_PROMPT = """你是一个专业的数学推理助手，具备以下能力：

1. **多领域覆盖**：代数、微积分、概率统计、线性代数、数论、几何、组合数学
2. **逐步推理**：展示完整的推理过程，每一步都有清晰的逻辑
3. **多种解法**：对同一问题尝试不同的解法
4. **自我验证**：对答案进行验算和验证
5. **难度评估**：评估问题的难度级别

请严格按照以下 JSON 格式输出：
{
    "domain": "数学领域",
    "steps": ["步骤1", "步骤2", ...],
    "answer": "最终答案",
    "verification": "验证过程",
    "alternative_methods": ["解法2说明", ...],
    "confidence": 0.95,
    "difficulty": "easy/medium/hard/expert"
}"""

    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com"):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.history: List[MathSolution] = []
    
    def solve(self, problem: str, domain: Optional[MathDomain] = None) -> MathSolution:
        """求解数学问题"""
        start_time = time.time()
        
        domain_hint = f"（提示：这可能是{domain.value}问题）" if domain else ""
        
        response = self.client.chat.completions.create(
            model="deepseek-reasoner",
            messages=[
                {"role": "system", "content": self.SYSTEM_PROMPT},
                {"role": "user", "content": f"请求解以下数学问题{domain_hint}：\n\n{problem}"}
            ],
            max_tokens=8192,
            temperature=0.0
        )
        
        message = response.choices[0].message
        reasoning = message.reasoning_content or ""
        content = message.content
        
        # 解析结构化输出
        try:
            result = json.loads(content)
        except json.JSONDecodeError:
            # 尝试从文本中提取 JSON
            json_match = re.search(r'\{[\s\S]*\}', content)
            if json_match:
                result = json.loads(json_match.group())
            else:
                result = {
                    "domain": "未分类",
                    "steps": [content],
                    "answer": content,
                    "verification": "未能自动验证",
                    "alternative_methods": [],
                    "confidence": 0.5
                }
        
        solution = MathSolution(
            problem=problem,
            domain=MathDomain(result.get("domain", "代数")) if domain is None else domain,
            steps=result.get("steps", []),
            answer=result.get("answer", ""),
            verification=result.get("verification", ""),
            alternative_methods=result.get("alternative_methods", []),
            confidence=result.get("confidence", 0.5),
            time_taken=time.time() - start_time
        )
        
        self.history.append(solution)
        return solution
    
    def verify(self, problem: str, answer: str) -> Dict:
        """验证答案正确性"""
        response = self.client.chat.completions.create(
            model="deepseek-reasoner",
            messages=[{
                "role": "user",
                "content": f"""请验证以下数学问题的答案是否正确：

问题：{problem}
待验证答案：{answer}

请给出：
1. 验证过程
2. 结论（正确/错误）
3. 如果错误，给出正确答案"""
            }],
            max_tokens=4096,
            temperature=0.0
        )
        
        return {
            "verification_result": response.choices[0].message.content,
            "reasoning": response.choices[0].message.reasoning_content
        }
    
    def explain_step(self, problem: str, step_number: int) -> str:
        """详细解释某个推理步骤"""
        if not self.history:
            return "请先求解一个问题"
        
        last_solution = self.history[-1]
        if step_number > len(last_solution.steps):
            return f"只有 {len(last_solution.steps)} 个步骤"
        
        step = last_solution.steps[step_number - 1]
        response = self.client.chat.completions.create(
            model="deepseek-reasoner",
            messages=[{
                "role": "user",
                "content": f"""在求解 "{problem}" 的过程中，第 {step_number} 步是：
"{step}"

请详细解释这一步的原理和推理逻辑。"""
            }],
            max_tokens=2048
        )
        
        return response.choices[0].message.content
    
    def print_solution(self, solution: MathSolution):
        """格式化输出解答"""
        print("=" * 60)
        print(f"📝 问题：{solution.problem}")
        print(f"📂 领域：{solution.domain.value}")
        print(f"⏱️  用时：{solution.time_taken:.2f}秒")
        print(f"🎯 置信度：{solution.confidence*100:.0f}%")
        print("-" * 60)
        
        print("\n📋 推理步骤：")
        for i, step in enumerate(solution.steps, 1):
            print(f"  {i}. {step}")
        
        print(f"\n✅ 最终答案：{solution.answer}")
        print(f"\n🔍 验证：{solution.verification}")
        
        if solution.alternative_methods:
            print("\n🔄 其他解法：")
            for method in solution.alternative_methods:
                print(f"  • {method}")
        
        print("=" * 60)


# 使用示例
if __name__ == "__main__":
    assistant = MathReasoningAssistant(api_key="your-api-key")
    
    # 求解问题
    solution = assistant.solve(
        "求不定积分 ∫ x²·e^x dx",
        domain=MathDomain.CALCULUS
    )
    assistant.print_solution(solution)
    
    # 详细解释某一步
    explanation = assistant.explain_step("求不定积分 ∫ x²·e^x dx", 2)
    print("\n步骤详解：", explanation)
    
    # 验证答案
    verification = assistant.verify(
        "求不定积分 ∫ x²·e^x dx",
        "e^x(x² - 2x + 2) + C"
    )
    print("\n验证结果：", verification["verification_result"])

实战项目二：代码自动修复系统

项目概述

构建一个自动代码修复系统，能够分析代码错误、定位 Bug、生成修复方案，并自动验证修复结果。

完整代码实现

"""
代码自动修复系统 - 基于 DeepSeek-R1
功能：错误分析、Bug 定位、修复方案生成、自动验证
"""

import ast
import json
import re
import subprocess
import tempfile
import os
from typing import Optional, List, Dict, Tuple
from dataclasses import dataclass, field
from openai import OpenAI


@dataclass
class BugReport:
    """Bug 报告"""
    location: str           # 代码位置
    bug_type: str          # Bug 类型
    description: str       # Bug 描述
    severity: str          # 严重程度：critical/major/minor
    fix_suggestion: str    # 修复建议


@dataclass
class FixResult:
    """修复结果"""
    original_code: str
    fixed_code: str
    bugs_found: List[BugReport]
    fixes_applied: List[str]
    test_passed: bool
    test_output: str
    explanation: str


class CodeAutoFixer:
    """代码自动修复系统"""
    
    ANALYSIS_PROMPT = """你是一个高级代码审查专家。请分析以下 Python 代码，找出所有潜在的 Bug 和问题。

请按以下 JSON 格式输出分析结果：
{
    "bugs": [
        {
            "location": "行号或代码片段",
            "bug_type": "类型（如：逻辑错误/类型错误/边界条件/异常处理等）",
            "description": "详细描述",
            "severity": "critical/major/minor",
            "fix_suggestion": "修复建议"
        }
    ],
    "code_quality_score": 0-100,
    "overall_assessment": "总体评价"
}"""

    FIX_PROMPT = """你是一个高级软件工程师。请根据 Bug 分析报告修复代码。

要求：
1. 修复所有发现的 Bug
2. 保持代码原有功能不变
3. 遵循 Python 最佳实践
4. 添加必要的错误处理
5. 保持代码可读性

请直接输出修复后的完整代码，不要包含其他内容。"""

    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com"):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
    
    def analyze(self, code: str) -> Dict:
        """分析代码中的 Bug"""
        response = self.client.chat.completions.create(
            model="deepseek-reasoner",
            messages=[
                {"role": "system", "content": self.ANALYSIS_PROMPT},
                {"role": "user", "content": f"请分析以下代码：\n\n```python\n{code}\n```"}
            ],
            max_tokens=4096,
            temperature=0.1
        )
        
        content = response.choices[0].message.content
        try:
            json_match = re.search(r'\{[\s\S]*\}', content)
            if json_match:
                return json.loads(json_match.group())
        except json.JSONDecodeError:
            pass
        
        return {"bugs": [], "code_quality_score": 50, "overall_assessment": content}
    
    def fix(self, code: str, error_message: Optional[str] = None) -> str:
        """修复代码"""
        context = f"\n\n错误信息：\n{error_message}" if error_message else ""
        
        response = self.client.chat.completions.create(
            model="deepseek-reasoner",
            messages=[
                {"role": "system", "content": self.FIX_PROMPT},
                {"role": "user", "content": f"请修复以下代码{context}：\n\n```python\n{code}\n```"}
            ],
            max_tokens=8192,
            temperature=0.1
        )
        
        content = response.choices[0].message.content
        
        # 提取代码块
        code_match = re.search(r'```python\n([\s\S]*?)\n```', content)
        if code_match:
            return code_match.group(1)
        
        return content
    
    def verify_fix(self, original_code: str, fixed_code: str, test_cases: List[str]) -> Tuple[bool, str]:
        """验证修复结果"""
        # 创建临时测试文件
        test_code = f"""
import sys

# 修复后的代码
{fixed_code}

# 测试用例
test_results = []
{chr(10).join(test_cases)}

print("ALL TESTS PASSED" if all(test_results) else "SOME TESTS FAILED")
"""
        
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(test_code)
            temp_file = f.name
        
        try:
            result = subprocess.run(
                ['python3', temp_file],
                capture_output=True,
                text=True,
                timeout=10
            )
            output = result.stdout + result.stderr
            passed = "ALL TESTS PASSED" in output
            return passed, output
        except subprocess.TimeoutExpired:
            return False, "测试超时"
        except Exception as e:
            return False, str(e)
        finally:
            os.unlink(temp_file)
    
    def auto_fix(self, code: str, error_message: Optional[str] = None, 
                 test_cases: Optional[List[str]] = None) -> FixResult:
        """完整的自动修复流程"""
        print("🔍 Step 1: 分析代码...")
        analysis = self.analyze(code)
        
        bugs = [
            BugReport(
                location=bug.get("location", "未知"),
                bug_type=bug.get("bug_type", "未知"),
                description=bug.get("description", ""),
                severity=bug.get("severity", "minor"),
                fix_suggestion=bug.get("fix_suggestion", "")
            )
            for bug in analysis.get("bugs", [])
        ]
        
        print(f"  发现 {len(bugs)} 个潜在问题")
        print(f"  代码质量评分：{analysis.get('code_quality_score', 'N/A')}/100")
        
        print("\n🔧 Step 2: 生成修复方案...")
        fixed_code = self.fix(code, error_message)
        
        print("\n✅ Step 3: 验证修复结果...")
        test_passed = True
        test_output = "无测试用例"
        
        if test_cases:
            test_passed, test_output = self.verify_fix(code, fixed_code, test_cases)
            print(f"  测试结果：{'通过 ✓' if test_passed else '失败 ✗'}")
        
        print("\n📝 Step 4: 生成修复说明...")
        explanation_response = self.client.chat.completions.create(
            model="deepseek-chat",
            messages=[{
                "role": "user",
                "content": f"""请用中文简要说明以下代码修复的主要改动：

原始代码：
```python
{code}

修复后代码：

{fixed_code}

发现的问题：{json.dumps([b.dict for b in bugs], ensure_ascii=False)}""" }], max_tokens=1024, temperature=0.3 )

    return FixResult(
        original_code=code,
        fixed_code=fixed_code,
        bugs_found=bugs,
        fixes_applied=[b.fix_suggestion for b in bugs],
        test_passed=test_passed,
        test_output=test_output,
        explanation=explanation_response.choices[0].message.content
    )

使用示例

if name == "main": fixer = CodeAutoFixer(api_key="your-api-key")

# 有 Bug 的代码
buggy_code = """

def calculate_average(numbers): total = 0 for num in numbers: total += num return total / len(numbers) # 可能除以零

def read_file(filepath): f = open(filepath, 'r') # 未关闭文件 content = f.read() return content

def find_max(arr): max_val = arr[0] # 空数组会报错 for i in range(1, len(arr)): if arr[i] > max_val: max_val = arr[i] return max_val

def parse_json(json_str): import json data = json.loads(json_str) # 未处理解析异常 return data['key'] # 可能键不存在 """

# 定义测试用例
test_cases = [
    'test_results.append(calculate_average([1, 2, 3, 4, 5]) == 3.0)',
    'test_results.append(calculate_average([]) is None or True)',  # 空列表应处理
    'test_results.append(find_max([3, 1, 4, 1, 5, 9]) == 9)',
]

# 执行自动修复
result = fixer.auto_fix(buggy_code, test_cases=test_cases)

print("\n" + "=" * 60)
print("📊 修复报告")
print("=" * 60)
print(f"\n🐛 发现的 Bug ({len(result.bugs_found)} 个)：")
for bug in result.bugs_found:
    print(f"  [{bug.severity}] {bug.bug_type}: {bug.description}")

print(f"\n✨ 修复说明：\n{result.explanation}")
print(f"\n📄 修复后的代码：\n{result.fixed_code}")


---

## 常见问题与故障排除

### Q1: API 返回 "Rate limit exceeded" 怎么办？

```python
import time
from openai import OpenAI, RateLimitError

client = OpenAI(api_key="your-key", base_url="https://api.deepseek.com")

def call_with_retry(func, max_retries=5, base_delay=1):
    """带重试的 API 调用"""
    for attempt in range(max_retries):
        try:
            return func()
        except RateLimitError as e:
            if attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt)  # 指数退避
                print(f"触发限流，{delay}秒后重试...")
                time.sleep(delay)
            else:
                raise e

# 使用
response = call_with_retry(
    lambda: client.chat.completions.create(
        model="deepseek-reasoner",
        messages=[{"role": "user", "content": "你好"}]
    )
)

Q2: R1 模型输出的思考过程太长怎么办？

# 方法 1：在 prompt 中限制推理深度
response = client.chat.completions.create(
    model="deepseek-reasoner",
    messages=[{
        "role": "user",
        "content": "请简洁地回答，不需要详细推理过程：1+1等于几？"
    }],
    max_tokens=256  # 限制输出长度
)

# 方法 2：使用 deepseek-chat (V3) 替代简单问题
response = client.chat.completions.create(
    model="deepseek-chat",  # 非推理模型，响应更快
    messages=[{"role": "user", "content": "1+1等于几？"}]
)

Q3: 本地部署 OOM（内存不足）怎么解决？

# 方案 1：使用更小的量化版本
ollama pull deepseek-r1:7b-q4_0  # 4-bit 量化

# 方案 2：限制上下文长度
ollama run deepseek-r1:7b --num-ctx 2048

# 方案 3：使用 CPU 推理（慢但可用）
# 设置 OLLAMA_NUM_GPU=0 强制使用 CPU
OLLAMA_NUM_GPU=0 ollama run deepseek-r1:7b

Q4: 如何提升 R1 的推理质量？

# 技巧 1：明确要求逐步推理
prompt = """请一步一步思考以下问题，每一步都解释你的推理逻辑：

问题：[你的问题]

要求：
1. 先分析问题的关键信息
2. 制定解题策略
3. 逐步执行并验证
4. 给出最终答案"""

# 技巧 2：使用 few-shot 示例
prompt = """请按照以下示例的推理方式解决问题：

示例：
问题：一个数的3倍加5等于20，求这个数。
推理：
- 设这个数为 x
- 根据题意：3x + 5 = 20
- 移项：3x = 20 - 5 = 15
- 解得：x = 15 / 3 = 5
- 验证：3 × 5 + 5 = 15 + 5 = 20 ✓
答案：5

现在请解决：
问题：[你的问题]"""

# 技巧 3：设置合适的温度参数
# 推理任务：temperature = 0.0 ~ 0.3
# 创作任务：temperature = 0.7 ~ 1.0

Q5: DeepSeek API 和 OpenAI SDK 兼容性问题？

# DeepSeek API 完全兼容 OpenAI SDK，只需修改 base_url
from openai import OpenAI

# DeepSeek 配置
client = OpenAI(
    api_key="your-deepseek-key",
    base_url="https://api.deepseek.com"  # 唯一需要修改的地方
)

# 注意事项：
# 1. R1 模型的响应包含 reasoning_content 字段（思考过程）
# 2. 部分 OpenAI 特有参数可能不支持（如 response_format）
# 3. Function Calling 支持但可能与 OpenAI 行为略有差异

Q6: 如何处理 R1 的 "token 浪费" 问题？

R1 模型会生成大量推理过程 token，但很多场景不需要输出这些 token：

# 方案 1：使用 R1 做"后台推理"，只提取最终答案
def get_concise_answer(question: str) -> str:
    response = client.chat.completions.create(
        model="deepseek-reasoner",
        messages=[{
            "role": "user",
            "content": f"{question}\n\n请只输出最终答案，不要输出推理过程。"
        }],
        max_tokens=1024
    )
    return response.choices[0].message.content

# 方案 2：简单问题用 V3，复杂问题用 R1
def smart_routing(question: str, complexity: str = "auto") -> str:
    if complexity == "auto":
        # 使用 V3 判断问题复杂度
        judge = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{
                "role": "user",
                "content": f"判断以下问题的复杂度（简单/中等/复杂），只回答一个词：\n{question}"
            }],
            max_tokens=10,
            temperature=0.0
        )
        complexity = judge.choices[0].message.content.strip()
    
    model = "deepseek-reasoner" if complexity == "复杂" else "deepseek-chat"
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": question}]
    )
    return response.choices[0].message.content

Q7: 蒸馏模型和完整 R1 的差距有多大？

任务类型 R1 完整版 R1-Distill-70B R1-Distill-32B R1-Distill-7B
数学推理 ★★★★★ ★★★★☆ ★★★★☆ ★★★☆☆
代码生成 ★★★★★ ★★★★☆ ★★★☆☆ ★★★☆☆
逻辑推理 ★★★★★ ★★★★☆ ★★★★☆ ★★★☆☆
通用问答 ★★★★☆ ★★★★☆ ★★★★☆ ★★★★☆
推理速度 ★★☆☆☆ ★★★★☆ ★★★★★ ★★★★★

建议：

  • 预算充足、追求最佳效果 → R1 完整版 API
  • 需要本地部署、有 GPU → R1-Distill-32B 或 70B
  • 资源有限、需要快速响应 → R1-Distill-7B
  • 移动端/边缘设备 → R1-Distill-1.5B

总结与下一步

恭喜你完成了 DeepSeek 推理模型完全教程的学习！以下是关键要点回顾：

  1. 架构理解：掌握了 MLA 注意力机制、MoE 混合专家架构、GRPO 强化学习算法
  2. API 实战：学会了使用兼容 OpenAI 的 SDK 调用 DeepSeek 模型
  3. 本地部署：了解了 Ollama、vLLM、llama.cpp 等多种部署方案
  4. 应用开发：掌握了 RAG 集成、Agent 开发等高级应用
  5. 实战项目：完成了数学推理助手和代码自动修复系统

推荐学习路径

入门 → API 调用基础 → 简单推理任务
     ↓
进阶 → Prompt 工程优化 → RAG 集成 → Agent 开发
     ↓
高级 → 本地部署 → 模型微调 → 多 Agent 系统
     ↓
专家 → 训练自己的推理模型 → 贡献开源社区

有用的资源

  • DeepSeek 官方文档：platform.deepseek.com
  • DeepSeek GitHub：github.com/deepseek-ai
  • HuggingFace 模型库：huggingface.co/deepseek-ai
  • 社区论坛：github.com/deepseek-ai/discussions

本教程持续更新中。如有问题或建议，欢迎反馈。

内容声明

本文内容为AI技术学习教程，仅供学习参考。如涉及技术问题，欢迎通过 xurj005@163.com 与我们交流。

相关教程

Python入门完全教程

29,193 次阅读

Java基础编程教程

116 次阅读

Go语言实战教程

111 次阅读

Python编程入门完全教程

93 次阅读
目录
快速导航
返回首页 全部教程 AI资讯