本文深入剖析了 LangChain 在检索增强生成（RAG）中的核心构成，包括 Retriever、Embedding 和 ChainType。文章详细阐述了 LangChain 如何安全、可扩展地整合检索与生成环节，并通过源码接口和实际示例，展示了 RAG 的端到端流程。从基础概念回顾到最小化本地复现实验，再到源码阅读要点和经验总结，为读者提供了构建和理解 RAG 系统的全面指南，强调了向量化策略、ChainType 选择以及安全注入 Prompt 的重要性。

🗂️ **核心组件职责明确**：LangChain 将 RAG 流程封装为可组合的 Chain，其中 Retriever 负责接收查询并返回相关文档，Embedding 负责将文本转换为向量以供检索，而 ChainType 则定义了将检索到的文档与用户问题结合并送入 LLM 的策略（如 stuff、map_reduce、refine）。这种模块化设计使得用户可以灵活替换不同的向量库、Embedding 模型和合并策略，构建个性化的 RAG 系统。

💡 **工程意义与解耦**：Retriever 组件通过实现 `_get_relevant_documents` 接口，将后端的检索逻辑与上层的 RAG 流程有效解耦。这意味着上层应用只需关注获取最相关的文档，而无需关心底层索引的具体实现细节，极大地提高了系统的灵活性和可维护性。

⚖️ **ChainType 的选择与影响**：不同的 ChainType 策略对 LLM 的输入和生成效果有着直接影响。`stuff` 策略简单但易受 token 限制；`map_reduce` 适合长文本，通过分块处理降低单次上下文大小；`refine` 则适用于需要逐步精炼答案的场景。选择合适的 ChainType 取决于文档长度、token 预算和任务的复杂性。

🛠️ **本地复现与源码理解**：文章提供了一个不依赖外部服务的最小化本地复现实验，通过简单的 Embedding、内存向量存储和 Mock LLM，清晰地展示了 RAG 的端到端流程。这有助于读者理解检索与生成是如何耦合在一起的，并指导了在 LangChain 源码中应关注的关键模块，如 `BaseRetriever` 和 `create_retrieval_chain`。

📈 **经验总结与实践建议**：在实践 RAG 系统时，优化向量化策略（如文本切分和 overlap）、选择合适的 ChainType、以及实现可追溯性（返回来源文档）至关重要。此外，缓存、批处理和检索器健康监控也是提升系统性能和稳定性的关键考量因素。

本文将拆解检索增强生成（Retrieval-Augmented Generation）的核心组成（Retriever、Embedding、ChainType），理解 LangChain 是如何把“检索 + 生成”这两部分安全、可扩展地拼接起来的。

---

一、核心概念快速回顾（为什么要 RAG？）

RAG 把检索（retrieval）和生成（generation）结合起来：先从知识库检索出相关文档片段，再把这些片段连同用户问题一并送给 LLM 做“基于上下文”的生成。这样既能显著提高生成回答的事实性，又能保持模型生成的自然语言能力。

LangChain 把这个流程封装为一套可组合的 Chain（链）与 Retriever/VectorStore 抽象，使得你能用不同的向量库、Embedding 模型、合并策略（chain_type）来搭建自己的 RAG 系统。关于 Retriever 的接口与职责，官方文档有明确说明：实现 _get_relevant_documents(query: str) 即可。([LangChain][1], [api.python.langchain.com][2])

---

二、关键组件与职责（源码角度）

1. Retriever（检索器）

职责：接收自然语言查询，返回 Document 列表（通常按相关度排序）。接口/约束：LangChain 要求自定义 retriever 实现 _get_relevant_documents(query: str)（同步）或 _aget_relevant_documents（异步）。这使得 Retriever 可以基于任意后端实现（向量库、数据库、网络 API）。([api.python.langchain.com][2])

工程意义：Retriever 将后端检索逻辑与上层 RAG 流程解耦；上层只关心“给我最相关的文档”，不需关心索引实现细节。

---

2. Embedding（向量化）

职责：把文本（文档段、query）转换为向量，供向量检索使用。注意点：Embedding 的语义空间、维度和文本切分策略直接影响检索质量；要保证对 query 与文档使用相同的 embedding 模型/参数。常见实现：OpenAIEmbeddings、local embedding 模型、sentence-transformers 等。向量库（FAISS / Pinecone / Milvus / Weaviate 等）通常提供 from_documents 或 add_documents 的便捷接口。([LangChain][3])

---

3. ChainType（合并策略）

LangChain 提供多种把检索到的文档送入 LLM 的策略（chain_type），主要有：

stuff：把所有检索到的片段直接拼接到 prompt 中（简单但易触发 token 限制）。map_reduce：先对每片段分别生成小答案，再把小答案汇总（降低单次上下文大小，适合长文本）。refine：先生成初始答案，再迭代用更多文档精炼答案（适合需要逐步打磨的场景）。

选择哪种策略取决于文档长度、token 预算和任务复杂度。关于迁移建议，LangChain 官方已经把老的 RetrievalQA 迁移到 create_retrieval_chain / 更现代的 LCEL 风格链上。([LangChain][4])

---

三、整体流程（代码级别的执行顺序）

典型 RAG 调用（高层伪流程）：

用户发起 query。Retriever.get_relevant_documents(query) → 返回 top-k Document（每个 Document 含 .page_content 与 .metadata）。按 chain_type 把这些文档拼成适合 LLM 的输入（如直接拼接、分块 map/reduce、或 refine 流程）。把拼好的 prompt 交给 LLM（LLMChain / ChatModelChain 等）执行生成。将 LLM 的回答和（可选的）source documents 一并返回。

在 LangChain 中，这条路径通常可通过 create_retrieval_chain(retriever, llm, chain_type=...) 构造并直接调用（或使用更细粒度的组合方式）。([LangChain][5])

---

四、可运行的 RAG 示例（需要 Embedding / Vectorstore / OpenAI 等）

说明：下面示例演示真实项目常用流程（使用 FAISS、OpenAIEmbeddings、ChatOpenAI）。如果你有 API Key 与依赖环境（faiss、openai），可直接运行。

准备文档 docs

代码

import osfrom langchain.chains.combine_documents import create_stuff_documents_chainfrom langchain.chains.retrieval import create_retrieval_chainfrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import DirectoryLoaderfrom langchain_community.vectorstores import FAISSfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_huggingface import HuggingFaceEmbeddingsfrom langchain_openai import ChatOpenAI# 1. 加载文档并切分成片段loader = DirectoryLoader("docs", glob="**/*.txt")  # 你的文档目录raw_docs = loader.load()splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)texts = splitter.split_documents(raw_docs)# 2. 嵌入并构建向量索引（FAISS）embedding_model_path = "BAAI/bge-small-zh-v1.5"embeddings = HuggingFaceEmbeddings(    model_name=embedding_model_path,    model_kwargs={'device': 'cpu'},  # 如果有 GPU 可改为 'cuda'    encode_kwargs={'normalize_embeddings': True})vectorstore = FAISS.from_documents(texts, embeddings)# 3. 构建 Retrieverretriever = vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": 2})# 4. LLMllm = ChatOpenAI(    temperature=0,    model="glm-4.5",    openai_api_key=os.getenv("ZAI_API_KEY"),    openai_api_base="https://open.bigmodel.cn/api/paas/v4/")# 5. Prompt + Combine Docs Chainprompt = ChatPromptTemplate.from_template("""你是一个问答助手，请基于以下上下文简要地使用纯文本回答用户问题：<context>{context}</context>问题: {input}""")combine_docs_chain = create_stuff_documents_chain(llm, prompt)# 6. 创建 Retrieval Chainchain = create_retrieval_chain(    retriever=retriever,    combine_docs_chain=combine_docs_chain)# 7. 调用out = chain.invoke({"input": "GPT-5的多模态能力怎么样？"})print("context:", out["context"])print("answer:", out["answer"])

输出：

说明：上面流程中 FAISS.from_documents 会调用 embeddings，将文本转为向量并建立索引；retriever 负责返回 top k 文档；create_retrieval_chain 把这些文档合并传给 LLM。

---

五、最小本地复现实验（零外部依赖）

为了便于阅读源码思想与调试，我们给出一个纯 Python、本地可运行的最小实现，包含：

一个非常简单的 Embedding（基于字符 hash）；一个内存向量 store + 简单余弦相似度检索（实现为 BaseRetriever 子类）；一个非常简单的 LLM 模拟器（MockLLM），把检索到的文档拼接并返回“基于文档的回答”。

这个版本不依赖 OpenAI / faiss 等，可以离线演示 RAG 的端到端流程。

from typing import Listimport math# LangChain 的 Document 类from langchain.schema import Documentfrom langchain_core.callbacks import CallbackManagerForRetrieverRunfrom langchain_core.retrievers import BaseRetriever# ---- 1) 最简单的 embedding（可替换为更复杂模型） ----def simple_embedding(text: str, dim: int = 32) -> List[float]:    # 非语义，仅做示例：字符码累加后散列到向量    v = [0.0] * dim    for i, ch in enumerate(text):        v[i % dim] += ord(ch)    # 归一化    norm = math.sqrt(sum(x * x for x in v)) or 1.0    return [x / norm for x in v]# ---- 2) 内存向量索引 ----class InMemoryVectorStore:    def __init__(self):        self.docs: List[Document] = []        self.vectors: List[List[float]] = []    def add_documents(self, docs: List[Document]):        for d in docs:            self.docs.append(d)            self.vectors.append(simple_embedding(d.page_content))    def similarity_search(self, query: str, k: int = 3) -> List[Document]:        qv = simple_embedding(query)        scores = []        for idx, v in enumerate(self.vectors):            # cosine similarity            dot = sum(a * b for a, b in zip(qv, v))            scores.append((dot, idx))        scores.sort(reverse=True)        return [self.docs[i] for _, i in scores[:k]]# ---- 3) 自定义 Retriever（继承 BaseRetriever） ----class InMemoryRetriever(BaseRetriever):    store: InMemoryVectorStore    k: int = 3    def _get_relevant_documents(        self, query: str, *, run_manager: CallbackManagerForRetrieverRun = None    ) -> list[Document]:        return self.store.similarity_search(query, k=self.k)# ---- 4) Mock LLM：把检索到的文档拼出来并返回“回答” ----class MockLLM:    def __init__(self):        pass    def answer(self, query: str, docs: List[Document]) -> str:        # 一个极其简单的回答函数：拼接 doc 摘要并回显        snippets = "\n---\n".join(d.page_content[:300] for d in docs)        return f"基于检索到的文档，我的回答是：\n{snippets}\n\n（以上为参考片段）"# ---- 5) 演示数据与流程 ----if __name__ == "__main__":    docs = [        Document(page_content="RAG 是检索增强生成，先检索相关片段，再由 LLM 生成答案。"),        Document(page_content="向量检索把文本映射为向量，用相似度来检索相关文档。"),        Document(            page_content="ChainType 决定如何把检索到的文档送入 LLM，例如 stuff/map_reduce/refine。"),        Document(page_content="FAISS 是常用的向量索引库，适合大规模相似度检索。"),    ]    store = InMemoryVectorStore()    store.add_documents(docs)    retriever = InMemoryRetriever(store=store, k=2)    llm = MockLLM()    query = "什么是 RAG？"    retrieved = retriever._get_relevant_documents(query)    print("检索到文档：")    for i, d in enumerate(retrieved, 1):        print(i, "-", d.page_content)    answer = llm.answer(query, retrieved)    print("\nLLM 返回：")    print(answer)

输出：

说明：上面示例把真实的 embedding/向量索引换成本地简化实现，但保留了 RAG 的关键构件：embedding → retriever → combine → LLM。这个实现便于理解检索与生成如何耦合。

---

六、实现细节与源码阅读要点（你在 LangChain 源码中该看的地方）

要深入理解 LangChain 的 RetrievalQA 实现，建议关注以下模块/函数：

langchain_core.retrievers.BaseRetriever：检索器基类，查看 _get_relevant_documents 的接口定义与文档。([api.python.langchain.com][6])langchain.vectorstores.*（FAISS / Pinecone wrapper）：这些模块实现把向量数据库封装为可直用的 VectorStore，并提供 as_retriever() 方法。([LangChain][3])langchain.chains.retrieval.create_retrieval_chain：新的推荐工厂函数，了解 combine_docs_chain 如何被传入与执行（stuff/map_reduce/refine）。迁移说明 / 文档页：RetrievalQA 被标记为 deprecated，官方建议迁移到 create_retrieval_chain。读迁移指南能帮你理解新/旧实现的差异。

---

七、经验总结

向量化策略：切分（chunking）+ 合理 overlap 能提高检索召回率，但会增加索引量与成本；先做小规模 A/B 测试。选择 ChainType：短文本、token 预算充足 → stuff；长文档或分布式计算 → map_reduce 或 refine。返回来源：若需要可追溯性（traceability），设置 return_source_documents=True 并把 Document.metadata 一并返回给用户。缓存与批处理：Embedding 调用可批量化，向量索引构建可离线化；检索热问可以做缓存。检索器健康监控：关注向量距离分布、召回与命中率，定期重训练/重嵌入文档。

---

八、如何把检索结果安全地注入 Prompt（提示工程注意点）

不要把完整原文直接注入（若含敏感或 PII），先做去敏感化处理或只注入摘要片段。控制 prompt 长度：在 chain_type 为 stuff 时，优先按相关性截断或采用滑窗切分。提供来源给用户：在答案末尾附上 source 列表以便核查与归因。

---

九、结语

本文从概念、源码接口、实现细节到运行示例，完整拆解了 RetrievalQA（RAG）在 LangChain 中的实现要点。你现在应该能够：

理解 Retriever / Embedding / ChainType 各自职责；在本地或云端构建简易 RAG 流程；选取适合的合并策略并调优 token 使用与检索策略。

---

接下来我们将深入讲解 FAISS / Pinecone / Milvus 等向量数据库的工作原理、索引构建策略、向量压缩/近邻查找算法（IVF / HNSW / PQ）与在 LangChain 中的集成方案。