Posts

GraphRAG vs 传统 RAG 的本质区别 传统 RAG 的核心问题是"只见树木不见森林"——它能找到局部相关的文本块，但无法理解全局关系。GraphRAG 通过构建知识图谱，让系统具备全局视角和推理能力。 维度 传统 RAG GraphRAG 检索单元 文本块 实体+关系+文本块 全局理解 ❌ ✅ 社区摘要 多跳推理 ❌ ✅ 图遍历 可解释性 低 高（路径溯源） 构建成本 低 高 查询延迟 1-2s 3-10s 架构设计 ┌─────────────────────────────────────────────────────┐ │ GraphRAG 架构 │ ├─────────────────────────────────────────────────────┤ │ │ │ 离线构建 Pipeline │ │ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ │ │ 文档解析 │→│ 实体抽取 │→│ 关系抽取 │ │ │ └──────────┘ └───────────┘ └──────────────┘ │ │ ↓ │ │ ┌───────────┐ ┌───────────┐ ┌──────────────┐ │ │ │ 图谱构建 │←│ 社区检测 │←│ Embedding │ │ │ └───────────┘ └───────────┘ └──────────────┘ │ │ ↓ │ │ ┌─────────────┐ │ │ │ 索引持久化 │ │ │ └─────────────┘ │ │ │ │ 在线查询 Engine │ │ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ │ │ Query │→│ 路由决策 │→│ 双路检索 │ │ │ │ 理解 │ │ │ │ 向量+图谱 │ │ │ └──────────┘ └───────────┘ └──────────────┘ │ │ ↓ │ │ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ │ │ 生成+引用 │←│ 信息整合 │←│ 重排序 │ │ │ └──────────┘ └───────────┘ └──────────────┘ │ │ │ └─────────────────────────────────────────────────────┘ 离线构建 Pipeline 1. 实体与关系抽取 from pydantic import BaseModel from typing import List class Entity(BaseModel): name: str type: str # Person, Organization, Concept, Event, etc. description: str source_chunk_id: str class Relation(BaseModel): source_entity: str target_entity: str relation_type: str description: str confidence: float source_chunk_id: str ENTITY_EXTRACTION_PROMPT = """ 你是一个信息抽取专家。从以下文本中抽取实体和关系。 实体类型：Person, Organization, Concept, Technology, Event, Location 关系类型：works_at, created, related_to, part_of, located_in, depends_on, competes_with 文本： {text} 请以 JSON 格式输出： {{ "entities": [ {{"name": "...", "type": "...", "description": "..."}} ], "relations": [ {{"source": "...", "target": "...", "type": "...", "description": "...", "confidence": 0.0-1.0}} ] }} """ class EntityExtractor: def __init__(self, llm): self.llm = llm def extract(self, text: str, chunk_id: str): prompt = ENTITY_EXTRACTION_PROMPT.format(text=text) result = self.llm.generate(prompt, response_format="json") entities = [ Entity(**e, source_chunk_id=chunk_id) for e in result["entities"] ] relations = [ Relation(**r, source_chunk_id=chunk_id) for r in result["relations"] ] return entities, relations 2. 知识图谱构建 import networkx as nx from community_detection import LeidenAlgorithm class KnowledgeGraph: def __init__(self): self.graph = nx.DiGraph() self.entity_index = {} # name -> node_id def add_entities(self, entities: List[Entity]): for entity in entities: if entity.name not in self.entity_index: node_id = len(self.entity_index) self.entity_index[entity.name] = node_id self.graph.add_node( node_id, name=entity.name, type=entity.type, description=entity.description, source_chunks=[entity.source_chunk_id] ) else: # 合并描述 node_id = self.entity_index[entity.name] self.graph.nodes[node_id]["source_chunks"].append( entity.source_chunk_id ) def add_relations(self, relations: List[Relation]): for rel in relations: if rel.source_entity in self.entity_index and rel.target_entity in self.entity_index: src = self.entity_index[rel.source_entity] tgt = self.entity_index[rel.target_entity] if self.graph.has_edge(src, tgt): # 合并关系 existing = self.graph[src][tgt] existing["relations"].append({ "type": rel.relation_type, "description": rel.description, "confidence": rel.confidence }) else: self.graph.add_edge( src, tgt, relations=[{ "type": rel.relation_type, "description": rel.description, "confidence": rel.confidence }] ) def detect_communities(self): """使用 Leiden 算法进行社区检测""" undirected = self.graph.to_undirected() communities = LeidenAlgorithm().fit(undirected) # 为每个社区生成摘要 for comm_id, nodes in communities.items(): subgraph = self.graph.subgraph(nodes) summary = self._summarize_community(subgraph) for node in nodes: self.graph.nodes[node]["community_id"] = comm_id self.graph.graph.setdefault("community_summaries", {})[comm_id] = summary return communities def _summarize_community(self, subgraph): entities_info = [] for node_id in subgraph.nodes(): node = self.graph.nodes[node_id] entities_info.append(f"{node['name']} ({node['type']}): {node['description']}") relations_info = [] for u, v, data in subgraph.edges(data=True): for rel in data["relations"]: relations_info.append( f"{self.graph.nodes[u]['name']} --{rel['type']}--> {self.graph.nodes[v]['name']}" ) prompt = f""" 请总结以下知识图谱社区的关键信息： 实体： {chr(10).join(entities_info)} 关系： {chr(10).join(relations_info)} 请生成一段简洁的摘要，涵盖主要实体和它们之间的关系。 """ return self.llm.generate(prompt) 3. 索引持久化 class GraphRAGIndex: def __init__(self): self.graph = KnowledgeGraph() self.vector_store = MilvusIndex(dim=1024) self.community_store = CommunityStore() def build(self, documents: List[Document]): # 1. 文本分块与向量化 chunks = document_aware_chunk(documents) for chunk in chunks: embedding = embed_model.encode(chunk.text) self.vector_store.add(id=chunk.id, embedding=embedding, metadata={"text": chunk.text}) # 2. 实体关系抽取 all_entities = [] all_relations = [] for chunk in chunks: entities, relations = extractor.extract(chunk.text, chunk.id) all_entities.extend(entities) all_relations.extend(relations) # 3. 构建知识图谱 self.graph.add_entities(all_entities) self.graph.add_relations(all_relations) # 4. 社区检测与摘要 communities = self.graph.detect_communities() # 5. 持久化 self._persist() def _persist(self): # 图谱存 Neo4j 或 NetworkX pickle nx.write_gpickle(self.graph.graph, "graph.gpickle") # 向量索引已在 Milvus 中持久化 # 社区摘要存数据库 self.community_store.save(self.graph.graph.graph.get("community_summaries", {})) 在线查询引擎 class GraphRAGQueryEngine: def __init__(self, index: GraphRAGIndex, llm): self.index = index self.llm = llm def query(self, question: str) -> str: # 1. 判断查询类型 query_type = self._classify_query(question) if query_type == "global": # 全局型问题 → 社区检索 context = self._global_search(question) elif query_type == "specific": # 具体型问题 → 向量+图遍历 context = self._local_search(question) else: # 混合型 → 双路检索 context = self._hybrid_search(question) # 2. 生成答案 answer = self.llm.generate( prompt=ANSWER_PROMPT.format(question=question, context=context), citations=True ) return answer def _local_search(self, question: str): # 向量检索找到相关文本块 query_emb = embed_model.encode(question) vector_hits = self.index.vector_store.search(query_emb, top_k=10) # 从命中块中提取实体 entities = set() for hit in vector_hits: entities.update(self._extract_entities_from_chunk(hit)) # 图遍历扩展上下文 graph_context = [] for entity_name in entities: if entity_name in self.index.graph.entity_index: node_id = self.index.graph.entity_index[entity_name] # 1-hop 和 2-hop 邻居 neighbors = nx.single_source_shortest_path_length( self.index.graph.graph, node_id, cutoff=2 ) for neighbor_id, hops in neighbors.items(): if hops > 0: node = self.index.graph.graph.nodes[neighbor_id] graph_context.append({ "entity": node["name"], "type": node["type"], "hops": hops, "description": node["description"] }) return { "vector_context": vector_hits, "graph_context": graph_context } def _global_search(self, question: str): # 从社区摘要中检索 community_summaries = self.index.community_store.get_all() # 用 LLM 判断哪些社区相关 relevant = self.llm.generate( prompt=f""" 以下问题与哪些社区摘要相关？ 问题：{question} 社区摘要： {json.dumps(community_summaries, ensure_ascii=False)} 返回最相关的 3 个社区 ID。 """ ) return {"community_context": relevant} 生产运维 监控指标 @dataclass class GraphRAGMetrics: # 构建阶段 entity_extraction_rate: float # 每分钟抽取实体数 relation_extraction_rate: float # 每分钟抽取关系数 community_detection_time: float # 社区检测耗时 # 查询阶段 query_latency_p50: float query_latency_p99: float vector_recall: float # 向量检索召回率 graph_coverage: float # 图遍历覆盖率 community_hit_rate: float # 社区命中率 # 质量指标 answer_accuracy: float # 答案准确率 citation_rate: float # 引用覆盖率 hallucination_rate: float # 幻觉率 增量更新策略 class IncrementalUpdater: """GraphRAG 增量更新：只处理新增/修改的文档""" def update(self, new_documents: List[Document], deleted_ids: List[str]): # 1. 删除过期数据 for doc_id in deleted_ids: self._remove_document(doc_id) # 2. 处理新文档 for doc in new_documents: entities, relations = extractor.extract(doc) self.graph.add_entities(entities) self.graph.add_relations(relations) # 3. 局部社区检测（只重新检测受影响的社区） affected_communities = self._find_affected_communities(new_documents, deleted_ids) self._redetect_communities(affected_communities) # 4. 更新向量索引 self.vector_store.upsert(new_documents) 成本与效果 以 10 万篇技术文档为例： ...

从 PPO 到 GRPO 的演进 PPO（Proximal Policy Optimization）是 RLHF 的传统方法，但它有几个问题：需要一个 Critic 模型（额外显存）、训练不稳定、超参数敏感。DeepSeek 提出的 GRPO（Group Relative Policy Optimization）通过组内相对比较消除了 Critic 模型，简化了训练同时提升了稳定性。 方法 需要Critic 显存占用 训练稳定性 实现复杂度 PPO ✅ 高 中 高 DPO ❌ 低 高 低 GRPO ❌ 中 高 中 GRPO 核心思想 传统 PPO 用 Critic 网络估计 baseline，GRPO 直接用同一 prompt 的多个采样结果的平均值作为 baseline： 对每个 prompt p: 1. 采样 G 个回复 {r_1, r_2, ..., r_G} 2. 计算每个回复的奖励 {R_1, R_2, ..., R_G} 3. 组内归一化：advantage_i = (R_i - mean(R)) / std(R) 4. 用归一化后的 advantage 更新策略 算法对比 # PPO 的 Advantage 计算 def ppo_advantage(rewards, values, gamma=0.99, lam=0.95): """需要 Critic 网络预测 values""" advantages = [] returns = [] gae = 0 for t in reversed(range(len(rewards))): delta = rewards[t] + gamma * values[t + 1] - values[t] gae = delta + gamma * lam * gae advantages.insert(0, gae) return advantages # GRPO 的 Advantage 计算 def grpo_advantage(rewards_per_group: list): """不需要 Critic，直接用组内统计""" group_mean = np.mean(rewards_per_group) group_std = np.std(rewards_per_group) advantages = [(r - group_mean) / (group_std + 1e-8) for r in rewards_per_group] return advantages GRPO 训练流程 import torch import torch.nn.functional as F from transformers import AutoModelForCausalLM, AutoTokenizer class GRPOTrainer: def __init__(self, model, ref_model, reward_model, tokenizer, config): self.model = model # 策略模型（训练） self.ref_model = ref_model # 参考模型（冻结） self.reward_model = reward_model # 奖励模型 self.tokenizer = tokenizer self.config = config def train_step(self, prompts: list): batch_size = len(prompts) group_size = self.config.group_size # G=8 或 16 all_advantages = [] all_old_logps = [] all_responses = [] # 1. 对每个 prompt 采样 G 个回复 for prompt in prompts: responses = [] rewards = [] for _ in range(group_size): # 生成回复 response = self._generate(prompt, temperature=0.8) responses.append(response) # 计算奖励 reward = self.reward_model.score(prompt, response) rewards.append(reward) # 2. GRPO Advantage 计算 advantages = self._compute_grpo_advantage(rewards) all_advantages.extend(advantages) all_responses.extend(responses) # 3. 计算旧策略的 log prob old_logps = self._compute_logps(self.model, prompts_expanded, all_responses) all_old_logps = old_logps.detach() # 4. PPO-style 更新（多轮 epoch） for epoch in range(self.config.ppo_epochs): new_logps = self._compute_logps(self.model, prompts_expanded, all_responses) ref_logps = self._compute_logps(self.ref_model, prompts_expanded, all_responses) # 5. 计算 Loss loss = self._grpo_loss( new_logps=new_logps, old_logps=all_old_logps, ref_logps=ref_logps, advantages=torch.tensor(all_advantages), beta=self.config.kl_coef # KL 散度惩罚 ) loss.backward() self.optimizer.step() self.optimizer.zero_grad() return loss.item() def _grpo_loss(self, new_logps, old_logps, ref_logps, advantages, beta=0.04): # Ratio ratio = torch.exp(new_logps - old_logps) # Clipped ratio (PPO clip) clipped_ratio = torch.clamp(ratio, 1 - self.config.clip_range, 1 + self.config.clip_range) # Policy loss policy_loss = -torch.min(ratio * advantages, clipped_ratio * advantages).mean() # KL penalty (GRPO 使用 k3 估计器) kl = (torch.exp(ref_logps - new_logps) - (ref_logps - new_logps) - 1).mean() return policy_loss + beta * kl def _compute_grpo_advantage(self, rewards: list): """GRPO 核心创新：组内相对优势""" mean_r = np.mean(rewards) std_r = np.std(rewards) advantages = [(r - mean_r) / (std_r + 1e-8) for r in rewards] return advantages 奖励函数设计 GRPO 的效果很大程度上取决于奖励函数。DeepSeek R1 使用了多种奖励信号： ...

KV Cache 是大模型推理的核心瓶颈。在 72B 模型上，一个 4K 上下文的请求需要 8GB 的 KV Cache；在 128K 上下文时，KV Cache 膨胀到 256GB——比模型权重还大。2026 年，KV Cache 优化已成为推理引擎的"第二战场"。本文将系统性解析从 PagedAttention 到 MLA 的全链路优化技术。 一、KV Cache 基础 为什么需要 KV Cache？ Transformer 自回归解码时，每生成一个 token 需要关注之前所有 token 的 Key 和 Value。如果不缓存，每步都要重算所有历史 token 的 K/V，复杂度 O(n²)；缓存后，每步只需计算新 token 的 K/V，复杂度 O(n)。 KV Cache 大小计算 KV Cache 大小 = 2 × num_layers × seq_len × num_kv_heads × head_dim × dtype_size 以 Qwen3.5-72B 为例： 参数 值 num_layers 80 num_kv_heads (GQA) 8 head_dim 128 dtype FP16 (2 bytes) 单 token KV Cache = 2 × 80 × 8 × 128 × 2 = 327,680 bytes ≈ 320KB 4K 上下文 = 320KB × 4096 = 1.25GB 32K 上下文 = 320KB × 32768 = 10GB 128K 上下文 = 320KB × 131072 = 40GB 关键问题：128K 上下文下，KV Cache（40GB）已接近模型权重大小（72B FP16 = 144GB），成为显存主要瓶颈。 ...

为什么选择 LoRA 全参数微调一个 70B 模型需要数百 GB 显存，而 LoRA（Low-Rank Adaptation）通过冻结原始权重、只训练低秩适配矩阵，将可训练参数减少到原来的 0.1%-1%，在消费级 GPU 上即可完成微调。 方法 可训练参数 显存需求 (7B) 显存需求 (70B) 全参数微调 100% 120GB 1200GB LoRA 0.1-1% 16GB 80GB QLoRA 0.1-1% 8GB 40GB 完整流程概览 数据准备 → 格式转换 → 训练配置 → LoRA训练 → 评估 → 合并 → 部署 1. 数据准备 数据格式 # 推荐格式：ShareGPT / OpenAI Messages { "messages": [ {"role": "system", "content": "你是一个专业的技术顾问。"}, {"role": "user", "content": "解释一下 RAG 的工作原理"}, {"role": "assistant", "content": "RAG（检索增强生成）是一种..."} ] } 数据构建脚本 import json from pathlib import Path class SFTDataBuilder: def __init__(self, output_dir: str): self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) def build_from_qa_pairs(self, qa_pairs: list, system_prompt: str): """从问答对构建训练数据""" samples = [] for qa in qa_pairs: sample = { "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": qa["question"]}, {"role": "assistant", "content": qa["answer"]} ] } samples.append(sample) # 划分训练/验证集 split = int(len(samples) * 0.95) train_path = self.output_dir / "train.jsonl" val_path = self.output_dir / "val.jsonl" with open(train_path, 'w', encoding='utf-8') as f: for s in samples[:split]: f.write(json.dumps(s, ensure_ascii=False) + '\n') with open(val_path, 'w', encoding='utf-8') as f: for s in samples[split:]: f.write(json.dumps(s, ensure_ascii=False) + '\n') print(f"训练集: {split} 条 → {train_path}") print(f"验证集: {len(samples) - split} 条 → {val_path}") def build_from_conversations(self, conversations: list): """从多轮对话构建训练数据""" samples = [] for conv in conversations: messages = [] for turn in conv: messages.append({"role": turn["role"], "content": turn["content"]}) samples.append({"messages": messages}) return samples 数据质量检查 class DataQualityChecker: def check(self, data_path: str): issues = [] with open(data_path, 'r', encoding='utf-8') as f: lines = f.readlines() for i, line in enumerate(lines): sample = json.loads(line) # 1. 检查消息格式 if "messages" not in sample: issues.append(f"Line {i}: 缺少 messages 字段") continue # 2. 检查角色顺序 roles = [m["role"] for m in sample["messages"]] if roles[-1] != "assistant": issues.append(f"Line {i}: 最后一条消息不是 assistant") # 3. 检查内容长度 for msg in sample["messages"]: if len(msg["content"]) < 5: issues.append(f"Line {i}: 消息内容过短") if len(msg["content"]) > 8000: issues.append(f"Line {i}: 消息内容过长 ({len(msg['content'])} chars)") # 4. 检查 assistant 回复质量 assistant_msgs = [m for m in sample["messages"] if m["role"] == "assistant"] for msg in assistant_msgs: if msg["content"].startswith("我是一个AI"): issues.append(f"Line {i}: assistant 回复包含模板化语言") if len(msg["content"]) < 20: issues.append(f"Line {i}: assistant 回复过短") # 5. 统计 stats = { "total_samples": len(lines), "avg_turns": np.mean([len(json.loads(l)["messages"]) for l in lines]), "avg_assistant_len": np.mean([ len(m["content"]) for l in lines for m in json.loads(l)["messages"] if m["role"] == "assistant" ]), "issues_found": len(issues), } return {"issues": issues[:20], "stats": stats} 2. 训练配置 # train_lora.py import torch from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments from peft import LoraConfig, get_peft_model, TaskType from trl import SFTTrainer, SFTConfig # 1. 加载模型和分词器 model_name = "Qwen/Qwen2.5-7B-Instruct" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.bfloat16, device_map="auto", attn_implementation="flash_attention_2" ) # 2. LoRA 配置 lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=64, # 秩，常用 8/16/32/64 lora_alpha=128, # alpha = 2 * r 是常见默认值 lora_dropout=0.05, target_modules=[ "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj" ], bias="none" ) # 3. 应用 LoRA model = get_peft_model(model, lora_config) model.print_trainable_parameters() # 输出: trainable params: 39,976,960 || all params: 7,621,836,800 || trainable%: 0.5247% # 4. 训练配置 training_args = SFTConfig( output_dir="./output/qwen2.5-7b-lora", num_train_epochs=3, per_device_train_batch_size=4, gradient_accumulation_steps=4, # 有效 batch_size = 16 learning_rate=2e-4, lr_scheduler_type="cosine", warmup_ratio=0.05, bf16=True, logging_steps=10, save_strategy="steps", save_steps=200, save_total_limit=3, eval_strategy="steps", eval_steps=200, load_best_model_at_end=True, metric_for_best_model="eval_loss", greater_is_better=False, gradient_checkpointing=True, max_seq_length=2048, dataset_text_field="messages", ) # 5. 加载数据 from datasets import load_dataset dataset = load_dataset("json", data_files={ "train": "data/train.jsonl", "validation": "data/val.jsonl" }) # 6. 启动训练 trainer = SFTTrainer( model=model, args=training_args, train_dataset=dataset["train"], eval_dataset=dataset["validation"], processing_class=tokenizer, ) trainer.train() 3. QLoRA：4bit 量化微调 显存不够？用 QLoRA 量化到 4bit： ...

在美中两国主导的大模型竞赛中，欧洲需要一个旗手。Mistral AI 从 2023 年的 Mistral 7B 起步，到 2026 年的 Mistral Large 3（123B），已成为欧洲 AI 最有分量的代表。但在 DeepSeek V4 和 Qwen3.5 的强势崛起下，Mistral Large 3 是否还能保持竞争力？本文将通过全面评测给出答案。 一、Mistral Large 3 概览 维度 Mistral Large 3 前代 Large 2 提升 参数量 123B 123B 持平 架构 Dense + GQA Dense + GQA 延续 上下文 128K 128K 持平 多模态 文本+图像 纯文本 新增 训练数据 18T 14T +29% 许可证 Apache 2.0 商业许可 全面开源 关键变化： 全面转向 Apache 2.0 开源——这是一个战略性决定 新增多模态能力（Pixtral 架构融合） 训练数据增加 29%，多语言覆盖扩展到 30+ 语言 Function Calling 原生支持 二、基础能力评测 通用能力 基准 Mistral Large 3 DeepSeek V4 Qwen3.5-72B Llama 4 Maverick GPT-5.5 MMLU-Pro 85.9 90.2 89.5 88.1 92.3 GPQA Diamond 69.8 78.3 76.1 74.5 83.6 MATH-500 71.8 82.6 80.3 77.8 89.2 HumanEval+ 89.7 94.1 93.8 92.3 96.1 MT-Bench 8.7 9.1 9.0 8.8 9.3 分析：Mistral Large 3 在所有基准上均落后于 DeepSeek V4 和 Qwen3.5-72B，差距在 4-11 分之间。最显著的差距在数学推理（-10.8 分 vs DeepSeek V4）。 ...

Mixture of Experts（MoE）已成为 2026 年大模型架构的事实标准。从 DeepSeek V4 的 256 专家设计到 Llama 4 Behemoth 的 16 专家稀疏路由，三大开源旗舰代表了 MoE 架构的三种不同哲学。本文将从架构细节、路由策略、推理效率与实际性能四个维度进行深度技术对比。 一、架构概览 维度 DeepSeek V4 (671B) Qwen3.5-Max (480B) Llama 4 Behemoth (2T) 总参数 671B 480B ~2T 激活参数 37B 42B ~120B 专家数量 256（共享+路由） 128（纯路由） 16（稀疏） 激活专家数 8 4 2 注意力机制 MLA GQA GQA 层数 61 64 96 隐藏维度 7168 6144 16384 训练数据 22T 18T 30T+ 二、路由机制深度解析 DeepSeek V4：细粒度专家 + 共享专家 DeepSeek V4 延续并升级了 V3 的架构理念，采用 256 个细粒度专家 + 2 个共享专家的设计： ...

从手工艺到工程化：Prompt 工程的范式转变 2026 年，Prompt 工程已经走过了"调参数式"的手工艺阶段。随着大模型能力边界的不断扩展，Prompt 的设计、测试、部署和维护已经形成了一套完整的工程化体系。根据 Gartner 2026 年 Q1 报告，超过 78% 的企业级 AI 应用已经建立了专门的 Prompt 工程团队，而 2024 年这一比例仅为 23%。 一、Prompt 工程化的核心要素 1.1 结构化 Prompt 架构 现代 Prompt 已从简单的文本指令发展为结构化的工程产物： # prompt-config.yaml prompt: metadata: name: "customer-support-v2" version: "2.3.1" author: "prompt-team" last_updated: "2026-06-15" system_prompt: role: "你是一个专业的客服助手" constraints: - "回答必须基于知识库内容" - "不确定时明确告知用户" - "单次回复不超过500字" knowledge_sources: - "product_docs" - "faq_database" user_template: | 用户问题：{{user_query}} 上下文信息：{{context}} 历史对话：{{history}} expected_output: format: "json" schema: answer: "string" confidence: "float" sources: "array" 1.2 Prompt 生命周期管理 设计 → 测试 → 评审 → 部署 → 监控 → 迭代 ↑ ↓ ←─────────── 反馈循环 ←─────────────← 二、2026 年 Prompt 设计核心原则 2.1 明确性原则（Clarity First） 维度 错误示例 正确示例 角色 “帮我写文章” “你是科技领域资深编辑，擅长将复杂技术概念转化为通俗易懂的文章” 任务 “总结一下” “用3段话总结以下内容，每段不超过100字，重点突出数据和结论” 约束 “写得好一点” “语气专业但不失亲和力，避免使用行业黑话，目标读者是产品经理” 格式 “列出来” “用Markdown表格输出，包含名称、描述、优缺点三列” 2.2 上下文工程（Context Engineering） 2026 年最重要的进化是"上下文工程"概念的兴起——不再只关注单条 Prompt 的措辞，而是整体上下文的构建： ...

RAG 故障诊断框架 RAG 系统出问题时，症状往往模糊——“回答不对”、“答非所问”、“太慢了”。2026 年的实践表明，几乎所有 RAG 故障都可以归因到三个核心环节：检索（Retrieval）、生成（Generation）、系统（System）。 故障报告 → 症状分类 → 根因定位 → 修复方案 → 验证 ↓ ┌─────────┼─────────┐ ↓ ↓ ↓ 检索层 生成层 系统层 漏检 幻觉 延迟 误检 偏题 超时 排序错 不完整 成本高 故障1：幻觉——“系统在编造事实” 症状 答案包含上下文中不存在的信息 引用了错误的来源 捏造数据或数字 诊断流程 class HallucinationDiagnostic: def diagnose(self, question: str, answer: str, contexts: list): # 1. 拆分答案为原子论断 claims = self._decompose_claims(answer) # 2. 逐条验证 results = [] for claim in claims: evidence = self._find_evidence(claim, contexts) results.append({ "claim": claim, "has_evidence": evidence["score"] > 0.5, "evidence_text": evidence["text"], "score": evidence["score"] }) hallucinated = [r for r in results if not r["has_evidence"]] return { "total_claims": len(claims), "hallucinated_claims": len(hallucinated), "hallucination_rate": len(hallucinated) / len(claims), "details": results, "likely_causes": self._identify_causes(hallucinated, contexts) } def _identify_causes(self, hallucinated, contexts): causes = [] # 原因1：上下文不足 if len(contexts) < 3: causes.append({ "cause": "context_insufficient", "description": "检索到的上下文数量太少", "fix": "增加 top_k 参数或扩大检索范围" }) # 原因2：上下文不相关 if contexts and all(not self._is_relevant(c) for c in contexts): causes.append({ "cause": "context_irrelevant", "description": "检索到的上下文与问题不相关", "fix": "检查 Embedding 模型质量或分块策略" }) # 原因3：LLM 忽略上下文 if contexts and any(self._is_relevant(c) for c in contexts): causes.append({ "cause": "llm_ignoring_context", "description": "上下文包含相关信息但 LLM 未使用", "fix": "优化 Prompt，强调'仅基于上下文回答'" }) return causes 常见根因与修复 根因 症状 修复方案 检索结果不相关 答案与文档无关 优化 Embedding 模型、加入重排序 上下文太短 缺少关键信息 增加 top_k、扩展上下文窗口 LLM 忽略上下文 编造而非引用 加强 Prompt 约束、降低 temperature 分块切断关键信息 信息碎片化 调整分块策略、增加重叠 模型本身幻觉倾向 即使上下文完整也编造 换用更可靠的模型 Prompt 修复示例 # 修复前（容易幻觉） BAD_PROMPT = "根据以下信息回答问题：{context}\n问题：{question}" # 修复后（抑制幻觉） GOOD_PROMPT = """请严格基于以下参考信息回答问题。 规则： 1. 只使用参考信息中的内容 2. 如果参考信息不足以回答，请说"根据现有信息无法回答" 3. 每个事实陈述后用 [1], [2] 标注来源 4. 不要编造任何信息 参考信息： {context} 问题：{question} 回答：""" 故障2：漏检——“该找到的没找到” 症状 知识库中明明有相关信息，但系统说"不知道" 回答不完整，遗漏关键信息 多文档对比时只引用了一部分 诊断流程 class RetrievalDiagnostic: def diagnose(self, question: str, retrieved: list, ground_truth_docs: list): report = {} # 1. 召回率分析 retrieved_ids = {r.id for r in retrieved} gt_ids = {d.id for d in ground_truth_docs} report["recall"] = len(retrieved_ids & gt_ids) / len(gt_ids) report["missed_docs"] = list(gt_ids - retrieved_ids) # 2. 分析漏检原因 for missed_id in report["missed_docs"]: missed_doc = self._get_doc(missed_id) root_cause = self._analyze_miss(question, missed_doc) report.setdefault("miss_reasons", []).append(root_cause) return report def _analyze_miss(self, question: str, missed_doc: Document): # 原因A: Embedding 相似度过低 q_emb = self.embedder.encode(question) d_emb = self.embedder.encode(missed_doc.content) similarity = cosine_similarity(q_emb, d_emb) if similarity < 0.5: return { "doc_id": missed_doc.id, "cause": "embedding_similarity_low", "similarity": similarity, "fix": "考虑使用 HyDE 或 Query 重写" } # 原因B: 分块导致信息碎片化 if len(missed_doc.content) < 50: return { "doc_id": missed_doc.id, "cause": "chunk_too_short", "fix": "调整分块策略，增大 chunk_size" } # 原因C: 元数据过滤排除了该文档 return { "doc_id": missed_doc.id, "cause": "metadata_filter_excluded", "similarity": similarity, "fix": "检查元数据过滤条件" } 常见漏检模式 模式1: 语义鸿沟 Query: "怎么提高模型准确率" 文档: "模型精度优化方法" ← 同义词但 Embedding 可能不够近 修复: 加入 Query 重写或同义词扩展 模式2: 多跳信息分散 Query: "A和B的区别" 文档: 分别有A和B的描述，但没有对比性文档 修复: 使用多查询策略或 Agentic RAG 模式3: 长尾知识 Query: 某罕见概念 文档: 存在但 Embedding 训练时未见 修复: 加入关键词检索做补充 模式4: 版本过期 Query: 最新信息 文档: 旧版本未更新 修复: 定期增量更新 + 时间戳过滤 故障3：延迟过高——“等了10秒才出结果” 延迟分解 class LatencyProfiler: def profile(self, query: str): timings = {} # 1. Query Embedding t0 = time.time() q_emb = self.embedder.encode(query) timings["embedding"] = time.time() - t0 # 2. 向量检索 t0 = time.time() results = self.vector_store.search(q_emb, top_k=50) timings["vector_search"] = time.time() - t0 # 3. 重排序 t0 = time.time() reranked = self.reranker.rerank(query, results, top_k=5) timings["rerank"] = time.time() - t0 # 4. Prompt 构建 t0 = time.time() prompt = self._build_prompt(query, reranked) timings["prompt_build"] = time.time() - t0 # 5. LLM 生成 t0 = time.time() answer = self.llm.generate(prompt) timings["llm_generate"] = time.time() - t0 timings["total"] = sum(timings.values()) return timings 典型延迟分布 环节 优化前 优化后 优化方法 Embedding 200ms 50ms 本地部署 + 批量 向量检索 500ms 80ms HNSW 参数调优 + 缓存 重排序 800ms 200ms 批量推理 + 减小候选集 LLM 生成 3000ms 1500ms 流式输出 + 更短 Prompt 总计 4500ms 1830ms 优化代码 class OptimizedRAG: def __init__(self): self.semantic_cache = SemanticCache( threshold=0.95, ttl=3600 ) self.query_classifier = QueryClassifier() def query(self, question: str): # 0. 语义缓存检查（~1ms） cached = self.semantic_cache.get(question) if cached: return cached, "cache_hit" # 1. 查询分类：简单 vs 复杂 complexity = self.query_classifier.predict(question) if complexity < 0.3: # 快速通道：跳过重排序 results = self._fast_query(question) else: # 标准通道：完整流程 results = self._full_query(question) # 写入缓存 self.semantic_cache.set(question, results) return results def _fast_query(self, question): """快速通道：向量检索 → LLM（跳过重排序）""" q_emb = self.embedder.encode(question) results = self.vector_store.search(q_emb, top_k=5) return self.llm.generate(question, context=results, stream=True) 故障4：答案偏题——“问的是A，答的是B” 诊断 def diagnose_off_topic(question, answer, contexts): # 检查问题覆盖 question_aspects = extract_question_aspects(question) answer_coverage = check_coverage(answer, question_aspects) # 检查检索相关性 retrieval_relevance = assess_relevance(question, contexts) return { "question_aspects": question_aspects, "covered_aspects": answer_coverage["covered"], "missed_aspects": answer_coverage["missed"], "retrieval_relevance": retrieval_relevance, "root_cause": identify_cause(answer_coverage, retrieval_relevance) } 常见根因 根因 检查方法 修复 Query 理解错误 检查检索 Query 是否与用户意图一致 加入 Query 重写 检索偏向热门内容 检查是否总是检索到相同文档 调整多样性参数 LLM 上下文窗口溢出 检查 Prompt 长度 压缩上下文 多意图问题未拆分 检查问题是否包含多个子问题 Agent 拆解 故障排查清单 □ 检索阶段 □ 检索结果是否相关？ → 查看原始检索结果 □ top_k 是否足够？ → 检查召回率 □ Embedding 模型是否合适？ → 测试语义相似度 □ 分块是否合理？ → 检查块边界 □ 是否需要 Query 重写？ → 检查 Query 与文档的语义鸿沟 □ 生成阶段 □ Prompt 是否清晰约束？ → 检查 Prompt 模板 □ 上下文是否足够？ → 检查上下文长度 □ temperature 是否过高？ → 降到 0-0.3 □ 模型能力是否足够？ → 换更强模型测试 □ 系统阶段 □ 延迟在哪个环节？ → Profile 各阶段耗时 □ 缓存是否生效？ → 检查缓存命中率 □ 是否有资源瓶颈？ → 检查 GPU/CPU/内存 总结 RAG 故障排查的核心是分层诊断：先定位是检索问题还是生成问题，再深入排查具体根因。最高效的排查方法是建立一个端到端的可观测系统，记录每个环节的输入输出和耗时，让故障无处遁形。 ...

分块为何如此重要 在 RAG 系统中，分块（Chunking）是影响检索质量的第一道关卡。一个糟糕的分块策略会让最好的 Embedding 模型和最贵的 LLM 都无济于事。2026 年的分块策略已经从简单的固定长度切分进化到了语义感知、文档结构感知的智能分块。 五大分块策略对比 策略1：固定长度分块（Fixed-Size Chunking） 最基础的方式，按固定 token 数切分，带重叠窗口。 def fixed_size_chunk(text: str, chunk_size: int = 512, overlap: int = 50): tokens = tokenizer.encode(text) chunks = [] for i in range(0, len(tokens), chunk_size - overlap): chunk = tokens[i:i + chunk_size] chunks.append(tokenizer.decode(chunk)) if i + chunk_size >= len(tokens): break return chunks 优点：简单、快速、可控 缺点：可能切断语义、忽略文档结构 适用：纯文本、均匀内容 策略2：语义分块（Semantic Chunking） 基于 Embedding 相似度，在语义跳变点切分。 import numpy as np from sentence_transformers import SentenceTransformer def semantic_chunk(text: str, model, threshold: float = 0.5): # 1. 先按句子切分 sentences = split_sentences(text) # 2. 逐句计算 Embedding embeddings = model.encode(sentences) # 3. 计算相邻句子的余弦相似度 similarities = [] for i in range(len(embeddings) - 1): sim = cosine_similarity(embeddings[i], embeddings[i+1]) similarities.append(sim) # 4. 在相似度低于阈值处切分 chunks = [] current_chunk = [sentences[0]] for i, sim in enumerate(similarities): if sim < threshold: chunks.append(" ".join(current_chunk)) current_chunk = [sentences[i+1]] else: current_chunk.append(sentences[i+1]) if current_chunk: chunks.append(" ".join(current_chunk)) return chunks 优点：语义完整、边界自然 缺点：计算开销大、块大小不均匀 适用：长文本、叙述性内容 策略3：文档感知分块（Document-Aware Chunking） 基于文档结构（标题、段落、列表）进行切分。 ...

RAG 架构的四代演进 检索增强生成（Retrieval-Augmented Generation, RAG）已经走过了三个年头。从 2023 年最初的概念验证，到 2026 年的今天，RAG 架构经历了四次重大范式转移。本文将系统梳理这条演进路线，并给出 2026 年的最佳实践。 第一代：Naive RAG（2023） 最朴素的 RAG 架构，核心流程是：文档分块 → 向量化 → 相似度检索 → 拼接 Prompt → LLM 生成。 用户提问 → Embedding → 向量检索 Top-K → 拼接上下文 → LLM 生成答案 问题：检索质量差、缺乏上下文理解、无法处理多跳推理。 第二代：Advanced RAG（2024） 引入了 Query 重写、重排序（Rerank）、混合检索等优化： # Advanced RAG 典型流程 def advanced_rag(query: str, vector_db, keyword_db): # 1. Query 重写 rewritten_query = llm.rewrite_query(query) # 2. 混合检索：向量 + 关键词 vector_results = vector_db.search(rewrite_query, top_k=20) keyword_results = keyword_db.search(rewrite_query, top_k=20) # 3. 合并去重 candidates = merge_and_dedupe(vector_results, keyword_results) # 4. 重排序 reranked = reranker.rerank(rewritten_query, candidates, top_k=5) # 5. 生成 return llm.generate(query, context=reranked) 提升：检索召回率提升 30-50%，但仍然缺乏全局视角。 ...