【大模型】- DeepSeek-V3 架构详解

算法

DeepSeek-V3 架构详解

分析 DeepSeek-V3 的创新设计

类型: 学习 | 语言: Python | 🏷 前置:《DualPipe 并行》(本系列第 18 篇)

学习目标

  • 了解 DeepSeek-V3 的架构创新
  • 实现 DeepSeek-V3 的关键组件
  • 分析 DeepSeek-V3 的设计选择
  • 理解 DeepSeek-V3 的优势
  • 应用 DeepSeek-V3 的技术

DeepSeek-V3 概述

DeepSeek-V3 是 DeepSeek 开发的最新大型语言模型,具有多项创新设计。

核心创新

  1. 多头潜在注意力(MLA):高效处理长上下文
  2. DeepSeekMoE:改进的混合专家架构
  3. 辅助损失无关的负载均衡:优化专家负载
  4. 多 token 预测:加速推理
  5. FP8 混合精度训练:提高训练效率

多头潜在注意力(MLA)

MLA 是 DeepSeek-V3 的核心创新之一,通过低秩压缩减少 KV 缓存。

MLA 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import torch
import torch.nn as nn
import torch.nn.functional as F

class MultiHeadLatentAttention(nn.Module):
def __init__(self, dim, n_heads, n_kv_heads, low_rank_dim):
super().__init__()
self.dim = dim
self.n_heads = n_heads
self.n_kv_heads = n_kv_heads
self.head_dim = dim // n_heads
self.low_rank_dim = low_rank_dim

# Q 投影
self.wq = nn.Linear(dim, dim, bias=False)

# KV 低秩投影
self.wkv = nn.Linear(dim, low_rank_dim * 2, bias=False)

# 从低秩恢复 K, V
self.wk = nn.Linear(low_rank_dim, n_kv_heads * self.head_dim, bias=False)
self.wv = nn.Linear(low_rank_dim, n_kv_heads * self.head_dim, bias=False)

# 输出投影
self.wo = nn.Linear(dim, dim, bias=False)

# 缩放因子
self.scale = self.head_dim ** -0.5

def forward(self, x, mask=None):
batch_size, seq_len, _ = x.shape

# Q 计算
q = self.wq(x).view(batch_size, seq_len, self.n_heads, self.head_dim)
q = q.transpose(1, 2)

# KV 低秩压缩
kv_low = self.wkv(x) # (batch, seq_len, low_rank_dim * 2)

# 分离 K, V
k_low = kv_low[:, :, :self.low_rank_dim]
v_low = kv_low[:, :, self.low_rank_dim:]

# 恢复 K, V
k = self.wk(k_low).view(batch_size, seq_len, self.n_kv_heads, self.head_dim)
v = self.wv(v_low).view(batch_size, seq_len, self.n_kv_heads, self.head_dim)

k = k.transpose(1, 2)
v = v.transpose(1, 2)

# 重复 K, V 以匹配 Q 的头数
if self.n_kv_heads != self.n_heads:
n_rep = self.n_heads // self.n_kv_heads
k = k.repeat_interleave(n_rep, dim=1)
v = v.repeat_interleave(n_rep, dim=1)

# 注意力计算
attn = torch.matmul(q, k.transpose(-2, -1)) * self.scale

if mask is not None:
attn = attn.masked_fill(mask == 0, float("-inf"))

attn = F.softmax(attn, dim=-1)

output = torch.matmul(attn, v)
output = output.transpose(1, 2).contiguous()
output = output.view(batch_size, seq_len, -1)

return self.wo(output)

KV 缓存优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class MLAKVCache:
def __init__(self, max_batch_size, max_seq_len, low_rank_dim, n_heads, head_dim):
self.max_batch_size = max_batch_size
self.max_seq_len = max_seq_len
self.low_rank_dim = low_rank_dim
self.n_heads = n_heads
self.head_dim = head_dim

# 低秩 KV 缓存
self.kv_cache = torch.zeros(
max_batch_size, max_seq_len, low_rank_dim * 2
)

self.cur_len = 0

def update(self, input_pos, kv_low):
"""更新缓存"""
batch_size = kv_low.shape[0]

self.kv_cache[:batch_size, input_pos:input_pos + 1, :] = kv_low
self.cur_len = input_pos + 1

def get(self, seq_len):
"""获取缓存"""
kv_low = self.kv_cache[:, :seq_len, :]

# 恢复 K, V
k = self.wk(kv_low[:, :, :self.low_rank_dim])
v = self.wv(kv_low[:, :, self.low_rank_dim:])

return k, v

DeepSeekMoE

DeepSeekMoE 是改进的混合专家架构。

架构设计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class DeepSeekMoE(nn.Module):
def __init__(self, dim, n_experts, n_shared_experts, top_k):
super().__init__()
self.dim = dim
self.n_experts = n_experts
self.n_shared_experts = n_shared_experts
self.top_k = top_k

# 共享专家
self.shared_experts = nn.ModuleList([
FeedForward(dim) for _ in range(n_shared_experts)
])

# 路由专家
self.routed_experts = nn.ModuleList([
FeedForward(dim) for _ in range(n_experts)
])

# 路由器
self.router = nn.Linear(dim, n_experts, bias=False)

def forward(self, x):
batch_size, seq_len, _ = x.shape

# 计算路由权重
router_logits = self.router(x) # (batch, seq_len, n_experts)

# 选择 top-k 专家
router_weights = F.softmax(router_logits, dim=-1)
topk_weights, topk_indices = torch.topk(router_weights, self.top_k, dim=-1)

# 归一化
topk_weights = topk_weights / topk_weights.sum(dim=-1, keepdim=True)

# 计算专家输出
expert_outputs = []
for i, expert in enumerate(self.routed_experts):
# 获取该专家的输入
expert_mask = (topk_indices == i).any(dim=-1)
if expert_mask.any():
expert_input = x[expert_mask]
expert_output = expert(expert_input)
expert_outputs.append((expert_mask, expert_output))

# 聚合输出
output = torch.zeros_like(x)
for expert_mask, expert_output in expert_outputs:
output[expert_mask] += expert_output

# 添加共享专家输出
for shared_expert in self.shared_experts:
output += shared_expert(x)

return output

辅助损失无关的负载均衡

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class AuxiliaryLossFreeLoadBalancing(nn.Module):
def __init__(self, n_experts, capacity_factor=1.25):
super().__init__()
self.n_experts = n_experts
self.capacity_factor = capacity_factor

# 专家容量
self.expert_capacity = None

def compute_capacity(self, n_tokens):
"""计算专家容量"""
self.expert_capacity = int(
n_tokens * self.capacity_factor / self.n_experts
)
return self.expert_capacity

def forward(self, router_weights, topk_indices):
"""负载均衡损失"""
n_tokens = router_weights.shape[0] * router_weights.shape[1]
capacity = self.compute_capacity(n_tokens)

# 计算每个专家的负载
expert负载 = torch.zeros(self.n_experts)
for i in range(self.n_experts):
expert负载[i] = (topk_indices == i).sum()

# 计算负载不均衡度
ideal负载 = n_tokens / self.n_experts
imbalance = ((expert负载 - ideal负载) ** 2).mean()

return imbalance

FP8 混合精度训练

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class FP8Training:
def __init__(self, model):
self.model = model

# FP8 配置
self.fp8_config = {
"e4m3": True, # 使用 E4M3 格式
"scale": 1.0,
}

def convert_to_fp8(self):
"""转换为 FP8"""
for name, param in self.model.named_parameters():
if param.dtype == torch.float32:
# 转换为 FP8
param.data = self.to_fp8(param.data)

def to_fp8(self, tensor):
"""转换为 FP8 格式"""
# 简化实现:实际应使用 CUDA 核心
return tensor.to(torch.float8_e4m3fn)

def forward(self, x):
"""FP8 前向传播"""
with torch.cuda.amp.autocast(dtype=torch.float8_e4m3fn):
return self.model(x)

多 Token 预测

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class DeepSeekMultiTokenPrediction(nn.Module):
def __init__(self, dim, vocab_size, n预测头=2):
super().__init__()
self.n预测头 = n预测头

# 基础 Transformer
self.transformer = Transformer(dim)

# 预测头
self.prediction_heads = nn.ModuleList([
nn.Linear(dim, vocab_size)
for _ in range(n预测头)
])

def forward(self, x):
hidden = self.transformer(x)

predictions = []
for head in self.prediction_heads:
pred = head(hidden)
predictions.append(pred)

return torch.stack(predictions, dim=1)

完整模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class DeepSeekV3(nn.Module):
def __init__(self, config):
super().__init__()
self.config = config

# 嵌入层
self.embed = nn.Embedding(config["vocab_size"], config["dim"])

# Transformer 块
self.layers = nn.ModuleList([
TransformerBlock(config) for _ in range(config["n_layers"])
])

# 输出层
self.norm = nn.LayerNorm(config["dim"])
self.head = nn.Linear(config["dim"], config["vocab_size"], bias=False)

# 多 token 预测头
self.multi_token_heads = nn.ModuleList([
nn.Linear(config["dim"], config["vocab_size"])
for _ in range(config["n预测头"])
])

def forward(self, idx, targets=None):
x = self.embed(idx)

for layer in self.layers:
x = layer(x)

x = self.norm(x)

# 标准预测
logits = self.head(x)

# 多 token 预测
multi_logits = []
for head in self.multi_token_heads:
multi_logits.append(head(x))

multi_logits = torch.stack(multi_logits, dim=1)

# 计算损失
loss = None
if targets is not None:
loss = F.cross_entropy(
logits.view(-1, logits.size(-1)),
targets.view(-1)
)

# 多 token 损失
for i in range(multi_logits.shape[1]):
shift_logits = multi_logits[:, i, :-1-1]
shift_targets = targets[:, 1+i:]
loss += F.cross_entropy(
shift_logits.reshape(-1, shift_logits.size(-1)),
shift_targets.reshape(-1)
)

return logits, loss

与其他模型比较

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def compare_architectures():
"""比较不同架构"""
architectures = {
"LLaMA-3": {
"attention": "MHA",
"mixture_of_experts": False,
"multi_token_prediction": False,
"fp8_training": False,
},
"Mistral-7B": {
"attention": "GQA",
"mixture_of_experts": False,
"multi_token_prediction": False,
"fp8_training": False,
},
"DeepSeek-V3": {
"attention": "MLA",
"mixture_of_experts": True,
"multi_token_prediction": True,
"fp8_training": True,
},
}

for name, arch in architectures.items():
print(f"\n{name}:")
for key, value in arch.items():
print(f" {key}: {value}")

总结

DeepSeek-V3 通过 MLA、DeepSeekMoE、辅助损失无关的负载均衡、多 token 预测和 FP8 训练等创新,实现了高效的LLM。这些技术共同提高了模型性能和训练效率。

下一步

下一课将介绍 Jamba 混合 SSM-Transformer 架构。

📚 本文改编自 AI Engineering from Scratch(MIT License · 作者 Rohit Ghumare),中文内容来自官方中文镜像。原课程共 503 课 · 20 阶段 · 免费开源,教程网站见 aiengineeringfromscratch.com

  • 标题: 【大模型】- DeepSeek-V3 架构详解
  • 作者:
  • 创建于 : 2026-08-19 09:20:00
  • 更新于 : 2026-08-21 16:20:11
  • 链接: https://sxl-space.tk/2026/08/19/010_LLM/010_LLM-20-DeepSeekV3/
  • 版权声明: 版权所有 © 宋,禁止转载。