RAG检索增强生成完全指南【2026】- 让AI更智能的关键技术

prompt_template = """
基于以下参考信息回答用户问题。如果参考信息中没有相关内容，请明确说明。

参考信息：
{context}

用户问题：{question}

回答：
"""

import numpy as np

def cosine_similarity(vec1, vec2):
    """余弦相似度：最常用的方法"""
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

def euclidean_distance(vec1, vec2):
    """欧氏距离：距离越小越相似"""
    return np.linalg.norm(vec1 - vec2)

def dot_product(vec1, vec2):
    """点积：向量已归一化时等价于余弦相似度"""
    return np.dot(vec1, vec2)

from langchain.text_splitter import RecursiveCharacterTextSplitter

def split_documents(documents, chunk_size=500, chunk_overlap=50):
    """
    文档分块策略
    - chunk_size: 每块大小，建议300-1000字符
    - chunk_overlap: 重叠部分，保持上下文连贯
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "！", "？", "；", " "]
    )
    return splitter.split_documents(documents)

from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings

openai_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

local_embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-base-zh-v1.5",
    model_kwargs={'device': 'cuda'}
)

from langchain.vectorstores import Chroma

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    return_source_documents=True
)

result = qa_chain.invoke({"query": "什么是RAG技术？"})
print(result["result"])

import os
from typing import List, Dict, Any
from langchain.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA

class RAGSystem:
    """完整的RAG系统实现"""
    
    def __init__(
        self,
        embedding_model: str = "text-embedding-3-small",
        llm_model: str = "gpt-4-turbo",
        persist_directory: str = "./rag_db"
    ):
        self.embeddings = OpenAIEmbeddings(model=embedding_model)
        self.llm = ChatOpenAI(model=llm_model, temperature=0)
        self.persist_directory = persist_directory
        self.vectorstore = None
        self.qa_chain = None
        
    def load_documents(self, directory: str, glob: str = "**/*.txt") -> List:
        """加载文档"""
        loader = DirectoryLoader(
            directory,
            glob=glob,
            loader_cls=TextLoader,
            loader_kwargs={'encoding': 'utf-8'}
        )
        return loader.load()
    
    def process_documents(
        self,
        documents: List,
        chunk_size: int = 500,
        chunk_overlap: int = 50
    ) -> List:
        """文档分块处理"""
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", " "]
        )
        return splitter.split_documents(documents)
    
    def build_vectorstore(self, chunks: List) -> None:
        """构建向量数据库"""
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_directory
        )
        
    def setup_qa_chain(self, k: int = 5) -> None:
        """设置问答链"""
        prompt_template = """你是一个专业的AI助手。请基于以下参考信息回答用户问题。
        
如果参考信息中没有相关内容，请明确说明"根据现有资料无法回答该问题"。
回答时请尽量引用具体来源。

参考信息：
{context}

用户问题：{question}

回答："""
        
        prompt = PromptTemplate(
            template=prompt_template,
            input_variables=["context", "question"]
        )
        
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(
                search_kwargs={"k": k}
            ),
            return_source_documents=True,
            chain_type_kwargs={"prompt": prompt}
        )
    
    def query(self, question: str) -> Dict[str, Any]:
        """执行查询"""
        if not self.qa_chain:
            raise ValueError("请先调用setup_qa_chain()初始化问答链")
        
        result = self.qa_chain.invoke({"query": question})
        return {
            "answer": result["result"],
            "sources": [
                {
                    "content": doc.page_content[:200] + "...",
                    "metadata": doc.metadata
                }
                for doc in result["source_documents"]
            ]
        }

if __name__ == "__main__":
    rag = RAGSystem()
    
    docs = rag.load_documents("./knowledge_base")
    chunks = rag.process_documents(docs)
    rag.build_vectorstore(chunks)
    rag.setup_qa_chain()
    
    result = rag.query("RAG技术的主要优势是什么？")
    print(f"回答：{result['answer']}")
    print(f"\n引用来源：")
    for i, source in enumerate(result['sources'], 1):
        print(f"{i}. {source['content']}")

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

class AdvancedRAGSystem(RAGSystem):
    """高级RAG系统：支持多种检索策略"""
    
    def setup_hybrid_retriever(self, k: int = 5):
        """混合检索：结合语义搜索和关键词搜索"""
        from langchain.retrievers import BM25Retriever, EnsembleRetriever
        
        bm25_retriever = BM25Retriever.from_documents(self.chunks)
        bm25_retriever.k = k
        
        dense_retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": k}
        )
        
        self.retriever = EnsembleRetriever(
            retrievers=[bm25_retriever, dense_retriever],
            weights=[0.3, 0.7]
        )
    
    def setup_reranking(self, k: int = 5):
        """重排序：使用LLM对检索结果进行重新排序"""
        base_retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": k * 2}
        )
        
        compressor = LLMChainExtractor.from_llm(self.llm)
        
        self.retriever = ContextualCompressionRetriever(
            base_compressor=compressor,
            base_retriever=base_retriever
        )
    
    def setup_multi_query(self):
        """多查询检索：生成多个查询变体提高召回率"""
        from langchain.retrievers.multi_query import MultiQueryRetriever
        
        self.retriever = MultiQueryRetriever.from_llm(
            retriever=self.vectorstore.as_retriever(),
            llm=self.llm
        )

optimized_prompt = """
你是{domain}领域的专家助手。

## 任务
基于提供的参考资料回答用户问题。

## 规则
1. 仅使用参考资料中的信息
2. 如果资料不足，明确说明
3. 引用具体来源增加可信度
4. 使用清晰、专业的语言

## 参考资料
{context}

## 用户问题
{question}

## 回答
"""