§1.3.13

LangChain / LlamaIndex / Haystack 的最小 RAG Demo？

好的，这是一份关于 LangChain / LlamaIndex / Haystack 最小 RAG Demo 的详细复习资料。

核心概念

检索增强生成（Retrieval-Augmented Generation, RAG）是一种将大型语言模型（LLM）与外部知识库相结合的架构。其核心思想是，当面对一个问题时，系统首先从一个庞大的文档语料库中检索出相关的上下文信息，然后将这些信息连同原始问题一起提供给 LLM，由 LLM 基于这些信息生成最终答案。这个过程分为两步：检索（Retrieval）和生成（Generation），旨在减少模型幻觉、提高答案的事实准确性，并使模型能够利用其训练数据之外的最新知识。LangChain、LlamaIndex 和 Haystack 是三个主流的开源框架，它们提供了构建 RAG 应用所需的工具链和抽象，极大地简化了开发流程。

原理与推导

RAG 的工作流程在根本上是一个信息检索与条件文本生成相结合的过程。

1. 离线索引 (Offline Indexing)

这是 RAG 的准备阶段，目标是创建一个可供快速检索的知识库。

分块 (Chunking): 将原始文档（如 PDF, TXT, HTML）切分成更小的、语义完整的文本块 $c_i$ 。
嵌入 (Embedding): 使用一个预训练的文本嵌入模型（如 Sentence-BERT, OpenAI Ada）将每个文本块 $c_i$ 映射到一个高维向量 $v_i$ 。 $v_i = \text{EmbeddingModel}(c_i) \in \mathbb{R}^d$ 其中 $d$ 是嵌入向量的维度。
索引 (Indexing): 将所有文本块的向量 $v_i$ 及其对应的原文 $c_i$ 存储在一个专门的数据结构中，通常是向量数据库（Vector Store）。这个数据库被优化用于高效地执行最近邻搜索。

2. 在线推理 (Online Inference)

这是用户与系统交互的阶段。

查询嵌入: 当用户提出一个问题（Query, $q$ ）时，使用与索引阶段相同的嵌入模型将其转换为查询向量 $v_q$ 。 $v_q = \text{EmbeddingModel}(q)$
检索 (Retrieval): 在向量数据库中，计算查询向量 $v_q$ 与所有已索引的文本块向量 $v_i$ 之间的相似度。最常用的相似度度量是余弦相似度： $\text{similarity}(v_q, v_i) = \frac{v_q \cdot v_i}{\|v_q\| \|v_i\|}$ 系统会找出与 $v_q$ 相似度最高的 Top-K 个向量，并取回它们对应的原始文本块 $\{c_1, c_2, ..., c_k\}$ 。
增强与生成 (Augmentation & Generation): 将检索到的文本块作为上下文（Context），与原始问题一起构建一个提示（Prompt）。这个提示的结构通常如下：
text
```
 1基于以下已知信息，请简洁地回答用户的问题。如果信息不足，请说“根据已知信息无法回答”。
 2已知信息：
 3{context}  <-- 检索到的文本块 c_1, c_2, ...
 4用户问题：
 5{query}    <-- 用户的原始问题 q
```
最后，将这个增强后的提示送入 LLM，生成最终的答案。从概率角度看，这一步是计算条件概率分布： $P(\text{Answer} | \text{Prompt}(\text{Query}, \text{Context}))$ $P (Answer ∣ Prompt (Query, Context))$

算法复杂度

索引阶段:
- 嵌入: 假设有 $N$ 个文本块，平均长度为 $L$ ，嵌入模型的复杂度为 $O(L)$ ，则总时间复杂度为 $O(N \cdot L)$ 。
- 向量索引构建: 对于像 FAISS 中基于 HNSW 的索引，构建时间复杂度大约为 $O(N \log N)$ 。
- 空间复杂度: $O(N \cdot d)$ ，用于存储 $N$ 个 $d$ 维的向量。
推理阶段:
- 查询嵌入: $O(L_q)$ ， $L_q$ 是查询长度。
- 检索: 使用近似最近邻（ANN）索引（如 HNSW），查询时间复杂度可以降至 $O(\log N)$ 。暴力精确搜索则是 $O(N \cdot d)$ 。
- 生成: 依赖于 LLM 的架构和输出长度，通常是推理成本中最高的部分。

信息论解释 标准的 LLM 在回答问题时，其答案是从整个语言知识的巨大熵空间中抽样的。而 RAG 通过提供高度相关的上下文，极大地降低了答案空间的熵。它将 LLM 的注意力限制在一个小的、与问题相关的知识子集上，从而显著提高生成正确、具体答案的概率，有效抑制了“一本正经地胡说八道”（即幻觉）。

代码实现

以下是使用三个主流框架实现 RAG 的最小可运行 Demo。请确保你已经安装了必要的库 (pip install langchain-openai langchain faiss-cpu llamaindex haystack-ai openaikeyring) 并设置了 OpenAI API 密钥（export OPENAI_API_KEY="..."）。

1. LangChain Demo

python

 1import os
 2from langchain_community.vectorstores import FAISS
 3from langchain_openai import OpenAIEmbeddings, ChatOpenAI
 4from langchain.text_splitter import RecursiveCharacterTextSplitter
 5from langchain_core.prompts import ChatPromptTemplate
 6from langchain.chains.combine_documents import create_stuff_documents_chain
 7from langchain.chains import create_retrieval_chain
 8
 9# 确保设置了 OpenAI API 密钥
10# os.environ["OPENAI_API_KEY"] = "sk-..."
11
12# 1. 定义 LLM, Embedding 和 Prompt 模板
13llm = ChatOpenAI(model="gpt-3.5-turbo")
14embeddings = OpenAIEmbeddings()
15prompt = ChatPromptTemplate.from_template("""Answer the following question based only on the provided context:
16
17<context>
18{context}
19</context>
20
21Question: {input}""")
22
23# 2. 准备示例文档
24documents_text = [
25    "The Eiffel Tower is a wrought-iron lattice tower on the Champ de Mars in Paris, France.",
26    "It is named after the engineer Gustave Eiffel, whose company designed and built the tower.",
27    "Constructed from 1887 to 1889 as the entrance to the 1889 World's Fair, it was initially criticized by some of France's leading artists and intellectuals for its design.",
28    "The tower is 330 metres (1,083 ft) tall, about the same height as an 81-storey building, and is the tallest structure in Paris."
29]
30
31# 3. 索引阶段：分块、嵌入、存入向量数据库
32# 在 LangChain 中，我们通常直接将文本列表传给 from_texts，它会处理嵌入和存储
33# 为了演示分块，我们先手动分块
34text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20)
35split_docs = text_splitter.create_documents(documents_text)
36
37# 为什么这样做：FAISS.from_documents 会自动处理文本的嵌入过程，并构建一个可供检索的内存向量索引。
38vector_store = FAISS.from_documents(split_docs, embeddings)
39
40# 4. 创建 RAG 链
41# 为什么这样做：create_stuff_documents_chain 将检索到的文档“塞入”到指定的 prompt 中。
42document_chain = create_stuff_documents_chain(llm, prompt)
43
44# 为什么这样做：create_retrieval_chain 将 retriever 和 document_chain 串联起来。
45# 它首先用 retriever 获取文档，然后将文档和原始输入传递给 document_chain。
46retriever = vector_store.as_retriever(search_kwargs={'k': 2}) # 设置为检索2个最相关的文档
47retrieval_chain = create_retrieval_chain(retriever, document_chain)
48
49# 5. 推理阶段：调用链进行问答
50query = "Who designed the Eiffel Tower?"
51response = retrieval_chain.invoke({"input": query})
52
53print("--- LangChain Demo ---")
54print("Query:", query)
55print("Answer:", response["answer"])
56# 打印检索到的上下文，用于调试
57print("Retrieved Context:")
58for i, doc in enumerate(response["context"]):
59    print(f"Doc {i+1}: {doc.page_content}")

2. LlamaIndex Demo

python

 1import os
 2from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
 3from llama_index.llms.openai import OpenAI
 4from llama_index.embeddings.openai import OpenAIEmbedding
 5
 6# 确保设置了 OpenAI API 密钥
 7# os.environ["OPENAI_API_KEY"] = "sk-..."
 8
 9# 1. 创建一个虚拟的 data 目录和文件
10if not os.path.exists("data"):
11    os.makedirs("data")
12with open("data/eiffel_tower.txt", "w") as f:
13    f.write("The Eiffel Tower is a wrought-iron lattice tower on the Champ de Mars in Paris, France. ")
14    f.write("It is named after the engineer Gustave Eiffel, whose company designed and built the tower. ")
15    f.write("Constructed from 1887 to 1889 as the entrance to the 1889 World's Fair. ")
16    f.write("The tower is 330 metres tall.")
17
18# 2. 配置 LLM 和 Embedding 模型 (LlamaIndex v0.10+ 推荐的全局设置方式)
19# 为什么这样做：通过 Settings 对象全局配置，可以使 LlamaIndex 的所有组件自动使用这些配置，简化代码。
20Settings.llm = OpenAI(model="gpt-3.5-turbo")
21Settings.embed_model = OpenAIEmbedding()
22
23# 3. 索引阶段：加载、分块、嵌入、索引
24# 为什么这样做：SimpleDirectoryReader 负责从指定目录加载所有文档。
25documents = SimpleDirectoryReader("./data").load_data()
26
27# 为什么这样做：VectorStoreIndex.from_documents 是一个高度封装的接口，
28# 它在内部自动完成了文本分块、调用嵌入模型、构建向量索引等所有 RAG 准备工作。
29index = VectorStoreIndex.from_documents(documents)
30
31# 4. 推理阶段：创建查询引擎并进行问答
32# 为什么这样做：as_query_engine() 从索引中创建一个即用型的查询接口，
33# 它封装了查询嵌入、检索、构建 Prompt 和调用 LLM 的完整 RAG 流程。
34query_engine = index.as_query_engine(similarity_top_k=2) # 检索2个最相关的文档
35
36query = "How tall is the Eiffel Tower?"
37response = query_engine.query(query)
38
39print("\n--- LlamaIndex Demo ---")
40print("Query:", query)
41print("Answer:", response.response)
42# 打印检索到的上下文，用于调试
43print("Retrieved Context:")
44for i, node in enumerate(response.source_nodes):
45    print(f"Node {i+1}: {node.get_content()}")
46
47# 清理创建的虚拟文件和目录
48os.remove("data/eiffel_tower.txt")
49os.rmdir("data")

3. Haystack Demo

python

 1import os
 2from haystack import Pipeline
 3from haystack.components.builders import PromptBuilder
 4from haystack.components.embedders import OpenAITextEmbedder
 5from haystack.components.generators import OpenAIGenerator
 6from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
 7from haystack.document_stores.in_memory import InMemoryDocumentStore
 8from haystack.utils import Secret
 9
10# 确保设置了 OpenAI API 密钥
11# os.environ["OPENAI_API_KEY"] = "sk-..."
12
13# 1. 准备文档并初始化 Document Store
14# 为什么这样做：Haystack 使用 DocumentStore 作为知识库的统一接口。InMemoryDocumentStore 用于快速原型验证。
15document_store = InMemoryDocumentStore()
16documents = [
17    {"content": "The Eiffel Tower is a wrought-iron lattice tower on the Champ de Mars in Paris, France."},
18    {"content": "It is named after the engineer Gustave Eiffel, whose company designed and built the tower."},
19    {"content": "Constructed from 1887 to 1889 as the entrance to the 1889 World's Fair."},
20    {"content": "The tower is 330 metres (1,083 ft) tall, about the same height as an 81-storey building."}
21]
22# 为什么这样做：document_store.write_documents 是将文档写入知识库的标准方法。
23# Haystack 2.0 中，嵌入过程与写入分离，在 pipeline 中动态进行。
24document_store.write_documents([{"content": d["content"]} for d in documents])
25
26# 2. 定义 RAG Pipeline 的组件
27# 为什么这样做：Haystack 2.0 的设计哲学是组件化。每个步骤都是一个独立的组件，可以自由组合。
28embedder = OpenAITextEmbedder(api_key=Secret.from_env_var("OPENAI_API_KEY"))
29retriever = InMemoryEmbeddingRetriever(document_store=document_store, top_k=2)
30template = """Answer the question based only on the following context:
31{{documents}}
32
33Question: {{query}}
34"""
35prompt_builder = PromptBuilder(template=template)
36generator = OpenAIGenerator(model="gpt-3.5-turbo", api_key=Secret.from_env_var("OPENAI_API_KEY"))
37
38# 3. 构建 Pipeline
39# 为什么这样做：Pipeline 对象定义了数据流。这里的数据流是：
40# query -> embedder -> retriever -> prompt_builder -> generator -> answer
41rag_pipeline = Pipeline()
42rag_pipeline.add_component("embedder", embedder)
43rag_pipeline.add_component("retriever", retriever)
44rag_pipeline.add_component("prompt_builder", prompt_builder)
45rag_pipeline.add_component("generator", generator)
46
47# 连接组件
48rag_pipeline.connect("embedder.embedding", "retriever.query_embedding")
49rag_pipeline.connect("retriever.documents", "prompt_builder.documents")
50rag_pipeline.connect("prompt_builder", "generator")
51
52# 4. 推理阶段：运行 Pipeline
53query = "When was the Eiffel Tower constructed?"
54# 为什么这样做：pipeline.run() 接受一个字典，key 对应 pipeline 中组件的输入端口。
55# 这里，query 被传递给 embedder 和 prompt_builder。
56result = rag_pipeline.run({
57    "embedder": {"text": query},
58    "prompt_builder": {"query": query}
59})
60
61print("\n--- Haystack Demo ---")
62print("Query:", query)
63print("Answer:", result['generator']['replies'][0])
64# 打印检索到的上下文，用于调试
65print("Retrieved Context:")
66for i, doc in enumerate(result['retriever']['documents']):
67    print(f"Doc {i+1}: {doc.content}")

工程实践

使用场景:
- 企业内部知识库问答: 员工可以查询公司政策、项目文档、技术手册等。
- 智能客服: 机器人可以基于产品手册和常见问题解答（FAQ）为客户提供 7x24 小时的支持。
- 个人研究助理: 辅助学者或分析师快速从大量论文或报告中提取和综合信息。
超参数选择:
- chunk_size / chunk_overlap: 这是最关键的参数之一。chunk_size 太小可能丢失关键的上下文，太大则可能引入过多噪声，降低检索精度。经验法则是 chunk_size 在 512-1024 之间，chunk_overlap 设为 chunk_size 的 10%-20%。
- top_k: 检索文档的数量。通常选择 3-5。太少可能错过答案，太多会增加 LLM 的处理负担和噪声，还可能超出上下文窗口限制。
- Embedding Model: 模型的选择直接影响检索质量。OpenAI 的 text-embedding-3-small 是一个性价比高的选择。对于更高要求，可使用 text-embedding-3-large 或开源的 bge-large-en-v1.5 等。
性能 / 显存 / 吞吐的权衡:
- 性能: RAG 的瓶颈通常在 LLM 生成步骤。使用更小的模型（如 GPT-3.5-Turbo vs GPT-4）或量化模型可以显著提高速度，但可能牺牲答案质量。
- 显存: 向量数据库是主要的显存消耗者。对于海量文档，内存索引（如 FAISS）可能不可行，需要使用基于磁盘的向量数据库（如 ChromaDB, Weaviate, Pinecone）。
- 吞吐: 批处理查询可以提高嵌入和检索的吞吐量。使用 vLLM 等推理服务器可以优化 LLM 的服务吞吐。
常见坑和调试技巧:
- 检索失败: 这是最常见的问题。调试时，务必检查 retriever 返回的文档内容。如果内容不相关，可以尝试：
  1. 调整分块策略。
  2. 更换或微调嵌入模型。
  3. 尝试混合搜索（Hybrid Search），即结合关键词搜索（如 BM25）和向量搜索。
- “大海捞针”问题 (Lost in the Middle): LLM 可能忽略了上下文中部的重要信息。可以引入一个 Reranker（如 CohereRerank）对 retriever 返回的 top_k 文档进行二次排序，将最相关的文档放在上下文的开头或结尾。
- 上下文污染: 检索到的文档虽然相关，但包含错误或过时的信息。这需要保证知识库的质量和时效性。

常见误区与边界情况

误区: RAG 就是简单的“复制粘贴”上下文。
- 正解: RAG 的核心是智能检索。其挑战在于如何从百万甚至数十亿的文档中，精确、快速地找到回答特定问题所需的那几片“知识拼图”。检索算法、分块策略和嵌入模型的质量是决定 RAG 系统成败的关键。
误区: 只要用了 RAG，LLM 就不会产生幻觉。
- 正解: RAG 显著减少了幻觉，但不能完全消除。如果检索到的上下文本身是模糊、矛盾或不完整的，LLM 仍然可能进行不准确的推断。此外，需要通过精巧的 Prompt Engineering 指示模型“必须只根据提供的上下文回答”。
边界情况:
- 知识库外的问题: 当用户提问的问题在知识库中完全没有相关信息时，理想的 RAG 系统应返回“我不知道”或“根据现有资料无法回答”，而不是强行从不相关的文档中编造答案。这需要对检索到的文档的相似度分数设置一个阈值。
- 需要综合推理: 有些问题需要综合多个文档、甚至进行多步推理才能回答。简单的 RAG（检索一次，然后生成）可能无法处理。这催生了更高级的 RAG 模式，如 Multi-hop RAG 或 Agentic RAG。
常见面试追问:
- "如何评估一个 RAG 系统的好坏？"
  - 回答要点: 需要从检索和生成两个维度进行评估。
    - 检索评估: Context Precision (检索到的文档有多少是相关的) 和 Context Recall (所有相关文档有多少被检索到了)。
    - 生成评估: Faithfulness (答案是否完全基于给定的上下文)、Answer Relevancy (答案是否切中问题要点)。
    - 端到端评估: 使用 RAGAS、ARES 等自动化评估框架，结合上述指标进行综合打分。
- "LangChain, LlamaIndex, Haystack，你如何选择？"
  - 回答要点:
    - LangChain: 最灵活、生态最庞大。适合需要高度定制化、集成多种工具的复杂应用。它的学习曲线可能稍陡，因为选择太多。
    - LlamaIndex: 专注于 RAG，特别是数据索引和检索。如果你主要做文档问答，它的高级 API 能让你用最少的代码实现强大的功能。它对 RAG 的优化非常深入。
    - Haystack: 面向生产和企业级应用。它的 Pipeline 设计非常清晰，支持 MLOps，并且对传统的 NLP 技术（如 BM25）和新潮的 LLM 技术都有很好的支持。适合构建需要长期维护和扩展的系统。