LLM应用开发——高级RAG技术详解

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)
retriever_from_llm = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(), llm=llm
)

# 检索时会并行执行多个改写后的查询
unique_docs = retriever_from_llm.get_relevant_documents(query="如何优化 RAG 的检索精度？")

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

# 使用 Cohere 或 BGE-Reranker
compressor = CohereRerank(model="rerank-english-v3.0", top_n=5)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor, base_retriever=vectorstore.as_retriever()
)

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma

class MultiQueryRAG:
    """多查询 RAG"""
    
    def __init__(self, vectorstore: Chroma):
        self.vectorstore = vectorstore
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    
    def generate_queries(self, query: str, n: int = 3) -> list:
        """生成多个查询变体"""
        prompt = f"""你是一个搜索查询优化专家。
给定用户问题，生成 {n} 个不同角度的搜索查询，以提高检索覆盖率。

用户问题：{query}

请生成 {n} 个查询，每行一个："""
        
        response = self.llm.invoke(prompt)
        queries = [q.strip() for q in response.content.strip().split('\n') if q.strip()]
        return [query] + queries[:n]  # 包含原始查询
    
    def retrieve(self, query: str, k: int = 5) -> list:
        """多查询检索"""
        queries = self.generate_queries(query)
        
        all_docs = []
        seen_ids = set()
        
        for q in queries:
            docs = self.vectorstore.similarity_search(q, k=k)
            for doc in docs:
                doc_id = hash(doc.page_content)
                if doc_id not in seen_ids:
                    seen_ids.add(doc_id)
                    all_docs.append(doc)
        
        return all_docs

# 使用
multi_rag = MultiQueryRAG(vectorstore)
docs = multi_rag.retrieve("如何提高模型性能？")

class QueryDecomposer:
    """查询分解器 - 处理复杂问题"""
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o-mini")
    
    def decompose(self, query: str) -> list:
        """将复杂问题分解为子问题"""
        prompt = f"""将以下复杂问题分解为可独立回答的子问题。

问题：{query}

分解规则：
1. 每个子问题应该可以独立检索和回答
2. 子问题之间应有逻辑顺序
3. 合并所有子问题的答案应能回答原问题

子问题列表（JSON 格式）："""
        
        response = self.llm.invoke(prompt)
        
        import json
        try:
            sub_queries = json.loads(response.content)
        except:
            # 降级处理
            sub_queries = [query]
        
        return sub_queries
    
    def retrieve_and_merge(
        self,
        query: str,
        retriever,
        k: int = 3
    ) -> dict:
        """分解检索并合并"""
        sub_queries = self.decompose(query)
        
        results = {}
        for i, sub_q in enumerate(sub_queries):
            docs = retriever.invoke(sub_q)[:k]
            results[f"子问题{i+1}: {sub_q}"] = docs
        
        return results

# 示例：复杂问题分解
decomposer = QueryDecomposer()
sub_queries = decomposer.decompose(
    "对比 GPT-4 和 Claude 3 在代码生成和数学推理方面的能力差异"
)
# 输出：
# ["GPT-4 的代码生成能力如何？",
#  "Claude 3 的代码生成能力如何？",
#  "GPT-4 的数学推理能力如何？",
#  "Claude 3 的数学推理能力如何？"]

class StepBackRetriever:
    """后退提问检索器"""
    
    def __init__(self, vectorstore):
        self.vectorstore = vectorstore
        self.llm = ChatOpenAI(model="gpt-4o-mini")
    
    def generate_step_back_query(self, query: str) -> str:
        """生成更抽象的后退问题"""
        prompt = f"""给定一个具体问题，生成一个更抽象、更基础的问题，
这个问题的答案将有助于回答原始问题。

原始问题：{query}

更抽象的问题："""
        
        response = self.llm.invoke(prompt)
        return response.content.strip()
    
    def retrieve(self, query: str, k: int = 5) -> dict:
        """后退提问检索"""
        # 原始查询检索
        original_docs = self.vectorstore.similarity_search(query, k=k)
        
        # 后退查询检索
        step_back_query = self.generate_step_back_query(query)
        step_back_docs = self.vectorstore.similarity_search(step_back_query, k=k)
        
        return {
            "original_query": query,
            "step_back_query": step_back_query,
            "original_docs": original_docs,
            "step_back_docs": step_back_docs
        }

# 示例
retriever = StepBackRetriever(vectorstore)
result = retriever.retrieve("为什么 Transformer 使用多头注意力而不是单头？")
# step_back_query: "什么是注意力机制？它的基本原理是什么？"

from langchain.chains import HypotheticalDocumentEmbedder
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

class HyDERetriever:
    """HyDE 检索器"""
    
    def __init__(self, vectorstore):
        self.vectorstore = vectorstore
        self.llm = ChatOpenAI(model="gpt-4o-mini")
        self.embeddings = OpenAIEmbeddings()
    
    def generate_hypothetical_document(self, query: str) -> str:
        """生成假设文档"""
        prompt = f"""请写一段文字，作为对以下问题的详细回答。
这段文字应该像是从一篇专业文档中摘录的。

问题：{query}

假设文档内容："""
        
        response = self.llm.invoke(prompt)
        return response.content
    
    def retrieve(self, query: str, k: int = 5) -> list:
        """HyDE 检索"""
        # 生成假设文档
        hypothetical_doc = self.generate_hypothetical_document(query)
        
        # 嵌入假设文档
        doc_embedding = self.embeddings.embed_query(hypothetical_doc)
        
        # 使用假设文档向量检索
        docs = self.vectorstore.similarity_search_by_vector(
            doc_embedding,
            k=k
        )
        
        return docs
    
    def retrieve_with_comparison(self, query: str, k: int = 5) -> dict:
        """对比 HyDE 和普通检索"""
        # HyDE 检索
        hyde_docs = self.retrieve(query, k)
        
        # 普通检索
        normal_docs = self.vectorstore.similarity_search(query, k)
        
        return {
            "hyde_docs": hyde_docs,
            "normal_docs": normal_docs,
            "hypothetical_doc": self.generate_hypothetical_document(query)
        }

# 使用 LangChain 内置实现
from langchain.chains import HypotheticalDocumentEmbedder

hyde_embeddings = HypotheticalDocumentEmbedder.from_llm(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    base_embeddings=OpenAIEmbeddings(),
    prompt_key="web_search"  # 内置模板
)

class MultiHyDERetriever:
    """多假设文档 HyDE"""
    
    def __init__(self, vectorstore, n_hypotheses: int = 3):
        self.vectorstore = vectorstore
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.8)
        self.embeddings = OpenAIEmbeddings()
        self.n_hypotheses = n_hypotheses
    
    def generate_multiple_hypotheses(self, query: str) -> list:
        """生成多个假设文档"""
        hypotheses = []
        
        for i in range(self.n_hypotheses):
            prompt = f"""从不同角度回答问题，生成第 {i+1} 个假设文档。

问题：{query}

假设文档（角度 {i+1}）："""
            
            response = self.llm.invoke(prompt)
            hypotheses.append(response.content)
        
        return hypotheses
    
    def retrieve(self, query: str, k: int = 5) -> list:
        """多假设 HyDE 检索"""
        hypotheses = self.generate_multiple_hypotheses(query)
        
        all_docs = []
        seen_ids = set()
        
        for hypothesis in hypotheses:
            embedding = self.embeddings.embed_query(hypothesis)
            docs = self.vectorstore.similarity_search_by_vector(embedding, k=k)
            
            for doc in docs:
                doc_id = hash(doc.page_content)
                if doc_id not in seen_ids:
                    seen_ids.add(doc_id)
                    all_docs.append(doc)
        
        return all_docs

from sentence_transformers import CrossEncoder
import numpy as np

class CrossEncoderReranker:
    """Cross-Encoder 重排序器"""
    
    def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.model = CrossEncoder(model_name)
    
    def rerank(
        self,
        query: str,
        documents: list,
        top_k: int = 5
    ) -> list:
        """重排序文档"""
        if not documents:
            return []
        
        # 构建查询-文档对
        pairs = [[query, doc.page_content] for doc in documents]
        
        # 计算相关性分数
        scores = self.model.predict(pairs)
        
        # 按分数排序
        scored_docs = list(zip(documents, scores))
        scored_docs.sort(key=lambda x: x[1], reverse=True)
        
        # 返回 top_k
        return [doc for doc, score in scored_docs[:top_k]]
    
    def rerank_with_scores(
        self,
        query: str,
        documents: list,
        top_k: int = 5
    ) -> list:
        """重排序并返回分数"""
        pairs = [[query, doc.page_content] for doc in documents]
        scores = self.model.predict(pairs)
        
        results = []
        for doc, score in sorted(zip(documents, scores), 
                                  key=lambda x: x[1], reverse=True)[:top_k]:
            results.append({
                "document": doc,
                "score": float(score)
            })
        
        return results

# 使用
reranker = CrossEncoderReranker()

# 初始检索（召回更多）
initial_docs = vectorstore.similarity_search(query, k=20)

# 重排序（精排）
reranked_docs = reranker.rerank(query, initial_docs, top_k=5)

import cohere

class CohereReranker:
    """Cohere Rerank API"""
    
    def __init__(self, api_key: str):
        self.client = cohere.Client(api_key)
    
    def rerank(
        self,
        query: str,
        documents: list,
        top_k: int = 5,
        model: str = "rerank-english-v3.0"
    ) -> list:
        """使用 Cohere 重排序"""
        # 提取文档内容
        doc_texts = [doc.page_content for doc in documents]
        
        # 调用 Cohere API
        results = self.client.rerank(
            query=query,
            documents=doc_texts,
            top_n=top_k,
            model=model
        )
        
        # 重组结果
        reranked = []
        for result in results.results:
            reranked.append({
                "document": documents[result.index],
                "score": result.relevance_score
            })
        
        return reranked

# LangChain 集成
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

compressor = CohereRerank(model="rerank-english-v3.0")
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)

class LLMReranker:
    """使用 LLM 进行重排序"""
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o-mini")
    
    def rerank(
        self,
        query: str,
        documents: list,
        top_k: int = 5
    ) -> list:
        """LLM 重排序"""
        # 构建提示
        doc_list = "\n".join([
            f"[{i}] {doc.page_content[:500]}"
            for i, doc in enumerate(documents)
        ])
        
        prompt = f"""根据与查询的相关性，对以下文档进行排序。

查询：{query}

文档列表：
{doc_list}

请返回最相关的 {top_k} 个文档编号，按相关性从高到低排列。
只返回数字，用逗号分隔，如：2,5,1,3,7"""
        
        response = self.llm.invoke(prompt)
        
        # 解析结果
        try:
            indices = [int(i.strip()) for i in response.content.split(',')]
            return [documents[i] for i in indices[:top_k] if i < len(documents)]
        except:
            return documents[:top_k]
    
    def rerank_with_explanation(
        self,
        query: str,
        documents: list,
        top_k: int = 3
    ) -> list:
        """带解释的重排序"""
        doc_list = "\n".join([
            f"[{i}] {doc.page_content[:300]}"
            for i, doc in enumerate(documents)
        ])
        
        prompt = f"""评估每个文档与查询的相关性。

查询：{query}

文档：
{doc_list}

对每个文档给出 0-10 的相关性分数和简短理由，JSON 格式：
rankings]}}"""
        
        response = self.llm.invoke(prompt)
        
        import json
        try:
            result = json.loads(response.content)
            rankings = sorted(result["rankings"], 
                            key=lambda x: x["score"], reverse=True)
            return [
                {
                    "document": documents[r["index"]],
                    "score": r["score"],
                    "reason": r["reason"]
                }
                for r in rankings[:top_k]
            ]
        except:
            return [{"document": doc} for doc in documents[:top_k]]

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    """混合检索器：BM25 + 向量检索"""
    
    def __init__(self, documents: list, vectorstore):
        self.documents = documents
        self.vectorstore = vectorstore
        
        # 构建 BM25 索引
        tokenized_docs = [doc.page_content.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)
    
    def bm25_search(self, query: str, k: int = 10) -> list:
        """BM25 稀疏检索"""
        tokenized_query = query.split()
        scores = self.bm25.get_scores(tokenized_query)
        
        # 获取 top-k 索引
        top_indices = np.argsort(scores)[::-1][:k]
        
        return [
            {"doc": self.documents[i], "score": scores[i]}
            for i in top_indices
        ]
    
    def vector_search(self, query: str, k: int = 10) -> list:
        """向量稠密检索"""
        docs_with_scores = self.vectorstore.similarity_search_with_score(
            query, k=k
        )
        return [
            {"doc": doc, "score": 1 / (1 + score)}  # 转换距离为相似度
            for doc, score in docs_with_scores
        ]
    
    def hybrid_search(
        self,
        query: str,
        k: int = 5,
        alpha: float = 0.5  # 向量权重
    ) -> list:
        """混合检索"""
        # 获取两种检索结果
        bm25_results = self.bm25_search(query, k=k*2)
        vector_results = self.vector_search(query, k=k*2)
        
        # 归一化分数
        bm25_scores = self._normalize_scores(bm25_results)
        vector_scores = self._normalize_scores(vector_results)
        
        # 合并分数
        combined = {}
        for item in bm25_scores:
            doc_id = hash(item["doc"].page_content)
            combined[doc_id] = {
                "doc": item["doc"],
                "score": (1 - alpha) * item["score"]
            }
        
        for item in vector_scores:
            doc_id = hash(item["doc"].page_content)
            if doc_id in combined:
                combined[doc_id]["score"] += alpha * item["score"]
            else:
                combined[doc_id] = {
                    "doc": item["doc"],
                    "score": alpha * item["score"]
                }
        
        # 排序返回
        sorted_results = sorted(
            combined.values(),
            key=lambda x: x["score"],
            reverse=True
        )
        
        return [r["doc"] for r in sorted_results[:k]]
    
    def _normalize_scores(self, results: list) -> list:
        """归一化分数到 0-1"""
        if not results:
            return results
        
        scores = [r["score"] for r in results]
        min_s, max_s = min(scores), max(scores)
        
        if max_s == min_s:
            return [{"doc": r["doc"], "score": 1.0} for r in results]
        
        return [
            {"doc": r["doc"], "score": (r["score"] - min_s) / (max_s - min_s)}
            for r in results
        ]

# 使用
hybrid = HybridRetriever(documents, vectorstore)
results = hybrid.hybrid_search("机器学习算法", k=5, alpha=0.6)

class RRFRetriever:
    """Reciprocal Rank Fusion 检索器"""
    
    def __init__(self, retrievers: list, k: int = 60):
        self.retrievers = retrievers
        self.k = k  # RRF 参数
    
    def retrieve(self, query: str, top_k: int = 5) -> list:
        """RRF 融合检索"""
        # 收集所有检索结果
        all_rankings = []
        for retriever in self.retrievers:
            docs = retriever.invoke(query)
            all_rankings.append(docs)
        
        # 计算 RRF 分数
        rrf_scores = {}
        
        for ranking in all_rankings:
            for rank, doc in enumerate(ranking):
                doc_id = hash(doc.page_content)
                
                if doc_id not in rrf_scores:
                    rrf_scores[doc_id] = {
                        "doc": doc,
                        "score": 0
                    }
                
                # RRF 公式: 1 / (k + rank)
                rrf_scores[doc_id]["score"] += 1 / (self.k + rank + 1)
        
        # 排序返回
        sorted_results = sorted(
            rrf_scores.values(),
            key=lambda x: x["score"],
            reverse=True
        )
        
        return [r["doc"] for r in sorted_results[:top_k]]

# 使用
from langchain.retrievers import BM25Retriever

bm25_retriever = BM25Retriever.from_documents(documents)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

rrf = RRFRetriever([bm25_retriever, vector_retriever])
results = rrf.retrieve("深度学习优化", top_k=5)

import uuid
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.schema import Document

class ParentChildRetriever:
    """父子文档检索器实现"""
    
    def __init__(self, child_vectorstore, parent_store: dict):
        self.child_vectorstore = child_vectorstore
        self.parent_store = parent_store  # parent_id -> parent_doc
    
    @classmethod
    def from_documents(cls, documents: list, child_size: int = 200):
        """从原始文档构建索引"""
        parent_store = {}
        child_docs = []
        
        child_splitter = RecursiveCharacterTextSplitter(chunk_size=child_size)
        
        for doc in documents:
            parent_id = str(uuid.uuid4())
            parent_store[parent_id] = doc
            
            # 将父文档切分为子块
            chunks = child_splitter.split_documents([doc])
            for chunk in chunks:
                chunk.metadata["parent_id"] = parent_id
                child_docs.append(chunk)
        
        # 仅对子块建立向量索引
        child_vectorstore = Chroma.from_documents(
            child_docs, OpenAIEmbeddings()
        )
        
        return cls(child_vectorstore, parent_store)
    
    def retrieve(self, query: str, k: int = 3) -> list:
        """检索子块，但返回父文档"""
        child_results = self.child_vectorstore.similarity_search(query, k=k*2)
        
        seen_parents = set()
        parent_docs = []
        
        for child in child_results:
            p_id = child.metadata.get("parent_id")
            if p_id and p_id not in seen_parents:
                seen_parents.add(p_id)
                parent_docs.append(self.parent_store[p_id])
                if len(parent_docs) >= k:
                    break
        return parent_docs

from typing import Literal
from pydantic import BaseModel, Field

class RouteQuery(BaseModel):
    """路由查询的模式"""
    datasource: Literal["vectorstore", "web_search", "none"] = Field(
        ...,
        description="根据用户问题选择最合适的知识源",
    )

class AdaptiveRouter:
    """自适应查询路由"""
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.structured_llm = self.llm.with_structured_output(RouteQuery)

    def route(self, query: str) -> str:
        prompt = f"""你是一个路由专家。根据用户问题，决定是搜索向量库、进行网页搜索还是直接回答。
问题：{query}"""
        result = self.structured_llm.invoke(prompt)
        return result.datasource

class SelfRAG:
    """带自我评估的 RAG 流程"""
    def __init__(self, retriever):
        self.retriever = retriever
        self.llm = ChatOpenAI(model="gpt-4o")

    def run(self, query: str):
        # 1. 检索
        docs = self.retriever.invoke(query)
        
        # 2. 评估相关性
        relevant_docs = self._grade_documents(query, docs)
        
        # 3. 生成
        if not relevant_docs:
            # 如果没有相关文档，尝试改写查询重新检索
            return self.run(self._rewrite_query(query))
            
        answer = self._generate(query, relevant_docs)
        
        # 4. 幻觉检测
        if self._check_hallucination(answer, relevant_docs):
            return "抱歉，我无法基于现有资料提供准确回答。"
            
        return answer