您的位置：首页 > > 教程攻略 > ai教程 >第6章检索增强生成：打造知识库驱动型Agent

第6章检索增强生成：打造知识库驱动型Agent

来源:互联网 更新时间:2026-07-27 22:15

第6章检索增强生成：打造知识库驱动型Agent

想象一下，一个Agent如果只靠LLM内置的知识工作，就像那种只靠记忆力的学生——知识储备有限，而且会随着时间推移逐渐过时。RAG，即检索增强生成，正是解决这个问题的关键方案。它让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, flags=re.IGNORECASE)
        return text
    def _normalize_whitespace(self, text: str) -> str:
        """规范化空白字符"""
        text = re.sub(r"n{3,}", "nn", text)  # 多个换行压缩为两个
        text = re.sub(r"[ t]+", " ", text)  # 多个空格压缩为一个
        return text
    def _remove_special_chars(self, text: str) -> str:
        """移除不可见字符和乱码"""
        text = re.sub(r"[x00-x08x0bx0cx0e-x1fx7f]", "", text)
        return text

文本分块策略

分块是整个RAG中最关键的环节，没有之一。块太大，检索精度低，容易把无关信息拽进来；块太小，上下文不完整，模型理解起来也费劲。这里介绍三种主流策略。

策略一：固定长度分块

最简单直接的方式，用RecursiveCharacterTextSplitter按设定字符数切分，并允许块间有重叠，避免切断关键语句。适合通用场景，但可能切断语义边界。

from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,  # 每块500字符
    chunk_overlap=50,  # 块间重叠50字符
    separators=["nn", "n", "。", "！", "？", "；", " "],
)

策略二：语义分块

按照语义边界，比如段落、章节来切分，而不是机械地按字符数切。对于Markdown或HTML文档，可以直接按标题层级分块，每个块自动携带所属的标题信息，上下文更完整。

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）

这是目前高质量RAG中比较常用的方案。检索时用小块，保证精度；返回时用大块，确保上下文完整。通过ParentDocumentRetriever实现，子块负责检索，父块负责提供完整信息。

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
# 子块分块器（检索用）
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)
# 父块分块器（返回用）
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
vectorstore = Chroma(embedding_function=embeddings)
docstore = InMemoryStore()  # 存储父块原文
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

这三种策略各有优劣，用一张表总结一下：

分块策略	优势	劣势	适用场景
固定长度	简单可控	可能切断语义	通用场景
语义分块	保留语义完整	依赖文档结构	Markdown/HTML文档
父-子分块	兼顾精度与上下文	实现复杂	高质量RAG

索引构建

完成分块后，需要将文本转换成向量并存入向量数据库。这里以Chroma为例，使用OpenAI的embedding模型，并指定余弦相似度作为距离度量。

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 从文档构建索引
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./knowledge_base",
    collection_metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
)

6.2 高级检索策略：混合搜索与重排序技术

基础向量检索的局限

纯向量检索虽然强大，但有一个根本性的问题：语义相似不等于答案相关。举个例子，用户问“Python如何安装”，向量检索可能返回一篇标题为“安装Python的各种方法”但内容过时的文章，而错过一篇标题不太相关但内容精准的教程。这种偏差在实际应用中经常出现。

混合搜索：向量 + 关键词

混合搜索的思路很直接：把向量检索（擅长语义匹配）和BM25关键词检索（擅长精确匹配）结合起来，取长补短。通过EnsembleRetriever可以轻松实现，为两种检索方式分配权重，比如各占50%。

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, k=10)
# 混合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.5, 0.5]  # 向量和关键词各占50%权重
)
results = ensemble_retriever.invoke("Python安装教程")

重排序（Re-ranking）

检索回来的文档通常是按相似度排序的，但相似度最高的不一定是最相关的。重排序模型能更精确地评估文档与查询的相关性，把真正有用的内容排到前面。这里用Cohere的rerank模型，先检索出一批候选文档，再重排序筛选出最相关的5条。

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
# 使用Cohere重排序模型
compressor = CohereRerank(model="rerank-v3.5", top_n=5)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever
)
# 先检索大量候选，再重排序筛选最相关的5条
results = compression_retriever.invoke("如何优化RAG的检索效果？")

查询改写与扩展

用户的原始查询往往不够精确，可能包含歧义或用词不准确。通过LLM对查询进行改写或扩展，可以有效提升检索召回率。这里定义了两种方法：一种是生成多个不同角度的子查询，比如同义词替换、细化版本、宽泛版本；另一种是HyDE方法，让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的人还创立了哪些公司？”——这种多跳关系查询，向量检索几乎无法处理，因为它缺乏对实体之间关系的理解。

知识图谱增强RAG

知识图谱正好弥补这个短板。通过Neo4j这样的图数据库，可以用Cypher查询语言表达复杂的多跳关系。这里用GraphCypherQAChain让LLM自动生成Cypher查询，把自然语言问题转化为图数据库查询，从而精准回答结构化问题。

from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain
# 连接知识图谱
graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="your_password"
)
# 用LLM生成Cypher查询
chain = GraphCypherQAChain.from_llm(
    llm=ChatOpenAI(model="gpt-4o", temperature=0),
    graph=graph,
    verbose=True,
)
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

一个被广泛验证的现象：模型在长上下文场景下，对中间位置的信息注意力最弱。当检索返回多个文档片段时，排在中间的那些片段容易被忽略，即使它们可能包含关键信息。解决方案之一是文档重排——把最相关的文档放在最前面，或者采用交替排列的方式，避免重要信息被淹没在中间。

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任总统是谁，再找到他的妻子，最后查出她的出生地。单次检索很难完成。迭代检索方案可以解决这个问题：每次检索后，让LLM判断当前信息是否足以回答原始问题，如果不够，则生成下一步需要检索的子问题，然后继续检索，直到信息完整。

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

痛点三：信息冲突

不同来源的信息可能互相矛盾，比如一个文档说A事件发生在2020年，另一个说发生在2021年。简单地把所有信息扔给模型，模型可能会产生困惑。信息冲突消解的思路是：让LLM分析冲突的来源，比如时效性差异、来源可靠性不同等，然后给出倾向性判断。

def resolve_conflicts(documents: list[str], query: str, llm) -> str:
    """处理信息冲突"""
    conflict_prompt = f"""问题：{query}
检索到的信息（可能有冲突）：
{chr(10).join(f'来源{i+1}：{d}' for i, d in enumerate(documents))}
请分析：
1. 哪些信息之间存在冲突？
2. 每个冲突的可能原因是什么（时效性、来源可靠性等）？
3. 你倾向采纳哪个版本？为什么？"""
    return llm.invoke(conflict_prompt).content

6.5 知识库动态更新：增量索引与版本管理策略

为什么需要动态更新？

知识库不是一次性工程。产品文档会更新，政策法规会调整，技术方案会演进。一个过时的知识库比没有知识库更危险——它会给出错误信息，误导用户。因此，动态更新能力是生产级RAG系统不可或缺的。

增量索引

增量索引的核心思路是：只对变更的文档进行更新，而不是全量重建。通过维护文档内容的哈希值，可以快速判断文档是否新增、变更或未变，然后只对需要更新的部分执行操作。这样既节省资源，又能保证索引的实时性。

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}

版本管理策略

版本管理是增量索引的进阶版。除了记录文档的最新版本，还保留历史版本，支持回滚操作。每个文档维护一个版本列表，新版本写入时标记为“current”，回滚时则恢复指定版本的内容。这在知识库质量出现问题时非常有用。

class VersionedKnowledgeBase:
    def __init__(self, vectorstore):
        self.vectorstore = vectorstore
        self.versions: dict[str, list[dict]] = {}  # doc_id -> 版本列表
    def add_version(self, doc_id: str, content: str, metadata: dict = None):
        """添加文档的新版本"""
        version = {
            "content": content,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat(),
            "version": len(self.versions.get(doc_id, [])) + 1
        }
        self.versions.setdefault(doc_id, []).append(version)
        # 只将最新版本写入向量库（标记为current）
        self.vectorstore.add_documents([{
            "page_content": content,
            "metadata": {**(metadata or {}), "doc_id": doc_id, "version": version["version"], "status": "current"}
        }])
    def rollback(self, doc_id: str, target_version: int):
        """回滚到指定版本"""
        if doc_id not in self.versions:
            return False
        versions = self.versions[doc_id]
        target = next((v for v in versions if v["version"] == target_version), None)
        if not target:
            return False
        # 删除当前版本，恢复目标版本
        self.add_version(doc_id, target["content"], target["metadata"])
        return True

自动更新触发机制

手动触发更新总归不是长久之计。一个自动更新机制可以监控源文件目录的变化，当文件新增、修改或删除时，自动触发增量索引更新。通过记录每个文件的哈希值，可以精确检测到哪些文件发生了变化，然后只对变更的文件执行更新操作。

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的整个流程从数据清洗到动态更新，每个环节都有其关键技术和核心要点。下面这张表可以帮你快速回顾：

RAG环节	关键技术	核心要点
数据清洗	正则去噪、格式规范化	垃圾进=垃圾出，清洗是基础
分块策略	固定长度/语义/父-子	父-子分块兼顾精度与上下文
混合检索	向量+BM25+重排序	语义匹配+精确匹配互补
查询优化	改写/扩展/HyDE	用LLM优化检索查询
知识图谱	Cypher+GraphRAG	结构化关系+多跳推理
痛点解决	文档重排/迭代检索/冲突消解	中间丢失/多跳/信息冲突
动态更新	增量索引/版本管理/自动扫描	知识库不是一次性工程

第6章检索增强生成：打造知识库驱动型Agent