RAG与微调

引言 RAG系统的第一步是将文档分块（Chunking）。这个看似简单的步骤，实际上对RAG质量有巨大影响——分块太大，检索不精确；分块太小，丢失上下文。 2026年，分块策略已经从简单的"按字数切分"发展出多种智能策略。本文将系统对比这些策略。 一、为什么分块很重要 1.1 检索精度 文档: "Python是一种解释型语言。它由Guido van Rossum于1991年创建。Python强调代码可读性..." 分块A（太大大）: 整个文档作为一个块 → 检索时可能返回很多不相关的内容 分块B（太小）: "Python是一种解释型语言。" → 缺少上下文，不知道在说什么 分块C（合适）: "Python是一种解释型语言。它由Guido van Rossum于1991年创建。" → 信息完整且精确 1.2 上下文完整性 好的分块应该保持语义完整性——一个块应该表达一个完整的意思。 1.3 检索效率 块的大小直接影响向量索引的大小和检索速度。 二、分块策略分类 2.1 固定大小分块 class FixedSizeChunker: def __init__(self, chunk_size=500, overlap=50): self.chunk_size = chunk_size # 字符数或token数 self.overlap = overlap def chunk(self, text): chunks = [] start = 0 while start < len(text): end = start + self.chunk_size chunk = text[start:end] chunks.append(chunk) start = end - self.overlap # 重叠部分 return chunks 优势：简单、快速、可预测。 劣势：可能在句子中间截断，破坏语义。 最佳实践：设置50-100 token的重叠，减少截断影响。 2.2 句子分块 class SentenceChunker: def __init__(self, max_sentences=5, max_tokens=500): self.max_sentences = max_sentences self.max_tokens = max_tokens def chunk(self, text): sentences = self.split_sentences(text) chunks = [] current_chunk = [] current_tokens = 0 for sentence in sentences: sent_tokens = self.count_tokens(sentence) if current_tokens + sent_tokens > self.max_tokens or \ len(current_chunk) >= self.max_sentences: if current_chunk: chunks.append(" ".join(current_chunk)) current_chunk = [sentence] current_tokens = sent_tokens else: current_chunk.append(sentence) current_tokens += sent_tokens if current_chunk: chunks.append(" ".join(current_chunk)) return chunks 优势：保持句子完整，语义更好。 劣势：块大小不均匀。 ...

引言 “我的RAG系统效果怎么样？“这是每个RAG开发者都会问的问题。但回答这个问题并不简单——RAG系统包含检索和生成两个环节，每个环节都有多个质量维度。 2026年，RAG评估已经形成了一套比较完整的指标体系。从检索的准确率和召回率，到生成的忠实度和相关性，再到端到端的用户满意度。本文将系统介绍这些指标。 一、RAG评估框架 1.1 评估维度 RAG系统 ├── 检索质量 │ ├── 准确率（Precision） │ ├── 召回率（Recall） │ ├── 排序质量（NDCG/MRR） │ └── 多样性（Diversity） ├── 生成质量 │ ├── 忠实度（Faithfulness） │ ├── 答案相关性（Answer Relevance） │ ├── 完整性（Completeness） │ └── 正确性（Correctness） └── 端到端质量 ├── 用户满意度 ├── 响应延迟 └── 成本效率 1.2 评估方法 人工评估：最准确但成本高 自动评估：使用LLM或规则自动评估 基准测试：在标准数据集上测试 A/B测试：线上对比不同版本 二、检索质量指标 2.1 准确率与召回率 class RetrievalMetrics: def precision_at_k(self, retrieved, relevant, k): """Top-K准确率""" retrieved_k = retrieved[:k] relevant_set = set(relevant) retrieved_set = set(retrieved_k) return len(relevant_set & relevant_set) / k def recall_at_k(self, retrieved, relevant, k): """Top-K召回率""" retrieved_k = retrieved[:k] relevant_set = set(relevant) retrieved_set = set(retrieved_k) return len(retrieved_set & relevant_set) / len(relevant_set) def average_precision(self, retrieved, relevant): """平均准确率""" relevant_set = set(relevant) precisions = [] num_relevant = 0 for i, doc in enumerate(retrieved): if doc in relevant_set: num_relevant += 1 precisions.append(num_relevant / (i + 1)) if not precisions: return 0.0 return np.mean(precisions) def mean_average_precision(self, queries_results): """MAP""" return np.mean([self.average_precision(r["retrieved"], r["relevant"]) for r in queries_results]) 2.2 排序质量 def ndcg_at_k(self, retrieved, relevance_scores, k): """NDCG@K""" # DCG dcg = sum( (2 ** relevance_scores.get(doc, 0) - 1) / np.log2(i + 2) for i, doc in enumerate(retrieved[:k]) ) # IDCG（理想排序的DCG） ideal_order = sorted(relevance_scores.values(), reverse=True)[:k] idcg = sum( (2 ** score - 1) / np.log2(i + 2) for i, score in enumerate(ideal_order) ) return dcg / idcg if idcg > 0 else 0 def mrr(self, queries_results): """平均倒数排名""" reciprocal_ranks = [] for result in queries_results: for i, doc in enumerate(result["retrieved"]): if doc in result["relevant"]: reciprocal_ranks.append(1 / (i + 1)) break else: reciprocal_ranks.append(0) return np.mean(reciprocal_ranks) 2.3 上下文相关性 class ContextRelevanceMetric: async def measure(self, question, retrieved_context): """评估检索到的上下文与问题的相关性""" prompt = f""" 问题: {question} 检索到的上下文: {self.format_context(retrieved_context)} 请评估上下文与问题的相关性: 1. 有多少段落直接相关？ 2. 有多少段落间接相关？ 3. 有多少段落完全不相关？ 4. 相关性评分（0-1） """ result = await self.llm.call(prompt) return result["relevance_score"] 三、生成质量指标 3.1 忠实度（Faithfulness） 答案是否忠实于检索到的上下文，不产生幻觉： ...

引言 真实世界的信息不只存在于文本中。图片、视频、音频、表格——这些非文本模态包含着文本无法替代的信息。传统的文本RAG无法处理"找一张包含红色汽车的产品图片"这类查询。 2026年，多模态RAG已经成为AI系统的标配能力。它能够跨模态检索信息、融合多模态上下文、生成多模态输出。本文将深入探讨多模态RAG的架构与实践。 一、多模态RAG的核心挑战 1.1 表示鸿沟 不同模态的数据需要不同的编码方式： 文本：文本embedding模型 图像：视觉embedding模型 音频：音频embedding模型 如何让不同模态的表示在同一个空间中可比？ 1.2 检索鸿沟 用户可能用文本查询图像，也可能用图像查询文本： 文本→图像: "找一张展示AI架构的图" 图像→文本: "这张图对应的技术文档在哪？" 图像→图像: "找类似的图片" 1.3 融合鸿沟 检索到的多模态信息如何融合？文本和图像的信息如何整合在一起喂给LLM？ 二、架构设计 2.1 整体架构 ┌─────────────────────────────────────────────┐ │ 多模态查询处理 │ │ (文本/图像/音频 → 统一表示) │ ├─────────────────────────────────────────────┤ │ 跨模态检索引擎 │ │ ┌─────────┬─────────┬─────────┐ │ │ │文本检索 │图像检索 │音频检索 │ │ │ └─────────┴─────────┴─────────┘ │ ├─────────────────────────────────────────────┤ │ 多模态融合层 │ │ (跨模态排序、上下文组装) │ ├─────────────────────────────────────────────┤ │ 多模态生成 │ │ (文本+图像+表格 → 统一回答) │ └─────────────────────────────────────────────┘ 2.2 多模态编码 class MultiModalEncoder: def __init__(self): self.text_encoder = load_model("text-embedding-3-large") self.image_encoder = load_model("clip-vit-large") self.audio_encoder = load_model("audio-embedding-model") async def encode(self, content, modality): """编码不同模态的内容""" if modality == "text": return await self.text_encoder.embed(content) elif modality == "image": return await self.image_encoder.encode(content) elif modality == "audio": return await self.audio_encoder.encode(content) elif modality == "mixed": # 混合内容：分别编码后融合 return await self.encode_mixed(content) async def encode_mixed(self, content): """编码混合模态内容""" embeddings = [] for item in content: emb = await self.encode(item["data"], item["modality"]) embeddings.append({"embedding": emb, "modality": item["modality"]}) # 使用跨模态对齐模型将不同模态的embedding映射到统一空间 unified = await self.align_embeddings(embeddings) return unified 2.3 跨模态对齐 使用CLIP等模型实现跨模态对齐： ...

引言 向量检索擅长语义匹配——“这段文字和我的问题有多相似”。但它不擅长关系推理——“A的上级的上级是谁”。知识图谱检索擅长关系推理，但不擅长模糊语义匹配。 2026年，混合RAG——将图检索和向量检索结合——已经成为处理复杂知识问答的最佳方案。本文将深入探讨这种混合架构。 一、为什么需要混合检索 1.1 向量检索的局限 问题: "爱因斯坦的博士导师是谁？" 向量检索: 搜索"爱因斯坦 博士 导师" → 可能找到: "爱因斯坦在苏黎世联邦理工学院学习" (语义相关但缺关键信息) 可能找不到: "Alfred Kleiner是爱因斯坦的博士论文导师" (语义距离较远) 1.2 图检索的局限 问题: "量子力学的哲学意义是什么？" 图检索: 需要遍历"量子力学"→"哲学意义"的边 → 但"哲学意义"是一个抽象概念，知识图谱中可能没有对应的节点 1.3 混合的优势 问题: "爱因斯坦的博士导师的研究领域是什么？" 混合检索: 1. 向量检索: "爱因斯坦 博士 导师" → 找到相关文档 2. 实体识别: 识别出"Alfred Kleiner" 3. 图检索: 查询"Alfred Kleiner" → "研究领域" → "实验物理学" 4. 向量检索: "Alfred Kleiner 实验物理学" → 找到详细描述 5. 综合答案: "爱因斯坦的博士导师Alfred Kleiner的研究领域是实验物理学..." 二、混合RAG架构 2.1 整体架构 用户问题 ↓ ┌─────────────────┐ │ 查询分析器 │ → 识别实体、关系、意图 └────────┬────────┘ ↓ ┌─────────────────────────────────────┐ │ 混合检索引擎 │ ├──────────┬──────────┬───────────────┤ │ 向量检索 │ 图谱检索 │ 关键词检索 │ └──────────┴──────────┴───────────────┘ ↓ ┌─────────────────┐ │ 结果融合器 │ → 排序、去重、互补 └────────┬────────┘ ↓ ┌─────────────────┐ │ 推理生成器 │ → 基于融合结果生成答案 └─────────────────┘ 2.2 查询分析器 class QueryAnalyzer: async def analyze(self, question): """分析查询，提取检索线索""" # 1. 实体识别 entities = await self.ner.extract(question) # 2. 关系识别 relations = await self.relation_extractor.extract(question, entities) # 3. 意图识别 intent = await self.intent_classifier.classify(question) # 4. 生成不同检索策略的查询 return { "entities": entities, "relations": relations, "intent": intent, "vector_query": await self.generate_vector_query(question, entities), "graph_query": self.generate_graph_query(entities, relations), "keyword_query": await self.generate_keyword_query(question) } 2.3 图谱构建 class KnowledgeGraphBuilder: async def build_from_documents(self, documents): """从文档构建知识图谱""" for doc in documents: # 1. 实体抽取 entities = await self.entity_extractor.extract(doc.text) # 2. 关系抽取 relations = await self.relation_extractor.extract(doc.text, entities) # 3. 添加到图谱 for entity in entities: await self.graph.add_node( id=entity.id, label=entity.text, type=entity.type, properties={"source": doc.id, "embedding": entity.embedding} ) for relation in relations: await self.graph.add_edge( source=relation.subject, target=relation.object, label=relation.predicate, properties={"confidence": relation.confidence} ) # 4. 同时添加到向量库 await self.vector_store.add( id=doc.id, text=doc.text, embedding=doc.embedding, metadata={"entities": [e.id for e in entities]} ) 2.4 混合检索引擎 class HybridRetrievalEngine: def __init__(self, vector_store, graph_store, keyword_index): self.vector_store = vector_store self.graph_store = graph_store self.keyword_index = keyword_index async def retrieve(self, query_analysis, top_k=10): # 1. 并行执行三种检索 vector_task = self.vector_search(query_analysis["vector_query"], top_k*2) graph_task = self.graph_search(query_analysis) keyword_task = self.keyword_search(query_analysis["keyword_query"], top_k*2) vector_results, graph_results, keyword_results = await asyncio.gather( vector_task, graph_task, keyword_task ) # 2. 融合结果 fused = self.fuse_results(vector_results, graph_results, keyword_results) # 3. 重排 reranked = await self.rerank(fused, query_analysis) return reranked[:top_k] async def graph_search(self, query_analysis): """图谱检索""" results = [] # 1. 实体匹配：在图谱中找到查询实体的对应节点 for entity in query_analysis["entities"]: matched_nodes = await self.graph_store.find_nodes( label=entity.text, fuzzy=True ) # 2. 关系遍历：沿着关系边扩展 for node in matched_nodes: for relation in query_analysis["relations"]: neighbors = await self.graph_store.traverse( start=node, edge_label=relation.predicate, max_depth=2 ) results.extend(neighbors) # 3. 获取关联文档 for result in results: if result.has_property("source"): doc = await self.document_store.get(result.source) result.content = doc.text return results def fuse_results(self, vector_results, graph_results, keyword_results): """融合三种检索结果""" fused = {} # 为每种检索结果分配权重 weights = {"vector": 0.4, "graph": 0.4, "keyword": 0.2} for result_type, results, weight in [ ("vector", vector_results, weights["vector"]), ("graph", graph_results, weights["graph"]), ("keyword", keyword_results, weights["keyword"]) ]: for rank, result in enumerate(results): doc_id = result.id if doc_id not in fused: fused[doc_id] = { "result": result, "score": 0, "sources": [] } # 倒数排名融合，加上权重 fused[doc_id]["score"] += weight * (1 / (60 + rank)) fused[doc_id]["sources"].append(result_type) # 按融合分数排序 sorted_results = sorted(fused.values(), key=lambda x: -x["score"]) # 标记同时被多种检索命中的文档（置信度更高） for item in sorted_results: item["multi_source"] = len(item["sources"]) > 1 return [item["result"] for item in sorted_results] 三、GraphRAG模式 3.1 社区检测 将知识图谱划分为社区，每个社区是一组紧密相关的实体： ...

引言 基础RAG（检索-增强-生成）已经成为了AI应用的标配：用户提问→检索相关文档→LLM基于文档生成回答。但这种简单模式在面对复杂问题时力不从心。 2026年，RAG架构已经演进出多种高级模式：多跳检索、自适应检索、推理增强检索、迭代检索等。这些模式让RAG系统能够处理更复杂的问题、提供更准确的答案。本文将系统介绍这些高级模式。 一、基础RAG的局限 1.1 单次检索不够 复杂问题需要多步推理和多轮检索： 问题: "对比GPT-4和Claude在代码生成任务上的表现" 基础RAG: 一次检索"GPT-4 Claude 代码生成对比" → 可能找不到直接的对比文档 高级RAG: 检索1: "GPT-4 代码生成性能" 检索2: "Claude 代码生成性能" 检索3: "GPT-4 vs Claude benchmark" → 综合多个来源生成对比 1.2 检索质量不稳定 简单向量检索可能返回语义相关但信息不足的文档。 1.3 缺乏推理能力 基础RAG是"检索+拼接"，缺乏对检索结果的推理和整合。 二、多跳检索 2.1 原理 多跳检索（Multi-hop Retrieval）通过多轮检索逐步收集信息，每轮检索基于前一轮的结果。 问题: "那位发明了Transformer的作者后来创立了哪家公司？" 跳1: 检索"Transformer论文作者" → 得到作者列表（包括Ashish Vaswani等） 跳2: 检索"Ashish Vaswani 创立公司" → 得到Adept AI 答案: Adept AI 2.2 实现 class MultiHopRetriever: def __init__(self, retriever, llm, max_hops=3): self.retriever = retriever self.llm = llm self.max_hops = max_hops async def retrieve(self, question): accumulated_context = [] for hop in range(self.max_hops): # 1. 根据当前上下文生成下一步检索查询 if hop == 0: query = question else: query = await self.generate_next_query( question, accumulated_context ) if query is None: # 不需要更多检索 break # 2. 执行检索 results = await self.retriever.search(query, top_k=5) # 3. 过滤和评估结果 relevant = await self.filter_relevant(results, question, accumulated_context) accumulated_context.extend(relevant) # 4. 检查是否已有足够信息回答 if await self.has_enough_info(question, accumulated_context): break return accumulated_context async def generate_next_query(self, question, context): """基于当前上下文生成下一步查询""" prompt = f""" 原始问题: {question} 已知信息: {context} 还需要什么信息才能回答问题？请生成一个搜索查询。 如果已经有足够信息，返回"DONE"。 """ response = await self.llm.call(prompt) if "DONE" in response: return None return response.strip() 三、自适应检索 3.1 原理 不是所有问题都需要检索，也不是所有问题都需要相同类型的检索。自适应检索根据问题特征选择最优的检索策略。 ...

引言 大模型的对齐（Alignment）是让模型输出符合人类期望和价值的关键技术。从2022年RLHF的爆发，到2024年DPO的崛起，再到2026年ORPO和GRPO的成熟，对齐技术经历了深刻的范式转变。本文系统梳理这一演进历程。 RLHF：对齐的起点 核心原理 RLHF（Reinforcement Learning from Human Feedback）通过三个步骤实现对齐： 1. SFT（监督微调）：用高质量数据训练基础模型 2. RM（奖励模型）：训练模型评估输出质量 3. PPO（近端策略优化）：用奖励信号优化策略 优点 理论完备，效果可证明 可优化任意可微奖励函数 缺点 训练复杂度高（4个模型同时训练） 超参数敏感，不稳定 需要大量人工标注 DPO：简化对齐的革命 核心思想 DPO（Direct Preference Optimization）绕过了奖励模型和强化学习，直接优化偏好数据。 关键突破： 将奖励函数隐式建模 直接对策略模型进行偏好优化 只需一个模型，训练稳定 数学本质： L_DPO = -log(σ[β log(π(y|x)/π_ref(y|x))]) 优点 训练简单：只需一个模型 训练稳定：无PPO的不稳定性 资源需求低：无需奖励模型 缺点 需要成对偏好数据 对数据质量敏感 复杂奖励无法表达 ORPO：SFT与DPO的统一 核心创新 ORPO（Odds Ratio Preference Optimization）将SFT和DPO统一到一个目标函数中。 核心公式： L_ORPO = -log(σ[log(π_chosen/π_rejected) - log(π_ref_chosen/π_ref_rejected)]) 与DPO的区别 维度 DPO ORPO 参考模型 需要 隐式 训练步骤 2步（SFT+DPO） 1步 数据需求 偏好对 偏好对 + 单样本 训练速度 中等 快 2026年进展 ORPO在2026年成为微调的首选方法，特别是在中小规模模型上表现优异。 ...

引言 检索增强生成（RAG）是2026年最实用的大模型应用架构之一。从基础的"检索-生成"流水线，到Agentic RAG的智能检索代理，RAG技术正在经历深刻的架构演进。本文系统梳理RAG的发展脉络和技术选型。 RAG 的演进阶段 第一阶段：基础 RAG（Naive RAG） 最原始的RAG架构，包含三个步骤： 将文档切分为片段 对片段进行向量化并存储 检索最相关的片段，拼接后输入模型 优点： 简单、快速、易于实现 缺点： 检索质量依赖单一向量相似度，无法处理复杂查询 第二阶段：增强 RAG（Enhanced RAG） 在基础RAG之上增加优化： 查询优化： 查询重写：将用户问题转化为更适合检索的形式 查询分解：将复杂问题拆分为多个子查询 多路检索：同时使用向量检索和关键词检索 后处理： 重排序（Reranking）：对检索结果进行精排 上下文压缩：去除冗余片段，保留关键信息 自适应窗口：根据问题类型调整上下文长度 第三阶段：Agentic RAG Agentic RAG的核心思想：让Agent自主决定如何检索、何时检索、检索什么。 ┌─────────────────────────────────────────────┐ │ Agentic RAG 架构 │ │ │ │ ┌──────────┐ ┌──────────┐ │ │ │ Agent │───▶│ 检索 │ │ │ │ 规划器 │ │ 策略 │ │ │ └──────────┘ └──────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ │ │ │ 反思 │◀───│ 多路 │ │ │ │ 与调整 │ │ 检索 │ │ │ └──────────┘ └──────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────────────────┐ │ │ │ 最终生成与验证 │ │ │ └──────────────────────────┘ │ └─────────────────────────────────────────────┘ 主流 RAG 框架对比 LangChain / LangGraph 特点： 生态最丰富，组件最全 适合： 快速原型开发、复杂RAG管线 ...

从Naive RAG到Agentic RAG的演进 RAG技术的发展经历了三个阶段： Naive RAG (2023) → Advanced RAG (2024) → Agentic RAG (2025-2026) 朴素检索 增强检索 智能体检索 单次查询 多次优化 自主决策 固定管道 模块化 动态规划 传统RAG是一个"固定管道"：查询→检索→拼接→生成，一锤子买卖。Agentic RAG将LLM作为"大脑"，让它自主决定是否需要检索、检索什么、检索几次、何时停止。这就像从"查字典"升级为"咨询一个会查资料的研究员"。 Agentic RAG的核心架构 ┌─────────────────────────────────────────────────────┐ │ Agentic RAG Architecture │ ├─────────────────────────────────────────────────────┤ │ │ │ ┌──────────┐ ┌──────────────────────────────┐ │ │ │ User │───▶│ Agent Controller │ │ │ │ Query │ │ (LLM + Planning + Memory) │ │ │ └──────────┘ └──────────┬───────────────────┘ │ │ │ │ │ ┌─────────┼─────────┐ │ │ ▼ ▼ ▼ │ │ ┌─────────┐ ┌───────┐ ┌─────────┐ │ │ │ Router │ │ Query │ │ Evaluator│ │ │ │ (路由) │ │Rewrite│ │ (评估器) │ │ │ └────┬────┘ └───┬───┘ └────┬────┘ │ │ │ │ │ │ │ ┌─────────┼──────────┼──────────┘ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌────────┐ ┌──────────┐ ┌─────────┐ │ │ │ Vector │ │ Web │ │ SQL │ │ Graph │ │ │ │ Store │ │ Search │ │ DB │ │ Store │ │ │ └──────────┘ └────────┘ └──────────┘ └─────────┘ │ │ │ └─────────────────────────────────────────────────────┘ 四大核心组件 组件 职责 关键能力 Agent Controller 任务规划与决策 分解问题、决定工具调用顺序 Router 工具路由 选择合适的数据源（向量库/搜索引擎/数据库） Query Rewriter 查询改写 将用户问题改写为更有效的检索查询 Evaluator 结果评估 判断检索结果是否充分，决定是否需要追加检索 核心模式解析 模式1：迭代检索（Iterative Retrieval） from langgraph.graph import StateGraph, END from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): question: str retrieved_docs: Annotated[List[str], operator.add] answer: str iteration: int sufficient: bool def retrieve_node(state: AgentState) -> dict: """基于当前问题+已有信息生成查询并检索""" query = query_rewriter(state["question"], state["retrieved_docs"]) docs = vector_store.similarity_search(query, k=5) return {"retrieved_docs": docs, "iteration": state["iteration"] + 1} def evaluate_node(state: AgentState) -> dict: """评估检索结果是否足以回答问题""" if state["iteration"] >= 3: return {"sufficient": True} prompt = f"""基于以下检索到的文档，能否完整回答用户问题？ 问题: {state['question']} 文档: {state['retrieved_docs']} 如果信息充分回答"SUFFICIENT"，否则回答"INSUFFICIENT"并说明缺什么。""" response = llm.invoke(prompt) sufficient = "SUFFICIENT" in response return {"sufficient": sufficient} def answer_node(state: AgentState) -> dict: """基于所有检索文档生成最终回答""" prompt = f"问题: {state['question']}\n\n参考资料:\n{state['retrieved_docs']}" return {"answer": llm.invoke(prompt)} # 构建工作流 workflow = StateGraph(AgentState) workflow.add_node("retrieve", retrieve_node) workflow.add_node("evaluate", evaluate_node) workflow.add_node("answer", answer_node) workflow.set_entry_point("retrieve") workflow.add_edge("retrieve", "evaluate") workflow.add_conditional_edges( "evaluate", lambda state: "answer" if state["sufficient"] else "retrieve" ) workflow.add_edge("answer", END) agent = workflow.compile() 模式2：多路检索（Multi-Route Retrieval） def route_query(state: AgentState) -> str: """根据问题类型选择检索路径""" routing_prompt = f""" 分析以下问题，选择最合适的数据源： 问题: {state['question']} 选项: - "vector": 适合从内部文档库检索事实性信息 - "web": 适合需要最新信息的时效性问题 - "sql": 适合结构化数据查询 - "graph": 适合关系推理和多方关联 返回JSON: {{"route": "...", "reason": "..."}} """ decision = llm.invoke(routing_prompt) return json.loads(decision)["route"] 模式3：自我纠正检索（Self-Corrective RAG） def grade_documents(state: AgentState) -> dict: """对检索文档打分，过滤低质量结果""" graded_docs = [] for doc in state["retrieved_docs"]: score_prompt = f"""评估文档与问题的相关性(0-10): 问题: {state['question']} 文档: {doc.page_content[:500]} 只返回数字。""" score = int(llm.invoke(score_prompt).strip()) if score >= 6: graded_docs.append(doc) # 如果所有文档都不相关，触发查询改写 if not graded_docs: new_query = llm.invoke(f"改写这个查询以获得更好的检索结果: {state['question']}") return {"retrieved_docs": [], "question": new_query} return {"retrieved_docs": graded_docs} 工程实现：基于LangGraph的完整Agentic RAG from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage # 定义工具 @tool def vector_search(query: str) -> str: """从内部知识库检索相关文档""" docs = vector_store.similarity_search(query, k=5) return "\n\n".join([d.page_content for d in docs]) @tool def web_search(query: str) -> str: """从网络搜索最新信息""" results = search_api.search(query, num_results=5) return "\n\n".join([r["content"] for r in results]) @tool def sql_query(question: str) -> str: """查询业务数据库""" sql = text2sql(question) return db.execute(sql) # Agent决策循环 tools = [vector_search, web_search, sql_query] llm = ChatOpenAI(model="gpt-4o", temperature=0) llm_with_tools = llm.bind_tools(tools) def agent_decision(state): messages = state["messages"] response = llm_with_tools.invoke(messages) return {"messages": [response]} def should_continue(state): last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END # 构建Graph graph = StateGraph(dict) graph.add_node("agent", agent_decision) graph.add_node("tools", ToolNode(tools)) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue) graph.add_edge("tools", "agent") # 工具执行后回到agent做下一轮决策 app = graph.compile() # 执行 result = app.invoke({ "messages": [HumanMessage(content="对比2026年Q1三大云厂商的AI服务收入")] }) 性能基准对比 我们在三个数据集上对比了不同RAG架构的表现： ...

从RLHF到DPO的范式转变 RLHF（基于人类反馈的强化学习）是ChatGPT成功的关键技术，但其训练过程极其复杂：需要训练Reward Model、使用PPO算法、维护多个模型副本。 DPO（Direct Preference Optimization）直接绕过Reward Model和强化学习，用一个简单的分类损失函数实现偏好对齐： RLHF Pipeline (复杂): SFT → Reward Model → PPO → 对齐模型 (3个模型，4个训练阶段) DPO Pipeline (简洁): SFT → DPO → 对齐模型 (1个模型，2个训练阶段) DPO核心原理 DPO的损失函数基于一个关键洞察：最优的策略模型可以通过偏好数据直接推导，不需要显式的Reward Model。 # DPO Loss公式 # L_DPO = -log σ(β * (log π(y_w|x)/π_ref(y_w|x) - log π(y_l|x)/π_ref(y_l|x))) # 其中： # π: 当前策略模型 # π_ref: 参考模型（SFT后的模型，冻结） # y_w: 偏好的回答 (winner) # y_l: 不偏好的回答 (loser) # β: 温度参数，控制偏离参考模型的程度 # σ: sigmoid函数 直觉理解：让模型对"好回答"的概率比参考模型更高，对"坏回答"的概率比参考模型更低。 ...

为什么Embedding模型决定RAG上限？ RAG系统的检索质量 = Embedding模型质量 × 向量库能力 × 检索策略。后两者已经相对成熟，Embedding模型成为拉开差距的关键变量。 一个差的Embedding模型，即使后面Reranker再强也无力回天——因为你根本检索不到相关文档。 测评模型清单 我们选取了2026年主流的10款Embedding模型进行测评： 模型 提供方 维度 参数量 最大长度 部署方式 text-embedding-3-large OpenAI 3072 - 8192 API text-embedding-3-small OpenAI 1536 - 8192 API BGE-large-zh-v1.5 BAAI 1024 326M 512 本地 BGE-M3 BAAI 1024 568M 8192 本地 GTE-large-zh 阿里 1024 326M 512 本地 GTE-Qwen2-7B-Instruct 阿里 3584 7B 32K 本地 Jina-embeddings-v3 Jina AI 1024 570M 8192 本地/API Cohere-embed-v4 Cohere 1024 - 512 API E5-large-v2 Microsoft 1024 335M 512 本地 UAE-large-v1 WhereIsAI 1024 335M 512 本地 测评设置 数据集 数据集 文档数 查询数 特点 C-Pack 100万 5000 中文通用检索基准 MTEB-ZH 10万 1000 中文检索评估集 行业知识库(自建) 5万 500 金融、法律、医疗领域 多跳QA(自建) 1万 200 需要多文档关联 评估指标 # 主指标：NDCG@10 (归一化折损累积增益) # 辅助指标：Recall@5, Recall@10, MRR@10 def evaluate_retrieval(qrels, results, k=10): """计算检索评估指标""" ndcg_scores = [] recall_scores = [] mrr_scores = [] for query_id, retrieved_docs in results.items(): relevant_docs = qrels.get(query_id, {}) # NDCG@k ndcg = compute_ndcg(retrieved_docs[:k], relevant_docs) ndcg_scores.append(ndcg) # Recall@k recall = len(set(retrieved_docs[:k]) & set(relevant_docs)) / len(relevant_docs) recall_scores.append(recall) # MRR@k mrr = compute_mrr(retrieved_docs[:k], relevant_docs) mrr_scores.append(mrr) return { "ndcg@10": np.mean(ndcg_scores), "recall@5": np.mean([compute_recall(r, q, 5) for r, q in zip(results, qrels)]), "recall@10": np.mean(recall_scores), "mrr@10": np.mean(mrr_scores), } 测评结果 总体排名（NDCG@10） 排名 模型 C-Pack MTEB-ZH 行业库 多跳QA 平均分 1 GTE-Qwen2-7B 0.823 0.801 0.782 0.698 0.801 2 text-embedding-3-large 0.815 0.792 0.768 0.671 0.787 3 BGE-M3 0.808 0.785 0.771 0.662 0.782 4 Jina-embeddings-v3 0.801 0.778 0.759 0.658 0.774 5 BGE-large-zh-v1.5 0.792 0.771 0.752 0.641 0.764 6 GTE-large-zh 0.788 0.768 0.748 0.635 0.760 7 Cohere-embed-v4 0.781 0.762 0.741 0.628 0.753 8 E5-large-v2 0.772 0.754 0.732 0.615 0.743 9 text-embedding-3-small 0.761 0.748 0.725 0.608 0.736 10 UAE-large-v1 0.752 0.738 0.718 0.601 0.727 关键发现 1. GTE-Qwen2-7B一骑绝尘 GTE-Qwen2-7B作为首款基于大语言模型（7B参数）的Embedding模型，在长文本理解上优势明显： ...