RAG与微调

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 使用了多种奖励信号： ...

为什么选择 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： ...

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%，但仍然缺乏全局视角。 ...

为什么 RAG 评估如此困难 RAG 系统的评估比单纯的 LLM 评估复杂得多，因为它涉及多个环节：检索质量、上下文相关性、生成质量、引用准确性。一个环节的优化可能影响另一个环节。2026 年的 RAG 评估已经形成了系统化的方法论。 RAG 评估的三层框架 ┌──────────────────────────────────────────────┐ │ 端到端评估（L3） │ │ 用户满意度 / 任务完成率 / 答案正确性 │ ├──────────────────────────────────────────────┤ │ 生成评估（L2） │ │ 答案相关性 / 忠实度 / 完整性 / 引用准确性 │ ├──────────────────────────────────────────────┤ │ 检索评估（L1） │ │ 召回率 / 精确率 / MRR / 上下文相关性 │ └──────────────────────────────────────────────┘ L1：检索层评估 基础指标 class RetrievalMetrics: @staticmethod def recall_at_k(retrieved_ids: list, relevant_ids: list, k: int = 5): """Top-K 召回率：相关文档是否出现在 Top-K 中""" retrieved_top_k = retrieved_ids[:k] hits = len(set(retrieved_top_k) & set(relevant_ids)) return hits / len(relevant_ids) if relevant_ids else 0.0 @staticmethod def precision_at_k(retrieved_ids: list, relevant_ids: list, k: int = 5): """Top-K 精确率""" retrieved_top_k = retrieved_ids[:k] hits = len(set(retrieved_top_k) & set(relevant_ids)) return hits / k @staticmethod def mrr(retrieved_ids: list, relevant_ids: list): """Mean Reciprocal Rank：第一个相关文档的排名倒数""" for i, doc_id in enumerate(retrieved_ids): if doc_id in relevant_ids: return 1.0 / (i + 1) return 0.0 @staticmethod def ndcg_at_k(retrieved_ids: list, relevance_scores: dict, k: int = 5): """Normalized Discounted Cumulative Gain""" dcg = sum( relevance_scores.get(doc_id, 0) / np.log2(i + 2) for i, doc_id in enumerate(retrieved_ids[:k]) ) ideal_scores = sorted(relevance_scores.values(), reverse=True)[:k] idcg = sum(s / np.log2(i + 2) for i, s in enumerate(ideal_scores)) return dcg / idcg if idcg > 0 else 0.0 上下文相关性评估 def context_relevance(question: str, contexts: list, llm) -> float: """评估检索到的上下文与问题的相关程度""" prompt = f""" 请评估以下检索上下文与问题的相关性。 问题：{question} 上下文： {chr(10).join([f'[{i+1}] {c[:200]}' for i, c in enumerate(contexts)])} 对每条上下文打分（0-3）： - 0: 完全无关 - 1: 部分相关，缺少关键信息 - 2: 相关，包含部分答案 - 3: 高度相关，直接回答问题 输出 JSON：{{"scores": [0-3, ...], "overall": 0.0-1.0}} """ result = llm.generate(prompt, response_format="json") return result["overall"] L2：生成层评估 RAGAS 框架 RAGAS（Retrieval Augmented Generation Assessment）是 2026 年最主流的 RAG 评估框架，核心指标包括： ...

数据管道是 RAG 系统的地基 RAG 系统的效果上限由数据质量决定。再好的 Embedding 模型和 LLM，如果喂进去的是垃圾数据，出来的也是垃圾答案。2026 年的 RAG 数据管道已经发展为一套完整的工程体系。 整体架构 数据源 解析层 处理层 索引层 ┌─────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ PDF │───→│ │ │ │ │ │ │ HTML │───→│ 文档解析 │───→│ 清洗+增强 │───→│ Embedding │ │ Word │───→│ Unified │ │ Pipeline │ │ + Index │ │ DB │───→│ Parser │ │ │ │ │ │ API │───→│ │ │ │ │ │ │ Web │───→│ │ │ │ │ │ └─────────┘ └───────────┘ └───────────┘ └───────────┘ ↓ ┌───────────┐ │ 质量监控 │ │ QC Layer │ └───────────┘ 1. 多源文档解析 统一解析层 from abc import ABC, abstractmethod from typing import List, Document import fitz # PyMuPDF from bs4 import BeautifulSoup from unstructured import partition_pdf class DocumentParser(ABC): @abstractmethod def parse(self, file_path: str) -> List[Document]: pass class PDFParser(DocumentParser): def parse(self, file_path: str) -> List[Document]: # 方案1: PyMuPDF（快，但结构感知弱） # 方案2: Unstructured（慢，但结构感知强） # 方案3: LLM 辅助解析（最准，但贵） elements = partition_pdf( file_path, strategy="hi_res", # 高精度模式 infer_table_structure=True, extract_image_block_types=["Image", "Table"], extract_image_block_output_dir="./images/" ) documents = [] for elem in elements: doc = Document( content=elem.text if hasattr(elem, 'text') else str(elem), metadata={ "source": file_path, "page": elem.metadata.page_number if elem.metadata else 0, "category": elem.category, "type": self._map_type(elem.category) } ) documents.append(doc) return documents class HTMLParser(DocumentParser): def parse(self, file_path: str) -> List[Document]: with open(file_path, 'r', encoding='utf-8') as f: soup = BeautifulSoup(f.read(), 'html.parser') # 移除无关元素 for tag in soup(['script', 'style', 'nav', 'footer', 'header']): tag.decompose() documents = [] # 按标题分块 current_section = {"title": "", "content": []} for element in soup.find_all(['h1', 'h2', 'h3', 'p', 'li', 'table', 'pre']): if element.name in ['h1', 'h2', 'h3']: if current_section["content"]: documents.append(Document( content=self._format_section(current_section), metadata={"source": file_path, "section": current_section["title"]} )) current_section = {"title": element.get_text(), "content": []} else: current_section["content"].append(self._extract_content(element)) if current_section["content"]: documents.append(Document( content=self._format_section(current_section), metadata={"source": file_path, "section": current_section["title"]} )) return documents class DatabaseParser(DocumentParser): def parse(self, conn_config: dict) -> List[Document]: """从数据库提取知识""" import sqlalchemy as sa engine = sa.create_engine(conn_config["url"]) documents = [] for table_name in conn_config["tables"]: query = f"SELECT * FROM {table_name}" df = pd.read_sql(query, engine) # 将每行转为一个文档 for _, row in df.iterrows(): content = self._row_to_text(row, table_name) documents.append(Document( content=content, metadata={ "source": f"db://{table_name}", "table": table_name, "primary_key": str(row.get('id', '')) } )) return documents class UnifiedParser: """统一解析入口""" def __init__(self): self.parsers = { ".pdf": PDFParser(), ".html": HTMLParser(), ".htm": HTMLParser(), ".docx": DocxParser(), ".md": MarkdownParser(), ".txt": TextParser(), } def parse(self, file_path: str) -> List[Document]: ext = Path(file_path).suffix.lower() parser = self.parsers.get(ext, TextParser()) return parser.parse(file_path) 2. 数据清洗 Pipeline class DataCleaner: def __init__(self): self.steps = [ self._remove_boilerplate, # 去除样板文字 self._normalize_whitespace, # 规范化空白 self._fix_encoding, # 修复编码 self._remove_duplicates, # 去重 self._filter_quality, # 质量过滤 self._add_metadata, # 补充元数据 ] def clean(self, documents: List[Document]) -> List[Document]: for step in self.steps: documents = step(documents) return documents def _remove_boilerplate(self, docs): """去除样板文字""" boilerplate_patterns = [ r"版权所有.*?保留所有权利", r"Cookie.*?设置", r"订阅我们的.*?通讯", r"© \d{4}.*?", ] cleaned = [] for doc in docs: text = doc.content for pattern in boilerplate_patterns: text = re.sub(pattern, "", text, flags=re.IGNORECASE) doc.content = text.strip() if len(doc.content) > 50: # 过滤过短的 cleaned.append(doc) return cleaned def _filter_quality(self, docs): """质量过滤""" filtered = [] for doc in docs: # 检查文本质量 quality_score = self._assess_quality(doc.content) if quality_score > 0.5: doc.metadata["quality_score"] = quality_score filtered.append(doc) return filtered def _assess_quality(self, text: str) -> float: score = 1.0 # 过短 if len(text) < 50: score -= 0.3 # 重复内容过多 words = text.split() if len(words) > 0: unique_ratio = len(set(words)) / len(words) if unique_ratio < 0.3: score -= 0.3 # 特殊字符过多 special_chars = sum(1 for c in text if not c.isalnum() and not c.isspace()) if len(text) > 0 and special_chars / len(text) > 0.2: score -= 0.2 # 乱码检测 if re.search(r'[\ufffd]{3,}', text): score -= 0.5 return max(0, score) 3. 数据增强 class DataAugmentor: """为原始文档增加结构化信息""" def augment(self, documents: List[Document]) -> List[Document]: for doc in documents: # 1. 生成摘要 doc.metadata["summary"] = self._generate_summary(doc.content) # 2. 提取关键词 doc.metadata["keywords"] = self._extract_keywords(doc.content) # 3. 生成问答对（用于 FAQ 式检索） doc.metadata["qa_pairs"] = self._generate_qa_pairs(doc.content) # 4. 实体标注 doc.metadata["entities"] = self._extract_entities(doc.content) # 5. 生成假设性问题（HyDE 优化） doc.metadata["hypothetical_questions"] = self._generate_hyde(doc.content) return documents def _generate_hyde(self, content: str) -> List[str]: """生成该内容可能回答的问题，用于提升检索召回""" prompt = f""" 基于以下内容，生成 3 个用户可能会问的问题，这些问题可以由这段内容回答： 内容：{content[:1000]} 输出 JSON 列表：["问题1", "问题2", "问题3"] """ return llm.generate(prompt, response_format="json") 4. 分块与索引 class ChunkingAndIndexing: def __init__(self): self.chunker = HybridChunker() self.embedder = EmbeddingModel("bge-m3") self.sparse_embedder = BM25Encoder() def process(self, documents: List[Document]): all_chunks = [] for doc in documents: # 1. 智能分块 chunks = self.chunker.chunk(doc.content) # 2. 为每个 chunk 添加上下文 for i, chunk in enumerate(chunks): chunk.metadata = { **doc.metadata, "chunk_index": i, "total_chunks": len(chunks), "context_before": chunks[i-1].text[-100:] if i > 0 else "", "context_after": chunks[i+1].text[:100] if i < len(chunks)-1 else "" } # 3. 生成 Embedding（稠密 + 稀疏） chunk.dense_embedding = self.embedder.encode(chunk.text) chunk.sparse_embedding = self.sparse_embedder.encode(chunk.text) # 4. HyDE embedding（用假设问题做额外 embedding） if "hypothetical_questions" in doc.metadata: hyde_embeddings = [ self.embedder.encode(q) for q in doc.metadata["hypothetical_questions"] ] chunk.hyde_embeddings = hyde_embeddings all_chunks.append(chunk) # 5. 写入向量数据库 self._write_to_index(all_chunks) return all_chunks 5. 质量监控 class PipelineQualityMonitor: def __init__(self): self.metrics = {} def monitor(self, pipeline_run): report = { "input": { "total_documents": len(pipeline_run.input_docs), "total_size_mb": pipeline_run.input_size, }, "parsing": { "success_rate": pipeline_run.parsed / pipeline_run.total, "failed_files": pipeline_run.failed, "avg_parse_time": pipeline_run.avg_parse_time, }, "cleaning": { "retention_rate": len(pipeline_run.cleaned) / len(pipeline_run.parsed), "avg_quality_score": np.mean([d.metadata.get("quality_score", 0) for d in pipeline_run.cleaned]), }, "chunking": { "total_chunks": len(pipeline_run.chunks), "avg_chunk_size": np.mean([len(c.text) for c in pipeline_run.chunks]), "chunks_per_doc": len(pipeline_run.chunks) / len(pipeline_run.cleaned), }, "indexing": { "index_time": pipeline_run.index_time, "index_size_mb": pipeline_run.index_size, "vectors_written": len(pipeline_run.chunks), } } # 告警 if report["parsing"]["success_rate"] < 0.95: alert("解析成功率低于 95%") if report["cleaning"]["avg_quality_score"] < 0.7: alert("平均质量分数低于 0.7") if report["chunking"]["avg_chunk_size"] > 2000: alert("平均块大小过大") return report 6. 增量更新与调度 class IncrementalPipeline: """支持增量更新的数据管道""" def run(self, data_source: str): # 1. 发现新增/修改/删除的文件 changes = self._detect_changes(data_source) if not changes["added"] and not changes["modified"] and not changes["deleted"]: return "No changes detected" # 2. 处理新增和修改 new_docs = [] for file_path in changes["added"] + changes["modified"]: docs = self.parser.parse(file_path) docs = self.cleaner.clean(docs) docs = self.augmentor.augment(docs) new_docs.extend(docs) # 3. 处理删除 for file_path in changes["deleted"]: self.index.delete_by_source(file_path) # 4. 更新索引 if new_docs: self.indexer.process(new_docs) # 5. 记录变更 self._update_file_registry(changes) # 6. 质量检查 quality_report = self.monitor.monitor(pipeline_run) return quality_report def _detect_changes(self, source: str) -> dict: """检测文件变更""" current_files = self._scan_files(source) registered_files = self._get_registered_files() added = set(current_files) - set(registered_files) deleted = set(registered_files) - set(current_files) modified = { f for f in current_files if f in registered_files and current_files[f] != registered_files[f] # 比较 hash } return {"added": list(added), "modified": list(modified), "deleted": list(deleted)} 总结 RAG 数据管道是系统工程，核心原则是： ...

为什么重排序是 RAG 的必备环节 向量检索（Bi-Encoder）速度快但精度有限，因为它将 query 和文档独立编码。重排序（Cross-Encoder）将 query 和文档拼接在一起送入模型，能捕获更精细的语义交互，显著提升检索精度。 向量检索 (Bi-Encoder) 重排序 (Cross-Encoder) Q → [Embedding] → ← [Embedding] ← Doc Q + Doc → [Cross-Encoder] → Score 速度快，精度中等 速度慢，精度高 召回阶段 (Top-50) 精排阶段 (Top-5) 三大重排序方案概览 特性 Cohere Rerank BGE-Reranker Jina Reranker 类型 闭源 API 开源模型 开源模型 + API 最大序列长度 4096 8192 8192 多语言 ✅ 100+语言 ✅ 中英文为主 ✅ 100+语言 部署方式 仅 SaaS 自托管 自托管/SaaS 延迟 (P99) 200ms 150ms (GPU) 180ms (GPU) 成本 $2/1K调用 GPU成本 GPU或$1/1K 实战对比 1. Cohere Rerank v4 import cohere client = cohere.Client(api_key="your-api-key") def cohere_rerank(query: str, documents: list, top_n: int = 5): response = client.rerank( model="rerank-multilingual-v3.0", query=query, documents=documents, top_n=top_n, max_tokens_per_doc=4096 ) return [ { "index": r.index, "document": documents[r.index], "relevance_score": r.relevance_score } for r in response.results ] 优点：开箱即用、多语言强、稳定可靠 缺点：依赖外部 API、有数据隐私顾虑、按调用计费 ...

在 RAG 系统中，Embedding 负责粗排，Reranker 负责精排。一个优质的 Reranker 可以将 RAG 的检索准确率提升 10-20%。2026 年，Reranker 市场形成了 Cohere（闭源标杆）、BGE（开源之光）、Jina（多语言专家）三足鼎立的格局。本文将在真实 RAG 场景中对三者进行全面对比。 一、Reranker 基础 Reranker vs Embedding 维度 Embedding Reranker 架构 双塔（Bi-Encoder） 交叉（Cross-Encoder） 计算方式 独立编码，余弦相似度 联合编码，输出相关性分数 速度 快（可预计算） 慢（每对实时计算） 精度 中（粗排） 高（精排） 在RAG中的角色 第一阶段召回 第二阶段重排 典型 RAG 检索流程： 查询 → Embedding检索(Top-100) → Reranker重排(Top-5) → 生成 为什么需要 Reranker？ 方案 Top-5 准确率 端到端准确率 仅 Embedding 78.3% 75.5% Embedding + Reranker 91.2% 87.3% 加入 Reranker 后 Top-5 准确率提升 12.9%，端到端准确率提升 11.8%——这是 RAG 系统中投入产出比最高的优化之一。 ...