Liquid AI LFM2.5-2.6B——26亿参数端侧大模型越级碾压的架构革命与部署实践

一、引言：当"参数军备竞赛"迎来终结

2026年8月4日，由MIT前计算机科学家创立的Liquid AI正式发布了LFM2.5-2.6B。这个仅有26亿参数的小模型，在指令遵循（IFBench 59.17）和工具调用（BFCLv4 56.88）基准上全面超越参数翻倍的Gemma 4-5.1B与Gemma 4-8B，Agent任务与97亿参数的Qwen3.5-9B打平，AIME25数学得分51.87逼近后者的56.07。

这不是一个简单的性能提升事件——它标志着AI行业从"参数量级竞赛"到"部署效率较量"的根本性范式转移。放在一个更大的语境中：当DeepSeek-V4-Flash、GLM-5.2、Kimi K2.6等千亿级模型在云端厮杀时，Liquid AI选择了一条截然不同的路——让Agent在手机上跑，在树莓派上跑，在2.5GB内存以内跑。

本文将深入剖析LFM2.5-2.6B的架构设计哲学、四阶段后训练流水线、端侧推理优化技术，并通过完整的Go/Python代码实践，展示如何亲手搭建一个端侧推理引擎。

二、架构解构：22个卷积块+8个注意力层的混合革命

2.1 整体架构一览

LFM2.5-2.6B共有30层，总参数量2.69B。其核心创新在于通过神经架构搜索（NAS）自动发现了最优的混合架构——22个双门控短卷积块（ConvBlock）与8个分组查询注意力层（GQA）的交替组合。

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
LFM2.5-2.6B 架构示意 (ASCII)
┌──────────────────────────────────────────────────┐
│ Input Embedding │
│ Vocab=128K, Dim=2048 │
├──────────────────────────────────────────────────┤
│ Layer 1: ConvBlock (short-conv, kernel=3) │
│ Layer 2: ConvBlock │
│ Layer 3: GQA (32Q-heads, 8KV-heads, RoPE=1e7) │
│ Layer 4: ConvBlock │
│ Layer 5: ConvBlock │
│ Layer 6: GQA │
│ … (每2-3个ConvBlock插入1个GQA) │
│ Layer 28: ConvBlock │
│ Layer 29: ConvBlock │
│ Layer 30: GQA │
├──────────────────────────────────────────────────┤
│ Output Embedding (Tied) │
│ SwiGLU FFN: 2048→10752→2048 │
├──────────────────────────────────────────────────┤
│ 128K Context Window | 16 Languages │
└──────────────────────────────────────────────────┘

2.2 ConvBlock：双门控短卷积的数学原理

ConvBlock的核心是双门控短卷积（Double-Gated Short Convolution）。与标准Transformer中的注意力机制不同，卷积操作的时间复杂度是O(n)而非O(n²)，这使得它在长序列场景下具有天然优势。

每个ConvBlock包含：

一个因果短卷积（kernel size=3），捕捉局部依赖
双门控机制，通过两个独立的门控信号控制信息流
RMSNorm归一化 + SwiGLU激活的FFN层

2.3 GQA：高效的分组查询注意力

LFM2.5-2.6B采用32个查询头（Q-heads）和8个键值头（KV-heads），GQA比例4:1。这意味着KV缓存的大小仅为标准MHA（Multi-Head Attention）的1/4，在128K上下文窗口下，KV缓存从_2GB降低到500MB。

2.4 NAS搜索：架构自动发现

Liquid AI没有手动设计层布局，而是通过神经架构搜索来确定最优的卷积/注意力比例。搜索空间包括：

每层选择ConvBlock或GQA
卷积核大小（3/5/7）
注意力头数配置
FFN中间维度缩放比

python
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
“”"
NAS搜索模拟：Liquid AI可能使用的架构搜索策略
使用进化算法在约束空间中找到最优层配置
“”"
import numpy as np
from dataclasses import dataclass, field
from typing import List, Optional
import random
import math
import json

@dataclass
class NASConfig:
“”“神经架构搜索配置空间”“”
total_layers: int = 30
hidden_dim: int = 2048
vocab_size: int = 128000

# 搜索空间 conv_kernel_sizes: List[int] = field(default_factory=lambda: [3, 5, 7]) gqa_head_options: List[int] = field(default_factory=lambda: [8, 16, 32]) ffn_scale_options: List[float] = field(default_factory=lambda: [2.5, 3.0, 3.5, 4.0]) min_gqa_layers: int = 4 max_gqa_layers: int = 12

@dataclass
class Architecture:
“”“单个架构编码”“”
layer_types: List[str] # ‘conv’ 或 ‘gqa’
conv_kernel_size: int
gqa_kv_heads: int
ffn_scale: float

def compute_kv_cache_mb(self, context_len: int = 131072) -> float: """估算KV缓存大小（MB）""" gqa_count = sum(1 for t in self.layer_types if t == 'gqa') if gqa_count == 0: return 0.0 # 每个GQA层: 2 (K+V) * kv_heads * (hidden_dim//q_heads) * context_len * 2bytes head_dim = self.hidden_dim // 32 # 固定32个Q头 bytes_per_layer = 2 * self.gqa_kv_heads * head_dim * context_len * 2 return (bytes_per_layer * gqa_count) / (1024 * 1024) def estimate_compute_cost(self) -> float: """估算计算成本（相对值）""" conv_cost = sum(1 for t in self.layer_types if t == 'conv') * self.conv_kernel_size * 0.3 gqa_cost = sum(1 for t in self.layer_types if t == 'gqa') * 1.0 ffn_cost = self.total_layers * self.ffn_scale * 0.4 return conv_cost + gqa_cost + ffn_cost

def random_architecture(config: NASConfig) -> Architecture:
“”“随机生成一个架构”“”
num_gqa = random.randint(config.min_gqa_layers, config.max_gqa_layers)
num_conv = config.total_layers - num_gqa

# 生成层类型数组，GQA尽量均匀分布 positions = sorted(random.sample(range(config.total_layers), num_gqa)) layer_types = ['conv'] * config.total_layers for pos in positions: layer_types[pos] = 'gqa' arch = Architecture( layer_types=layer_types, conv_kernel_size=random.choice(config.conv_kernel_sizes), gqa_kv_heads=random.choice(config.gqa_head_options), ffn_scale=random.choice(config.ffn_scale_options), hidden_dim=config.hidden_dim, total_layers=config.total_layers ) return arch

def mutate_architecture(arch: Architecture, config: NASConfig) -> Architecture:
“”“变异架构”“”
new_types = arch.layer_types.copy()

# 随机交换一个conv和一个gqa if random.random() < 0.3: conv_indices = [i for i, t in enumerate(new_types) if t == 'conv'] gqa_indices = [i for i, t in enumerate(new_types) if t == 'gqa'] if conv_indices and gqa_indices: ci = random.choice(conv_indices) gi = random.choice(gqa_indices) new_types[ci], new_types[gi] = new_types[gi], new_types[ci] # 随机修改超参数 new_kernel = arch.conv_kernel_size if random.random() < 0.2: new_kernel = random.choice([k for k in config.conv_kernel_sizes if k != arch.conv_kernel_size] or config.conv_kernel_sizes) new_kv = arch.gqa_kv_heads if random.random() < 0.2: new_kv = random.choice([h for h in config.gqa_head_options if h != arch.gqa_kv_heads] or config.gqa_head_options) new_ffn = arch.ffn_scale if random.random() < 0.2: new_ffn = random.choice([s for s in config.ffn_scale_options if abs(s - arch.ffn_scale) > 0.1] or config.ffn_scale_options) return Architecture( layer_types=new_types, conv_kernel_size=new_kernel, gqa_kv_heads=new_kv, ffn_scale=new_ffn, hidden_dim=arch.hidden_dim, total_layers=arch.total_layers )

def crossover(a1: Architecture, a2: Architecture) -> Architecture:
“”“交叉两个架构”“”
child_types = []
for i in range(len(a1.layer_types)):
child_types.append(random.choice([a1.layer_types[i], a2.layer_types[i]]))

return Architecture( layer_types=child_types, conv_kernel_size=random.choice([a1.conv_kernel_size, a2.conv_kernel_size]), gqa_kv_heads=random.choice([a1.gqa_kv_heads, a2.gqa_kv_heads]), ffn_scale=random.choice([a1.ffn_scale, a2.ffn_scale]), hidden_dim=a1.hidden_dim, total_layers=a1.total_layers )

def fitness(arch: Architecture, target_kv_mb: float = 500.0) -> float:
“”“适应度函数：平衡性能与资源约束”“”
kv_cache = arch.compute_kv_cache_mb()
compute_cost = arch.estimate_compute_cost()

# KV缓存不能超过目标 if kv_cache > target_kv_mb * 1.5: return -float('inf') # GQA层数越多，长程能力越强（但成本越高） gqa_count = sum(1 for t in arch.layer_types if t == 'gqa') gqa_benefit = gqa_count * 1.5 # 卷积层提供效率 conv_count = sum(1 for t in arch.layer_types if t == 'gqa') conv_benefit = conv_count * 0.8 # 总得分 = 能力 - 成本 score = (gqa_benefit + conv_benefit) - compute_cost * 0.3 # 偏好GQA均匀分布（避免所有注意力集中在开头或结尾） gqa_positions = [i for i, t in enumerate(arch.layer_types) if t == 'gqa'] if gqa_positions: spread = max(gqa_positions) - min(gqa_positions) spread_score = spread / len(arch.layer_types) * 2.0 score += spread_score return score

def evolutionary_search(config: NASConfig,
population_size: int = 50,
generations: int = 100,
elite_ratio: float = 0.2) -> List[Architecture]:
“”"
进化算法搜索最优架构
模拟Liquid AI的NAS过程
“”"
# 初始化种群
population = [random_architecture(config) for _ in range(population_size)]

best_archs = [] for gen in range(generations): # 计算适应度 scored = [(arch, fitness(arch)) for arch in population] scored.sort(key=lambda x: x[1], reverse=True) # 记录最优 if scored[0][1] > -float('inf'): best_archs.append(scored[0][0]) # 精英选择 elite_count = int(population_size * elite_ratio) elites = [arch for arch, _ in scored[:elite_count]] # 填充下一代 next_gen = elites.copy() while len(next_gen) < population_size: parent1 = random.choice(elites) if random.random() < 0.7: parent2 = random.choice(elites) child = crossover(parent1, parent2) else: child = parent1 # 变异概率 if random.random() < 0.4: child = mutate_architecture(child, config) next_gen.append(child) population = next_gen if (gen + 1) % 20 == 0: print(f"Generation {gen+1}: Best fitness = {scored[0][1]:.2f}, " f"GQA layers = {sum(1 for t in scored[0][0].layer_types if t == 'gqa')}, " f"KV cache = {scored[0][0].compute_kv_cache_mb():.1f}MB") return best_archs

ifname== “main”:
config = NASConfig()
print(“开始NAS架构搜索模拟…”)
print(f"搜索空间: {config.total_layers}层, "
f"卷积核={config.conv_kernel_sizes}, "
f"GQA头数={config.gqa_head_options}")
print()

best_archs = evolutionary_search(config, population_size=60, generations=80) if best_archs: final = best_archs[-1] gqa_count = sum(1 for t in final.layer_types if t == 'gqa') conv_count = sum(1 for t in final.layer_types if t == 'conv') print(f"\n最佳架构:") print(f" 总层数: {final.total_layers}") print(f" ConvBlock: {conv_count}层") print(f" GQA: {gqa_count}层") print(f" 卷积核大小: {final.conv_kernel_size}") print(f" KV头数: {final.gqa_kv_heads}") print(f" FFN缩放: {final.ffn_scale}") print(f" KV缓存: {final.compute_kv_cache_mb():.1f}MB") print(f" 层分布: {''.join('C' if t == 'conv' else 'A' for t in final.layer_types)}")

三、四阶段后训练：从基座模型到Agent的蜕变

3.1 训练流水线全景

LFM2.5-2.6B的预训练数据量为约34万亿token，词汇表从LFM2.5的65K扩展至128K，以更好地支持非拉丁文字。中训练阶段将上下文窗口从32K扩展至128K。

真正让这个模型与众不同的是其四阶段后训练流水线：

plaintext
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
四阶段后训练流水线
┌────────────────────────────────────────────────────────────┐
│ 阶段1: SFT (监督微调) │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 两轮SFT，聚焦Agent数据：工具调用、网页搜索、Harness轨迹 │ │
│ └──────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段2: Teacher Specialization (教师特化) │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 数学教师 │ 代码教师 │ 工具使用教师 │ 推理教师 │ │ │
│ └──────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段3: MOPD (多域同策略蒸馏) │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 将多个专家教师模型蒸馏到单个学生模型中 │ │
│ └──────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段4: Agentic RL (Agent强化学习) │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 在真实Agent Harness中多轮RL训练 │ │
│ │ OpenClaw / Hermes Agent / Pi │ │
│ │ GRPO + 沙箱环境 + Harness Proxy │ │
│ └──────────────────────────────────────────────────────┘ │
└────────────────────────────────────────────────────────────┘

3.2 MOPD：多域同策略蒸馏的技术细节

MOPD（Multi-Domain On-Policy Distillation）是这套流水线的核心创新。传统蒸馏通常使用固定的教师输出，而MOPD让教师和学生模型在相同的策略下生成数据，从而保持分布一致性。

python
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
“”"
MOPD (Multi-Domain On-Policy Distillation) 实现
多域同策略蒸馏的核心算法
“”"
import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass
import math

@dataclass
class MOPDConfig:
“”“MOPD配置”“”
vocab_size: int = 128000
hidden_dim: int = 2048
num_layers: int = 30
num_teachers: int = 4 # 数学、代码、工具、推理
kl_weight: float = 0.5
ce_weight: float = 1.0
distill_temperature: float = 2.0
domain_weights: List[float] = None

class MOPDDistiller:
“”"
多域同策略蒸馏器
核心思想：在策略采样(on-policy)过程中，同时用教师和学生生成logits，
然后通过KL散度+交叉熵的联合损失进行蒸馏
“”"

def __init__(self, config: MOPDConfig): self.config = config if config.domain_weights is None: self.config.domain_weights = [1.0, 1.0, 1.0, 1.0] def compute_distill_loss( self, student_logits: torch.Tensor, # [batch, seq_len, vocab] teacher_logits_list: List[torch.Tensor], # 4个教师的logits labels: torch.Tensor, # [batch, seq_len] domain_ids: torch.Tensor, # [batch], 每个样本所属领域 attention_mask: Optional[torch.Tensor] = None ) -> Dict[str, torch.Tensor]: """ 计算蒸馏损失 Args: student_logits: 学生模型输出logits teacher_logits_list: 四个教师模型的logits labels: 目标token ids domain_ids: 领域标签 (0=math, 1=code, 2=tool, 3=reasoning) attention_mask: 注意力掩码 """ batch_size, seq_len, vocab_size = student_logits.shape if attention_mask is None: attention_mask = torch.ones(batch_size, seq_len, dtype=torch.bool) # 1. 交叉熵损失（标准语言建模） ce_loss = F.cross_entropy( student_logits.view(-1, vocab_size), labels.view(-1), reduction='none' ).view(batch_size, seq_len) ce_loss = (ce_loss * attention_mask).sum() / attention_mask.sum() # 2. KL散度损失（蒸馏） # 对每个样本，只使用对应领域的教师 kl_loss = 0.0 student_log_probs = F.log_softmax( student_logits / self.config.distill_temperature, dim=-1 ) for domain_idx in range(self.config.num_teachers): domain_mask = (domain_ids == domain_idx) if domain_mask.sum() == 0: continue # 获取该领域教师logits teacher_logits = teacher_logits_list[domain_idx] # 教师概率分布 teacher_probs = F.softmax( teacher_logits / self.config.distill_temperature, dim=-1 ) # KL(P_teacher || P_student) domain_kl = F.kl_div( student_log_probs[domain_mask], teacher_probs[domain_mask], reduction='sum', log_target=False ) domain_weight = self.config.domain_weights[domain_idx] kl_loss += domain_weight * domain_kl kl_loss = kl_loss / attention_mask.sum() # 3. 联合损失 total_loss = (self.config.ce_weight * ce_loss + self.config.kl_weight * kl_loss * (self.config.distill_temperature ** 2)) return { 'total_loss': total_loss, 'ce_loss': ce_loss, 'kl_loss': kl_loss, }

class GRPOTrainer:
“”"
GRPO (Group Relative Policy Optimization) Agent训练器
用于Agentic RL阶段
“”"

def __init__( self, model: nn.Module, tokenizer: Callable, clip_epsilon: float = 0.2, kl_coeff: float = 0.01, group_size: int = 8 ): self.model = model self.tokenizer = tokenizer self.clip_epsilon = clip_epsilon self.kl_coeff = kl_coeff self.group_size = group_size @dataclass class Trajectory: """单次Agent交互轨迹""" observations: List[str] actions: List[str] tool_calls: List[Dict] rewards: List[float] log_probs: List[float] def compute_grpo_loss( self, trajectories: List[Trajectory], old_log_probs: torch.Tensor, advantages: torch.Tensor ) -> torch.Tensor: """ 计算GRPO损失 GRPO = -E[ min(r * A, clip(r, 1-ε, 1+ε) * A) ] 其中 r = exp(log_prob_new - log_prob_old) """ # 当前策略的log概率 current_log_probs = self._compute_log_probs(trajectories) # 概率比 ratios = torch.exp(current_log_probs - old_log_probs) # 裁剪后的替代目标 surr1 = ratios * advantages surr2 = torch.clamp(ratios, 1.0 - self.clip_epsilon, 1.0 + self.clip_epsilon) * advantages policy_loss = -torch.min(surr1, surr2).mean() # KL惩罚（防止策略偏离太远） kl_div = (old_log_probs - current_log_probs).mean() return policy_loss + self.kl_coeff * kl_div def _compute_log_probs(self, trajectories: List[Trajectory]) -> torch.Tensor: """计算轨迹的log概率""" # 简化实现：实际中需要完整的模型前向传播 log_probs = [] for traj in trajectories: for log_prob in traj.log_probs: log_probs.append(log_prob) return torch.tensor(log_probs)

def run_mopd_pipeline():
“”"
演示完整的MOPD训练流程
“”"
config = MOPDConfig()
distiller = MOPDDistiller(config)

print("MOPD训练流程演示") print("=" * 60) print(f"词汇表大小: {config.vocab_size}") print(f"教师模型数量: {config.num_teachers}") print(f"蒸馏温度: {config.distill_temperature}") print(f"KL权重: {config.kl_weight}, CE权重: {config.ce_weight}") print() # 模拟训练数据 batch_size = 4 seq_len = 512 dummy_student_logits = torch.randn(batch_size, seq_len, config.vocab_size) dummy_teacher_logits = [ torch.randn(batch_size, seq_len, config.vocab_size) for _ in range(config.num_teachers) ] dummy_labels = torch.randint(0, config.vocab_size, (batch_size, seq_len)) dummy_domains = torch.randint(0, config.num_teachers, (batch_size,)) # 计算损失 losses = distiller.compute_distill_loss( dummy_student_logits, dummy_teacher_logits, dummy_labels, dummy_domains ) print(f"总损失: {losses['total_loss']:.4f}") print(f"交叉熵损失: {losses['ce_loss']:.4f}") print(f"KL散度损失: {losses['kl_loss']:.4f}") print(f"蒸馏温度^2缩放: {config.distill_temperature ** 2:.2f}") # 演示GRPO grpo_trainer = GRPOTrainer(model=None, tokenizer=None) print(f"\nGRPO组大小: {grpo_trainer.group_size}") print(f"裁剪ε: {grpo_trainer.clip_epsilon}") print(f"KL系数: {grpo_trainer.kl_coeff}")

ifname== “main”:
run_mopd_pipeline()

四、端侧推理引擎：从零实现一个轻量级推理框架

4.1 推理性能全景

LFM2.5-2.6B的推理性能令人印象深刻：

表格
硬件平台 解码速度 内存占用
Apple M5 Max 220 tok/s <2.5 GB
AMD Ryzen AI Max+ 395 113 tok/s <2.5 GB
智能手机 ~30 tok/s <2.5 GB
NVIDIA H100 (高并发) ~15,000 tok/s -

这意味着同一套权重既可以在边缘设备上运行，也可以在服务器端进行批量推理。

4.2 Go实现：端侧推理引擎核心

下面我们使用Go语言实现一个端侧推理引擎的核心组件，重点关注KV Cache优化和内存管理。

go
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
24