分类流映射(CFMs)大规模扩展：从理论到工程实践

在机器学习领域，生成模型的核心任务之一是从简单分布（如高斯噪声）中采样，并将其转化为复杂的数据分布（如图像、文本）。近年来，扩散模型凭借其强大的生成能力成为主流，但其迭代去噪过程往往导致采样速度缓慢。分类流映射（Categorical Flow Maps, CFMs）作为一种新兴的生成建模框架，提供了一种基于常微分方程（ODE）的、理论上可逆的流匹配方法，旨在实现快速、高质量的采样。然而，将CFMs扩展到大规模、高维分类数据（如语言建模中的词汇表）时，面临着计算复杂度、内存消耗和训练稳定性的严峻挑战。

本文旨在深入探讨如何将分类流映射的规模进行有效扩展，使其能够处理像大型词汇表语言建模这样的复杂任务。我们将从CFMs的基本原理出发，逐步构建一个可实践的扩展方案，涵盖理论理解、模型架构设计、训练策略优化以及关键的工程实现细节。无论你是希望理解流匹配与扩散模型的内在联系，还是计划在文本生成等任务中应用CFMs，本文都将提供一个从理论到实践的完整路径。

1. 理解分类流映射（CFMs）与流匹配的核心思想

在深入扩展之前，我们必须先厘清CFMs试图解决的根本问题及其工作原理。这有助于我们在后续扩展时做出正确的设计决策。

1.1 从扩散模型到流匹配：一个效率视角

扩散模型通过一个前向过程逐渐向数据中添加噪声，再训练一个神经网络学习反向的去噪过程。采样时，需要从纯噪声开始，逐步迭代（通常需要数十甚至上百步）去噪才能生成样本。这个过程虽然质量高，但计算成本巨大。

流匹配（Flow Matching）提供了一种不同的视角。它旨在学习一个从简单先验分布（如标准高斯分布）到复杂数据分布的确定性变换，这个变换由一个速度场（Velocity Field）定义，并可以通过求解一个常微分方程（ODE）来实现。一旦学好了这个速度场，理论上可以通过单次或少数几次ODE求解就完成从噪声到数据的转换，从而极大提升采样速度。CFMs是流匹配思想在离散分类数据（如one-hot向量表示的词）上的具体实现。

1.2 分类流映射（CFMs）的数学表述

对于分类数据，每个样本通常表示为一个在K个类别上的one-hot向量。CFMs的目标是构建一个连续的时间依赖的概率路径p_t(x)，其中t从0到1。在t=0时，p_0(x)是一个简单的先验分布（例如，在所有类别上均匀分布的分类分布）；在t=1时，p_1(x)就是我们的目标数据分布。

关键点在于定义一个向量场u_t(x)，使得沿着这个场从t=0到t=1积分，就能将先验分布的样本“流动”成数据分布的样本。这个向量场需要满足连续性方程（Continuity Equation），以确保概率质量在流动过程中守恒。

训练CFMs的核心损失函数是流匹配损失（Flow Matching Loss）：

L(θ) = E_{t, p_t(x)} [ || v_θ(t, x) - u_t(x) ||^2 ]

其中，v_θ是我们用神经网络参数化的速度场，目标是让它逼近真实的目标向量场u_t。在实际操作中，u_t(x)是未知的，但我们可以通过构造一个条件向量场u_t(x | x_1)来绕过这个问题，该场定义了从某个先验样本x_0到某个目标数据样本x_1的直线路径（或其他简单路径）。最终的训练目标变为：

L(θ) = E_{t, p_data(x_1), p_prior(x_0)} [ || v_θ(t, x_t) - (x_1 - x_0) ||^2 ]

这里x_t = (1 - t) * x_0 + t * x_1是线性插值点。对于分类数据，x_0和x_1都是one-hot向量，但它们的线性组合x_t不再是one-hot，而是一个概率单纯形（Simplex）上的点。这就是CFMs处理离散数据的核心技巧：在连续的时间空间中，用连续的概率向量来表示离散状态。

2. 扩展CFMs规模的核心挑战与应对策略

当词汇表规模K从几百上升到数万甚至数十万（如现代语言模型常见的3万到10万词表）时，原始的CFMs框架会面临几个致命瓶颈。

2.1 挑战一：高维输出空间与计算成本

神经网络的输出层需要预测一个K维的速度场v_θ。这意味着最后一层线性层的参数数量是hidden_dim * K。当K很大时，这个矩阵会变得极其庞大，导致：

1. 巨大的参数量：消耗大量GPU内存。
2. 高昂的计算成本：前向和反向传播中矩阵乘法的计算复杂度为O(batch_size * hidden_dim * K)。

应对策略：参数共享与分解

• 共享嵌入矩阵：一个常见的做法是让速度场预测头与输入词嵌入层共享权重。在语言模型中，输入层通常有一个K x hidden_dim的嵌入矩阵E。我们可以让输出层使用同一个矩阵的转置E^T进行投影，将隐藏状态映射回词汇表空间。这不仅能减半大矩阵的参数，还能在输入输出之间建立对称性。
• 自适应Softmax与层次化Softmax：这是从经典语言模型借鉴的技术。不是直接计算所有K个类别的得分，而是将词汇表组织成一棵树。网络只需要预测路径上的节点，将计算复杂度从O(K)降低到O(log K)。虽然这增加了模型结构的复杂性，但对于超大词表是必要的。

2.2 挑战二：训练稳定性与梯度问题

在CFMs的损失函数中，我们计算预测速度场与目标方向(x_1 - x_0)的均方误差。当K很大时：

• x_1 - x_0在绝大多数维度上为0（因为x_1和x_0都是one-hot，且通常不是同一个词）。这会导致一个非常稀疏的监督信号。
• 神经网络容易倾向于将所有输出预测为接近0，从而学到一种“平均”但无用的速度场。

应对策略：改进的损失函数与训练技巧

• 焦点损失（Focal Loss）或加权MSE：可以对目标为0和不为0的维度赋予不同的权重，让模型更关注那些需要发生“流动”的维度（即x_1对应的维度）。
• 标签平滑（Label Smoothing）的变体：与其使用硬性的one-hot目标x_1，可以考虑使用一个平滑后的分布（如0.9的概率给目标词，0.1的概率均匀分给其他词）。这同样可以作用于x_1 - x_0的目标构造，为模型提供更丰富的梯度信号。
• 时间步采样策略：不均匀地采样时间t。在t接近0或1时，x_t更接近x_0或x_1，变化相对简单。在t接近0.5时，插值点最“模糊”，预测难度最大。可以更多地采样中间区域的时间点，以增强模型在复杂情况下的学习能力。

2.3 挑战三：先验分布p_prior(x_0)的选择

在原始公式中，x_0从先验分布采样。对于图像等连续数据，标准高斯分布是自然选择。对于分类数据，均匀分布是常见选择。但在大规模词表下，均匀先验可能不是最优的，因为它与真实语言数据的分布差异极大，可能导致学习到的流路径非常扭曲和困难。

应对策略：数据驱动的先验

• 使用一元语言模型（Unigram LM）作为先验：统计训练语料中每个词的频率，用这个频率分布作为p_prior(x_0)。这样，先验分布就包含了数据的基本统计信息，从高频词流向目标词可能比从均匀随机词流向目标词更平滑、更容易学习。
• 可学习的先验：将先验分布p_prior也参数化（例如，另一个小的神经网络或一个可学习的概率向量），并与CFM联合训练。这增加了模型灵活性，但也增加了训练难度。

3. 构建大规模CFMs的工程实践

理论策略需要落地到具体的代码和配置中。下面我们以一个基于Transformer架构的大规模文本CFM为例，说明关键实现步骤。

3.1 环境准备与依赖配置

假设我们使用PyTorch进行开发。核心依赖如下：

# requirements.txt 或环境配置 torch>=2.0.0 transformers>=4.30.0 # 用于Tokenizer和基础Transformer组件 datasets>=2.10.0 # 数据加载 accelerate>=0.20.0 # 分布式训练 tensorboard # 可视化 scipy # 用于ODE求解器（如果需要）

项目目录结构建议如下：

cfm_large_scale/ ├── config/ │ └── model_config.yaml # 模型超参数配置 ├── data/ │ ├── tokenizer/ # 存放分词器 │ └── dataset.py # 数据加载与处理逻辑 ├── model/ │ ├── __init__.py │ ├── transformer_cfm.py # CFM模型核心定义 │ └── layers.py # 自定义层（如自适应Softmax） ├── training/ │ ├── train.py # 训练主循环 │ └── loss.py # 自定义损失函数 ├── inference/ │ └── sample.py # 采样（生成）脚本 ├── utils/ │ └── ode_solver.py # ODE求解器封装 └── main.py # 程序入口

3.2 模型架构设计关键代码

以下是一个简化的CFM-Transformer模型核心部分，重点展示如何处理大规模词表。

# model/transformer_cfm.py import torch import torch.nn as nn from transformers import AutoConfig, AutoModel class LargeScaleCFM(nn.Module): def __init__(self, vocab_size, hidden_dim, num_layers, num_heads, max_seq_len, prior_type='unigram'): super().__init__() self.vocab_size = vocab_size self.hidden_dim = hidden_dim self.prior_type = prior_type # 1. 词嵌入层 - 同时作为输入嵌入和输出投影的共享权重 self.token_embedding = nn.Embedding(vocab_size, hidden_dim) self.position_embedding = nn.Embedding(max_seq_len, hidden_dim) # 2. Transformer编码器骨干 encoder_config = AutoConfig.from_pretrained('bert-base-uncased') # 示例，可自定义 encoder_config.hidden_size = hidden_dim encoder_config.num_hidden_layers = num_layers encoder_config.num_attention_heads = num_heads encoder_config.intermediate_size = hidden_dim * 4 self.transformer = AutoModel.from_config(encoder_config) # 3. 输出层：使用共享嵌入矩阵的转置进行投影 # 这是应对大词表的关键，避免了单独的 K*hidden_dim 矩阵 self.output_bias = nn.Parameter(torch.zeros(vocab_size)) # 4. 先验分布参数 if prior_type == 'unigram': # 初始化为均匀分布，训练中会更新 self.register_buffer('log_prior', torch.zeros(vocab_size)) elif prior_type == 'learnable': self.log_prior = nn.Parameter(torch.zeros(vocab_size)) # 5. 时间步编码 self.time_embedding = nn.Sequential( nn.Linear(1, hidden_dim), nn.SiLU(), nn.Linear(hidden_dim, hidden_dim) ) def get_velocity(self, input_ids, t): """ 计算速度场 v_θ(t, x_t) Args: input_ids: [batch, seq_len] 在时间t的token索引（由x_t经过argmax或采样得到？） 注意：这里有一个概念难点。x_t是连续概率向量，不是整数索引。 实际实现中，我们通常用Gumbel-Softmax或Straight-Through技巧从x_t得到可微的“软”索引。 为简化，此处假设input_ids是x_1（目标词）的索引，用于获取嵌入。 t: [batch, 1] 时间步标量 """ batch_size, seq_len = input_ids.shape # 获取目标词嵌入 (x_1 的表示) token_embeds = self.token_embedding(input_ids) # [batch, seq_len, hidden] pos_ids = torch.arange(seq_len, device=input_ids.device).unsqueeze(0).expand(batch_size, -1) pos_embeds = self.position_embedding(pos_ids) # 时间编码，广播到每个token time_emb = self.time_embedding(t).unsqueeze(1) # [batch, 1, hidden] # 将词嵌入、位置编码和时间编码结合 # 这里是一个简单的相加，更复杂的做法可以是拼接或门控 combined_input = token_embeds + pos_embeds + time_emb # 通过Transformer transformer_outputs = self.transformer(inputs_embeds=combined_input) hidden_states = transformer_outputs.last_hidden_state # [batch, seq_len, hidden] # 使用共享嵌入矩阵进行投影，计算logits # logits = hidden_states @ self.token_embedding.weight.T + self.output_bias # 更高效的做法： logits = torch.nn.functional.linear(hidden_states, self.token_embedding.weight, self.output_bias) # logits 形状: [batch, seq_len, vocab_size] # 关键：CFM预测的是速度场，而不是下一个token的概率。 # 我们需要将logits解释为对速度方向（x1 - x0）的预测。 # 一种实现方式是让模型直接预测这个差值向量在词汇空间的方向。 # 这里，我们简单地将logits视为未归一化的速度分量。 # 实际训练时，损失函数会将其与 (x1 - x0) 进行比较。 return logits def sample_prior(self, batch_size, seq_len, device): """从先验分布 p_prior 中采样 x_0""" if self.prior_type == 'uniform': # 均匀分布采样 return torch.randint(0, self.vocab_size, (batch_size, seq_len), device=device) elif self.prior_type == 'unigram': # 根据一元分布采样 probs = torch.softmax(self.log_prior, dim=-1) return torch.multinomial(probs.expand(batch_size*seq_len, -1), 1).view(batch_size, seq_len) else: # learnable probs = torch.softmax(self.log_prior, dim=-1) return torch.multinomial(probs.expand(batch_size*seq_len, -1), 1).view(batch_size, seq_len) def forward(self, x1_ids, t): """训练过程的前向传播""" # 1. 采样先验 x0 x0_ids = self.sample_prior(x1_ids.size(0), x1_ids.size(1), x1_ids.device) # 2. 构造连续时间点 x_t = (1-t)*x0 + t*x1 # 注意：x0和x1是整数索引，需要转为one-hot才能插值 x0_onehot = torch.nn.functional.one_hot(x0_ids, num_classes=self.vocab_size).float() x1_onehot = torch.nn.functional.one_hot(x1_ids, num_classes=self.vocab_size).float() # 为了可微，我们使用连续表示。这里t是标量，需要扩展维度以进行广播。 t_expanded = t.view(-1, 1, 1) # [batch, 1, 1] xt = (1 - t_expanded) * x0_onehot + t_expanded * x1_onehot # [batch, seq_len, vocab] # 3. 从 xt 得到“软”输入。这里使用Gumbel-Softmax松弛的argmax。 # temperature = 1.0 # xt_soft = torch.nn.functional.gumbel_softmax(xt.log(), tau=temperature, hard=False) # 可微采样 # 为简化，我们直接用 xt 的加权平均嵌入作为输入表示。 # 这是CFM处理分类数据的核心：在连续空间操作。 xt_embedding = xt @ self.token_embedding.weight # [batch, seq_len, hidden] # 4. 将 xt_embedding（而非x1_ids）与时间编码结合，输入网络 pos_ids = torch.arange(x1_ids.size(1), device=x1_ids.device).unsqueeze(0).expand(x1_ids.size(0), -1) pos_embeds = self.position_embedding(pos_ids) time_emb = self.time_embedding(t).unsqueeze(1) combined_input = xt_embedding + pos_embeds + time_emb transformer_outputs = self.transformer(inputs_embeds=combined_input) hidden_states = transformer_outputs.last_hidden_state v_theta_logits = torch.nn.functional.linear(hidden_states, self.token_embedding.weight, self.output_bias) # 5. 计算目标向量场 u_t = x1_onehot - x0_onehot u_target = x1_onehot - x0_onehot # [batch, seq_len, vocab] return v_theta_logits, u_target

3.3 训练循环与损失函数实现

训练循环需要集成上述策略，特别是处理稀疏目标和可能的大词表损失计算。

# training/loss.py import torch.nn.functional as F class FocalFlowMatchingLoss(nn.Module): """带焦点权重的流匹配损失，用于缓解大词表下的稀疏目标问题""" def __init__(self, alpha=0.25, gamma=2.0, reduction='mean'): super().__init__() self.alpha = alpha self.gamma = gamma self.reduction = reduction def forward(self, v_pred_logits, u_target): """ v_pred_logits: 网络预测的logits，形状 [batch, seq_len, vocab] u_target: 目标向量场 (x1 - x0)，形状 [batch, seq_len, vocab] """ # 将目标向量场二值化（非零即目标方向） target_mask = (u_target != 0).float() # 1 where flow is needed, 0 otherwise # 计算逐元素的MSE mse_loss = F.mse_loss(v_pred_logits, u_target, reduction='none') # 焦点权重：让模型更关注那些需要流动（target_mask=1）的位置 # 对于target_mask=1的位置，权重为alpha，否则为1-alpha weights = self.alpha * target_mask + (1 - self.alpha) * (1 - target_mask) # 进一步根据预测误差调整权重（gamma参数） # 这里简化处理，直接应用固定权重 weighted_loss = weights * mse_loss if self.reduction == 'mean': return weighted_loss.mean() elif self.reduction == 'sum': return weighted_loss.sum() else: return weighted_loss # training/train.py (关键片段) def train_step(model, batch, optimizer, loss_fn, device): token_ids = batch['input_ids'].to(device) # x1 batch_size, seq_len = token_ids.shape # 随机采样时间步t，可以非均匀采样以侧重中间区域 # t ~ U(0,1) 或 t ~ Beta(2,2) 使得中间值更多 t = torch.rand(batch_size, 1, device=device) # 模型前向传播 v_pred_logits, u_target = model(token_ids, t) # 计算损失 loss = loss_fn(v_pred_logits, u_target) # 反向传播与优化 optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # 梯度裁剪 optimizer.step() return loss.item()

3.4 采样（生成）过程

训练完成后，我们可以通过求解ODE从先验分布生成数据。

# inference/sample.py from utils.ode_solver import solve_ode @torch.no_grad() def generate_text(model, prompt_ids, num_steps=10, solver='euler', device='cuda'): """ 使用训练好的CFM模型生成文本。 Args: model: 训练好的LargeScaleCFM模型 prompt_ids: 可选的提示词ID序列，[1, prompt_len] num_steps: ODE求解的步数（影响生成质量与速度） solver: 求解器类型，'euler'（欧拉法，快）或'dopri5'（自适应步长，慢但准） """ model.eval() batch_size = 1 if prompt_ids is not None: seq_len = prompt_ids.size(1) # 将提示词部分作为x1的已知部分，其余部分从先验开始 x0 = model.sample_prior(batch_size, seq_len, device) # 将提示词位置的目标设为已知词 x0[:, :prompt_ids.size(1)] = prompt_ids # 构造一个mask，标记哪些位置是提示词（不需要流动） fixed_mask = torch.zeros_like(x0, dtype=torch.bool) fixed_mask[:, :prompt_ids.size(1)] = True else: seq_len = model.max_seq_len # 或指定长度 x0 = model.sample_prior(batch_size, seq_len, device) fixed_mask = torch.zeros_like(x0, dtype=torch.bool) # 定义速度场函数，供ODE求解器调用 def velocity_func(t, x_flat): """t: 标量, x_flat: [batch*seq_len*vocab?, ...] 展平的连续表示""" # 这里需要将x_flat重塑为[batch, seq_len, vocab]的连续概率向量 # 然后通过模型计算速度场，再展平返回。 # 实现细节较复杂，涉及连续表示的维护。 # 一种简化：我们只在离散的token空间操作，使用“概率向量插值+argmax”的近似。 pass # 简化版采样：使用训练损失中定义的直线路径的逆过程 # 这不是真正的ODE求解，而是一种启发式采样，适用于快速验证。 # 更严谨的做法需要实现上述velocity_func并使用ODE求解器。 generated_ids = x0.clone() for i in range(num_steps): t = torch.ones(batch_size, 1, device=device) * (i / num_steps) # 从1到0反向？ # 注意：采样是反向过程，时间方向与训练相反。 # 这里仅为示意，完整实现需要仔细设计。 with torch.no_grad(): v_pred, _ = model(generated_ids, t) # 这里输入是当前状态，不是x1 # 使用预测的速度场更新状态（欧拉步） # 更新逻辑需要根据CFM的具体采样公式实现 # generated_ids = update_function(generated_ids, v_pred, step_size=1.0/num_steps) # 将fixed_mask位置的id重置为提示词 generated_ids[fixed_mask] = prompt_ids.view(-1) # 将生成的ID序列解码为文本 # tokenizer.decode(generated_ids[0].tolist()) return generated_ids

4. 训练、验证与常见问题排查

大规模CFMs的训练是一个资源密集型过程，需要系统的监控和问题排查。

4.1 训练流程与关键监控指标

1. 数据准备：使用大规模文本语料库（如C4、Wikipedia）。确保分词器与模型词表一致。
2. 初始化：先验分布如果使用unigram，需要在训练前遍历一次数据计算词频并初始化log_prior。
3. 训练循环：
   • 使用accelerate库支持混合精度训练和分布式数据并行。
   • 每N步计算一次验证集损失。
   • 监控以下指标：
     • train_loss: 流匹配损失。
     • grad_norm: 梯度范数，防止梯度爆炸或消失。
     • param_norm: 模型参数范数。
     • prior_entropy: 先验分布的熵，观察其变化。
4. 采样验证：定期（如每5000步）运行采样函数，生成文本样本，人工评估生成质量（连贯性、多样性、与提示的相关性）。

4.2 常见问题、现象与排查路径

问题现象	可能原因	检查与排查步骤	解决建议
训练损失不下降或震荡	学习率设置不当；梯度稀疏（大词表）；损失函数权重不平衡。	1. 绘制损失曲线，观察初始几个step是否下降。 2. 检查梯度统计信息（均值、方差），看是否有很多零梯度。 3. 分析损失值中“流动维度”（目标非零）和“静止维度”（目标为零）的贡献比例。	1. 尝试更小的学习率（如1e-5）或使用学习率预热。 2. 使用FocalFlowMatchingLoss并调整alpha和gamma。 3. 尝试对时间步t进行非均匀采样（更多中间值）。
生成文本重复或无意义	模型坍缩（Mode Collapse）；先验分布过于尖锐；采样步数太少或求解器不准确。	1. 检查验证集损失是否也停滞或上升。 2. 可视化生成样本的多样性（不同随机种子）。 3. 检查先验分布log_prior，是否少数词概率极高。 4. 增加采样步数num_steps或换用更精确的ODE求解器（如dopri5）。	1. 在损失中加入轻微的正则项（如权重衰减）。 2. 如果使用learnable先验，用KL散度约束其不要偏离均匀分布太远。 3. 在采样时加入少量噪声（类似于扩散模型的随机性）。
GPU内存溢出（OOM）	批次过大；序列过长；词表过大导致输出层矩阵巨大。	1. 使用nvidia-smi监控GPU内存使用。 2. 使用梯度累积（Gradient Accumulation）来减小有效批次大小。 3. 检查模型输出层的参数数量。	1. 减小batch_size和max_seq_len。 2. 启用梯度检查点（Gradient Checkpointing）。 3.必须使用共享嵌入权重。对于超大词表（>10万），考虑自适应Softmax。
采样速度极慢	ODE求解器步数过多；求解器本身效率低；未启用torch.no_grad()。	1. 分析采样代码各步骤耗时。 2. 对比不同求解器（欧拉法 vs 龙格-库塔法）的速度和质量。	1. 对于初步验证，使用欧拉法（euler）并减少步数（如20步）。 2. 确保采样时模型处于eval()模式，并使用@torch.no_grad()装饰器。 3. 研究知识蒸馏，训练一个更小的“教师”网络来模拟ODE流。
提示词条件生成效果差	训练时未充分暴露条件生成任务；固定mask逻辑有误。	1. 检查训练数据中是否包含各种上下文-续写对。 2. 调试采样代码，确认fixed_mask是否正确阻止了提示词位置的更新。	1. 在训练目标中，可以随机mask掉序列后半部分，让模型学习从前文预测后文（类似BERT的MLM，但是CFM形式）。 2. 在采样时，对于提示词位置，直接将速度场v_theta设为零。

4.3 性能优化与生产环境考量

1. 混合精度训练：使用torch.cuda.amp自动混合精度，显著减少显存占用并加速训练。
2. 分布式训练：对于超大规模模型和数据，使用accelerate或deepspeed进行多卡、多机训练。
3. 高效的DataLoader：使用datasets库和自定义迭代器，确保数据加载不成为瓶颈。
4. 模型量化与推理优化：训练完成后，可以使用动态量化或静态量化来压缩模型，并使用TorchScript或ONNX进行导出，以优化推理速度。
5. 监控与日志：集成TensorBoard或W&B，记录损失曲线、生成样本、硬件利用率等，便于长期追踪和问题诊断。

5. 总结与扩展方向

扩展分类流映射的规模，本质上是将一种优雅的生成建模理论应用于现实世界的高维离散数据问题。成功的关键在于平衡理论保真度与工程可行性。通过共享嵌入权重、改进损失函数、设计数据驱动的先验以及采用层次化输出结构，我们可以克服大词表带来的计算挑战。

本文提供的实现是一个起点。要将其应用于真正的生产级语言模型，还需要在以下几个方面进行深入探索：

• 更高效的采样算法：研究基于线性多步法或预测-校正器的快速ODE求解器，在保证质量的前提下将采样步数降至个位数。
• 与其他生成范式的结合：探索CFMs与自回归（Autoregressive）模型的结合，例如用CFM生成段落或句子的全局轮廓，再用自回归模型进行细节填充。
• 条件生成与控制：扩展模型以接受更复杂的条件输入，如情感标签、文体风格、关键词等，实现可控文本生成。
• 跨模态应用：将CFMs的思想应用于图像-文本、音频-文本等多模态生成任务，学习跨域的概率流。

流匹配为生成模型提供了一条通向快速、高质量采样的新路径。尽管在扩展过程中会遇到诸多挑战，但通过持续的技术优化和工程实践，分类流映射有望在文本生成、代码生成等大规模离散数据建模领域发挥越来越重要的作用。在实际项目中，建议从一个中等规模的词表（如1万词）开始，验证整个流程，再逐步向更大规模扩展，并密切关注训练动态和生成质量之间的平衡。