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:
Flaw
How RAG Solves It
Training Cutoff — LLM only knows data up to its training date
RAG retrieves live documents at query time
No Private Data Access — LLM doesn’t know your internal docs, Slack, Jira
RAG connects to your private knowledge base
Hallucinations — LLM makes up plausible but wrong answers
RAG grounds answers in retrieved facts
Context Window Limits — can’t paste entire company wiki
RAG 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
Stage
EN
CN
What happens
1. Load
Load documents
加载文档
Read PDFs, TXT, HTML, DB records
2. Split
Chunk documents
切分文档
Split long docs into smaller semantic pieces
3. Embed
Convert to vectors
向量化
Turn text chunks into numerical vectors
4. Store
Store in vector DB
存储向量
Save vectors in FAISS/Chroma/Milvus
5. Retrieve + Generate
Search + 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()
