如何用 LangChain 和 Elasticsearch 构建 RAG 系统

大语言模型令人印象深刻——直到它一本正经地胡编乱造。这种现象被称为幻觉（hallucination），是 LLM 的致命弱点。模型根据你的提示词预测下一个 token，当它缺乏可靠数据时，就会用听起来煞有介事的胡话来填补空白。

**RAG（检索增强生成，Retrieval Augmented Generation）**是目前解决这一问题最实用的方案。与其寄希望于模型自己”知道”答案，不如在查询时把相关文档喂给它，让它基于真实来源来组织回答。

本文将带你从零开始，使用 LangChain、以 Elasticsearch 作为向量存储、以 AWS Bedrock（Claude）作为 LLM，完整构建一个 RAG 系统。每一段代码示例都面向生产环境，并使用当前版本的库。

---

什么是 RAG？

RAG（检索增强生成）是一种分两个阶段的文本生成方法：

1. 检索（Retrieval）——给定用户查询，在外部知识库中搜索并检索出最相关的文档。
2. 生成（Generation）——将检索到的文档作为上下文喂给 LLM，由它生成一个基于这些证据的答案。

其核心思想很简单：与其依赖模型的参数化知识（它在预训练期间学到的东西），不如在推理时提供非参数化知识（你的文档）。这能显著减少幻觉，并让你能够处理模型从未见过的数据——专有文档、最新更新、内部 wiki。

下面是详细的流程：

1. 用户向 RAG 系统提交一个问题。
2. **检索器（retriever）**在知识库（向量存储）中搜索相关文档。
3. 知识库返回**最相似的前 K 个（top-K）**文档块。
4. **提示词构建器（prompt builder）**将用户的原始问题与检索到的上下文组合成一个增强后的提示词。
5. LLM 基于所提供的上下文生成答案。
6. 系统返回答案——理想情况下还附带指向源文档的引用。

---

文档嵌入流水线

在系统能够检索任何内容之前，你需要先构建知识库。这就是索引阶段（indexing phase）——一个一次性（或周期性）的过程，把你的文档转换为可搜索的向量。

2.1 加载文档

LangChain 为不同格式提供了数十种文档加载器——PDF、Markdown、HTML、纯文本、Notion 导出等等。下面是一个加载 Markdown 文件的简单示例：

from langchain_community.document_loaders import DirectoryLoader, TextLoader

loader = DirectoryLoader(
    "./knowledge_base/",
    glob="**/*.md",
    loader_cls=TextLoader,
    loader_kwargs={"encoding": "utf-8"},
)
documents = loader.load()
print(f"Loaded {len(documents)} documents")

2.2 分块：为什么大小很重要

原始文档通常太长，无法作为单个向量来嵌入。你需要把它们切分成块（chunk）——每个块成为存储中的一个向量的较小段落。

分块大小是 RAG 系统中影响最大的参数之一。一旦设错，无论你的嵌入模型多好，检索质量都会大打折扣。

分块大小	优点	缺点
小（128-256 tokens）	检索精准，每个块主题聚焦	丢失周边上下文，可能返回碎片
中（512-1024 tokens）	精度与上下文兼顾，平衡良好	可能包含一些无关内容
大（1024-2048 tokens）	保留完整上下文和推理链	稀释信号，检索精度较低

实用建议：

• 以 512 tokens 作为基准分块大小起步。
• 在块之间使用 10-20% 的重叠，防止边界处的信息丢失。
• 对于结构化文档（Markdown、HTML），使用**标题感知的切分（header-aware splitting）**来尊重文档结构。
• 对于代码文档，考虑使用更大的块（1024+），因为代码片段需要周边上下文才有意义。

下面是一个针对 Markdown 文档的结构感知切分器：

from langchain.text_splitter import (
    MarkdownHeaderTextSplitter,
    RecursiveCharacterTextSplitter,
)
from langchain_core.documents import Document


def split_markdown(
    markdown_text: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50,
) -> list[Document]:
    """Split a Markdown document by headers first, then by size."""

    # Phase 1: Split on Markdown headers to preserve structure
    headers_to_split_on = [
        ("#", "Header 1"),
        ("##", "Header 2"),
        ("###", "Header 3"),
    ]
    md_splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=headers_to_split_on
    )
    header_splits = md_splitter.split_text(markdown_text)

    # Phase 2: Further split large sections by character count
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )
    final_chunks = text_splitter.split_documents(header_splits)
    return final_chunks

2.3 嵌入模型的选择

嵌入模型将文本块转换为稠密向量。这些向量的质量直接决定了检索的准确性。以下是 AWS 上的主要选项：

Amazon Titan Text Embeddings v2——Bedrock 上的托管选项。生成 1024 维向量，支持最多 8,192 个输入 token，且无需任何基础设施。如果你已经在 AWS 上并且追求简单，这是最佳选择。

Cohere Embed v3——同样在 Bedrock 上可用。支持多种语言，并有 search_document / search_query 的输入类型区分，可以提升检索质量。

开源替代方案（例如 sentence-transformers/all-MiniLM-L6-v2、BGE、GTE）——自托管，无按请求计费的成本，但你需要自己管理基础设施（SageMaker 端点或 EC2）。如果你有极高的吞吐量或成本约束，可以考虑这些。

在本教程中，我们将通过 Bedrock 使用 Amazon Titan Embeddings：

from langchain_aws import BedrockEmbeddings

embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name="us-east-1",
)

# Quick test
test_vector = embeddings.embed_query("What is RAG?")
print(f"Vector dimensions: {len(test_vector)}")  # 1024

提示： 索引和查询务必使用同一个嵌入模型。混用模型（例如用 Titan 索引却用 Cohere 查询）会产生毫无意义的相似度分数，因为它们的向量空间是不同的。

---

用 Elasticsearch 构建知识库

Elasticsearch 自 7.3 版本起就支持稠密向量搜索，而 8.x 版本通过原生 kNN 搜索使其成为一等公民特性。它是 RAG 的绝佳选择，因为你能在同一个系统里同时获得向量搜索和传统关键词搜索——从而支持混合检索策略。

AWS 替代方案： 如果你更倾向于全托管服务，Amazon OpenSearch Serverless 支持相同的向量搜索能力，且运维开销为零。其 LangChain 集成（langchain_aws.OpenSearchVectorSearch）与下面展示的 Elasticsearch 集成用法几乎完全一致。

3.1 安装依赖

pip install langchain>=0.3.0 langchain-aws langchain-elasticsearch
pip install langchain-community  # document loaders
pip install boto3                # AWS SDK

3.2 索引文档

让我们把它们串联起来。我们将使用 AWS 服务文档作为知识库——比通用文本更实用的一个例子。

from langchain_aws import BedrockEmbeddings
from langchain_elasticsearch import ElasticsearchStore
from langchain_core.documents import Document

# ── 1. Initialize the embedding model ──────────────────────
embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name="us-east-1",
)

# ── 2. Connect to Elasticsearch ────────────────────────────
ES_URL = "https://your-elasticsearch-host:9200"
ES_API_KEY = "your-api-key"

vector_store = ElasticsearchStore(
    embedding=embeddings,
    index_name="aws_docs_index",
    es_url=ES_URL,
    es_api_key=ES_API_KEY,
)

# ── 3. Prepare sample documents (AWS knowledge base) ──────
aws_docs_text = """
# Amazon S3 Storage Classes

## S3 Standard
S3 Standard offers high durability, availability, and performance object storage for
frequently accessed data. It delivers low latency and high throughput, making it
suitable for a wide variety of use cases including cloud applications, dynamic websites,
content distribution, mobile and gaming applications, and big data analytics.

## S3 Intelligent-Tiering
S3 Intelligent-Tiering is the only cloud storage class that delivers automatic storage
cost savings when data access patterns change, without performance impact or operational
overhead. It monitors access patterns and moves objects that have not been accessed for
30 consecutive days to the Infrequent Access tier, delivering 40% cost savings. Objects
that have not been accessed for 90 days move to the Archive Instant Access tier with
68% savings.

## S3 Glacier
Amazon S3 Glacier is a secure, durable, and extremely low-cost Amazon S3 storage class
for data archiving and long-term backup. It is designed to deliver 99.999999999% durability
and provides query-in-place functionality. Retrieval times range from minutes to hours
depending on the retrieval tier selected: Expedited (1-5 minutes), Standard (3-5 hours),
or Bulk (5-12 hours).

## S3 Transfer Acceleration
S3 Transfer Acceleration enables fast, easy, and secure transfers of files over long
distances between your client and an S3 bucket. It leverages Amazon CloudFront globally
distributed edge locations. Data arriving at an edge location is routed to S3 over an
optimized network path, providing 50-500% improvement for cross-region uploads.

## Cross-Region Replication (CRR)
Cross-Region Replication automatically replicates data between buckets across AWS Regions.
CRR helps meet compliance requirements, minimize latency, and increase operational
efficiency. You can replicate objects to a single destination bucket or to multiple
destination buckets in different AWS Regions.
"""

# ── 4. Split and index ─────────────────────────────────────
chunks = split_markdown(aws_docs_text, chunk_size=512, chunk_overlap=50)
print(f"Created {len(chunks)} chunks")

# Add documents to the vector store
vector_store.add_documents(chunks)
print("Documents indexed successfully")

3.3 测试检索

# Configure the retriever with a similarity threshold
retriever = vector_store.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "score_threshold": 0.6,  # Only return results above this confidence
        "k": 3,                   # Return top 3 matches
    },
)

# Test with a natural language query
results = retriever.invoke("How can I reduce storage costs for infrequently accessed data?")

for i, doc in enumerate(results):
    print(f"\n── Result {i+1} ──")
    print(doc.page_content[:200])
    print(f"Metadata: {doc.metadata}")

预期输出：S3 Intelligent-Tiering 这个块应该排名最高，其次是 S3 Glacier——两者都与降低存储成本直接相关。

---

查询时的检索流程

现在我们已经有了一个填充好的向量存储，让我们追踪一下用户提问时会发生什么。

1. 使用与索引文档时相同的模型对用户查询进行嵌入（embed）。
2. 使用余弦相似度（cosine similarity）（或点积）将查询向量与所有文档向量进行比较。
3. 返回最相似的前 K 个块作为上下文。
4. 将原始查询和检索到的上下文组装进一个提示词模板（prompt template）。
5. LLM 生成一个基于所提供上下文的响应。

关键洞见：LLM 从不直接搜索向量存储。它只看到包含查询和预先检索好的上下文的最终提示词。检索步骤与生成步骤是完全分离的。

---

检索增强生成：完整流水线

让我们用 LangChain 和 AWS Bedrock 构建完整的 RAG 链。

5.1 定义提示词模板

提示词模板至关重要。它告诉 LLM 如何使用检索到的上下文以及应遵循何种行为：

from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.documents import Document

ANSWER_TEMPLATE = """You are a helpful technical assistant. Answer the user's question
based ONLY on the provided context. If the context does not contain enough information
to answer the question, say "I don't have enough information to answer this question."

Do not make up facts or use knowledge outside the provided context.

[Context]
{context}

[Question]
{question}

Provide a clear, well-structured answer:"""

ANSWER_PROMPT = ChatPromptTemplate.from_template(ANSWER_TEMPLATE)

5.2 初始化 LLM

from langchain_aws import ChatBedrock

llm = ChatBedrock(
    model_id="us.anthropic.claude-sonnet-4-20250514",
    region_name="us-east-1",
    model_kwargs={
        "max_tokens": 2048,
        "temperature": 0.1,  # Low temperature for factual answers
    },
)

5.3 构建 RAG 链

# Helper: combine multiple documents into a single context string
DEFAULT_DOCUMENT_PROMPT = PromptTemplate.from_template(template="{page_content}")

def combine_documents(
    docs: list[Document],
    document_prompt=DEFAULT_DOCUMENT_PROMPT,
    separator: str = "\n\n",
) -> str:
    """Combine retrieved documents into a single context string."""
    doc_strings = [document_prompt.format(page_content=doc.page_content) for doc in docs]
    return separator.join(doc_strings)


# The RAG chain: retriever -> combine -> prompt -> LLM -> parse
rag_chain = (
    {
        "context": retriever | combine_documents,
        "question": RunnablePassthrough(),
    }
    | ANSWER_PROMPT
    | llm
    | StrOutputParser()
)

# A baseline chain WITHOUT retrieval — for comparison
baseline_chain = (
    ANSWER_PROMPT
    | llm
    | StrOutputParser()
)

5.4 对比结果

question = "What are the retrieval time options for S3 Glacier?"

# ── With RAG ────────────────────────────────────────────
rag_answer = rag_chain.invoke(question)
print("=== RAG Answer ===")
print(rag_answer)

# ── Without RAG (baseline) ──────────────────────────────
baseline_answer = baseline_chain.invoke({
    "context": "No context available.",
    "question": question,
})
print("\n=== Baseline Answer (no retrieval) ===")
print(baseline_answer)

RAG 的答案会直接从已索引的文档中引用具体的检索层级（Expedited：1-5 分钟，Standard：3-5 小时，Bulk：5-12 小时）。基线答案如果 LLM 恰好在训练中见过这些信息，也可能是正确的，但它也可能凭空捏造出具体的数字——而你将无从核实。

5.5 将 RAG 链作为 API 提供服务

from fastapi import FastAPI
from langserve import add_routes

app = FastAPI(
    title="RAG API Server",
    version="1.0",
    description="RAG system powered by LangChain, Elasticsearch, and AWS Bedrock",
)

# Expose the RAG chain as a REST endpoint
add_routes(app, rag_chain, path="/rag")

# Expose the baseline chain for comparison
add_routes(app, baseline_chain, path="/baseline")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

运行服务器并测试它：

# Start the server
python server.py

# Test the RAG endpoint
curl -X POST http://localhost:8080/rag/invoke \
  -H "Content-Type: application/json" \
  -d '{"input": "How does S3 Transfer Acceleration work?"}'

---

分块大小策略及其影响

分块大小不是一个”设好就不管”的参数。不同的使用场景需要不同的策略：

固定大小分块

最简单的方法——每 N 个 token 切分一次，并带上一些重叠。适用于同质化文档（例如散文、文章）。

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""],  # Try to split at natural boundaries
)

语义分块

更进阶的方法——基于句子之间的语义相似度来切分。相似度高的相邻句子留在一起；相似度显著下降时触发切分。这会产生与主题边界对齐的可变长度块。

from langchain_experimental.text_splitter import SemanticChunker

semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95,  # Split at the top 5% similarity drops
)

父子分块

用小块来做精准检索，但返回更大的父文档来提供上下文。LangChain 通过 ParentDocumentRetriever 支持这一点：

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# Small chunks for retrieval, full documents for context
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

store = InMemoryStore()
parent_retriever = ParentDocumentRetriever(
    vectorstore=vector_store,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

何时用哪一种：

• 固定大小——默认的起点。简单、可预测、易于调优。
• 语义分块——当你的文档在同一节内混杂了多个主题时。
• 父子分块——当你既需要精准检索，又需要为生成提供宽泛上下文时。

---

嵌入模型对比

选对嵌入模型和分块大小同样重要。以下是一份实用对比：

模型	维度	最大 tokens	成本	最适合
Titan Embeddings v2	1024	8,192	约 $0.02/百万 tokens	AWS 上的通用场景
Cohere Embed v3	1024	512	约 $0.10/百万 tokens	多语言、搜索优化
all-MiniLM-L6-v2	384	256	免费（自托管）	低延迟、预算友好
BGE-large-en-v1.5	1024	512	免费（自托管）	高准确率、专注英文
GTE-large	1024	512	免费（自托管）	质量与速度均衡

关键考量：

• 维度数量影响存储和搜索速度。384 维（MiniLM）对大多数场景都够用；1024 维（Titan、BGE）能提供略高的准确率。
• 最大输入 token 数决定了你能使用的最大分块大小。Titan 的 8,192-token 上限格外慷慨。
• 一致性没有商量余地。 索引和查询必须使用同一个模型。如果切换模型，你就必须重新索引所有内容。
• 在你自己的数据上评估。 在正式选定某个模型之前，用一批真实查询样本跑一遍检索基准测试。MTEB 排行榜很有用，但你所在领域的具体表现可能会有所不同。

---

RAG vs. 微调：何时用哪种

RAG 很强大，但它并不总是正确的选择。以下是一份客观的对比：

RAG 胜出的场景

• 频繁变化的知识——你的文档每周或每天都在更新。RAG 让你无需重新训练即可更新知识库。
• 需要来源归属——RAG 天然支持引用，因为你确切知道是哪些文档支撑了答案。
• 需要整合专有数据——内部 wiki、客户数据、产品目录。RAG 让这些数据始终掌握在你自己手中。
• 成本敏感——没有 GPU 训练成本。你只需为嵌入和推理付费。
• 上线速度——一个 RAG 系统可以在几天内搭建完成。微调通常需要数周的实验。

微调胜出的场景

• 特定的输出格式或风格——如果模型需要按某个框架的惯用法生成代码，或以特定的品牌口吻写作。
• 隐性知识——当”知识”不是事实性的，而是行为性的（例如，“像一名医疗专业人士那样回答”）。
• 对延迟极其敏感的应用——RAG 会增加检索延迟。微调则把知识固化进了模型里。
• 小而稳定的知识领域——如果你的知识库很少变化，且能容纳在训练数据的限制之内。

混合方法

在实践中，许多生产系统会两者兼用：

1. 对基础模型进行微调，使其掌握特定领域的风格和术语。
2. 用 RAG 来做事实性的、最新的知识检索。

这样你就能兼得两者之长：一个”说你的话”、又始终立足于当前事实的模型。

需要注意的 RAG 局限

• 检索质量天花板——如果检索器找不到正确的文档，LLM 就无法给出好的答案。垃圾进，垃圾出。
• 上下文窗口限制——你能塞进提示词里的检索块数量是有限的。对于需要在数十个文档之间综合推理的复杂问题，RAG 会力不从心。
• 延迟开销——每次查询都需要一次嵌入调用 + 向量搜索 + LLM 调用。相比直接调用 LLM，这会增加 200-500ms。
• 块边界问题——跨越两个块的重要信息可能只被部分检索到。重叠和父子策略有所帮助，但并不能完全解决这个问题。

---

生产环境清单

在部署你的 RAG 系统之前，请逐项检查以下清单：

• 评估数据集——构建一组 50+ 个带标准答案（ground truth）的问答对。分别衡量检索召回率（是否找到了正确的文档？）和答案准确率。
• 分块大小调优——在你的评估集上至少测试三种分块大小（256、512、1024）。最佳大小取决于你的文档结构。
• 相似度阈值——设得太高会没有结果；太低则会引入噪声。对于余弦相似度，0.5-0.7 是一个合理的起始区间。
• 兜底行为——当检索返回零结果时会发生什么？LLM 应该明确地说”我不知道”，而不是产生幻觉。
• 监控——记录检索分数、延迟和用户反馈。追踪用户报告答案错误的案例——这些是你评估工作的金矿。
• 重新索引流水线——设置在源文档变化时自动重新索引。过时的嵌入比没有嵌入更糟糕。
• 混合搜索——考虑将向量搜索与关键词（BM25）搜索结合起来。Elasticsearch 原生支持这一点，而混合检索往往优于任何单一方法。

---

结语

RAG 不是魔法——它是工程。核心思想很直白：给 LLM 正确的上下文，它就会给你正确的答案。但魔鬼藏在细节里：分块大小、嵌入模型选择、检索阈值、提示词设计和兜底行为层层叠加，共同决定了你的系统是可靠还是令人抓狂。

我们在本文中搭建的技术栈——用 LangChain 做编排、用 Elasticsearch 做向量存储、用 AWS Bedrock（Claude）做生成——已经具备生产就绪的能力，并能良好地扩展。Elasticsearch 赋予你混合搜索（向量 + 关键词）的灵活性，而 Bedrock 让你无论做嵌入还是生成都无需管理 GPU 基础设施。

从简单开始。先用固定大小的块和基础的余弦相似度跑通一个可用的流水线。用真实查询衡量它的表现。然后再迭代：尝试语义分块、试验重排序（re-ranking）、加入元数据过滤。每一次改进都应由数据驱动，而非直觉。

本教程的完整源码可在 GitHub 上获取。

---

参考资料

• LangChain Documentation — RAG
• Elasticsearch Vector Search
• AWS Bedrock — Titan Embeddings
• Hands-On with RAG: Step-by-Step Guide to Integrating Retrieval Augmented Generation in LLMs
• MTEB Embedding Benchmark Leaderboard