10分钟带你从零搭建本地RAG：向量 + BM25 + RRF 混合检索，一篇就够+github完整源码

用户提问 ──→ 在你的文档中检索 ──→ 把相关片段发给 LLM ──→ 生成答案

langchain-rag-tutorial/
├── main.py                    # 核心代码（包含混合检索）
├── pyproject.toml             # 依赖配置
├── data/
│   └── knowledge_base.txt     # 知识库文件
├── faiss_index/               # 运行后自动生成
│   ├── index.faiss            # FAISS 向量索引
│   └── index.pkl              # FAISS 索引元数据（原写法遗漏）
├── bm25_index.pkl             # BM25 索引（持久化）
└── tests/
    └── test_main.py

# macOS
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

# 下载所需模型（ollama serve 会自动后台运行，无需手动执行）
ollama pull deepseek-r1:1.5b    # LLM 模型
ollama pull nomic-embed-text    # 向量嵌入模型

git clone https://github.com/tangcheng/langchain-rag-tutorial.git
cd langchain-rag-tutorial
uv sync

from langchain_core.embeddings import Embeddings
from typing import List
import ollama

class OllamaEmbeddings(Embeddings):
    def __init__(self, model: str = "nomic-embed-text"):
        self.model_name = model

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return [self.embed_query(text) for text in texts]

    def embed_query(self, text: str) -> List[float]:
        return ollama.embeddings(model=self.model_name, prompt=text)["embedding"]

from langchain_core.messages import BaseMessage

class OllamaChat:
    def __init__(self, model: str = "deepseek-r1:1.5b"):
        self.model = model

    def invoke(self, messages: List[BaseMessage]) -> str:
        # 拼接多轮消息
        prompt = "\n".join([msg.content for msg in messages])
        response = ollama.chat(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        return response["message"]["content"]

import os
from langchain_text_splitters import CharacterTextSplitter
from langchain_core.documents import Document as LCDocument

def load_documents():
    with open("./data/knowledge_base.txt", "r", encoding="utf-8") as f:
        content = f.read()

    text_splitter = CharacterTextSplitter(
        separator="\n\n",
        chunk_size=500,      # 每块最多 500 字符
        chunk_overlap=50,    # 块之间重叠 50 字符（防止割裂语义）
        length_function=len
    )
    chunks = text_splitter.split_text(content)

    documents = [
        LCDocument(page_content=chunk, metadata={"source": "knowledge_base.txt"})
        for chunk in chunks if chunk.strip()
    ]
    return documents

from langchain_community.vectorstores import FAISS

INDEX_DIR = "./faiss_index"
BM25_PKL = "./bm25_index.pkl"
RETRIEVE_K = 5
FUISON_K = 5
RRF_K = 60

def build_index(documents, embed_model):
    # 已有则直接加载
    if os.path.exists(INDEX_DIR):
        print("📦 检测到已有 FAISS 索引，加载中...")
        return FAISS.load_local(
            INDEX_DIR, embed_model,
            allow_dangerous_deserialization=True
        )

    print("🔍 构建 FAISS 向量索引...")
    vectorstore = FAISS.from_documents(
        documents=documents,
        embedding=embed_model   # OllamaEmbeddings 把文本转成向量
    )
    vectorstore.save_local(INDEX_DIR)
    return vectorstore

import pickle
from rank_bm25 import BM25Okapi

def build_bm25_index(documents: List[LCDocument]):
    """构建 BM25 倒排索引，支持持久化"""
    if os.path.exists(BM25_PKL):
        print("📦 检测到已有 BM25 索引，加载中...")
        with open(BM25_PKL, "rb") as f:
            return pickle.load(f)

    print("🔍 构建 BM25 索引...")
    doc_texts = [doc.page_content for doc in documents]
    # 简单分词：空格分割 + 小写化
    tokenized_corpus = [doc.lower().split() for doc in doc_texts]
    bm25 = BM25Okapi(tokenized_corpus)

    # 持久化到本地
    bm25_data = {"bm25": bm25, "doc_texts": doc_texts}
    with open(BM25_PKL, "wb") as f:
        pickle.dump(bm25_data, f)
    print("   BM25 索引创建完成！")
    return bm25_data

def reciprocal_rank_fusion(results: List[List[dict]], k: int = 60) -> List[dict]:
    """
    RRF 算法核心思想：
    对每个检索器返回的结果，按排名分配分数：score = 1 / (k + rank)
    排名越靠前，分数贡献越大。不同检索器中同一文档的分数累加。
    """
    doc_scores = {}

    for retriever_results in results:
        for rank, item in enumerate(retriever_results, start=1):
            doc_key = item["doc"].page_content
            if doc_key not in doc_scores:
                doc_scores[doc_key] = {"doc": item["doc"], "rrf_score": 0.0}
            # 累加 RRF 分数（排名越靠前，贡献越大）
            doc_scores[doc_key]["rrf_score"] += 1.0 / (k + rank)

    # 按综合分数降序排列
    fused = sorted(doc_scores.values(),
                   key=lambda x: x["rrf_score"], reverse=True)
    return fused

def dense_search(vectorstore, query: str, k: int) -> List[dict]:
    docs = vectorstore.similarity_search(query, k=k)
    return [{"doc": doc, "score": 1.0} for doc in docs]

def hybrid_search(vectorstore, bm25_data, documents, query: str) -> List[LCDocument]:
    """
    混合检索完整流程：
    1. FAISS 向量检索 → Top-5
    2. BM25 稀疏检索 → Top-5
    3. RRF 融合 → 最终 Top-5 排名
    """
    bm25 = bm25_data["bm25"]
    doc_texts = bm25_data["doc_texts"]

    # 两路并行检索
    dense_results = dense_search(vectorstore, query, k=RETRIEVE_K)
    sparse_results = sparse_search(bm25, doc_texts, query, documents, k=RETRIEVE_K)

    print(f"   📊 FAISS 命中: {len(dense_results)} 条 | BM25 命中: {len(sparse_results)} 条")

    if not sparse_results:
        # 无 BM25 时，降级为纯向量检索
        return [item["doc"] for item in dense_results[:FUISON_K]]

    # RRF 融合
    fused = reciprocal_rank_fusion([dense_results, sparse_results], k=RRF_K)

    # 打印融合排名（方便调试理解）
    print("   🏆 RRF 融合排名（Top-5）：")
    for i, item in enumerate(fused[:5], 1):
        preview = item["doc"].page_content[:40].replace("\n", " ")
        print(f"      [{i}] 分数={item['rrf_score']:.3f}  |  {preview}...")

    return [item["doc"] for item in fused[:FUISON_K]]

def sparse_search(bm25, doc_texts, query, documents, k):
    if bm25 is None:
        return []
    tokenized_query = query.lower().split()
    scores = bm25.get_scores(tokenized_query)
    top_indices = sorted(range(len(scores)),
                        key=lambda i: scores[i], reverse=True)[:k]
    return [
        {"doc": documents[idx], "score": float(scores[idx])}
        for idx in top_indices if scores[idx] > 0
    ]

def answer_question(question, vectorstore, bm25_data, documents, llm):
    # 混合检索 Top-5
    docs = hybrid_search(vectorstore, bm25_data, documents, question)

    # 拼 Prompt，发给 LLM
    context = "\n\n".join([doc.page_content for doc in docs])
    prompt = f"""基于以下上下文回答问题。如果无法从上下文找到答案，请如实说明。

上下文:
{context}

问题: {question}

回答:"""

    print(f"\n❓ 问题: {question}")
    print(f"📖 参考 {len(docs)} 个文档片段：")
    for i, doc in enumerate(docs, 1):
        print(f"   [{i}] {doc.page_content[:60].replace(chr(10), ' ')}...")

    response = llm.invoke([prompt])
    print(f"\n✅ 答案:\n{response}")

uv run python main.py

🔧 初始化 Ollama LLM...
🔧 初始化 Ollama Embedding...
📚 加载知识库文档...
   已加载 8 个文档块
📦 检测到已有 FAISS 索引，加载中...
📦 检测到已有 BM25 索引，加载中...

==================================================
💬 开始混合检索问答测试（FAISS + BM25 + RRF）
==================================================

❓ 问题: 什么是 LlamaIndex？
   🔎 混合检索: query =「什么是 LlamaIndex？」
   📊 FAISS 命中: 5 条 | BM25 命中: 5 条
   🏆 RRF 融合排名（Top-5）：
      [1] 分数=0.0325  |  LlamaIndex（formerly GPT Index）是一...
      [2] 分数=0.0246  |  RAG（检索增强生成）是 LlamaIndex 的...
      [3] 分数=0.0245  |  核心概念包括文档加载器、索引、查询引...
      [4] 分数=0.0164  |  LlamaIndex 为构建 LLM 应用提供了...
      [5] 分数=0.0161  |  与 LangChain 的关系：两者可以结合使用
📖 参考 5 个文档片段：
   [1] LlamaIndex（formerly GPT Index）是一...
   ...

✅ 答案:
 LlamaIndex 是一个强大的数据框架...

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

history = InMemoryChatMessageHistory()
chain_with_history = RunnableWithMessageHistory(
    rag_chain,
    history,
    input_messages_key="question",
    history_messages_key="chat_history"
)

import streamlit as st

question = st.text_input("问一个问题：")
if question:
    docs = hybrid_search(vectorstore, bm25_data, documents, question)
    st.write(llm.invoke([build_prompt(docs, question)]))

from fastapi import FastAPI
app = FastAPI()

@app.post("/ask")
def ask(question: str):
    docs = hybrid_search(vectorstore, bm25_data, documents, question)
    return {"answer": llm.invoke([build_prompt(docs, question)])}

文档 → 切块 → FAISS 向量库 + BM25 倒排索引

问题 → FAISS Top-5 + BM25 Top-5 → RRF 融合 → Top-5

Top-5 文档 + 问题 → 拼 Prompt → DeepSeek-R1 → 答案