Posts

大模型训练四阶段：从原始语料到对齐智能 大语言模型的训练是一个多阶段的复杂过程。从原始互联网文本到能与人类对齐的智能助手，需要经过预训练、监督微调、人类反馈强化学习（RLHF）和直接偏好优化（DPO）四个阶段。本文将全面解析这条训练链路。 一、训练流程总览 ┌─────────────────────────────────────────────────────────────┐ │ 大模型训练全流程 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 阶段1: 预训练 (Pre-training) │ │ 输入: 万亿 Token 原始语料 │ │ 目标: Next Token Prediction │ │ 产出: 基座模型 (Base Model) │ │ │ │ 阶段2: 监督微调 (SFT) │ │ 输入: 高质量指令-回复对 (10K-1M) │ │ 目标: 模仿专家回复 │ │ 产出: 指令模型 (Instruct Model) │ │ │ │ 阶段3: 人类反馈强化学习 (RLHF) │ │ 输入: 人类偏好数据 (A > B 对比) │ │ 目标: 最大化人类偏好奖励 │ │ 产出: 对齐模型 (Aligned Model) │ │ │ │ 阶段4: 直接偏好优化 (DPO) │ │ 输入: 同 RLHF 的偏好数据 │ │ 目标: 直接优化策略, 无需奖励模型 │ │ 产出: 进一步优化的对齐模型 │ │ │ └─────────────────────────────────────────────────────────────┘ 二、预训练（Pre-training） 2.1 训练目标 预训练使用Next Token Prediction（NTP）作为训练目标： ...

Ring Attention：突破 GPU 显存墙 当上下文窗口从 4K 扩展到 1M Token 时，单 GPU 的 KV Cache 存储和注意力计算已经无法满足需求。Ring Attention（环形注意力）通过将注意力计算分布到多 GPU，实现了超长上下文的训练和推理。 一、长上下文的挑战 1.1 注意力计算复杂度 标准注意力的计算和显存需求都是 $O(n^2)$： 对于 $n = 1,048,576$（1M Token）： 注意力矩阵：$10^{12}$ 元素 = 8 TB（FP16） 单 GPU A100 80GB 显存：只能放下 0.01 的注意力矩阵 即使使用 Flash Attention 减少 HBM 访问，单 GPU 也无法存储完整的注意力矩阵。 1.2 KV Cache 显存需求 每个 Token 需要存储 Key 和 Value 向量。以 DeepSeek V4（MLA 压缩后）为例： $$\text{KV Cache/Token} = 512 \times 2 \times 2 \text{ bytes} = 2 \text{ KB}$$ ...

连续批处理：推理吞吐的范式革命 在 LLM 推理系统中，如何高效处理并发请求是决定服务成本的核心问题。连续批处理（Continuous Batching）配合 PagedAttention，让 vLLM 实现了比传统推理框架高 5-20 倍的吞吐量。本文深入解析这项技术的原理与工程实现。 一、批处理的演进 1.1 静态批处理（Static Batching） 最简单的批处理方式：等待凑齐一批请求，然后一起推理，直到所有请求完成。 ┌────────────────────────────────────────────────────┐ │ 静态批处理 │ ├────────────────────────────────────────────────────┤ │ │ │ 时间→ 0 1 2 3 4 5 6 7 │ │ │ │ Req1 [G][G][G][G][G][D] ✓ │ │ Req2 [G][G][G][G][G][G][G][G] ✓ │ │ Req3 [G][G][D] ✓ │ │ Req4 [G][G][G][G][G][G][G][D] ✓ │ │ │ │ G=生成, D=完成(停止生成) │ │ │ │ 问题: Req3 在 t=2 就完成了, │ │ 但要等 Req2 到 t=7 整批才结束 │ │ GPU 利用率: ~35% │ └────────────────────────────────────────────────────┘ 核心问题：Padding 浪费 + 等待浪费。不同请求长度差异大，短请求要等长请求完成。 ...

投机解码：用小模型的"猜测"加速大模型 大模型推理的最大瓶颈是自回归生成的串行性——每生成一个 Token，都需要完整的前向传播。投机解码（Speculative Decoding）通过引入一个小型 Draft Model 来"猜测"多个 Token，再由大模型并行验证，打破了这一串行瓶颈。 一、标准推理的瓶颈 1.1 自回归推理 标准自回归推理中，生成 $N$ 个 Token 需要 $N$ 次串行前向传播： 步骤1: [prompt] → 大模型 → token_1 步骤2: [prompt, token_1] → 大模型 → token_2 步骤3: [prompt, token_1, token_2] → 大模型 → token_3 ... 步骤N: [prompt, token_1, ..., token_{N-1}] → 大模型 → token_N 总延迟 = N × 单步前向延迟 每次前向传播只生成 1 个 Token，但需要计算所有参数。对于 70B 模型，单步前向约 30ms，生成 1000 个 Token 需要约 30 秒。 1.2 为什么不能并行 自回归的数学约束： ...

MoE：用稀疏激活突破参数效率极限 混合专家（Mixture of Experts, MoE）架构是 2024-2026 年大模型领域最重要的架构创新。它让模型在不增加推理计算量的前提下大幅扩展参数量，实现了"大模型的能力，小模型的速度"。本文将从原理到工程，全面解析 MoE 的演进。 一、MoE 基本原理 1.1 稀疏激活的核心思想 标准 Transformer 中，每个 Token 都通过所有参数计算（密集激活）。MoE 则让每个 Token 只激活部分参数（稀疏激活）： $$\text{MoE}(x) = \sum_{i \in \text{TopK}(G(x))} G(x)_i \cdot E_i(x)$$ 其中： $G(x) = \text{softmax}(W_g \cdot x)$ 是路由器（Gate/Router） $\text{TopK}(G(x))$ 选择概率最高的 $K$ 个专家 $E_i(x)$ 是第 $i$ 个专家的输出 ┌─────────────────────────────────────────────────────┐ │ MoE 层结构 │ ├─────────────────────────────────────────────────────┤ │ │ │ Input x ──► Router G(x) ──► Top-K 选择 │ │ │ │ │ ┌─────┬─────┬─────┬───┴───┐ │ │ │ │ │ │ │ │ │ ▼ ▼ ▼ ▼ ▼ │ │ ┌────┐┌────┐┌────┐┌────┐ ┌────┐ │ │ │ E1 ││ E2 ││ E3 ││ E4 │...│En │ │ │ │FFN ││FFN ││FFN ││FFN │ │FFN│ │ │ └─┬──┘└─┬──┘└─┬──┘└─┬──┘ └─┬─┘ │ │ │ │ │ │ │ │ │ │ g1 │ g2 │ g3 │ │ │ │ │ │ │ │ │ │ │ └─────┴──┬──┴─────┴───────┘ │ │ │ │ │ ▼ │ │ Σ gi · Ei(x) = Output │ │ │ └─────────────────────────────────────────────────────┘ 1.2 为什么 MoE 有效 MoE 的优势在于参数解耦： ...

引言 Agent 的"状态"是什么？是对话历史、是工作流进度、是工具调用结果、是用户偏好、是 Agent 的"记忆"。无状态 Agent 简单但健忘；有状态 Agent 智能但复杂。2026年，随着 Agent 处理的任务越来越长（从分钟级到天级），状态管理成为架构设计的核心挑战。 一、Agent 状态的分类 ┌─────────────────────────────────────────────────────┐ │ Agent 状态全景图 │ ├──────────────┬──────────────────┬───────────────────┤ │ 状态类型 │ 生命周期 │ 存储介质 │ ├──────────────┼──────────────────┼───────────────────┤ │ 会话状态 │ 单次会话 │ 内存 / Redis │ │ (消息历史) │ 30分钟-24小时 │ │ ├──────────────┼──────────────────┼───────────────────┤ │ 工作流状态 │ 任务执行期间 │ Redis / 数据库 │ │ (执行进度) │ 分钟-天 │ │ ├──────────────┼──────────────────┼───────────────────┤ │ 用户状态 │ 用户生命周期 │ 数据库 │ │ (偏好/画像) │ 永久 │ │ ├──────────────┼──────────────────┼───────────────────┤ │ 检查点状态 │ 可恢复期间 │ 对象存储/数据库 │ │ (快照) │ 可配置 │ │ ├──────────────┼──────────────────┼───────────────────┤ │ 共享状态 │ 多Agent协作期间 │ Redis/共享存储 │ │ (黑板/消息) │ 会话级 │ │ └──────────────┴──────────────────┴───────────────────┘ 二、会话状态管理 2.1 会话状态模型 from dataclasses import dataclass, field from datetime import datetime from enum import Enum class MessageRole(Enum): SYSTEM = "system" USER = "user" ASSISTANT = "assistant" TOOL = "tool" @dataclass class Message: role: MessageRole content: str tool_calls: list | None = None tool_call_id: str | None = None timestamp: datetime = field(default_factory=datetime.now) metadata: dict = field(default_factory=dict) @dataclass class SessionState: """完整的会话状态""" session_id: str user_id: str agent_id: str messages: list[Message] = field(default_factory=list) context: dict = field(default_factory=dict) # 上下文变量 active_tools: list[str] = field(default_factory=list) pending_tool_calls: list[dict] = field(default_factory=list) created_at: datetime = field(default_factory=datetime.now) updated_at: datetime = field(default_factory=datetime.now) expires_at: datetime | None = None status: str = "active" # active / paused / completed / error 2.2 会话存储实现 class SessionStore: """会话状态存储——分层缓存策略""" def __init__(self, redis, postgres): self.redis = redis # 热数据 self.postgres = postgres # 冷数据/持久化 self.ttl = 86400 * 7 # 7天过期 async def get(self, session_id: str) -> SessionState | None: # L1: Redis 缓存 cached = await self.redis.get(f"session:{session_id}") if cached: return SessionState.from_json(cached) # L2: PostgreSQL row = await self.postgres.fetchrow( "SELECT * FROM agent_sessions WHERE session_id = $1", session_id ) if not row: return None state = SessionState.from_db(row) # 回填缓存 await self.redis.setex( f"session:{session_id}", 3600, # 缓存1小时 state.to_json() ) return state async def save(self, state: SessionState): state.updated_at = datetime.now() # 写入 Redis（热路径） await self.redis.setex( f"session:{state.session_id}", 3600, state.to_json() ) # 异步写入 PostgreSQL（冷路径） asyncio.create_task(self._persist_to_db(state)) async def _persist_to_db(self, state: SessionState): await self.postgres.execute(""" INSERT INTO agent_sessions (session_id, user_id, agent_id, state, updated_at) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (session_id) DO UPDATE SET state = $4, updated_at = $5 """, state.session_id, state.user_id, state.agent_id, state.to_json(), state.updated_at) 2.3 上下文窗口管理 class ContextWindowManager: """管理 Agent 的上下文窗口""" def __init__(self, max_tokens: int = 128000): self.max_tokens = max_tokens self.reserved_for_output = 4096 self.available = max_tokens - self.reserved_for_output def prepare_context( self, system_prompt: str, messages: list[Message], tools_schema: list[dict] ) -> list[dict]: """在 Context Window 内准备消息""" # 计算各部分 Token system_tokens = self._count_tokens(system_prompt) tools_tokens = self._count_tokens(json.dumps(tools_schema)) remaining = self.available - system_tokens - tools_tokens # 从最新消息向前保留 prepared = [] total = 0 for msg in reversed(messages): msg_tokens = self._count_tokens(msg.content) if total + msg_tokens > remaining: break prepared.insert(0, msg) total += msg_tokens # 如果截断了，添加摘要提示 if len(prepared) < len(messages): summary = self._generate_summary(messages[:len(messages) - len(prepared)]) prepared.insert(0, Message( role=MessageRole.SYSTEM, content=f"[Earlier conversation summary: {summary}]" )) return prepared def _count_tokens(self, text: str) -> int: # 使用 tiktoken 精确计算 import tiktoken enc = tiktoken.encoding_for_model("gpt-5") return len(enc.encode(text)) 三、工作流状态与检查点 3.1 检查点机制 class CheckpointManager: """Agent 执行检查点管理""" def __init__(self, storage): self.storage = storage async def save_checkpoint( self, execution_id: str, state: WorkflowState, step_index: int, step_name: str ): """保存执行检查点""" checkpoint = Checkpoint( execution_id=execution_id, step_index=step_index, step_name=step_name, state=state, timestamp=datetime.now() ) await self.storage.save(checkpoint) # 保留最近 N 个检查点 await self._prune_old_checkpoints(execution_id, keep=10) async def restore(self, execution_id: str) -> tuple[WorkflowState, int]: """从最新检查点恢复""" latest = await self.storage.get_latest(execution_id) if not latest: raise NoCheckpointError(execution_id) logger.info( f"Restoring from checkpoint: step={latest.step_index}, " f"name={latest.step_name}" ) return latest.state, latest.step_index async def list_checkpoints(self, execution_id: str) -> list[Checkpoint]: """列出所有检查点（用于调试）""" return await self.storage.list(execution_id) @dataclass class WorkflowState: """工作流状态——可序列化""" execution_id: str current_step: str step_index: int results: dict # {step_name: result} variables: dict # 工作流变量 pending_actions: list # 待执行操作 error: str | None # 错误信息（如果有的话） iteration: int # 循环计数 def serialize(self) -> bytes: return pickle.dumps(self) # 或使用 JSON @classmethod def deserialize(cls, data: bytes) -> "WorkflowState": return pickle.loads(data) 3.2 可恢复的 Agent 扥行器 class ResumableAgentExecutor: """支持断点续传的 Agent 执行器""" def __init__(self, checkpoint_mgr: CheckpointManager): self.checkpoints = checkpoint_mgr async def execute( self, workflow: Workflow, initial_state: WorkflowState, execution_id: str | None = None ) -> WorkflowState: execution_id = execution_id or str(uuid.uuid4()) # 尝试从检查点恢复 try: state, start_step = await self.checkpoints.restore(execution_id) logger.info(f"Resuming from step {start_step}") except NoCheckpointError: state = initial_state start_step = 0 # 获取工作流步骤 steps = workflow.get_steps() for i, step in enumerate(steps[start_step:], start=start_step): try: # 执行前保存检查点 state.current_step = step.name state.step_index = i await self.checkpoints.save_checkpoint( execution_id, state, i, step.name ) # 执行步骤 result = await step.execute(state) # 更新状态 state.results[step.name] = result state.variables.update(result.get("variables", {})) # 条件分支 if step.condition: next_step = step.condition(result) if next_step: state.pending_actions = [next_step] except Exception as e: state.error = str(e) await self.checkpoints.save_checkpoint( execution_id, state, i, step.name ) # 重试逻辑 if step.retry_policy: retry_count = state.variables.get(f"retry_{step.name}", 0) if retry_count < step.retry_policy.max_attempts: state.variables[f"retry_{step.name}"] = retry_count + 1 await asyncio.sleep(step.retry_policy.backoff(retry_count)) # 重新执行当前步骤 continue raise return state 四、多 Agent 共享状态 4.1 黑板模式 class SharedBlackboard: """多 Agent 共享黑板""" def __init__(self, redis_client): self.redis = redis_client self.namespace = "blackboard" async def write( self, key: str, value: any, agent_id: str, ttl: int = 3600 ): """写入共享状态""" entry = { "value": value, "writer": agent_id, "timestamp": time.time(), } await self.redis.hset( f"{self.namespace}:{key}", mapping={k: json.dumps(v) for k, v in entry.items()} ) await self.redis.expire(f"{self.namespace}:{key}", ttl) # 通知订阅者 await self.redis.publish( f"{self.namespace}:updates", json.dumps({"key": key, "writer": agent_id}) ) async def read(self, key: str) -> any: """读取共享状态""" data = await self.redis.hgetall(f"{self.namespace}:{key}") if not data: return None return json.loads(data.get("value", "null")) async def subscribe( self, key_pattern: str, callback: callable ): """订阅状态变更""" pubsub = self.redis.pubsub() await pubsub.subscribe(f"{self.namespace}:updates") async for message in pubsub.listen(): if message["type"] == "message": data = json.loads(message["data"]) if fnmatch.fnmatch(data["key"], key_pattern): await callback(data) 4.2 Agent 间消息传递 class AgentMessageBus: """Agent 间异步消息总线""" def __init__(self, redis_client): self.redis = redis_client self.queues = {} # {agent_id: Queue} async def send( self, from_agent: str, to_agent: str, message_type: str, payload: dict, reply_to: str | None = None ): """发送消息给另一个 Agent""" msg = AgentMessage( id=str(uuid.uuid4()), from_agent=from_agent, to_agent=to_agent, type=message_type, payload=payload, reply_to=reply_to, timestamp=datetime.now() ) # 推入接收者的队列 await self.redis.lpush( f"agent:inbox:{to_agent}", msg.to_json() ) async def receive( self, agent_id: str, timeout: int = 30 ) -> AgentMessage | None: """接收消息""" result = await self.redis.brpop( f"agent:inbox:{agent_id}", timeout=timeout ) if result: return AgentMessage.from_json(result[1]) return None async def request_reply( self, from_agent: str, to_agent: str, message_type: str, payload: dict, timeout: int = 60 ) -> dict | None: """请求-回复模式""" reply_channel = f"reply:{uuid.uuid4()}" await self.send( from_agent, to_agent, message_type, payload, reply_to=reply_channel ) # 等待回复 result = await self.redis.brpop(reply_channel, timeout=timeout) if result: return json.loads(result[1]) return None 五、状态序列化与迁移 class StateSerializer: """状态序列化器""" SCHEMA_VERSION = "2.0" def serialize(self, state: any) -> str: """序列化状态为 JSON""" data = { "schema_version": self.SCHEMA_VERSION, "type": type(state).__name__, "data": self._to_dict(state), "timestamp": datetime.now().isoformat() } return json.dumps(data, ensure_ascii=False, default=str) def deserialize(self, raw: str) -> any: """反序列化""" data = json.loads(raw) # 版本迁移 if data["schema_version"] != self.SCHEMA_VERSION: data = self._migrate(data) return self._from_dict(data["type"], data["data"]) def _migrate(self, data: dict) -> dict: """状态版本迁移""" migrations = [ ("1.0", "1.1", self._migrate_1_0_to_1_1), ("1.1", "2.0", self._migrate_1_1_to_2_0), ] current = data["schema_version"] for from_v, to_v, migrator in migrations: if current == from_v: data = migrator(data) current = to_v return data 六、状态管理架构选型 场景 推荐方案 原因 短对话（< 30min） 内存 + Redis 低延迟、自动过期 长对话（> 1h） Redis + PostgreSQL 持久化 + 快速访问 长流程工作流 Redis + 检查点 + DB 断点续传 多 Agent 协作 Redis 黑板 + 消息总线 实时共享 用户画像 PostgreSQL + 向量DB 持久化 + 语义检索 跨设备同步 CRDT + Redis 冲突解决 七、状态管理 Checklist □ 会话状态分层存储（内存 → Redis → 数据库） □ 上下文窗口管理策略（滑动窗口 + 摘要） □ 工作流检查点定期保存 □ 检查点支持断点续传 □ 多 Agent 共享状态通过消息总线 □ 状态序列化支持版本迁移 □ 过期状态自动清理 □ 状态加密敏感字段 □ 状态变更审计日志 □ 状态一致性测试（并发读写） 结语 状态管理是 Agent 从"玩具"到"产品"的分水岭。无状态 Agent 是函数——输入即输出；有状态 Agent 是伙伴——它记得你、理解上下文、能从中断处继续。但状态也带来了复杂性：一致性、持久化、恢复、迁移。好的状态管理架构是透明的——开发者不需要关心状态的存储和恢复，Agent 始终如丝般顺滑地运行。这是工程的艺术。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

Flash Attention 3：让 GPU 跑满的注意力计算 Flash Attention 系列是近年来大模型工程领域最重要的优化之一。从 Flash Attention 1 到 3，每代都在逼近 GPU 硬件的理论极限。2026 年，Flash Attention 3 已经成为所有主流大模型推理和训练的标配。本文将深入解析其原理。 一、问题：标准注意力的内存瓶颈 1.1 GPU 内存层次 现代 GPU（如 H100）有复杂的内存层次： ┌─────────────────────────────────────────────┐ │ GPU 内存层次 │ ├─────────────────────────────────────────────┤ │ │ │ ┌─────────┐ 延迟: ~20 cycles │ │ │Register │ 带宽: ~30 TB/s │ │ │(SRAM) │ 容量: 256 KB/SM │ │ └─────────┘ │ │ ↑ │ │ ┌─────────┐ 延迟: ~200 cycles │ │ │L2 Cache │ 带宽: ~12 TB/s │ │ │ │ 容量: 50 MB │ │ └─────────┘ │ │ ↑ │ │ ┌─────────┐ 延迟: ~400+ cycles │ │ │ HBM │ 带宽: ~3.35 TB/s (H100) │ │ │(显存) │ 容量: 80 GB │ │ └─────────┘ │ │ │ └─────────────────────────────────────────────┘ 关键洞察：HBM 带宽只有 SRAM 的 1/9，但标准注意力几乎完全在 HBM 上操作。 ...

KV Cache：大模型推理的内存瓶颈 如果说 Transformer 架构决定了一个模型的智力上限，那么 KV Cache 就决定了它在实际应用中的性能下限。在 2026 年，KV Cache 已经成为大模型推理系统最核心的优化对象——理解它，是掌握大模型工程的第一课。 一、什么是 KV Cache 1.1 问题：为什么需要缓存 在 Transformer 的自回归推理中，生成第 $t$ 个 Token 时需要计算它与之前所有 Token 的注意力： $$\text{Attention}(q_t, K_{1:t}, V_{1:t}) = \text{softmax}\left(\frac{q_t K_{1:t}^T}{\sqrt{d_k}}\right) V_{1:t}$$ 注意：$K_{1:t}$ 和 $V_{1:t}$ 包含了从第 1 到第 $t$ 个位置的 Key 和 Value。 如果没有缓存：每生成一个新 Token，都需要重新计算所有历史 Token 的 $K$ 和 $V$，这导致了 $O(t^2)$ 的重复计算。 有了缓存：只需计算当前 Token 的 $q_t$，从缓存中读取 $K_{1:t-1}$ 和 $V_{1:t-1}$，再计算新的 $k_t, v_t$ 并追加到缓存。 ┌─────────────────────────────────────────────┐ │ 自回归推理第 t 步 │ ├─────────────────────────────────────────────┤ │ │ │ 无缓存: X_1..t → 全部重新计算 → Y_t │ │ (计算量: O(t)) │ │ │ │ 有缓存: X_t → Q_t ───────────→ Y_t │ │ ↑ │ │ KV Cache (K_1..t-1, V_1..t-1) │ │ (计算量: O(1)) │ │ │ └─────────────────────────────────────────────┘ 1.2 KV Cache 的存储内容 每个 Transformer 层、每个 Token 需要存储： ...

分词器：大模型的语言基础 分词器（Tokenizer）是大语言模型的第一个组件，决定了文本如何被切分为 Token。分词质量直接影响模型的词汇覆盖、多语言能力、编码效率乃至推理速度。本文深入解析 2026 年主流分词算法的原理与实践。 一、为什么分词很重要 1.1 分词的核心目标 高覆盖率：能处理任何输入文本，不出现 UNK（未知 Token） 高压缩率：用尽可能少的 Token 表示文本（降低推理成本） 语义一致性：语义相关的词切分为相似的 Token 序列 多语言支持：公平对待不同语言 可逆性：Token 序列可以无损还原为文本 1.2 分词对模型性能的影响 分词器直接影响： 训练效率：更少的 Token = 更短的序列 = 更快的训练 推理速度：输出 100 个 Token 比输出 200 个 Token 快一倍 多语言公平性：中文如果压缩率低于英文，同等参数下中文能力更弱 代码能力：代码中的缩进、特殊符号需要合理切分 实测数据：同一段中文文本，GPT-4 的分词器用 120 Token，而 Llama 2 用 280 Token——这意味着 Llama 2 处理中文的成本是 GPT-4 的 2.3 倍。 二、BPE（Byte Pair Encoding） 2.1 算法原理 BPE 从字符级别开始，迭代合并最高频的相邻 Token 对： 初始词表: a, b, c, d, e, f, g, h, i, j, k, l, m, n, o, p, q, r, s, t, u, v, w, x, y, z, ... 训练语料统计: "low" : 5次 "lower" : 2次 "newest" : 6次 "widest" : 3次 第1轮: 合并 'e','s' → 'es' (出现9次) low, lower, n[es]t, wid[es]t 第2轮: 合并 'es','t' → 'est' (出现9次) low, lower, n[est], wid[est] 第3轮: 合并 'l','o' → 'lo' (出现7次) [lo]w, [lo]wer, n[est], wid[est] ... 直到词表大小达到目标 2.2 数学描述 给定训练语料 $\mathcal{D}$，初始词表为所有字符。每轮选择使以下目标最大化的合并对 $(a, b)$： ...

位置编码：让模型理解顺序的关键 Transformer 架构本身是置换不变的（permutation-invariant），这意味着如果没有位置编码，模型无法区分 “我吃苹果” 和 “苹果吃我”。位置编码为序列注入位置信息，是理解文本顺序和层级结构的关键。 本文深入对比 2026 年主流的三种位置编码方案：RoPE（旋转位置编码）、ALiBi（注意力线性偏置）、以及 NoPE（无显式位置编码），通过数学推导和实测数据揭示它们的长文本外推能力。 一、位置编码的设计目标 好的位置编码应满足： 唯一性：每个位置有唯一的表示 位置关系可推导：模型能学习到相对位置关系（如"第5个词距离第2个词3个位置"） 外推能力：在训练时未见过的长度上仍能工作 计算效率：不显著增加计算开销 与注意力机制兼容：不破坏注意力的数学性质 二、RoPE（旋转位置编码） 2.1 数学原理 RoPE 的核心思想是将位置信息编码为旋转操作。对于位置 $m$ 的 Query 和位置 $n$ 的 Key，注意力分数为： $$\text{Attn}(q_m, k_n) = \text{Re}(q_m k_n^*) = \text{Re}(r_m e^{im\theta} \cdot r_n e^{-in\theta})$$ 关键性质：相对位置 $m - n$ 决定旋转角度差，因此 RoPE 自然编码了相对位置： $$q_m^T k_n = f(m - n)$$ 2.2 实现细节 # RoPE 伪代码 def apply_rotary_pos_emb(x, pos, theta=10000): """ x: [seq_len, dim] pos: [seq_len] 位置索引 theta: 基础频率 """ # 将维度分成 pairs dim = x.shape[-1] freqs = 1.0 / (theta ** (torch.arange(0, dim, 2) / dim)) angles = pos[:, None] * freqs[None, :] # 应用旋转 cos, sin = angles.cos(), angles.sin() x_rot = rotate_half(x) return x * cos + x_rot * sin 2.3 外推挑战与解决方案 RoPE 的外推能力受限于训练时的最大长度。如果训练长度为 4096，推理时扩展到 128K 会导致： ...