关于自学AI工程师技术体系的总结

总参数：671B激活参数：37B/token核心机制：动态路由 - 每个token仅激活TopK个专家优势：推理成本降低约18倍（相比稠密模型）

# 简化的路由逻辑def expert_routing(x, gate_network, num_experts=8, top_k=2):    """    x: 输入token的embedding [batch, seq_len, hidden_dim]    返回：加权专家输出    """    gate_logits = gate_network(x)  # [batch, seq_len, num_experts]    top_k_gates, top_k_indices = torch.topk(gate_logits, top_k)    top_k_gates = softmax(top_k_gates, dim=-1)    output = 0    for i in range(top_k):        expert_id = top_k_indices[:, :, i]        expert_output = experts[expert_id](x)        output += top_k_gates[:, :, i:i+1] * expert_output    return output

aux_loss = load_balancing_loss(gate_logits)  # 鼓励均匀分配total_loss = task_loss + α * aux_loss

IFEval（Instruction Following Eval）├── 精确格式要求（如JSON输出）├── 多步骤约束（如"先总结再翻译"）└── 反事实指令（测试是否盲目服从）HumanEval+├── 隐藏测试用例├── 边界条件检测└── 代码可读性评分

class FinancialComplianceEval:    def __init__(self):        self.criteria = {            'hallucination_rate': self.check_factual_grounding,            'risk_disclosure': self.verify_disclaimers,            'regulatory_alignment': self.check_compliance_terms        }    def check_factual_grounding(self, response, knowledge_base):        """验证每个声明是否有文档支持"""        claims = extract_claims(response)        supported = [kb_search(claim, knowledge_base) for claim in claims]        return sum(supported) / len(claims)

任务：{问题描述}要求：1. 列出解决该问题需要的关键信息2. 逐步推导（标注每步的中间结果）3. 验证答案的合理性开始推理：

def tree_of_thought(problem, model, max_branches=3, max_depth=4):    root = ThoughtNode(problem)    for depth in range(max_depth):        # 生成候选思路        candidates = model.generate_thoughts(root.state, n=max_branches)        # 自我评估打分        scores = [model.evaluate_thought(c) for c in candidates]        # 选择最优路径        best_idx = argmax(scores)        root = root.expand(candidates[best_idx])        if is_terminal(root.state):            return root.get_solution()

RAG Pipeline:Query → [查询改写] → [向量检索] → [重排序] → [上下文压缩] → LLM生成关键模块性能指标：├── 检索召回率（Recall@K）：相关文档是否被检索到├── 精确率（Precision@K）：检索结果的噪声比例├── 延迟分解：│   ├── 嵌入计算：~50ms│   ├── 向量搜索：~30ms（FAISS GPU索引）│   └── LLM生成：~2000ms└── 成本：主要在嵌入API调用（每次查询$0.0001-0.0004）

# 传统RAGquery = "DeepSeek-V3的MoE架构细节"results = vector_db.search(embed(query))# HyDE改进hypothetical_doc = llm.generate(    f"写一段详细解释{query}的文档")results = vector_db.search(embed(hypothetical_doc))  # 用生成文档检索

def self_rag_generate(query, llm, retriever):    response = ""    current_query = query    for step in range(max_steps):        # 判断是否需要检索        need_retrieval = llm.decide_retrieval(current_query)        if need_retrieval:            docs = retriever.search(current_query)            context = rerank_and_filter(docs)        else:            context = None        # 生成片段        chunk = llm.generate(current_query, context)        # 自我验证        if llm.verify_factuality(chunk, context):            response += chunk            current_query = update_query(chunk)  # 基于已生成内容更新查询        else:            # 重新检索或修正            continue    return response

Thought: 分析当前状态，决定下一步行动Action: 调用工具（搜索/计算/API调用）Observation: 获取行动结果... (循环直到完成任务)Answer: 最终输出

class AgentExecutor:    def __init__(self, llm, tools, max_iterations=10):        self.llm = llm        self.tools = {t.name: t for t in tools}        self.max_iterations = max_iterations        self.memory = []  # 防止重复行动    def run(self, task):        for i in range(self.max_iterations):            # 生成思考+行动            response = self.llm.generate(                self.construct_prompt(task, self.memory)            )            thought, action, action_input = self.parse(response)            # 检测循环            if (action, action_input) in self.memory:                return self.handle_loop()            # 执行行动            observation = self.tools[action].run(action_input)            self.memory.append((thought, action, observation))            # 判断是否完成            if self.is_final_answer(observation):                return observation

Memory Hierarchy:├── Core Memory（类似热数据缓存）│   ├── 人物设定（Persona）│   └── 关键事实（Human Info）├── Archival Memory（类似磁盘持久化）│   └── 长期知识库（向量数据库）└── Recall Memory（最近交互）    └── 滑动窗口（最近N轮对话）

class MemoryManager:    def update_memory(self, new_info):        # 使用LLM判断重要性（而非传统LRU的访问频率）        importance_score = self.llm.rate_importance(new_info)        if importance_score > threshold:            # Core已满，需要替换            if len(self.core_memory) >= MAX_CORE:                victim = self.select_victim()                self.archival_memory.store(victim)            self.core_memory.add(new_info)        else:            self.archival_memory.store(new_info)

# 原始HumanEvaldef test_function(candidate):    assert candidate([1,2,3]) == 6# HumanEval+增强def test_function_plus(candidate):    # 基础测试    assert candidate([1,2,3]) == 6    # 边界情况    assert candidate([]) == 0    assert candidate([-1,-2]) == -3    # 类型检查    with pytest.raises(TypeError):        candidate("not a list")    # 性能测试    assert time_execution(candidate, large_input) < 1.0

class SafeCodeGenerator:    def __init__(self, base_model):        self.model = base_model        self.sanitizer = CodeSanitizer()        self.license_checker = LicenseChecker()    def generate(self, prompt):        code = self.model.generate(prompt)        # 静态分析        if self.sanitizer.has_dangerous_patterns(code):            return self.regenerate_safe(prompt)        # 许可证检测        license = self.license_checker.identify(code)        if license in INCOMPATIBLE_LICENSES:            code = self.add_attribution(code, license)        return code

# InfoNCE loss的正确实现def contrastive_loss(image_embeddings, text_embeddings, temperature=0.07):    """    image_embeddings: [batch_size, embed_dim]    text_embeddings: [batch_size, embed_dim]    """    # 计算相似度矩阵    logits = (image_embeddings @ text_embeddings.T) / temperature    # 对角线为正样本（匹配的图文对）    batch_size = logits.shape[0]    labels = torch.arange(batch_size)    # 双向对比损失    loss_i2t = F.cross_entropy(logits, labels)  # 图像→文本    loss_t2i = F.cross_entropy(logits.T, labels)  # 文本→图像    return (loss_i2t + loss_t2i) / 2

输入处理：├── 图像 → Vision Encoder → 视觉tokens [N个]├── 文本 → Text Tokenizer → 文本tokens [M个]└── 融合：视觉tokens作为特殊tokens插入文本序列Transformer处理：[BOS] <img_1> ... <img_N> 描述这张图片 [EOS]       ↑ 视觉tokens ↑

传统ASR（如Kaldi）：声学模型 → 发音词典 → 语言模型 → 解码器（需要大量人工设计）Whisper：音频 → Encoder → Decoder → 文本（纯数据驱动，多任务训练）

# Whisper的特殊token设计<|startoftranscript|><|zh|>  # 语言标识<|transcribe|>  # 任务类型（或translate）<|notimestamps|>  # 是否需要时间戳实际转写内容...<|endoftext|>

class StreamingWhisper:    def __init__(self, chunk_duration=2.0):        self.model = load_whisper_model()        self.chunk_duration = chunk_duration  # 每次处理2秒音频        self.overlap = 0.5  # 重叠0.5秒避免截断    def transcribe_stream(self, audio_stream):        buffer = []        for audio_chunk in audio_stream:            buffer.append(audio_chunk)            # 累积到足够长度            if len(buffer) >= self.chunk_duration:                audio = np.concatenate(buffer)                # 转写                result = self.model.transcribe(audio)                yield result['text']                # 保留重叠部分                buffer = buffer[-int(self.overlap * len(buffer)):]

架构流程：文本 → Text Encoder → 条件变分自编码器 → Flow-based生成器 → HiFi-GAN声码器 → 音频核心创新：1. 变分推断：学习文本到语音的概率分布p(audio|text)2. 对抗训练：判别器提升音质自然度3. 端到端：无需手工标注韵律/音素时长

class VITS(nn.Module):    def __init__(self):        self.text_encoder = TextEncoder()  # 处理音素序列        self.posterior_encoder = PosteriorEncoder()  # 从真实音频提取隐变量        self.flow = ResidualCouplingBlock()  # 标准化流模型        self.decoder = HiFiGANGenerator()  # 生成波形        self.discriminator = MultiPeriodDiscriminator()  # 对抗训练    def forward(self, text, audio):        # 文本编码        text_hidden = self.text_encoder(text)        # 后验编码（训练时）        z_posterior = self.posterior_encoder(audio)        # Flow转换：后验 → 先验        z_prior = self.flow(z_posterior, text_hidden)        # 生成音频        audio_pred = self.decoder(z_posterior)  # 训练时用真实z        # 对抗损失        disc_loss = self.discriminator(audio, audio_pred)        return audio_pred, kl_loss(z_posterior, z_prior), disc_loss    def infer(self, text):        """推理时：文本 → 先验采样 → 生成"""        text_hidden = self.text_encoder(text)        z = self.sample_prior(text_hidden)  # 从先验分布采样        audio = self.decoder(z)        return audio

传统TTS：文本 → 确定性特征 → 音频（缺乏多样性）VITS：文本 → 概率分布 → 采样 → 音频      └─ 变分自编码器学习这个分布

最大化：p(audio|text) = ∫ p(audio|z,text) p(z|text) dz通过变分推断优化：ELBO = E[log p(audio|z,text)] - KL(q(z|audio,text) || p(z|text))      ↑ 重构质量              ↑ 正则化项

def clone_voice(reference_audio, target_text):    """    reference_audio: 3-10秒目标说话人音频    target_text: 要合成的文本    """    # 1. 提取说话人嵌入    speaker_embedding = speaker_encoder(reference_audio)    # 2. 条件生成（无需微调）    audio = vits_model.infer(        text=target_text,        speaker_emb=speaker_embedding    )    return audio

# 流式生成（边合成边播放）class StreamingVITS:    def __init__(self, chunk_size=512):        self.model = load_vits()        self.chunk_size = chunk_size  # 每次生成的采样点数    def generate_stream(self, text):        phonemes = text_to_phoneme(text)        # 分块处理        for i in range(0, len(phonemes), self.chunk_size):            chunk = phonemes[i:i+self.chunk_size]            audio_chunk = self.model.infer_chunk(chunk)            yield audio_chunk  # 流式返回

q(x_t | x_{t-1}) = N(x_t; √(1-β_t)·x_{t-1}, β_t·I)其中：- β_t: 噪声调度（noise schedule），通常β_1=0.0001, β_T=0.02- x_0: 原始图像- x_T: 纯高斯噪声

q(x_t | x_0) = N(x_t; √ᾱ_t·x_0, (1-ᾱ_t)·I)其中 ᾱ_t = ∏_{i=1}^t (1-β_i)这意味着可以直接从x_0一步到达任意x_t：x_t = √ᾱ_t·x_0 + √(1-ᾱ_t)·ε,  ε ~ N(0,I)

p_θ(x_{t-1} | x_t) = N(x_{t-1}; μ_θ(x_t, t), Σ_θ(x_t, t))简化：固定方差 Σ_θ = σ_t²·I，只学习均值μ_θ

def ddpm_loss_simplified(model, x_0, timesteps=1000):    """    核心思想：预测添加的噪声，而非直接预测x_0    """    # 1. 随机采样时间步    t = torch.randint(0, timesteps, (x_0.shape[0],))    # 2. 采样噪声    noise = torch.randn_like(x_0)    # 3. 前向加噪（一步到位）    alpha_bar_t = get_alpha_bar(t)  # 预计算的ᾱ_t    x_t = torch.sqrt(alpha_bar_t) * x_0 + torch.sqrt(1 - alpha_bar_t) * noise    # 4. 预测噪声    noise_pred = model(x_t, t)    # 5. MSE损失    loss = F.mse_loss(noise_pred, noise)    return loss

@torch.no_grad()def ddpm_sample(model, shape, timesteps=1000):    """    从纯噪声逐步去噪生成图像    """    device = next(model.parameters()).device    # 初始化为纯噪声    x_t = torch.randn(shape, device=device)    # 逆向迭代    for t in reversed(range(timesteps)):        # 预测噪声        t_batch = torch.full((shape[0],), t, device=device, dtype=torch.long)        noise_pred = model(x_t, t_batch)        # 计算去噪后的x_{t-1}        alpha_t = get_alpha(t)        alpha_bar_t = get_alpha_bar(t)        # 均值        mu = (x_t - (1 - alpha_t) / torch.sqrt(1 - alpha_bar_t) * noise_pred) / torch.sqrt(alpha_t)        # 添加随机性（t>0时）        if t > 0:            sigma_t = get_sigma(t)            z = torch.randn_like(x_t)            x_t = mu + sigma_t * z        else:            x_t = mu  # 最后一步不加噪声    return x_t

原始DDPM问题：- 512×512×3图像 = 786,432维空间- 每步去噪需要大量计算Stable Diffusion解决方案：图像(512×512×3) → VAE Encoder → 潜在表示(64×64×4)                                       ↓ (压缩64×)                                  扩散过程(UNet)                                      ↓                  生成图像 ← VAE Decoder ← 去噪后的潜在表示

class StableDiffusionPipeline:    def __init__(self):        self.vae = AutoencoderKL()  # 压缩比8×8=64        self.unet = UNet2DConditionModel()  # 扩散模型        self.text_encoder = CLIPTextModel()  # 文本条件        self.scheduler = DDPMScheduler()  # 噪声调度器    def encode_prompt(self, prompt):        """文本 → 嵌入向量"""        tokens = self.tokenizer(prompt)        text_embeddings = self.text_encoder(tokens)        return text_embeddings    def __call__(self, prompt, num_steps=50, guidance_scale=7.5):        # 1. 文本编码        text_emb = self.encode_prompt(prompt)        # 2. 初始化潜在噪声        latents = torch.randn(1, 4, 64, 64)        # 3. 去噪循环        for t in self.scheduler.timesteps:            # Classifier-Free Guidance            latent_input = torch.cat([latents] * 2)            noise_pred = self.unet(                latent_input,                 t,                 encoder_hidden_states=text_emb  # 条件注入            )            # 分离条件/无条件预测            noise_pred_uncond, noise_pred_text = noise_pred.chunk(2)            noise_pred = noise_pred_uncond + guidance_scale * (                noise_pred_text - noise_pred_uncond            )            # 更新latents            latents = self.scheduler.step(noise_pred, t, latents).prev_sample        # 4. 解码为图像        image = self.vae.decode(latents)        return image

联合训练条件/无条件模型：- 训练时随机丢弃条件（10-20%概率）- 推理时计算：ε̃ = ε_uncond + s·(ε_cond - ε_uncond)  其中 s 是guidance scale（通常7-15）

DDPM：随机采样过程DDIM：确定性采样，可跳步关键公式：x_{t-Δt} = √ᾱ_{t-Δt}·(x_t - √(1-ᾱ_t)·ε_θ(x_t,t))/√ᾱ_t            + √(1-ᾱ_{t-Δt})·ε_θ(x_t,t)允许Δt>1，例如t=[999,899,799,...,99,0]仅需10步

传统微调：W_{新} = W_{预训练} + ΔW其中 ΔW ∈ R^{d×k} 是全秩矩阵LoRA假设：ΔW实际上是低秩的ΔW ≈ B·A^T其中 A ∈ R^{d×r}, B ∈ R^{k×r}, r << min(d,k)

模型结构：- 隐藏维度 d_model = 4096- 32层Transformer- 每层有4个权重矩阵：Q, K, V, O（attention）+ 2个MLP单层参数（仅对attention应用LoRA）：- 原始：4 × (4096 × 4096) = 67.1M- LoRA (r=8)：4 × (4096×8 + 8×4096) = 262K全模型LoRA参数：- 32层 × 262K = 8.4M- 占比：8.4M / 7B = 0.12%

class LoRALinear(nn.Module):    def __init__(        self,         in_features,         out_features,         rank=8,         alpha=16,        dropout=0.0    ):        super().__init__()        self.rank = rank        self.alpha = alpha        # LoRA权重（初始化策略很重要）        self.lora_A = nn.Parameter(torch.zeros(in_features, rank))        self.lora_B = nn.Parameter(torch.zeros(rank, out_features))        # Kaiming初始化A，B初始化为0        nn.init.kaiming_uniform_(self.lora_A, a=math.sqrt(5))        # Dropout防止过拟合        self.dropout = nn.Dropout(dropout)        # 缩放因子（平衡预训练权重与LoRA权重）        self.scaling = alpha / rank    def forward(self, x, pretrained_weight):        # 冻结的预训练输出        base_out = F.linear(x, pretrained_weight)        # LoRA调整        lora_out = self.dropout(x) @ self.lora_A @ self.lora_B * self.scaling        return base_out + lora_out

NormalFloat4 (NF4)量化：- 将FP16权重压缩到4bit- 信息论最优：假设权重服从正态分布双重量化（Double Quantization）：- 连量化常数也进行量化- 额外节省0.37bit/参数分页优化器（Paged Optimizers）：- 利用CPU-GPU统一内存- 避免OOM崩溃

configs = {    'full_finetune': {        'trainable_params': '7B',        'learning_rate': 2e-5,        'epochs': 3    },    'lora_r8': {        'trainable_params': '8.4M',        'rank': 8,        'learning_rate': 1e-4,        'epochs': 10    },    'lora_r8_dropout': {        'trainable_params': '8.4M',        'rank': 8,        'dropout': 0.1,        'learning_rate': 1e-4,        'epochs': 10    }}

class AdaptiveLoRATrainer:    def __init__(self, model, train_data, val_data):        self.model = model        self.best_val_loss = float('inf')        self.patience = 3        self.patience_counter = 0    def train(self, epochs=10):        for epoch in range(epochs):            train_loss = self.train_epoch()            val_loss = self.validate()            # Early stopping            if val_loss < self.best_val_loss:                self.best_val_loss = val_loss                self.save_checkpoint()                self.patience_counter = 0            else:                self.patience_counter += 1                if self.patience_counter >= self.patience:                    print(f"Early stopping at epoch {epoch}")                    break            # 动态调整学习率            if self.patience_counter == 2:                self.reduce_lr()    def apply_weight_decay_only_to_lora(self):        """仅对LoRA权重应用L2正则"""        return [            {'params': [p for n, p in self.model.named_parameters()                        if 'lora' in n], 'weight_decay': 0.01},            {'params': [p for n, p in self.model.named_parameters()                        if 'lora' not in n], 'weight_decay': 0.0}        ]

# 释义增强（paraphrase augmentation）augmented = backtranslation(original_text, src='zh', pivot='en')

def mixout(lora_weight, pretrained_weight, p=0.5):    """随机替换部分LoRA输出为预训练输出"""    mask = torch.bernoulli(torch.full_like(lora_weight, p))    return mask * pretrained_weight + (1 - mask) * lora_weight

# 先微调最后几层，逐步解冻更多层for stage in [24, 16, 0]:  # 从第24层开始解冻    freeze_layers_before(model, stage)    train(epochs=2)

def evaluate_instruction_quality(instruction, output):    scores = {        'diversity': check_diversity(instruction),  # 与现有数据的差异        'complexity': count_reasoning_steps(output),  # 推理深度        'verifiability': has_clear_answer(instruction),  # 可验证性        'safety': toxicity_score(output),  # 安全性    }    return weighted_sum(scores)def check_diversity(new_inst, existing_insts):    """基于嵌入的语义去重"""    new_emb = embed(new_inst)    similarities = [cosine_sim(new_emb, embed(exist))                    for exist in existing_insts]    return 1 - max(similarities)  # 与最相似样本的差异度

def self_instruct_pipeline(seed_tasks, model, n_target=50000):    """    从175个种子任务生成52K指令数据    """    dataset = seed_tasks.copy()    while len(dataset) < n_target:        # 1. 随机采样8个示例（包含种子和已生成）        examples = random.sample(dataset, k=8)        # 2. 提示模型生成新任务        prompt = f"""你是一个帮助研究者创建任务的助手。以下是一些任务示例：{format_examples(examples)}请创建1个新的、不同类型的任务。要求：1. 与上述示例不重复2. 具有明确的输入输出格式3. 可以被智能助手完成输出格式：###指令：<你的任务描述>"""        new_task = model.generate(prompt)        # 3. 质量过滤        if is_valid_task(new_task, dataset):            # 4. 生成输入输出            instance = generate_instance(new_task, model)            dataset.append({                'instruction': new_task,                'input': instance['input'],                'output': instance['output']            })    return datasetdef is_valid_task(new_task, existing):    """多层过滤"""    # Rouge-L相似度 < 0.7    if max_similarity(new_task, existing) > 0.7:        return False    # 关键词过滤（避免有害内容）    if contains_unsafe_keywords(new_task):        return False    # 长度合理性    if not (10 < len(new_task.split()) < 150):        return False    return True

EVOLUTION_STRATEGIES = [    "添加约束条件（如字数限制、格式要求）",    "增加推理深度（需要多步推理）",    "复杂化输入（更长的上下文）",    "具体化描述（添加细节）",    "增加难度（需要领域知识）"]def evolve_instruction(base_inst, strategy, model):    prompt = f"""原始指令：{base_inst}请根据以下策略改写指令：{strategy}改写后的指令："""    evolved = model.generate(prompt)    # 验证是否确实提升了复杂度    if complexity_score(evolved) > complexity_score(base_inst):        return evolved    else:        return None  # 拒绝未提升的结果

# 阶段1：自动预筛（保留50-60%）def auto_filter(instruction, output):    checks = [        perplexity(output) < 100,  # 流畅性        not contains_repetition(output),  # 无重复        factual_consistency_score(instruction, output) > 0.8,  # 一致性    ]    return all(checks)# 阶段2：人工精筛annotation_interface = {    'display': (instruction, output),    'questions': [        "任务描述是否清晰？(1-5)",        "答案是否正确？(是/否/不确定)",        "是否存在有害内容？(是/否)",        "总体质量：(接受/修改/拒绝)"    ]}

┌─────────────────────────────────────────────────┐│               用户接入层                          ││  Web/App/微信 → API Gateway → 负载均衡           │└─────────────────────────────────────────────────┘                      ↓┌─────────────────────────────────────────────────┐│             对话理解模块                          ││  ┌─────────┐      ┌──────────┐                  ││  │ ASR引擎  │ →   │ 意图分类器 │ (Qwen-7B微调)    ││  │(Whisper) │      │NLU Parser│                  ││  └─────────┘      └──────────┘                  │└─────────────────────────────────────────────────┘                      ↓┌─────────────────────────────────────────────────┐│             知识检索层(RAG)                       ││  ┌──────────┐  ┌──────────┐  ┌──────────┐      ││  │ 向量检索  │  │ 关键词检索 │  │  Reranker │      ││  │(Milvus)  │  │(Elastic) │  │ (bge-rerank)│   ││  └──────────┘  └──────────┘  └──────────┘      │└─────────────────────────────────────────────────┘                      ↓┌─────────────────────────────────────────────────┐│          生成与编排层(Agent)                       ││  ┌──────────────────────────────────────┐       ││  │ ReAct Agent (基于LangGraph)          │       ││  │  ├─ 工具1: 订单查询API               │       ││  │  ├─ 工具2: 知识库检索                │       ││  │  ├─ 工具3: 情感分析                  │       ││  │  └─ 工具4: 人工转接判断              │       ││  └──────────────────────────────────────┘       ││  生成模型: Qwen-14B (4bit量化)                   │└─────────────────────────────────────────────────┘                      ↓┌─────────────────────────────────────────────────┐│               输出层                             ││  ┌──────────┐  ┌──────────┐  ┌──────────┐      ││  │ 安全过滤  │  │ TTS合成   │  │ 日志记录  │      ││  │(毒性检测) │  │(VITS)    │  │(追踪)     │      ││  └──────────┘  └──────────┘  └──────────┘      │└─────────────────────────────────────────────────┘

class HybridRetriever:    def __init__(self):        self.dense_index = MilvusClient()  # 向量检索        self.sparse_index = ElasticsearchClient()  # BM25检索        self.reranker = CrossEncoder('BAAI/bge-reranker-large')    def retrieve(self, query, top_k=20):        # 并行检索        dense_results = self.dense_index.search(            embed(query),             top_k=top_k        )        sparse_results = self.sparse_index.search(            query,             top_k=top_k        )        # 结果融合（Reciprocal Rank Fusion）        combined = self.rrf_fusion(dense_results, sparse_results)        # 重排序        reranked = self.reranker.rerank(            query,             [doc.content for doc in combined[:50]]        )        return reranked[:10]    def rrf_fusion(self, list1, list2, k=60):        """RRF: 1/(k+rank)加权"""        scores = defaultdict(float)        for rank, doc in enumerate(list1):            scores[doc.id] += 1 / (k + rank + 1)        for rank, doc in enumerate(list2):            scores[doc.id] += 1 / (k + rank + 1)        # 按分数排序        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)        return [self.get_doc(doc_id) for doc_id, _ in sorted_docs]

from langgraph.graph import StateGraph, ENDclass CustomerServiceAgent:    def __init__(self):        self.llm = Qwen14BChat()        self.tools = {            'search_order': self.search_order_tool,            'query_kb': self.knowledge_base_tool,            'transfer_human': self.transfer_tool        }        # 构建工作流        workflow = StateGraph()        workflow.add_node("classify_intent", self.classify)        workflow.add_node("use_tool", self.execute_tool)        workflow.add_node("generate_response", self.generate)        workflow.add_edge("classify_intent", "use_tool")        workflow.add_conditional_edges(            "use_tool",            self.should_continue,            {                "continue": "use_tool",                "generate": "generate_response"            }        )        workflow.add_edge("generate_response", END)        self.app = workflow.compile()    def classify(self, state):        """意图分类"""        intent = self.llm.classify(            state['user_input'],            labels=['订单查询', '产品咨询', '投诉建议', '闲聊']        )        state['intent'] = intent        return state    def execute_tool(self, state):        """执行工具"""        # LLM决定使用哪个工具        tool_call = self.llm.generate_tool_call(            state['user_input'],            available_tools=self.tools.keys()        )        result = self.tools[tool_call['name']](**tool_call['args'])        state['tool_results'].append(result)        return state    def should_continue(self, state):        """判断是否需要继续调用工具"""        if len(state['tool_results']) >= 3:  # 最多3次            return "generate"        decision = self.llm.decide(            f"已获取信息：{state['tool_results']}，"            f"用户问题：{state['user_input']}，"            f"是否需要更多信息？"        )        return "continue" if decision else "generate"    def generate(self, state):        """生成最终回复"""        context = "\n".join([str(r) for r in state['tool_results']])        response = self.llm.generate(            f"用户问题：{state['user_input']}\n"            f"参考信息：{context}\n"            f"请生成专业、友好的客服回复："        )        # 添加来源引用        response_with_source = self.add_citations(response, state['tool_results'])        return {'response': response_with_source}

class HallucinationDetector:    def __init__(self):        self.nli_model = NLIModel('MoritzLaurer/mDeBERTa-v3-base-mnli')    def verify_response(self, response, retrieved_docs):        """验证生成内容是否有文档支持"""        # 提取生成文本中的断言        claims = self.extract_claims(response)        verification_results = []        for claim in claims:            # 对每个断言，检查是否有文档支持            supported = False            for doc in retrieved_docs:                entailment_score = self.nli_model.predict(                    premise=doc.content,                    hypothesis=claim                )                if entailment_score > 0.8:  # 蕴含阈值                    supported = True                    break            verification_results.append({                'claim': claim,                'supported': supported            })        # 计算幻觉率        hallucination_rate = sum(            1 for r in verification_results if not r['supported']        ) / len(verification_results)        return hallucination_rate, verification_results    def extract_claims(self, text):        """使用LLM提取可验证断言"""        prompt = f"""从以下文本中提取所有事实性断言（忽略观点和建议）：{text}断言列表（每行一个）："""        claims = self.llm.generate(prompt).strip().split('\n')        return [c.strip('- ') for c in claims if c.strip()]

推理集群：├── 4× NVIDIA A10 (24GB)│   ├── 2卡: Qwen-14B (4bit量化，vLLM部署)│   ├── 1卡: Embedding模型 + Reranker│   └── 1卡: Whisper + VITS向量数据库：├── Milvus集群 (3节点)│   ├── 每节点: 16核CPU, 64GB内存│   └── SSD存储: 500GB应用服务器：├── 8× Kubernetes Pods│   └── 每个: 4核CPU, 8GB内存

# vLLM配置（相比HuggingFace Transformers提升5-10×吞吐量）from vllm import LLM, SamplingParamsllm = LLM(    model="Qwen/Qwen-14B-Chat",    tensor_parallel_size=2,  # 跨2卡并行    gpu_memory_utilization=0.9,    max_num_seqs=64,  # 最大batch size    quantization="awq"  # 4bit量化)sampling_params = SamplingParams(    temperature=0.7,    top_p=0.9,    max_tokens=512)# Continuous Batching：动态组batchoutputs = llm.generate(prompts, sampling_params)

class SemanticCache:    def __init__(self, similarity_threshold=0.95):        self.cache = {}        self.embeddings = {}        self.threshold = similarity_threshold    def get(self, query):        query_emb = embed(query)        # 语义相似度匹配        for cached_query, cached_emb in self.embeddings.items():            if cosine_similarity(query_emb, cached_emb) > self.threshold:                return self.cache[cached_query]        return None    def set(self, query, response):        self.cache[query] = response        self.embeddings[query] = embed(query)

class SemanticCache:    def __init__(self, similarity_threshold=0.95):        self.cache = {}        self.embeddings = {}        self.threshold = similarity_threshold    def get(self, query):        query_emb = embed(query)        # 语义相似度匹配        for cached_query, cached_emb in self.embeddings.items():            if cosine_similarity(query_emb, cached_emb) > self.threshold:                return self.cache[cached_query]        return None    def set(self, query, response):        self.cache[query] = response        self.embeddings[query] = embed(query)

[SSM层] → [Attention层] → [SSM层] → [Attention层] → ...  ↑           ↑ 效率      上下文学习

训练：大量计算 → 固定模型推理：快速前向传播

训练：学习"如何思考"推理：根据任务难度分配计算

class AdaptiveReasoningModel:    def generate(self, problem, max_compute_budget=1000):        difficulty = self.estimate_difficulty(problem)        if difficulty < 0.3:  # 简单问题            return self.fast_path(problem)        # 复杂问题：树搜索        best_solution = None        best_score = -inf        for _ in range(int(difficulty * max_compute_budget)):            # 生成候选推理路径            thought_chain = self.generate_chain_of_thought(problem)            # 自我评估            score = self.evaluate_reasoning(thought_chain, problem)            if score > best_score:                best_score = score                best_solution = thought_chain        return self.synthesize_answer(best_solution)

训练对：(图像, 文本) - CLIP对比学习(图像, 音频) - 视频对比学习(图像, 深度) - RGB-D数据(图像, IMU) - 视频元数据结果：无需直接配对也能对齐（如文本 ↔ 音频）

# 跨模态检索query_audio = load_audio("狗叫声.wav")results = imagebind.search(    query=query_audio,    database=image_database,    modality='image')# 返回：狗的图片# 跨模态生成text = "海浪声"audio = imagebind.generate(    source=text,    target_modality='audio')

模型在任务A上微调 → 任务B → 任务A性能大幅下降

class ContinualLearner:    def __init__(self, model, buffer_size=10000):        self.model = model        self.memory_buffer = []    def learn_task(self, new_data, task_id):        # 混合新数据与记忆样本        if self.memory_buffer:            replay_samples = random.sample(                self.memory_buffer,                 k=min(len(self.memory_buffer), len(new_data) // 2)            )            training_data = new_data + replay_samples        else:            training_data = new_data        # 训练        self.model.train(training_data)        # 更新记忆（保留代表性样本）        self.update_memory(new_data, task_id)

为每个任务训练独立的LoRA推理时根据任务ID激活对应LoRA优势：- 无遗忘（参数隔离）- 高效（每任务仅8M参数）

# 必须掌握的库import torch  # 深度学习框架import transformers  # HuggingFace生态import numpy as npimport pandas as pd# 实践项目1. 从零实现一个2层MLP（手写反向传播）2. 复现MNIST分类（使用PyTorch）3. 微调一个BERT模型（文本分类任务）

# Week 1-2: 基础调用tasks = [    "使用GPT-4完成10种不同类型的任务",    "对比3个模型在同一任务上的表现",    "测试不同temperature/top_p的效果"]# Week 3-4: 高级Promptingadvanced_tasks = [    "设计一个Few-shot学习提示",    "实现一个Chain-of-Thought推理流程",    "使用ReAct框架调用外部工具"]

# 功能需求1. 导入Markdown/PDF文档2. 智能分块（考虑语义完整性）3. 混合检索（向量+关键词）4. 生成带引用的答案# 技术栈- LangChain/LlamaIndex- ChromaDB（轻量级向量库）- 开源嵌入模型- Qwen-7B（本地部署）

tools = [    PythonREPLTool(),  # 执行代码    WikipediaSearchTool(),  # 查询知识    ArxivSearchTool(),  # 查论文]# 任务示例user_query = """分析2020-2023年全球新能源汽车销量趋势，并预测2024年中国市场占比"""# Agent需要：1. 搜索相关数据源2. 编写Python代码清洗数据3. 生成可视化图表4. 撰写分析报告

from peft import LoRAConfig, get_peft_model# 配置lora_config = LoRAConfig(    r=8,    lora_alpha=16,    target_modules=["q_proj", "v_proj"],    lora_dropout=0.05,    task_type="CAUSAL_LM")# 数据准备（1000条高质量样本）dataset = prepare_instruction_dataset(    domain="customer_service",    format="alpaca")# 训练trainer.train()

# 简化的RLHF流程1. 收集对比数据（1000对"好回复 vs 坏回复"）2. 训练Reward Model3. 使用PPO/DPO优化策略

数据层：├── 医学文献库（PubMed）├── 临床指南（UpToDate）└── 病例数据库（脱敏）模型层：├── 基座：Qwen-14B-Med（医疗领域预训练）├── 微调：LoRA（2000条诊断对话）└── RAG：混合检索医学知识应用层：├── 症状分析Agent├── 用药建议系统└── 医学影像辅助（多模态）

# 论文复现SOP1. 精读论文3遍   - 第1遍：理解大意   - 第2遍：推导公式   - 第3遍：找实现细节2. 查找官方代码（GitHub）   - 理解数据预处理   - 定位核心算法3. 最小化复现   - 仅实现关键创新点   - 在小数据集上验证4. 消融实验   - 测试各组件的贡献   - 记录详细日志

# 快速对比scenarios = {    '原型验证': 'ChromaDB (本地、免费)',    '中小规模(<1M)': 'Qdrant (易用性好)',    '大规模生产': 'Milvus (性能最强)',    '混合检索': 'Weaviate (支持关键词+向量)',    '极致性能': 'FAISS GPU (需自行管理)'}

# Weights & Biases配置import wandbwandb.init(    project="my-llm-project",    config={        "model": "qwen-7b",        "lora_r": 8,        "learning_rate": 1e-4    })# 自动记录wandb.log({"loss": loss, "accuracy": acc})# 可视化对比多次实验

场景1：快速验证MVP└─> API调用（GPT-4/Claude） + LangChain场景2：成本敏感型应用└─> 开源模型（Qwen/DeepSeek） + LoRA微调 + RAG场景3：高隐私要求└─> 本地部署 + 私有化向量库 + 端到端加密场景4：极致性能需求└─> vLLM推理 + 混合检索 + 模型蒸馏场景5：多模态应用└─> Qwen-VL / GPT-4V + 专用预处理管道