RAG 技术方案深度调研（2025-2026）

# pip install llama-index llama-index-llms-openai llama-index-embeddings-openai

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

# 1. 加载文档
documents = SimpleDirectoryReader("./data").load_data()

# 2. 构建索引（自动分块 + 嵌入）
index = VectorStoreIndex.from_documents(documents)

# 3. 查询
query_engine = index.as_query_engine()
response = query_engine.query("什么是 RAG?")
print(response)

# pip install langchain langchain-openai langchain-community
# pip install faiss-cpu rank_bm25 sentence-transformers

from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import EnsembleRetriever
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# ---------- 1. 准备文档 ----------
from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,       # 推荐 400-512 tokens
    chunk_overlap=50,     # 10% 重叠
    separators=["\n\n", "\n", "。", ".", " "]
)
# 假设 documents 已经加载
chunks = text_splitter.split_documents(documents)

# ---------- 2. 构建双路检索器 ----------
# 稠密向量检索
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = FAISS.from_documents(chunks, embeddings)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})

# 稀疏 BM25 检索
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 20

# 混合检索（RRF 融合）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]   # BM25 权重 0.4，向量权重 0.6
)

# ---------- 3. Cross-Encoder 重排序 ----------
cross_encoder = HuggingFaceCrossEncoder(
    model_name="cross-encoder/ms-marco-MiniLM-L-6-v2"
)
reranker = CrossEncoderReranker(
    model=cross_encoder,
    top_n=5               # 重排后取 Top 5
)
retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=ensemble_retriever
)

# ---------- 4. 生成答案 ----------
llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_template("""
基于以下检索到的上下文回答问题。如果上下文中没有相关信息，请说明无法回答。

上下文：
{context}

问题：{question}

答案：
""")

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = chain.invoke("RAG 系统的最佳分块策略是什么?")
print(answer)

# pip install langgraph langchain-openai langchain-community

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# ---------- 1. 定义状态 ----------
class AgentState(TypedDict):
    question: str
    documents: list
    generation: str
    retry_count: int

# ---------- 2. 定义节点 ----------
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def retrieve(state: AgentState) -> AgentState:
    """检索相关文档"""
    question = state["question"]
    # 假设 retriever 已经定义（如上面的混合检索器）
    documents = retriever.invoke(question)
    return {"documents": documents, "retry_count": state.get("retry_count", 0)}

def grade_documents(state: AgentState) -> AgentState:
    """评估检索文档的相关性"""
    question = state["question"]
    documents = state["documents"]

    graded_docs = []
    for doc in documents:
        response = llm.invoke([HumanMessage(content=f"""
            判断以下文档是否与问题相关。只回答 "yes" 或 "no"。
            问题：{question}
            文档：{doc.page_content}
        """)])
        if "yes" in response.content.lower():
            graded_docs.append(doc)

    return {"documents": graded_docs}

def decide_to_generate(state: AgentState) -> Literal["generate", "rewrite"]:
    """决定是生成答案还是改写查询"""
    if len(state["documents"]) > 0:
        return "generate"
    elif state.get("retry_count", 0) >= 2:
        return "generate"  # 重试上限，强制生成
    return "rewrite"

def generate(state: AgentState) -> AgentState:
    """基于检索文档生成答案"""
    question = state["question"]
    documents = state["documents"]
    context = "\n\n".join(doc.page_content for doc in documents)

    response = llm.invoke([HumanMessage(content=f"""
        基于以下上下文回答问题。如果上下文不足，请说明。
        上下文：{context}
        问题：{question}
    """)])
    return {"generation": response.content}

def rewrite_query(state: AgentState) -> AgentState:
    """改写查询以获得更好的检索结果"""
    question = state["question"]
    response = llm.invoke([HumanMessage(content=f"""
        原始问题的检索结果不佳，请改写以下问题使其更适合检索：
        原始问题：{question}
        改写后的问题：
    """)])
    return {
        "question": response.content,
        "retry_count": state.get("retry_count", 0) + 1
    }

# ---------- 3. 构建图 ----------
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("rewrite_query", rewrite_query)

# 添加边
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {"generate": "generate", "rewrite": "rewrite_query"}
)
workflow.add_edge("rewrite_query", "retrieve")  # 改写后重新检索
workflow.add_edge("generate", END)

# 编译
app = workflow.compile()

# ---------- 4. 运行 ----------
result = app.invoke({
    "question": "Agentic RAG 和传统 RAG 有什么区别?",
    "documents": [],
    "generation": "",
    "retry_count": 0
})
print(result["generation"])

# pip install graphrag neo4j

import graphrag
from graphrag.query.structured_search.local_search import LocalSearch
from graphrag.query.structured_search.global_search import GlobalSearch

# 使用 Neo4j 作为图存储后端
from neo4j import GraphDatabase

driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=("neo4j", "password")
)

# Local Search: 从实体出发的子图检索
local_search = LocalSearch(
    llm=llm,
    context_builder=context_builder,
    token_encoder=token_encoder,
    llm_params={"max_tokens": 2000, "temperature": 0},
)
result = local_search.search("Agentic RAG 的核心设计模式有哪些?")
print(result.response)

# pip install ragas langchain-openai

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

# ---------- 1. 准备评估数据 ----------
samples = [
    SingleTurnSample(
        user_input="RAG 的核心架构是什么?",
        retrieved_contexts=[
            "RAG 由检索器和生成器两部分组成...",
            "检索器负责从知识库中找到相关文档..."
        ],
        response="RAG 的核心架构包含检索器和生成器...",
        reference="RAG 由检索器（Retriever）和生成器（Generator）组成..."
    ),
    SingleTurnSample(
        user_input="什么是混合检索?",
        retrieved_contexts=[
            "混合检索结合 BM25 稀疏检索和向量稠密检索...",
        ],
        response="混合检索是将 BM25 和向量检索结合的技术...",
        reference="混合检索结合稀疏（BM25）和稠密（向量）两种检索方式..."
    ),
]

dataset = EvaluationDataset(samples=samples)

# ---------- 2. 运行评估 ----------
results = evaluate(
    dataset=dataset,
    metrics=[
        faithfulness,        # 答案是否忠实于上下文
        answer_relevancy,    # 答案是否与问题相关
        context_precision,   # 检索内容的精确度
        context_recall,      # 检索内容的召回率
    ],
)

# ---------- 3. 查看结果 ----------
print(results)
# {
#     "faithfulness": 0.92,
#     "answer_relevancy": 0.88,
#     "context_precision": 0.85,
#     "context_recall": 0.80,
# }

# 转为 DataFrame 查看详细结果
df = results.to_pandas()
print(df)