知识库与检索增强

检索增强生成（RAG）基础

检索增强生成（RAG）是一种技术，通过从知识库中检索相关信息来增强语言模型的能力。

知识库的组织

1. 文档加载

from langchain.document_loaders import PyPDFLoader, TextLoader

# 加载PDF
pdf_loader = PyPDFLoader("document.pdf")
pdf_docs = pdf_loader.load()

# 加载文本文件
text_loader = TextLoader("knowledge.txt")
text_docs = text_loader.load()

# 加载多种格式
documents = pdf_docs + text_docs

2. 文档分割

from langchain.text_splitters import RecursiveCharacterTextSplitter

# 分割文档
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每个块的大小
    chunk_overlap=200,    # 块之间的重叠
    separators=["\n\n", "\n", "。", "，"]  # 优先分割点
)

chunks = splitter.split_documents(documents)

3. 向量化和存储

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

# 创建嵌入
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建向量存储
vector_store = FAISS.from_documents(
    chunks,
    embeddings
)

# 保存到磁盘
vector_store.save_local("vector_db")

检索策略

1. 相似度检索

def retrieve_similar_documents(query: str, vector_store, k: int = 5):
    """检索与查询最相似的文档"""
    
    # 基本相似度检索
    results = vector_store.similarity_search(
        query,
        k=k
    )
    
    return results

# 使用
query = "如何安装项目?"
results = retrieve_similar_documents(query, vector_store)

for doc in results:
    print(f"相关度: {doc.metadata.get('relevance_score', 'N/A')}")
    print(f"内容: {doc.page_content[:200]}...")

2. 最大边际相关性（MMR）检索

def retrieve_diverse_documents(query: str, vector_store, k: int = 5):
    """检索多样化的相关文档"""
    
    # 使用MMR避免检索到重复信息
    results = vector_store.max_marginal_relevance_search(
        query,
        k=k,
        fetch_k=20  # 从最近的20条中选择
    )
    
    return results

3. 元数据过滤

def retrieve_with_filters(query: str, vector_store, filters: dict):
    """基于元数据过滤的检索"""
    
    # 检索并应用过滤器
    results = vector_store.similarity_search(
        query,
        k=10,
        filter=filters  # 例如 {"category": "API", "version": "v2"}
    )
    
    return results

知识库集成到代理

1. 创建检索工具

from langchain.tools import tool

@tool
def search_knowledge_base(query: str, k: int = 5) -> str:
    """在知识库中搜索信息
    
    Args:
        query: 搜索查询
        k: 返回结果数
    """
    # 加载存储的向量数据库
    from langchain.vectorstores import FAISS
    from langchain.embeddings import OpenAIEmbeddings
    
    embeddings = OpenAIEmbeddings()
    vector_store = FAISS.load_local("vector_db", embeddings)
    
    # 检索文档
    results = vector_store.similarity_search(query, k=k)
    
    # 格式化结果
    formatted_results = []
    for i, doc in enumerate(results, 1):
        formatted_results.append(
            f"{i}. {doc.page_content}\n"
            f"   来源: {doc.metadata.get('source', 'Unknown')}"
        )
    
    return "\n".join(formatted_results)

# 在代理中使用
agent = create_agent(
    model="anthropic:claude-sonnet-4",
    tools=[search_knowledge_base],
    system_prompt="你是一个知识库助手。回答问题前，使用search_knowledge_base工具检索相关信息。"
)

2. RAG代理流程

高级RAG技术

1. 多级检索（Hierarchical Retrieval）

def hierarchical_retrieval(query: str, vector_store):
    """分层检索：先查摘要，后查详细内容"""
    
    # 第一级：检索摘要文档
    summary_results = vector_store.similarity_search(
        query,
        k=3,
        filter={"type": "summary"}
    )
    
    # 第二级：基于摘要，检索详细文档
    detail_results = []
    for summary in summary_results:
        source = summary.metadata.get("source")
        details = vector_store.similarity_search(
            query,
            k=5,
            filter={"source": source, "type": "detail"}
        )
        detail_results.extend(details)
    
    return {
        "summaries": summary_results,
        "details": detail_results
    }

2. 上下文压缩

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMCompressor

# 创建压缩检索器
compressor = LLMCompressor.from_llm_and_prompt(
    llm=ChatAnthropic(model="claude-sonnet-4"),
    prompt=compression_prompt
)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_store.as_retriever()
)

# 使用压缩检索
compressed_docs = compression_retriever.get_relevant_documents(query)

3. 混合检索（Hybrid Retrieval）

def hybrid_retrieval(query: str, vector_store, text_index):
    """结合向量搜索和关键词搜索"""
    
    # 向量相似度搜索
    vector_results = vector_store.similarity_search(query, k=5)
    
    # 关键词搜索
    keyword_results = text_index.search(query, k=5)
    
    # 合并并去重
    all_results = {}
    
    for doc in vector_results:
        doc_id = doc.metadata.get("id")
        all_results[doc_id] = {
            "doc": doc,
            "vector_score": 1.0,
            "keyword_score": 0.0
        }
    
    for doc in keyword_results:
        doc_id = doc.metadata.get("id")
        if doc_id in all_results:
            all_results[doc_id]["keyword_score"] = 1.0
        else:
            all_results[doc_id] = {
                "doc": doc,
                "vector_score": 0.0,
                "keyword_score": 1.0
            }
    
    # 按综合分数排序
    sorted_results = sorted(
        all_results.items(),
        key=lambda x: x[1]["vector_score"] + x[1]["keyword_score"],
        reverse=True
    )
    
    return [item[1]["doc"] for item in sorted_results[:5]]

知识库维护

1. 增量更新

def update_knowledge_base(new_documents, vector_store):
    """增量添加新文档到知识库"""
    
    from langchain.embeddings import OpenAIEmbeddings
    
    embeddings = OpenAIEmbeddings()
    
    # 分割新文档
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200
    )
    chunks = splitter.split_documents(new_documents)
    
    # 添加到向量存储
    vector_store.add_documents(chunks)
    
    # 保存更新
    vector_store.save_local("vector_db")

2. 版本管理

import datetime

class VersionedKnowledgeBase:
    def __init__(self, base_path: str):
        self.base_path = base_path
        self.versions = []
    
    def save_version(self, vector_store, version_note: str):
        """保存知识库的版本"""
        timestamp = datetime.datetime.now().isoformat()
        version = {
            "timestamp": timestamp,
            "note": version_note,
            "path": f"{self.base_path}_v{len(self.versions)}"
        }
        
        vector_store.save_local(version["path"])
        self.versions.append(version)
    
    def list_versions(self):
        """列出所有版本"""
        for i, version in enumerate(self.versions):
            print(f"版本 {i}: {version['timestamp']} - {version['note']}")
    
    def load_version(self, version_index: int):
        """加载特定版本的知识库"""
        version = self.versions[version_index]
        # 加载向量存储
        from langchain.vectorstores import FAISS
        from langchain.embeddings import OpenAIEmbeddings
        return FAISS.load_local(version["path"], OpenAIEmbeddings())

常见问题

Q: 知识库应该有多大？ A: 取决于应用场景。从小规模（10-100文档）开始，根据需要扩展。

Q: 如何处理过时的信息？ A: 定期更新知识库，使用版本管理，标记过时内容。

Q: 检索的准确性不高怎么办？ A: 优化文档分割策略，改进查询重写，使用混合检索。