RAG 高级优化技巧

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

eval_data = {
    "question": ["什么是 RAG？"],
    "answer": ["RAG 是检索增强生成技术..."],
    "contexts": [["RAG 文档内容"]],
    "ground_truth": ["RAG 的正确定义..."]
}

dataset = Dataset.from_dict(eval_data)

result = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)

print(result)

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain.vectorstores import Chroma

# 向量检索器
vectorstore = Chroma.from_documents(documents, embeddings)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# BM25 检索器
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# 混合检索
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.5, 0.5]
)

from langchain_openai import ChatOpenAI

def expand_query(query: str, llm: ChatOpenAI) -> list:
    """扩展查询为多个变体"""
    prompt = f"生成 3 个语义相关的查询变体：{query}"
    response = llm.invoke(prompt)

    queries = [query]
    for line in response.content.split('\n'):
        if line.strip():
            queries.append(line.strip())

    return queries

# 使用扩展查询检索
queries = expand_query("什么是 RAG？", llm)
all_docs = []
for q in queries:
    docs = retriever.get_relevant_documents(q)
    all_docs.extend(docs)

def hyde_retrieval(query: str, retriever, llm):
    """使用 HyDE 检索"""
    # 生成假设性文档
    prompt = f"基于问题生成详细答案：{query}"
    hypothetical_doc = llm.invoke(prompt).content

    # 使用假设性文档检索
    results = retriever.get_relevant_documents(hypothetical_doc)

    return results

from sentence_transformers import CrossEncoder

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

def rerank_documents(query: str, documents: list, top_k: int = 10):
    """重排序文档"""
    pairs = [[query, doc.page_content] for doc in documents]
    scores = reranker.predict(pairs)

    for doc, score in zip(documents, scores):
        doc.metadata["rerank_score"] = float(score)

    reranked = sorted(documents, key=lambda x: x.metadata["rerank_score"], reverse=True)

    return reranked[:top_k]

class AdvancedRAG:
    """带重排序的 RAG"""

    def __init__(self, vectorstore, reranker=None):
        self.vectorstore = vectorstore
        self.reranker = reranker
        self.llm = ChatOpenAI(model="gpt-4o")

    def retrieve(self, query: str, top_k: int = 5, retrieve_num: int = 50):
        """检索文档（带重排序）"""
        # 第一阶段：向量检索
        docs = self.vectorstore.similarity_search(query, k=retrieve_num)

        # 第二阶段：重排序
        if self.reranker:
            docs = self.reranker(query, docs, top_k=top_k)

        return docs

    def query(self, query: str):
        """完整查询流程"""
        docs = self.retrieve(query)
        context = "\\n\\n".join([d.page_content for d in docs])

        response = self.llm.invoke(f"基于文档：{context}\\n\\n问题：{query}")

        return response.content

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import tool

@tool
def search_docs(query: str) -> str:
    """搜索文档"""
    docs = vectorstore.similarity_search(query, k=3)
    return "\\n".join([d.page_content for d in docs])

@tool
def search_code(query: str) -> str:
    """搜索代码"""
    docs = code_store.similarity_search(query, k=3)
    return "\\n".join([d.page_content for d in docs])

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o")
tools = [search_docs, search_code]

agent = create_openai_functions_agent(llm, tools)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({"input": "用 Python 实现 RAG 系统"})

# 使用 TruLens 可视化 RAG
from trulens_eval import Tru
from trulens_eval.instruments import instrument

class InstrumentedRAG:
    """带监控的 RAG"""

    @instrument
    def retrieve(self, query: str):
        return self.retriever.get_relevant_documents(query)

    @instrument
    def generate(self, context: str, query: str):
        return self.llm.invoke(f"基于文档：{context}\\n\\n问题：{query}")

# 初始化
tru = Tru()
rag = InstrumentedRAG(retriever, llm)

# 记录
tru_recorder = tru.run_app(app=rag, query="什么是 RAG？")