Agent如果只依赖LLM的内置知识就像一个只靠记忆力的学生——知识有限且会过时。RAGRetrieval-Augmented Generation让Agent能从外部知识库中检索专业、实时的信息大幅提升回答的准确性和深度。本章将从全流程解析到高级策略系统讲解RAG技术。6.1 RAG全流程解析数据清洗、分块与索引构建RAG的核心流程离线阶段文档 - 清洗 - 分块 - Embedding - 向量数据库 在线阶段问题 - Embedding - 检索 - 拼入提示词 - LLM生成数据清洗原始文档通常格式混乱、包含噪音清洗质量直接影响检索效果import re class DocumentCleaner: 文档清洗工具 def clean(self, text: str) - str: text self._remove_boilerplate(text) text self._normalize_whitespace(text) text self._remove_special_chars(text) return text.strip() def _remove_boilerplate(self, text: str) - str: 移除页眉页脚、版权声明等模板文本 patterns [ r版权所有.*?保留一切权利, r本文档仅供参考.*?不构成任何建议, r第\s*\d\s*页\s*/\s*\d, ] for pattern in patterns: text re.sub(pattern, , text, flagsre.IGNORECASE) return text def _normalize_whitespace(self, text: str) - str: 规范化空白字符 text re.sub(r\n{3,}, \n\n, text) # 多个换行压缩为两个 text re.sub(r[ \t], , text) # 多个空格压缩为一个 return text def _remove_special_chars(self, text: str) - str: 移除不可见字符和乱码 text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f], , text) return text文本分块策略分块是RAG中最关键的环节。块太大则检索精度低块太小则上下文不完整。策略一固定长度分块from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, # 每块500字符 chunk_overlap50, # 块间重叠50字符 separators[\n\n, \n, 。, , , , ], )策略二语义分块按语义边界段落、章节分块而非机械切割from langchain.text_splitter import MarkdownHeaderTextSplitter # 按Markdown标题层级分块 md_splitter MarkdownHeaderTextSplitter( headers_to_split_on[ (#, h1), (##, h2), (###, h3), ] ) chunks md_splitter.split_text(markdown_doc) # 每个chunk会自动携带其所属的标题层级信息策略三父-子分块Small-to-Big Retrieval检索时用小块精度高返回时用大块上下文完整from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore # 子块分块器检索用 child_splitter RecursiveCharacterTextSplitter(chunk_size200) # 父块分块器返回用 parent_splitter RecursiveCharacterTextSplitter(chunk_size1000) vectorstore Chroma(embedding_functionembeddings) docstore InMemoryStore() # 存储父块原文 retriever ParentDocumentRetriever( vectorstorevectorstore, docstoredocstore, child_splitterchild_splitter, parent_splitterparent_splitter, )分块策略优势劣势适用场景固定长度简单可控可能切断语义通用场景语义分块保留语义完整依赖文档结构Markdown/HTML文档父-子分块兼顾精度与上下文实现复杂高质量RAG索引构建from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 从文档构建索引 vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./knowledge_base, collection_metadata{hnsw:space: cosine} # 使用余弦相似度 )6.2 高级检索策略混合搜索与重排序技术基础向量检索的局限纯向量检索存在一个根本性问题语义相似不等于答案相关。比如用户问Python如何安装向量检索可能返回一篇标题为安装Python的各种方法但内容过时的文章而错过一篇标题不太相关但内容精确的教程。混合搜索向量 关键词混合搜索将向量检索语义匹配和BM25关键词检索精确匹配结合取长补短from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever # 向量检索器 vector_retriever vectorstore.as_retriever(search_kwargs{k: 10}) # BM25关键词检索器 bm25_retriever BM25Retriever.from_documents(chunks, k10) # 混合检索器 ensemble_retriever EnsembleRetriever( retrievers[vector_retriever, bm25_retriever], weights[0.5, 0.5] # 向量和关键词各占50%权重 ) results ensemble_retriever.invoke(Python安装教程)重排序Re-ranking检索回来的文档按相似度排序但相似度最高的不一定是最相关的。重排序模型能更精确地评估文档与查询的相关性from langchain.retrievers import ContextualCompressionRetriever from langchain_cohere import CohereRerank # 使用Cohere重排序模型 compressor CohereRerank(modelrerank-v3.5, top_n5) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverensemble_retriever ) # 先检索大量候选再重排序筛选最相关的5条 results compression_retriever.invoke(如何优化RAG的检索效果)查询改写与扩展用户的原始查询可能不够精确。通过LLM改写或扩展查询可以提升检索召回率class QueryRewriter: def __init__(self, llm): self.llm llm def expand_query(self, original_query: str) - list[str]: 将原始查询扩展为多个子查询 prompt f 原始查询{original_query} 请生成3个不同角度的子查询帮助检索到更全面的信息 1. 同义词替换版本 2. 更具体的细化版本 3. 更宽泛的上位版本 response self.llm.invoke(prompt) return [original_query] self._parse_queries(response.content) def hyde_query(self, original_query: str) - str: HyDE让LLM先假设性回答用回答来检索 prompt f请回答以下问题即使不确定也给出你的最佳猜测{original_query} hypothetical_answer self.llm.invoke(prompt).content return hypothetical_answer # 用假设性答案作为检索查询6.3 知识图谱与Agent的结合结构化数据的利用向量检索的盲区向量检索擅长语义匹配但不擅长处理实体关系和结构化查询。比如和马斯克共同创立PayPal的人还创立了哪些公司——这类多跳关系查询向量检索几乎无法处理。知识图谱增强RAGfrom langchain_community.graphs import Neo4jGraph from langchain.chains import GraphCypherQAChain # 连接知识图谱 graph Neo4jGraph( urlbolt://localhost:7687, usernameneo4j, passwordyour_password ) # 用LLM生成Cypher查询 chain GraphCypherQAChain.from_llm( llmChatOpenAI(modelgpt-4o, temperature0), graphgraph, verboseTrue, ) result chain.run(和马斯克共同创立PayPal的人还创立了哪些公司) # LLM自动生成MATCH (p:Person)-[:COFOUNDED]-(c:Company {name: PayPal})-[:COFOUNDED]-(other:Person) # MATCH (other)-[:COFOUNDED]-(other_c:Company) # RETURN other.name, collect(other_c.name)GraphRAG微软的图谱增强方案微软的GraphRAG方案流程抽取从文档中抽取实体和关系构建构建社区图谱发现实体集群索引为每个社区生成摘要检索先定位相关社区再在社区内检索# 简化版实体抽取 def extract_entities_and_relations(text: str, llm) - dict: prompt f从以下文本中抽取实体和关系以JSON格式返回 文本{text} 格式{{ entities: [{{name: 实体名, type: 类型}}], relations: [{{source: 实体1, target: 实体2, relation: 关系}}] }} response llm.invoke(prompt) return json.loads(response.content)6.4 解决RAG痛点丢失中间内容与多跳推理问题痛点一Lost in the Middle研究表明LLM对上下文中间位置的信息注意力最弱。当检索返回多个文档片段时排在中间的片段容易被忽略。解决方案文档重排def relevance_ordered_placement(documents: list[str], strategy: str descending) - list[str]: 按相关性重新排列文档避免重要信息被放在中间 # 方案1递减排列 - 最相关的在前 if strategy descending: return documents # 保持检索排序 # 方案2交替排列 - 最相关和次相关的交替放置 if strategy alternating: result [] left, right 0, len(documents) - 1 while left right: result.append(documents[left]) if left ! right: result.append(documents[right]) left 1 right - 1 return result return documents痛点二多跳推理谁是美国第46任总统的妻子的出生地——这需要两跳推理第46任总统是谁 - 他的妻子是谁 - 她的出生地。解决方案迭代检索class MultiHopRetriever: def __init__(self, retriever, llm): self.retriever retriever self.llm llm def retrieve(self, query: str, max_hops: int 3) - list[str]: all_docs [] current_query query for hop in range(max_hops): docs self.retriever.invoke(current_query) all_docs.extend(docs) # 判断是否需要继续检索 judge_prompt f 原始问题{query} 已检索到的信息{[d.page_content[:200] for d in all_docs]} 问题是否已经可以被完整回答 如果否请生成下一步需要检索的子问题。 response self.llm.invoke(judge_prompt).content if 是 in response and 完整 in response: break # 提取下一步检索的子问题 current_query response.split(子问题)[-1].strip() if 子问题 in response else query return all_docs痛点三信息冲突不同来源的信息可能互相矛盾def resolve_conflicts(documents: list[str], query: str, llm) - str: 处理信息冲突 conflict_prompt f 问题{query} 检索到的信息可能有冲突 {chr(10).join(f来源{i1}{d} for i, d in enumerate(documents))} 请分析 1. 哪些信息之间存在冲突 2. 每个冲突的可能原因是什么时效性、来源可靠性等 3. 你倾向采纳哪个版本为什么 return llm.invoke(conflict_prompt).content6.5 知识库动态更新增量索引与版本管理策略为什么需要动态更新知识库不是一成不变的。产品文档会更新、政策法规会调整、技术方案会演进。一个过时的知识库比没有知识库更危险——它会给出错误信息。增量索引class IncrementalIndexer: def __init__(self, vectorstore, embeddings): self.vectorstore vectorstore self.embeddings embeddings self.doc_hashes: dict[str, str] {} # doc_id - content_hash def _content_hash(self, content: str) - str: import hashlib return hashlib.md5(content.encode()).hexdigest() def update(self, documents: list) - dict: 增量更新索引 added, updated, unchanged 0, 0, 0 for doc in documents: doc_id doc.metadata.get(doc_id, str(id(doc))) new_hash self._content_hash(doc.page_content) if doc_id not in self.doc_hashes: # 新文档 self.vectorstore.add_documents([doc]) self.doc_hashes[doc_id] new_hash added 1 elif self.doc_hashes[doc_id] ! new_hash: # 文档有变更删除旧版本再添加新版本 self.vectorstore.delete(ids[doc_id]) self.vectorstore.add_documents([doc]) self.doc_hashes[doc_id] new_hash updated 1 else: unchanged 1 return {added: added, updated: updated, unchanged: unchanged}自动更新触发机制import hashlib from datetime import datetime class AutoUpdater: 监控源文件变化自动触发索引更新 def __init__(self, indexer: IncrementalIndexer, source_dir: str): self.indexer indexer self.source_dir source_dir self.file_hashes: dict[str, str] {} def scan_and_update(self) - dict: 扫描源目录检测变化并更新 changes {added: [], modified: [], deleted: []} # 扫描当前文件 current_files {} for root, dirs, files in os.walk(self.source_dir): for f in files: if f.endswith((.md, .txt, .pdf, .docx)): filepath os.path.join(root, f) with open(filepath, rb) as file: file_hash hashlib.md5(file.read()).hexdigest() current_files[filepath] file_hash # 检测新增和修改 for filepath, file_hash in current_files.items(): if filepath not in self.file_hashes: changes[added].append(filepath) elif self.file_hashes[filepath] ! file_hash: changes[modified].append(filepath) # 检测删除 for filepath in self.file_hashes: if filepath not in current_files: changes[deleted].append(filepath) # 执行更新 # ... (将变更文件加载、分块、写入索引) self.file_hashes current_files return changes本章小结RAG环节关键技术核心要点数据清洗正则去噪、格式规范化垃圾进垃圾出清洗是基础分块策略固定长度/语义/父-子父-子分块兼顾精度与上下文混合检索向量BM25重排序语义匹配精确匹配互补查询优化改写/扩展/HyDE用LLM优化检索查询知识图谱CypherGraphRAG结构化关系多跳推理痛点解决文档重排/迭代检索/冲突消解中间丢失/多跳/信息冲突动态更新增量索引/版本管理/自动扫描知识库不是一次性工程学习资源推荐如果你想更深入地学习大模型以下是一些非常有价值的学习资源这些资源将帮助你从不同角度学习大模型提升你的实践能力。一、全套AGI大模型学习路线AI大模型时代的学习之旅从基础到前沿掌握人工智能的核心技能​因篇幅有限仅展示部分资料需要点击文章最下方名片即可前往获取二、640套AI大模型报告合集这套包含640份报告的合集涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师还是对AI大模型感兴趣的爱好者这套报告合集都将为您提供宝贵的信息和启示​因篇幅有限仅展示部分资料需要点击文章最下方名片即可前往获取三、AI大模型经典PDF籍随着人工智能技术的飞速发展AI大模型已经成为了当今科技领域的一大热点。这些大型预训练模型如GPT-3、BERT、XLNet等以其强大的语言理解和生成能力正在改变我们对人工智能的认识。 那以下这些PDF籍就是非常不错的学习资源。因篇幅有限仅展示部分资料需要点击文章最下方名片即可前往获取四、AI大模型商业化落地方案作为普通人入局大模型时代需要持续学习和实践不断提高自己的技能和认知水平同时也需要有责任感和伦理意识为人工智能的健康发展贡献力量。