RAG：检索增强生成技术深度解析¶

from sentence_transformers import SentenceTransformer

# 加载嵌入模型
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 文本转向量
texts = [
    "RAG是一种检索增强生成技术",
    "大语言模型可以理解和生成自然语言",
    "向量数据库用于存储和检索向量"
]

# 生成嵌入向量
embeddings = model.encode(texts)
print(f"向量维度: {embeddings.shape}")  # (3, 1024)

import numpy as np
from numpy.linalg import norm

def cosine_similarity(vec1, vec2):
    """计算余弦相似度"""
    return np.dot(vec1, vec2) / (norm(vec1) * norm(vec2))

# 查询向量与文档向量的相似度计算
query_embedding = model.encode("什么是RAG技术？")
similarities = [cosine_similarity(query_embedding, doc_emb) for doc_emb in doc_embeddings]

from rank_bm25 import BM25Okapi

# BM25检索示例
corpus = [doc.split() for doc in documents]
bm25 = BM25Okapi(corpus)

query = "RAG检索增强"
tokenized_query = query.split()
scores = bm25.get_scores(tokenized_query)

def hybrid_search(query, alpha=0.5):
    """
    混合检索：结合向量检索和BM25
    alpha: 向量检索权重 (0-1)
    """
    # 向量检索得分
    dense_scores = vector_search(query)
    # BM25检索得分
    sparse_scores = bm25_search(query)

    # 归一化并加权融合
    dense_scores_norm = normalize(dense_scores)
    sparse_scores_norm = normalize(sparse_scores)

    final_scores = alpha * dense_scores_norm + (1 - alpha) * sparse_scores_norm
    return final_scores

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 推荐的分块配置
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,           # 每块约500字符
    chunk_overlap=50,         # 块之间重叠50字符，保持上下文连贯
    separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""],
    length_function=len,
)

chunks = text_splitter.split_text(document)

from sentence_transformers import CrossEncoder

# 加载重排序模型
reranker = CrossEncoder('BAAI/bge-reranker-large')

# 对检索结果重排序
query = "什么是RAG？"
passages = ["RAG是检索增强生成...", "LLM是大语言模型...", ...]

# 计算相关性得分
pairs = [[query, passage] for passage in passages]
scores = reranker.predict(pairs)

# 按得分排序
ranked_results = sorted(zip(passages, scores), key=lambda x: x[1], reverse=True)

RAG_PROMPT_TEMPLATE = """你是一个专业的AI助手。请基于以下提供的参考资料回答用户的问题。

## 参考资料
{context}

## 回答要求
1. 仅基于上述参考资料进行回答
2. 如果参考资料中没有相关信息，请明确告知用户
3. 回答要准确、简洁、有条理
4. 如适用，请引用具体的参考资料来源

## 用户问题
{question}

## 回答
"""

def query_rewrite(original_query, llm):
    """使用LLM改写查询"""
    prompt = f"""请将以下用户查询改写为更适合检索的形式，生成3个不同角度的查询：

原始查询：{original_query}

改写后的查询（每行一个）："""

    rewritten_queries = llm.generate(prompt)
    return rewritten_queries.split('\n')

def hyde_retrieval(query, llm, retriever):
    """HyDE检索策略"""
    # 1. 生成假设性答案
    hypothetical_answer = llm.generate(
        f"请回答以下问题（即使你不确定）：{query}"
    )

    # 2. 用假设性答案进行检索
    results = retriever.search(hypothetical_answer)

    return results

def multi_hop_retrieval(query, retriever, llm, max_hops=3):
    """多跳检索"""
    context = []
    current_query = query

    for hop in range(max_hops):
        # 检索相关文档
        docs = retriever.search(current_query)
        context.extend(docs)

        # 判断是否需要继续检索
        need_more = llm.generate(
            f"基于当前信息，是否需要更多信息来回答'{query}'？"
        )

        if "不需要" in need_more:
            break

        # 生成下一轮查询
        current_query = llm.generate(
            f"为了回答'{query}'，还需要查询什么信息？"
        )

    return context

def adaptive_retrieval(query, retriever):
    """自适应检索策略"""
    # 分析查询类型
    query_type = classify_query(query)

    if query_type == "factual":
        # 事实性问题：精确检索
        return retriever.search(query, top_k=3, threshold=0.8)
    elif query_type == "exploratory":
        # 探索性问题：广泛检索
        return retriever.search(query, top_k=10, threshold=0.5)
    elif query_type == "comparative":
        # 比较性问题：多角度检索
        return multi_perspective_search(query, retriever)
    else:
        return retriever.search(query, top_k=5)

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)

# 准备评估数据
eval_data = {
    "question": ["什么是RAG？"],
    "answer": ["RAG是检索增强生成技术..."],
    "contexts": [["RAG全称是...", "RAG的核心是..."]],
    "ground_truth": ["RAG是一种结合检索和生成的AI技术"]
}

# 执行评估
result = evaluate(
    eval_data,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)

print(result)

"""
RAG系统完整实现示例
"""

from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Milvus
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

class RAGSystem:
    def __init__(self, 
                 embedding_model="BAAI/bge-large-zh-v1.5",
                 llm_model="gpt-4",
                 collection_name="my_knowledge_base"):

        # 初始化嵌入模型
        self.embeddings = HuggingFaceEmbeddings(
            model_name=embedding_model,
            model_kwargs={'device': 'cuda'}
        )

        # 初始化LLM
        self.llm = ChatOpenAI(
            model=llm_model,
            temperature=0
        )

        # 初始化向量数据库
        self.vector_store = Milvus(
            embedding_function=self.embeddings,
            collection_name=collection_name,
            connection_args={"host": "localhost", "port": "19530"}
        )

        # 初始化文本分割器
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )

        # 设置Prompt模板
        self.prompt_template = PromptTemplate(
            template="""基于以下参考资料回答问题。如果资料中没有相关信息，请说明。

参考资料：
{context}

问题：{question}

回答：""",
            input_variables=["context", "question"]
        )

    def add_documents(self, file_paths):
        """添加文档到知识库"""
        all_chunks = []

        for file_path in file_paths:
            # 加载文档
            loader = PyPDFLoader(file_path)
            documents = loader.load()

            # 分块
            chunks = self.text_splitter.split_documents(documents)
            all_chunks.extend(chunks)

        # 添加到向量数据库
        self.vector_store.add_documents(all_chunks)
        print(f"成功添加 {len(all_chunks)} 个文档块")

    def query(self, question, top_k=5):
        """查询知识库"""
        # 创建检索链
        qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vector_store.as_retriever(
                search_kwargs={"k": top_k}
            ),
            chain_type_kwargs={"prompt": self.prompt_template},
            return_source_documents=True
        )

        # 执行查询
        result = qa_chain({"query": question})

        return {
            "answer": result["result"],
            "sources": [doc.page_content for doc in result["source_documents"]]
        }


# 使用示例
if __name__ == "__main__":
    # 初始化RAG系统
    rag = RAGSystem()

    # 添加文档
    rag.add_documents(["document1.pdf", "document2.pdf"])

    # 查询
    result = rag.query("什么是RAG技术？")
    print("回答:", result["answer"])
    print("来源:", result["sources"])