Simple RAG Pipeline 

RAG stands for Retrieval-Augmented Generation. It’s a technique that combines information retrieval with LLM generation — instead of asking the LLM to answer from memory alone, you first retrieve relevant documents from an external knowledge base, then feed those documents as context to the LLM, and let the LLM generate an answer based on that context.

Why It Matters? 

Traditional LLMs have four fatal flaws that RAG solves:

FlawHow RAG Solves It
Training Cutoff — LLM only knows data up to its training dateRAG retrieves live documents at query time
No Private Data Access — LLM doesn’t know your internal docs, Slack, JiraRAG connects to your private knowledge base
Hallucinations — LLM makes up plausible but wrong answersRAG grounds answers in retrieved facts
Context Window Limits — can’t paste entire company wikiRAG retrieves only the most relevant chunks

Simple RAG Pipeline

A simple RAG pipeline has two phases and five stages:

  • Phase 1: Indexing (Offline / 离线阶段)
    Build the searchable knowledge base before users ask questions.
    离线阶段(索引):先把所有文档加载进来,切分成小块,调用 Embedding 模型转成向量,最后存进 FAISS 等向量数据库里。
  • Phase 2: Retrieval & Generation (Online / 在线阶段)
    Execute at query time for each user question.
    在线阶段(检索+生成):用户提问时,先把问题也转成向量,去库里找最相似的 Top-K 个块,把这几个块作为“参考资料”连同问题一起扔给 GPT,让它写出最终回答。

Stage-by-Stage Breakdown

StageENCNWhat happens
1. LoadLoad documents加载文档Read PDFs, TXT, HTML, DB records
2. SplitChunk documents切分文档Split long docs into smaller semantic pieces
3. EmbedConvert to vectors向量化Turn text chunks into numerical vectors
4. StoreStore in vector DB存储向量Save vectors in FAISS/Chroma/Milvus
5. Retrieve + GenerateSearch + Answer检索+生成Query → search similar vectors → LLM answers

Code Implementation

Below is a complete, self-contained simple RAG pipeline. We’ll use:

  • LangChain for orchestration
  • FAISS as vector database (local, no server needed)
  • OpenAI for embeddings and chat completion

Runtime Sequence

#!/usr/bin/env python3
"""
================================================================================
B02 升级版:完整集成 (Code A + Code B + 智能缓存)
================================================================================
这是一个 100% 完整的、可直接复制粘贴运行的 Python 脚本。

包含:
1. 数据加载 (Load)
2. 文本切分 (Split)  
3. FAISS 专业配置 (支持 Flat / IVF 索引,带持久化)
4. RAG 检索生成链 (Retrieve + Generate)
5. 智能缓存逻辑 (首次构建,后续秒级加载)

使用方法:
1. 安装依赖:pip install faiss-cpu langchain langchain-community langchain-openai openai numpy
2. 设置环境变量:export OPENAI_API_KEY="你的Key"
3. 运行:python rag_complete.py
================================================================================
"""

# ============================================================
# 0. 依赖导入 (全部列出,一个不少)
# ============================================================
import os
import tempfile
import shutil
from typing import List, Dict, Any, Optional

import numpy as np
import faiss

# LangChain 核心
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# LangChain 社区
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores.utils import DistanceStrategy

# LangChain 集成
from langchain_openai import ChatOpenAI

# 文本切分
from langchain_text_splitters import RecursiveCharacterTextSplitter


# ============================================================
# 第一部分:CODE A —— 原版 RAG 流程 (加载、切分、生成链)
# ============================================================

def load_documents(file_paths: List[str]) -> List[Document]:
    """
    从文本文件加载文档。
    """
    all_docs: List[Document] = []
    for path in file_paths:
        loader = TextLoader(path, encoding="utf-8")
        docs = loader.load()
        all_docs.extend(docs)
        print(f"✅ 加载完成: {path} ({len(docs)} 个文档)")
    return all_docs


def chunk_documents(documents: List[Document]) -> List[Document]:
    """
    将大文档切分成语义块。
    使用 RecursiveCharacterTextSplitter 保证上下文连贯。
    """
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,          # 每块最大字符数
        chunk_overlap=200,        # 块间重叠,保留跨块上下文
        length_function=len,
        separators=["\n\n", "\n", " ", ""],
        add_start_index=True,     # 标记在原文中的位置
    )
    chunks = text_splitter.split_documents(documents)
    print(f"✅ 切分完成: {len(documents)} 个文档 → {len(chunks)} 个块")
    return chunks


def retrieve_context(vector_store: FAISS, query: str, k: int = 4) -> List[Document]:
    """
    从向量库中检索最相似的 k 个块。
    """
    retrieved_docs = vector_store.similarity_search(query, k=k)
    print(f"🔍 检索到 {len(retrieved_docs)} 个块")
    return retrieved_docs


def format_context(documents: List[Document]) -> str:
    """
    将检索到的文档格式化成上下文字符串。
    """
    return "\n\n---\n\n".join([
        f"[来源 {i+1}]\n{doc.page_content}"
        for i, doc in enumerate(documents)
    ])


def create_rag_chain(vector_store: FAISS):
    """
    使用 LCEL 构建完整的 RAG 生成链。
    这是 Code A 的核心生成逻辑,完全不需要改动。
    """
    # 提示词模板:强制 LLM 仅基于上下文回答
    prompt_template = ChatPromptTemplate.from_template("""
    你是一个只根据提供的上下文来回答问题的助手。
    如果上下文中没有答案,请直接说"我没有足够的信息回答这个问题"。

    上下文 (Context):
    {context}

    问题 (Question):
    {question}

    答案 (Answer):
    """)

    # 使用 gpt-4o-mini,temperature=0 保证事实性回答
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

    # 内部函数:检索 + 格式化
    def retrieve_and_format(inputs: Dict[str, Any]) -> Dict[str, str]:
        question = inputs["question"]
        docs = retrieve_context(vector_store, question)
        context = format_context(docs)
        return {"context": context, "question": question}

    # 构建 LCEL 链
    rag_chain = (
        RunnablePassthrough()
        | RunnableLambda(retrieve_and_format)
        | prompt_template
        | llm
        | StrOutputParser()
    )
    return rag_chain


# ============================================================
# 第二部分:CODE B —— FAISS 专业配置工厂 (带索引工程 + 持久化)
# ============================================================

class FAISSAdvancedConfig:
    """
    FAISS 高级配置类。
    控制距离度量、索引类型、检索参数。
    """
    def __init__(
        self,
        index_factory_string: str = "Flat",
        nprobe: int = 10
    ):
        self.distance_strategy = DistanceStrategy.COSINE  # 余弦相似度
        self.index_factory_string = index_factory_string  # 例如 "Flat", "IVF100,Flat"
        self.nprobe = nprobe                              # IVF 查询时探测的簇数


class FAISSFactory:
    """
    FAISS 工厂类。
    职责:构建索引、保存到磁盘、从磁盘加载。
    """
    def __init__(self, config: FAISSAdvancedConfig):
        self.config = config
        # 嵌入模型:text-embedding-3-small (1536维,性价比高)
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    def build_from_documents(self, chunks: List[Document]) -> FAISS:
        """
        从文档块构建 FAISS 索引。
        支持 Flat / IVF / HNSW 等所有 FAISS 原生索引类型。
        """
        print(f"🚀 正在构建 FAISS 索引 (类型: {self.config.index_factory_string})...")
        
        # 1. 提取文本并生成向量
        texts = [doc.page_content for doc in chunks]
        vectors = self.embeddings.embed_documents(texts)  # 返回 List[List[float]]
        vectors_np = np.array(vectors).astype('float32')  # FAISS 要求 float32
        
        dim = vectors_np.shape[1]   # 向量维度 (1536)
        n_vectors = vectors_np.shape[0]
        print(f"   向量维度: {dim}, 向量数量: {n_vectors}")

        # 2. 使用工厂字符串创建原生 FAISS 索引
        #    "Flat"      -> 暴力搜索 (精确)
        #    "IVF100,Flat" -> 100个聚类的倒排索引 (快速)
        index = faiss.index_factory(dim, self.config.index_factory_string)

        # 3. 如果使用 IVF,需要训练 (聚类) 并设置 nprobe
        if "IVF" in self.config.index_factory_string:
            print(f"   正在训练 IVF (nlist={self.config.index_factory_string.split('IVF')[1].split(',')[0]})...")
            index.train(vectors_np)
            # nprobe 越大,召回率越高,速度越慢
            index.nprobe = self.config.nprobe
            print(f"   设置 nprobe = {self.config.nprobe}")

        # 4. 添加向量到索引
        index.add(vectors_np)
        print(f"   成功添加 {n_vectors} 个向量")

        # 5. 包装成 LangChain 的 FAISS 对象
        #    需要构建 docstore (存储原始文本) 和 id 映射
        docstore = {str(i): chunk for i, chunk in enumerate(chunks)}
        index_to_docstore_id = {i: str(i) for i in range(len(chunks))}

        vector_store = FAISS(
            embedding_function=self.embeddings.embed_query,  # 查询时用的函数
            index=index,
            docstore=docstore,
            index_to_docstore_id=index_to_docstore_id,
            relevance_score_fn=None  # LangChain 自动根据 distance_strategy 处理
        )
        print("✅ FAISS 索引构建完成")
        return vector_store

    def save(self, vector_store: FAISS, path: str) -> None:
        """
        将 FAISS 索引保存到磁盘。
        会生成两个文件: {path}.pkl 和 {path}.faiss
        """
        vector_store.save_local(path)
        print(f"💾 索引已保存到: {path}")

    def load(self, path: str) -> FAISS:
        """
        从磁盘加载 FAISS 索引。
        """
        print(f"📂 正在从磁盘加载索引: {path}")
        # allow_dangerous_deserialization=True 是因为本地开发环境安全
        vector_store = FAISS.load_local(
            folder_path=path,
            embeddings=self.embeddings,
            allow_dangerous_deserialization=True
        )
        print("✅ 索引加载成功")
        return vector_store


# ============================================================
# 第三部分:集成逻辑 (智能缓存 + 自动切换)
# ============================================================

def get_vector_store_with_cache(
    chunks: List[Document],
    cache_path: str = "./faiss_cache",
    force_rebuild: bool = False
) -> FAISS:
    """
    智能获取向量存储:
    - 如果 cache_path 存在且 force_rebuild=False -> 直接加载 (秒级启动)
    - 否则 -> 用专业配置构建,并保存到 cache_path
    
    自动索引选型:
    - 向量数 <= 10000 -> 使用 Flat (暴力精确)
    - 向量数 > 10000  -> 使用 IVF (倒排加速)
    """
    # 根据数据量自动选择索引类型
    if len(chunks) <= 10000:
        index_type = "Flat"
    else:
        index_type = "IVF100,Flat"  # 100个聚类,适合中等规模
    
    print(f"⚙️  数据量: {len(chunks)} 条,选用索引: {index_type}")
    
    # 创建配置和工厂
    config = FAISSAdvancedConfig(index_factory_string=index_type, nprobe=10)
    factory = FAISSFactory(config)

    # 决策:加载缓存 or 重新构建
    if not force_rebuild and os.path.exists(cache_path):
        # 路径存在 -> 直接加载
        return factory.load(cache_path)
    else:
        # 路径不存在 或 强制重建 -> 构建并保存
        if force_rebuild:
            print("🔄 强制重建索引...")
        vector_store = factory.build_from_documents(chunks)
        factory.save(vector_store, cache_path)
        return vector_store


# ============================================================
# 第四部分:主程序入口 (包含完整测试数据)
# ============================================================

def main():
    """
    完整流程演示:
    1. 生成临时测试文档
    2. 加载 + 切分
    3. 获取向量存储 (带缓存)
    4. 执行 RAG 问答
    5. 清理临时文件
    """
    print("=" * 70)
    print("B02 升级版 RAG Pipeline (Code A + Code B 完整集成)")
    print("=" * 70)

    # ----- 1. 准备测试数据 (临时文件) -----
    print("\n📝 创建临时测试文档...")
    with tempfile.NamedTemporaryFile(
        mode="w",
        suffix=".txt",
        delete=False,
        encoding="utf-8"
    ) as f:
        f.write("""
        Retrieval-Augmented Generation (RAG) is a technique that combines
        information retrieval with large language model generation.
        
        RAG has two main phases: indexing and retrieval-generation.
        
        The indexing phase involves loading documents, splitting them into chunks,
        converting chunks to embeddings, and storing them in a vector database.
        
        The retrieval-generation phase involves converting a user query to an
        embedding, searching the vector database for similar chunks, and using
        those chunks as context for the LLM to generate an answer.
        
        RAG solves the problem of LLM hallucinations by grounding answers in
        retrieved facts. It also allows LLMs to access private or up-to-date
        information that wasn't in their training data.
        
        FAISS (Facebook AI Similarity Search) is a library for efficient
        similarity search and clustering of dense vectors. It is widely used
        as the vector database in RAG systems.
        """)
        temp_path = f.name

    # ----- 2. 加载 + 切分 -----
    print("\n📂 阶段 1: 加载与切分")
    print("-" * 40)
    docs = load_documents([temp_path])
    chunks = chunk_documents(docs)

    # ----- 3. 获取向量存储 (智能缓存) -----
    print("\n💾 阶段 2: 向量存储初始化")
    print("-" * 40)
    # 首次运行会构建,第二次运行会直接加载 (秒级)
    cache_dir = "./demo_faiss_cache"
    vector_store = get_vector_store_with_cache(
        chunks=chunks,
        cache_path=cache_dir,
        force_rebuild=False  # 设为 True 可强制重建
    )

    # ----- 4. 执行 RAG 问答 -----
    print("\n🤖 阶段 3: RAG 问答")
    print("-" * 40)
    
    # 创建 RAG 链 (完全使用 Code A 的逻辑)
    rag_chain = create_rag_chain(vector_store)
    
    # 测试问题列表
    questions = [
        "What is RAG?",
        "What are the two main phases of RAG?",
        "How does RAG solve the hallucination problem?",
        "What is FAISS?"
    ]
    
    for i, question in enumerate(questions, 1):
        print(f"\nQ{i}: {question}")
        answer = rag_chain.invoke({"question": question})
        print(f"A{i}: {answer}")

    # ----- 5. 清理临时文件 -----
    print("\n🧹 清理临时文件...")
    os.unlink(temp_path)  # 删除临时 txt
    # 保留缓存目录,方便下次测试秒启;如果想删掉,取消注释下面一行
    # shutil.rmtree(cache_dir, ignore_errors=True)
    
    print("\n" + "=" * 70)
    print("✅ RAG Pipeline 运行完毕")
    print(f"💡 缓存目录: {cache_dir} (保留以加速下次启动)")
    print("=" * 70)


if __name__ == "__main__":
    main()

逐阶段详解

阶段 0:环境准备 (Pre-flight)

步骤做什么为什么重要
0.1pip install faiss-cpu langchain...确保运行环境就绪
0.2export OPENAI_API_KEY="sk-..."LLM 和 Embedding 服务需要鉴权
0.3导入 osnumpyfaisslangchain 全套代码依赖的基础设施

阶段 1:数据加载 (Data Ingestion)

步骤做什么对应代码函数输入 → 输出
1.1创建临时 .txt 文件(或指定真实文件路径)tempfile.NamedTemporaryFile文本字符串 → 物理文件
1.2读取文件内容,包装成 LangChain Document 对象load_documents()文件路径 → List[Document]

为什么Document 是 LangChain 的标准数据单元,包含 page_content(文本)和 metadata(元数据)。

阶段 2:文本切分 (Chunking)

步骤做什么对应代码函数关键参数
2.1初始化递归切分器RecursiveCharacterTextSplitterchunk_size=1000overlap=200
2.2执行切分split_documents()长文档 → 多个短块

为什么

  • 防止 Embedding 模型截断(通常最大输入 8192 tokens,但 1000 字符更稳妥)。
  • 200 字符重叠保证跨段落的语义连贯(比如一句话被切到两段时,重叠部分能保留完整语义)。

阶段 3:智能缓存路由 (Cache Router) —— 这是集成代码 B 的核心价值

步骤做什么决策逻辑
3.1检查磁盘目录 ./faiss_cache 是否存在os.path.exists(cache_path)
3.2分支 A (首次运行):目录不存在 → 执行阶段 4A构建全新的索引
3.2分支 B (后续运行):目录存在 → 执行阶段 4B直接加载,跳过昂贵的 Embedding 计算

阶段 4A:索引构建 (Build) —— 离线阶段,耗时但只做一次

子步骤做什么对应代码
4A.1调用 OpenAI Embedding API,把每个 chunk 变成 1536 维向量embeddings.embed_documents(texts)
4A.2根据数据量自动选型:≤1万用 Flat(精确),>1万用 IVF100,Flat(加速)faiss.index_factory(dim, index_type)
4A.3如果是 IVF,执行 K-Means 聚类训练 (index.train)把向量划分到 100 个簇里
4A.4将所有向量添加到索引 (index.add)构建倒排表
4A.5将索引和原始文本打包成 LangChain FAISS 对象FAISS(...) 构造函数
4A.6保存到磁盘:./faiss_cache.pkl + ./faiss_cache.faissfactory.save()

阶段 4B:索引加载 (Load) —— 在线阶段,毫秒级恢复

子步骤做什么对应代码
4B.1读取磁盘上的 .pkl 文件(包含 docstore 和元数据)pickle.load()
4B.2读取 .faiss 二进制文件(包含索引结构)faiss.read_index()
4B.3重建内存中的 LangChain FAISS 对象直接返回,无需调用任何 API

为什么这很重要:如果不做持久化,每次重启服务,阶段 4A 都要重新调用 OpenAI API 嵌入所有文档,对于 10 万条数据,耗费数小时且产生巨额费用。


阶段 5:构建 RAG 链 (Chain Assembly)

步骤做什么对应代码本质
5.1定义 System Prompt(约束 LLM 只能基于上下文回答)ChatPromptTemplate.from_template提示词工程
5.2将“检索+格式化”封装成可调用函数RunnableLambda(retrieve_and_format)定制化检索逻辑
5.3用 LCEL 的 | 运算符串联所有组件... | prompt | llm | parser构建流水线(Pipeline)
5.4编译成可执行的 Runnable 对象rag_chain等待输入

阶段 6:问答执行 (Inference) —— 在线阶段,每次查询触发

子步骤做什么具体动作耗时占比
6.1检索 (Retrieve)vector_store.similarity_search(query, k=4) → 返回 4 个最相似的 chunks~10%
6.2增强 (Augment)把 4 个 chunks 拼接成一个长字符串(Context)~1%
6.3生成 (Generate)调用 gpt-4o-mini API,传入 Context + Question~89% (网络IO)
6.4解析 (Parse)提取 LLM 返回的文本内容<1%

阶段 7:资源回收 (Cleanup)

步骤做什么原因
7.1os.unlink(temp_path)删除测试用的临时 .txt 文件,保持工作区整洁
7.2(可选) shutil.rmtree(cache_path)如果想完全重置,删除缓存目录;否则保留加速下次启动

How to run?


# 1. 安装依赖(如果还没装)
pip install faiss-cpu langchain langchain-community langchain-openai openai numpy

# 2. 设置 OpenAI API Key
export OPENAI_API_KEY="sk-你的Key"

# 3. 把上面的完整代码保存为 rag_complete.py,然后运行
python rag_complete.py

第一次运行:会构建索引并保存到 ./demo_faiss_cache 目录(大约需要 3-5 秒)。
第二次运行:会直接加载缓存(不到 1 秒就启动),你可以把 force_rebuild=False 改成 True 来测试强制重建。

Key Takeaways

要点 (Topic)ENCN
RAG定义RAG = Retrieval + Augmented + GenerationRAG = 检索 + 增强 + 生成
核心比喻Open-book exam vs Closed-book exam开卷考试 vs 闭卷考试
两大阶段Indexing (offline) + Retrieval & Generation (online)索引构建(离线)+ 检索与生成(在线)
五个步骤Load → Split → Embed → Store → Retrieve + Generate加载 → 切分 → 向量化 → 存储 → 检索 + 生成
为什么需要RAGSolves: training cutoff, private data, hallucinations, context limits解决:训练截止、私有数据、幻觉、上下文限制
Embedding作用Converts text to vectors; similar texts = similar vectors将文本转成向量;相似文本 = 相似向量
Chunk参数chunk_size=1000, overlap=200 is a good baselinechunk_size=1000, overlap=200 是好的基准
Top-K检索Retrieve top-k most similar chunks (k=3-5 typical)检索top-k个最相似的chunks(通常k=3-5)
LCEL管道Use | operator to compose: retrieve → prompt → llm → parse用 | 运算符组合:检索 → 提示词 → LLM → 解析
TemperatureSet to 0 for factual RAG answersRAG事实性回答设为0

速查表:逻辑流程 vs 代码行号对照

逻辑阶段主函数调用代码文件中的位置
0. 环境准备import ... + OPENAI_API_KEY顶部 1-40 行
1. 数据加载load_documents([temp_path])main() 函数内部
2. 文本切分chunk_documents(docs)main() 函数内部
3. 缓存路由get_vector_store_with_cache(chunks, cache_path)main() 函数内部
4A. 构建索引factory.build_from_documents(chunks)FAISSFactory 类内部
4B. 加载索引factory.load(cache_path)FAISSFactory 类内部
5. 构建链create_rag_chain(vector_store)main() 函数内部
6. 问答循环rag_chain.invoke({"question": q})for 循环内部
7. 资源回收os.unlink(temp_path)main() 函数末尾

总结一句话

这个流程本质上是一个 ETL Pipeline 的 AI 变体

传统 ETL这个 RAG 流程
Extract (抽取)阶段 1:加载文档
Transform (转换)阶段 2:切分文本 + 阶段 4A:向量化
Load (加载)阶段 4A/4B:存入/读取 FAISS 向量库
查询 (Query)阶段 6:检索 + LLM 生成

现在你手里既有完整的代码,又有清晰的流程图,完全可以按图索骥,把每一块代码对应到具体的业务步骤上。你可以把这个流程图画在你的工程文档里,作为系统设计的一部分。