第2章：预备知识速览

        [CLS]  机器  学习  教程
[CLS]    0.25  0.35  0.35  0.05
机器     0.10  0.30  0.50  0.10
学习     0.10  0.40  0.40  0.10  
教程     0.20  0.20  0.20  0.40

for query_type in ['short', 'long', 'exact', 'semantic']:
    for model in [Model_C, Model_B, Model_H]:
        results = evaluate(model, query_type)
        record_metrics(results)

def find_optimal_dimension():
    dimensions = [128, 256, 512, 768, 1024, 1536, 2048, 3072]
    results = {}

    for d in dimensions:
        model = train_embedding_model(dim=d)

        # 评估指标
        metrics = {
            'retrieval_acc': evaluate_retrieval(model),
            'clustering_score': evaluate_clustering(model),
            'inference_time': measure_latency(model),
            'storage_gb': calculate_storage(model, num_docs=1e9),
            'anisotropy': measure_anisotropy(model)
        }

        # 综合得分（加权）
        score = (
            0.4 * metrics['retrieval_acc'] +
            0.2 * metrics['clustering_score'] -
            0.2 * normalize(metrics['inference_time']) -
            0.1 * normalize(metrics['storage_gb']) +
            0.1 * (1 - metrics['anisotropy'])
        )

        results[d] = (metrics, score)

    return results

维度  | 检索精度 | 延迟(ms) | 存储(GB) | 综合得分
128   | 0.78    | 5        | 0.5      | 0.72
256   | 0.84    | 8        | 1.0      | 0.78  
512   | 0.89    | 15       | 2.0      | 0.82
768   | 0.91    | 22       | 3.0      | 0.83
1024  | 0.92    | 30       | 4.0      | 0.84
1536  | 0.93    | 45       | 6.0      | 0.83 ← OpenAI选择
2048  | 0.935   | 60       | 8.0      | 0.81
3072  | 0.94    | 90       | 12.0     | 0.78

class MultiModalInput:
    def __init__(self, d_model=768):
        # 文本编码器
        self.text_tokenizer = BPETokenizer()
        self.text_embed = nn.Embedding(vocab_size, d_model)

        # 图像编码器
        self.image_encoder = ViT(patch_size=16, d_model=d_model)

        # 模态类型嵌入
        self.modality_embed = nn.Embedding(2, d_model)  # 0:text, 1:image

    def forward(self, text, image):
        # 文本处理
        text_tokens = self.text_embed(self.text_tokenizer(text))
        text_tokens += self.modality_embed(torch.zeros(...))

        # 图像处理（分割成patches）
        image_patches = self.image_encoder(image)
        image_patches += self.modality_embed(torch.ones(...))

        # 拼接
        return torch.cat([text_tokens, image_patches], dim=1)

class MultiModalPositionalEncoding:
    def __init__(self):
        # 1D位置编码用于文本
        self.text_pos = SinusoidalPE()

        # 2D位置编码用于图像
        self.image_pos = Learned2DPE()

    def forward(self, tokens, modality_mask):
        # 根据模态类型应用不同的位置编码
        text_mask = (modality_mask == 0)
        image_mask = (modality_mask == 1)

        tokens[text_mask] += self.text_pos(tokens[text_mask])
        tokens[image_mask] += self.image_pos(tokens[image_mask])

        return tokens

class CrossModalAttention(nn.Module):
    def __init__(self, d_model, n_heads):
        super().__init__()
        self.n_heads = n_heads
        self.d_head = d_model // n_heads

        # 模态特定的投影
        self.q_proj = nn.ModuleDict({
            'text': nn.Linear(d_model, d_model),
            'image': nn.Linear(d_model, d_model)
        })
        self.k_proj = nn.Linear(d_model, d_model)
        self.v_proj = nn.Linear(d_model, d_model)

        # 模态交互门控
        self.gate = nn.Linear(d_model * 2, d_model)

    def forward(self, x, modality_mask):
        # 根据模态类型使用不同的查询投影
        q_text = self.q_proj['text'](x[modality_mask == 0])
        q_image = self.q_proj['image'](x[modality_mask == 1])

        # 统一的键值投影
        k = self.k_proj(x)
        v = self.v_proj(x)

        # 计算注意力...
        # 应用门控机制调节跨模态交互强度

输入层:
├── 文本: "红色跑车" → Tokenize → Embed
└── 图像: [224×224×3] → Patches → Linear Projection

早期融合层 (Layer 1-4):

- 模态内自注意力为主
- 轻量级跨模态交互

中期交互层 (Layer 5-8):

- 平衡的模态内/跨模态注意力
- 学习对齐表示

后期融合层 (Layer 9-12):

- 深度跨模态理解
- 统一的多模态表示

输出层:

- 池化策略：[CLS] token或平均池化
- 生成文档ID序列

# 多模态对比学习
loss = (
    0.4 * text_to_doc_loss +
    0.3 * image_to_doc_loss +
    0.3 * multimodal_to_doc_loss
)

# 模态缺失训练（提高鲁棒性）
if random.random() < 0.2:
    # 随机mask掉一个模态
    if random.random() < 0.5:
        image = None
    else:
        text = None

def scheduled_sampling_training(model, epoch, max_epochs):
    # 线性衰减采样概率
    teacher_forcing_ratio = max(0.5, 1.0 - epoch / max_epochs)

    for batch in dataloader:
        outputs = []
        input_seq = batch['doc_id']

        for t in range(len(input_seq)):
            if random.random() < teacher_forcing_ratio:
                # Teacher forcing: 使用真实token
                next_input = input_seq[t]
            else:
                # Student forcing: 使用模型预测
                next_input = model.predict(outputs)

            output = model(next_input, encoder_output)
            outputs.append(output)

        loss = compute_loss(outputs, input_seq)

def prefix_constrained_training(model):
    """训练时模拟推理时的前缀树约束"""

    # 构建文档ID前缀树
    prefix_tree = build_prefix_tree(all_doc_ids)

    for batch in dataloader:
        # 随机选择错误注入点
        error_position = random.randint(1, max_length)

        # 正常训练到错误点
        outputs = teacher_forcing(model, batch, until=error_position)

        # 从错误点开始，只允许生成有效的续接
        for t in range(error_position, max_length):
            logits = model(outputs[:t])

            # 应用前缀树约束
            valid_tokens = prefix_tree.get_valid_continuations(outputs[:t])
            masked_logits = mask_invalid_tokens(logits, valid_tokens)

            next_token = sample(masked_logits)
            outputs.append(next_token)

def reinforce_finetuning(model, reward_fn):
    """使用检索质量作为奖励信号"""

    for query in queries:
        # 采样生成多个文档ID
        generated_ids = []
        log_probs = []

        for _ in range(num_samples):
            doc_id, log_p = model.sample(query, return_log_prob=True)
            generated_ids.append(doc_id)
            log_probs.append(log_p)

        # 计算奖励（检索质量）
        rewards = []
        for doc_id in generated_ids:
            if doc_id in valid_doc_ids:
                doc = retrieve(doc_id)
                reward = compute_relevance(query, doc)
            else:
                reward = -1  # 无效ID惩罚
            rewards.append(reward)

        # REINFORCE更新
        baseline = np.mean(rewards)
        loss = -sum((r - baseline) * log_p 
                   for r, log_p in zip(rewards, log_probs))

def mixed_objective_training(model):
    """结合多个训练信号"""

    # 1. 标准的teacher forcing损失
    tf_loss = teacher_forcing_loss(model, batch)

    # 2. 序列级别的损失（BLEU/编辑距离）
    generated = model.generate(batch['query'])
    seq_loss = sequence_level_loss(generated, batch['doc_id'])

    # 3. 对比学习损失（正确ID vs 错误ID）
    pos_score = model.score(batch['query'], batch['doc_id'])
    neg_scores = model.score(batch['query'], negative_samples)
    contrastive_loss = max(0, margin - pos_score + max(neg_scores))

    # 组合损失
    total_loss = (
        0.5 * tf_loss +
        0.3 * seq_loss +
        0.2 * contrastive_loss
    )