Doing

简易自注意力机制

要求：手写 Scaled Dot-Product Attention，含 Q/K/V 线性变换及 mask 处理。

核心代码（PyTorch风格，但底层可用矩阵乘法实现）：

import torch
import torch.nn.functional as F
import math

def scaled_dot_product_attention(Q, K, V, mask=None):
    # Q,K,V 形状: (batch, heads, seq_len, d_k)
    d_k = Q.size(-1)
    scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k)
    if mask is not None:
        # mask中True位置置为极小值
        scores = scores.masked_fill(mask == 0, float('-inf'))
    attn = F.softmax(scores, dim=-1)
    output = torch.matmul(attn, V)
    return output, attn

class SimpleSelfAttention(torch.nn.Module):
    def __init__(self, d_model, n_heads):
        super().__init__()
        assert d_model % n_heads == 0
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads
        self.W_q = torch.nn.Linear(d_model, d_model)
        self.W_k = torch.nn.Linear(d_model, d_model)
        self.W_v = torch.nn.Linear(d_model, d_model)
        self.out_proj = torch.nn.Linear(d_model, d_model)

    def split_heads(self, x):
        batch, seq, _ = x.size()
        return x.view(batch, seq, self.n_heads, self.d_k).transpose(1, 2)

    def forward(self, x, mask=None):
        Q = self.split_heads(self.W_q(x))
        K = self.split_heads(self.W_k(x))
        V = self.split_heads(self.W_v(x))
        attn_out, _ = scaled_dot_product_attention(Q, K, V, mask)
        # 合并多头
        attn_out = attn_out.transpose(1, 2).contiguous().view(x.size(0), -1, self.d_model)
        return self.out_proj(attn_out)

关键点：mask 在解码器自回归时使用下三角，值为 -inf 让 softmax 归零；多头拆分与合并时注意维度变换。

---

手写 BPE 分词器

步骤：

1. 初始化词汇表为所有字符。

2. 统计所有相邻符号对的出现频次。

3. 合并频率最高的符号对，加入词汇表。

4. 重复直到词汇表大小达标。 核心代码：

import re, collections

def get_vocab(text):
    # 将文本切分为单词和空格序列，每个单词末尾加</w>
    words = re.findall(r'\w+|\S', text)
    vocab = collections.Counter()
    for word in words:
        vocab[' '.join(list(word)) + ' </w>'] += 1
    return vocab

def learn_bpe(vocab, num_merges):
    merges = {}
    for i in range(num_merges):
        pairs = collections.defaultdict(int)
        for word, freq in vocab.items():
            symbols = word.split()
            for j in range(len(symbols)-1):
                pairs[symbols[j], symbols[j+1]] += freq
        if not pairs:
            break
        best_pair = max(pairs, key=pairs.get)
        merges[best_pair] = i
        # 合并所有出现该对的地方
        new_vocab = {}
        bigram = re.escape(' '.join(best_pair))
        p = re.compile(r'(?<!\S)' + bigram + r'(?!\S)')
        for word in vocab:
            w_out = p.sub(''.join(best_pair), word)
            new_vocab[w_out] = vocab[word]
        vocab = new_vocab
    return merges, vocab

def encode(text, merges):
    # 从字符开始，按学习到的合并顺序依次合并
    symbols = list(text) + ['</w>']
    # 需要将merges按顺序排序，然后循环应用直到无法合并
    sorted_merges = sorted(merges.items(), key=lambda x: x[1])  # 按合并次序
    # 简化版：反复扫描合并
    while True:
        i = 0
        merged = False
        while i < len(symbols)-1:
            if (symbols[i], symbols[i+1]) in merges:
                symbols = symbols[:i] + [symbols[i]+symbols[i+1]] + symbols[i+2:]
                merged = True
            else:
                i += 1
        if not merged:
            break
    return symbols

关键点：处理标点与单词边界（</w>），编码时按学习到的合并顺序进行贪心合并。

---

大模型流式推理接口

要求：逐 token 返回并用 SSE 推送。

代码（使用 FastAPI 示例）：

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import torch

app = FastAPI()

async def token_generator(prompt: str):
    # 模拟模型推理：每次产生一个token
    inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
    past_key_values = None
    input_ids = inputs.input_ids
    for _ in range(max_new_tokens):
        with torch.no_grad():
            outputs = model(input_ids, past_key_values=past_key_values, use_cache=True)
        logits = outputs.logits[:, -1, :]
        next_token_id = torch.argmax(logits, dim=-1).item()
        # 当生成停止词时退出
        if next_token_id == eos_token_id:
            break
        token_text = tokenizer.decode([next_token_id])
        yield f"data: {token_text}\n\n"
        # 更新past_key_values用于下一次迭代
        past_key_values = outputs.past_key_values
        input_ids = torch.tensor([[next_token_id]]).to("cuda")
    yield "data: [DONE]\n\n"

@app.post("/stream")
async def stream(request: Request):
    data = await request.json()
    prompt = data["prompt"]
    return StreamingResponse(token_generator(prompt), media_type="text/event-stream")

关键点：SSE 格式为 data: ...\n\n，利用 KV Cache 避免重复计算，通过生成器实现流式。

---

实现 KV Cache 并估算显存

KV Cache 原理：自回归生成时，将历史 K、V 保存在 past_key_values 元组中。 显存计算：

• 单层 KV Cache 大小 = 2 × batch_size × num_heads × seq_length × d_head × dtype_bytes

• 总大小 = 层数 × 单层大小 例如 LLaMA-7B：32层，heads=32，d_head=128，fp16（2字节），batch=1，seq_len=2048 → 单层 2×1×32×2048×128×2 = 33,554,432 字节 ≈ 32 MB，总约 1 GB。 代码片段（展示 KV Cache 使用）：

past_key_values = None
for _ in range(max_len):
    outputs = model(input_ids, past_key_values=past_key_values, use_cache=True)
    logits = outputs.logits[:, -1, :]
    past_key_values = outputs.past_key_values
    # 采样...

关键点：use_cache=True 启用缓存，在每一轮将新的 K、V 拼接到缓存中。

---

VLLM 部署与压测吞吐

部署命令：

python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Meta-Llama-3-8B \
    --tensor-parallel-size 1 \
    --max-model-len 4096 \
    --gpu-memory-utilization 0.95

压测脚本（使用 vllm 的 benchmark）：

from vllm import LLM, SamplingParams
import time

llm = LLM(model="meta-llama/Meta-Llama-3-8B")
prompts = ["Explain the theory of relativity."] * 100
sampling_params = SamplingParams(temperature=0, max_tokens=256)
start = time.time()
outputs = llm.generate(prompts, sampling_params)
end = time.time()
tokens = sum(len(out.outputs[0].token_ids) for out in outputs)
print(f"Throughput: {tokens/(end-start):.2f} tokens/s")

关键点：Continuous batching 动态合并请求，gpu-memory-utilization 控制显存使用。

---

手写 LoRA 适配器

原理：在预训练权重旁路插入低秩矩阵 A 和 B，前向为 Wx + BAx。 代码：

import torch
import torch.nn as nn

class LoRALayer(nn.Module):
    def __init__(self, in_features, out_features, rank=4, alpha=1):
        super().__init__()
        std_dev = 1 / torch.sqrt(torch.tensor(rank).float())
        self.A = nn.Parameter(torch.randn(in_features, rank) * std_dev)
        self.B = nn.Parameter(torch.zeros(rank, out_features))
        self.alpha = alpha
        self.scaling = alpha / rank

    def forward(self, x):
        # x: (batch, in_features)
        return (x @ self.A @ self.B) * self.scaling

class LoRALinear(nn.Module):
    def __init__(self, linear, rank, alpha):
        super().__init__()
        self.linear = linear  # 原始冻结层
        self.lora = LoRALayer(linear.in_features, linear.out_features, rank, alpha)

    def forward(self, x):
        return self.linear(x) + self.lora(x)

    def merge(self):
        # 训练后将 lora 权重合并回原层
        self.linear.weight.data += (self.lora.A @ self.lora.B).T * self.lora.scaling

关键点：初始化时 A 用高斯噪声，B 为零；合并权重时做转置和缩放。

---

DeepSpeed ZeRO-3 配置

配置文件示例 (ds_config.json)：

{
  "train_batch_size": 16,
  "gradient_accumulation_steps": 4,
  "fp16": {
    "enabled": true
  },
  "zero_optimization": {
    "stage": 3,
    "offload_optimizer": {
      "device": "cpu",
      "pin_memory": true
    },
    "offload_param": {
      "device": "cpu",
      "pin_memory": true
    },
    "overlap_comm": true,
    "contiguous_gradients": true,
    "sub_group_size": 1e9,
    "reduce_bucket_size": 5e8,
    "stage3_prefetch_bucket_size": 5e8,
    "stage3_param_persistence_threshold": 1e6,
    "stage3_max_live_parameters": 1e9,
    "stage3_max_reuse_distance": 1e9,
    "stage3_gather_16bit_weights_on_model_save": true
  },
  "gradient_clipping": 1.0,
  "steps_per_print": 10
}

参数说明：

• stage: 3：ZeRO-3 将优化器状态、梯度和参数都分片。

• offload_optimizer/param：卸载到 CPU，适合显存不足时训练大模型。

• reduce_bucket_size 等：控制通信粒度。 启动命令：deepspeed --num_gpus=8 train.py --deepspeed ds_config.json

---

RAG 问答 Pipeline

使用 LangChain 实现：

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.llms import HuggingFacePipeline
from langchain.chains import RetrievalQA

# 1. 加载文档
loader = TextLoader("knowledge.txt")
documents = loader.load()
# 2. 文本切分
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
docs = splitter.split_documents(documents)
# 3. 向量化并存储
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
vectorstore = Chroma.from_documents(docs, embeddings)
# 4. 构建 QA chain
llm = HuggingFacePipeline.from_model_id(model_id="gpt2", task="text-generation")
qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=vectorstore.as_retriever())
result = qa.run("What is the capital of France?")

关键点：chain_type 可选 "stuff"（直接拼接）、"map_reduce"等；检索器可设定 top_k。

---

滑动窗口长文本分块（段落感知）

要求：保持 Markdown 段落完整性，支持 chunk size 和 overlap。

实现思路：先按段落分割（双换行），再贪心合并段落直到接近 chunk size，重叠部分用上一块最后几个段落实现 overlap。

代码：

def markdown_splitter(text, chunk_size=1000, chunk_overlap=200):
    paragraphs = text.split('\n\n')
    chunks = []
    current_chunk = []
    current_len = 0
    for para in paragraphs:
        para_len = len(para)
        if current_len + para_len > chunk_size and current_chunk:
            # 保存当前 chunk
            chunks.append('\n\n'.join(current_chunk))
            # 重叠处理：保留最后几个段落使得总字符数接近 overlap
            overlap_paras = []
            overlap_len = 0
            for p in reversed(current_chunk):
                if overlap_len + len(p) <= chunk_overlap:
                    overlap_paras.insert(0, p)
                    overlap_len += len(p)
                else:
                    break
            current_chunk = overlap_paras
            current_len = overlap_len
        current_chunk.append(para)
        current_len += para_len
    if current_chunk:
        chunks.append('\n\n'.join(current_chunk))
    return chunks

关键点：确保 chunk 边界不会切断句子，overlap 通过保留尾部段落实现上下文衔接。

---

评估大模型幻觉 & 事实性检测

方案：给定问答对，用大模型判断答案是否与参考文本一致或包含虚构内容。

Prompt 设计：

You are a factuality checker. Given the question and the answer, determine if the answer contains any fabricated information that is not supported by common knowledge. Answer "Yes" if fabricated, "No" if factual.
Question: {question}
Answer: {answer}

批量评估脚本：

def check_factuality(questions, answers, model, tokenizer):
    results = []
    for q, a in zip(questions, answers):
        prompt = f"...{q}...{a}..."
        inputs = tokenizer(prompt, return_tensors="pt")
        outputs = model.generate(**inputs, max_new_tokens=5)
        decision = tokenizer.decode(outputs[0], skip_special_tokens=True).strip().lower()
        results.append("yes" in decision)
    return sum(results)/len(results)

指标：幻觉率 = 判为“虚构”的比例；可结合人工标注计算准确率。

---

奖励模型训练（RLHF Pairwise Loss）

损失函数：

def pairwise_ranking_loss(rewards_chosen, rewards_rejected):
    # rewards: shape (batch_size,)
    return -torch.mean(torch.log(torch.sigmoid(rewards_chosen - rewards_rejected)))

# 训练循环核心
for batch in dataloader:
    chosen_inputs, rejected_inputs = batch
    # 获取奖励分数，通常取最后一个token的表示经过线性层
    rewards_chosen = reward_model(chosen_inputs)
    rewards_rejected = reward_model(rejected_inputs)
    loss = pairwise_ranking_loss(rewards_chosen, rewards_rejected)
    loss.backward()

关键点：奖励模型常基于基座模型，最后加一层 nn.Linear(hidden_size, 1)。

---

支持 Function Calling 的 Agent 循环

伪代码：

messages = [system_prompt, user_query]
while True:
    response = model.generate(messages, tools=tools)
    if response has no tool_calls:
        break
    for tool_call in response.tool_calls:
        func_name = tool_call.name
        args = tool_call.arguments
        result = call_function(func_name, args)
        messages.append({"role": "tool", "content": result, "tool_call_id": tool_call.id})
    messages.append(response.assistant_message)

代码关键：解析模型返回的 JSON 工具调用，执行本地函数，将结果注入消息列表，继续循环直到模型输出最终答案。

---

Top-P (Nucleus) 采样

算法：

1. 将 logits 转换为概率（softmax）。

2. 按概率降序排序，累加概率直到超过阈值 p。

3. 将剩余 token 的概率置零，重新归一化，然后多项式采样。 代码：

def top_p_sampling(logits, p):
    probs = F.softmax(logits, dim=-1)
    sorted_probs, sorted_indices = torch.sort(probs, descending=True)
    cumulative_probs = torch.cumsum(sorted_probs, dim=-1)
    # 保留累积概率<=p的token，但至少保留一个
    cutoff_idx = (cumulative_probs > p).nonzero(as_tuple=True)
    if len(cutoff_idx[0]) > 0:
        last_included = cutoff_idx[0][0]
    else:
        last_included = sorted_probs.size(-1) - 1
    sorted_probs[last_included+1:] = 0
    sorted_probs /= sorted_probs.sum(dim=-1, keepdim=True)
    sampled_idx = torch.multinomial(sorted_probs, 1)
    return sorted_indices.gather(-1, sampled_idx).item()

---

P-Tuning v2 给 ChatGLM 添加指令遵循

原理：在每层 Transformer 的输入前添加可训练的连续 prefix token，只优化这些 prefix。

核心代码（使用 transformers）：

from transformers import AutoModel, AutoConfig
import torch.nn as nn

class PtuningV2Model(nn.Module):
    def __init__(self, model_name, pre_seq_len=128):
        super().__init__()
        config = AutoConfig.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name)
        for param in self.model.parameters():
            param.requires_grad = False
        self.pre_seq_len = pre_seq_len
        self.prefix_encoder = nn.Embedding(pre_seq_len, config.hidden_size)
        self.dropout = nn.Dropout(0.1)

    def forward(self, input_ids, attention_mask):
        batch_size = input_ids.size(0)
        prefix = self.prefix_encoder.weight.unsqueeze(0).expand(batch_size, -1, -1)
        # 获取模型的嵌入层
        inputs_embeds = self.model.embeddings(input_ids)
        # 在序列前拼接 prefix
        inputs_embeds = torch.cat([prefix, inputs_embeds], dim=1)
        # 调整 attention_mask
        prefix_mask = torch.ones(batch_size, self.pre_seq_len).to(input_ids.device)
        attention_mask = torch.cat([prefix_mask, attention_mask], dim=1)
        outputs = self.model(inputs_embeds=inputs_embeds, attention_mask=attention_mask)
        return outputs

关键点：prefix encoder 可更复杂（如 LSTM），但 Embedding 已可工作；训练时只更新 prefix 参数。

---

BERT MLM 与 GPT 自回归损失

MLM 损失：

def mlm_loss(logits, labels, mask):
    # logits: (batch, seq, vocab), labels: (batch, seq) with -100 for ignored
    loss_fct = nn.CrossEntropyLoss(ignore_index=-100)
    return loss_fct(logits.view(-1, vocab_size), labels.view(-1))

GPT 自回归损失：

def lm_loss(logits, labels):
    # logits: (batch, seq, vocab), labels: (batch, seq) 目标为下一个token
    shift_logits = logits[..., :-1, :].contiguous()
    shift_labels = labels[..., 1:].contiguous()
    return F.cross_entropy(shift_logits.view(-1, vocab_size), shift_labels.view(-1))

差异：MLM 预测被 mask 的 token，双向注意力；GPT 单向预测下一个 token。

---

大规模数据去重 (MinHash)

步骤：

1. 将文本转换为 n-gram 集合（如 5-gram）。

2. 使用多个哈希函数计算 MinHash 签名。

3. 使用 LSH（Locality Sensitive Hashing）分桶，桶内相似对计算 Jaccard 相似度。

4. 相似度超过阈值则去重。 核心（使用 datasketch 库）：

from datasketch import MinHash, MinHashLSH
lsh = MinHashLSH(threshold=0.8, num_perm=128)
for i, text in enumerate(corpus):
    m = MinHash(num_perm=128)
    for word in text.split():
        m.update(word.encode('utf8'))
    lsh.insert(i, m)
# 查询重复
for i, text in enumerate(corpus):
    m = MinHash(num_perm=128)
    for word in text.split():
        m.update(word.encode('utf8'))
    result = lsh.query(m)
    # result 包含相似文档 id

关键：num_perm（哈希函数个数）影响精度；threshold 一般 0.8～0.9。

---

AWQ 量化部署与评测

AWQ 量化步骤：

# 使用 AutoAWQ 库
from awq import AutoAWQForCausalLM

quant_config = { "zero_point": True, "q_group_size": 128, "w_bit": 4, "version": "GEMM" }
model = AutoAWQForCausalLM.from_pretrained("meta-llama/Llama-2-7b")
model.quantize(tokenizer, quant_config=quant_config)
model.save_quantized("llama-2-7b-awq")

评测：在同一测试集上对比原始模型和量化模型的 PPL 和生成质量。

from transformers import AutoModelForCausalLM
import time
original = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b", device_map="auto")
# 计算困惑度
def evaluate_ppl(model, dataloader):
    ...

关键：AWQ 通过激活感知权重量化保护重要通道；群组量化 group_size 影响精度与速度。

---

安全护栏：检测并拦截有害输出

模块设计：

import re
class SafetyGuard:
    def __init__(self):
        self.blacklist = ["violence", "hack", ...]
        self.regex_pattern = re.compile(r'\b(harmful pattern)\b')
        self.secondary_prompt = "Please review the following content and determine if it is safe. Answer YES or NO."

    def check(self, text):
        if any(word in text.lower() for word in self.blacklist):
            return False, "blacklist"
        if self.regex_pattern.search(text):
            return False, "regex"
        # 二次过滤用小型分类模型
        score = safety_classifier(text)
        if score > 0.8:
            return False, "classifier"
        return True, "safe"

关键：多层防护，黑名单+正则+模型审核，可结合提示词工程进行内容过滤。

---

多轮对话数据清洗

从原始日志清洗为 ShareGPT 格式：

import json
def clean_conversation(raw_log, min_turns=2, max_len=2048):
    turns = []
    for msg in raw_log:
        if not msg['text'] or len(msg['text']) < 2:
            continue
        # 去广告、无意义内容
        if 'click here' in msg['text'].lower():
            break
        turns.append({"from": msg['role'], "value": msg['text']})
    if len(turns) < min_turns * 2:
        return None
    # 检查总长度
    total_len = sum(len(t['value']) for t in turns)
    if total_len > max_len:
        # 从后往前裁剪到 max_len
        ...
    return {"conversations": turns}

# 批量处理
cleaned = []
for session in raw_sessions:
    conv = clean_conversation(session)
    if conv:
        cleaned.append(conv)
with open("sharegpt.json", "w") as f:
    json.dump(cleaned, f, indent=2)

关键：过滤广告、截断过长、保证最小轮次。

---

RoPE 外推：NTK-aware 缩放

原理：将旋转位置编码的频率进行动态缩放，使模型能处理更长的上下文。

代码实现：

import torch
def ntk_aware_scaling(seq_len, base=10000.0, alpha=1.0):
    # 原始频率：base^{-2i/d}
    # 缩放：new_base = base * alpha ^ (d / (d-2))
    d = 128  # head dimension
    new_base = base * (alpha ** (d / (d - 2)))
    # 生成位置编码矩阵，使用new_base
    inv_freq = 1.0 / (new_base ** (torch.arange(0, d, 2).float() / d))
    # 应用RoPE
    ...
    return cos, sin

# 在推理时根据目标长度动态设置alpha = target_length / original_length

在 Transformers 中使用：LlamaRotaryEmbedding 可设置 scaling_factor 或重写 forward 实现 NTK。

---

以上每道题的解答都包含了核心实现思路、关键代码和工程细节。你可以根据面试岗位的不同（推理优化、数据工程、微调对齐等）选择重点深入追问。

如果需要某道题更完整的项目级代码或深度理论解析，随时告诉我。

---