AI大模型工程师-应用篇 系列目录：
├── LangChain
│　├── 提示词模板、对话管理与结构化输出核心用法
│　├── LCEL 表达式语法
│　└── 多模态聊天机器人实战
├── Embedding
│　└── Embedding 与向量数据库简单应用——从文本向量化到 RAG 检索增强生成（本文）
├── LangGraph
│　└── LangGraph 工作流与 Agent 开发实战
└── MCP
　　└── 协议与 Agent 通信实战（即将发布）

本文从"什么是 RAG"出发，先解释为什么需要 Embedding，再逐步介绍 OpenAI Embedding、本地私有化部署 Qwen3/BGE 嵌入模型、自定义 Embedding 与 LangChain 整合，接着实操 FAISS 与 Chroma 向量数据库的增删改查，最后结合 Web 文档加载与文本分割，完成一个支持上下文感知的完整 RAG 对话链路。

文章目录

环境准备

本文使用 uv 作为 Python 包管理工具。将以下内容保存为项目根目录下的 pyproject.toml 文件，然后在终端执行 uv sync，即可自动创建虚拟环境并安装所有依赖：

[project]
name = "embedding-rag-demo"
version = "0.1.0"
description = "Embedding 与向量数据库简单应用——从文本向量化到 RAG 检索增强生成"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "beautifulsoup4>=4.14.3",
    "dotenv>=0.9.9",
    "faiss-cpu>=1.13.2",
    "langchain>=1.2.10",
    "langchain-chroma>=1.1.0",
    "langchain-community>=0.4.1",
    "langchain-huggingface>=1.2.1",
    "langchain-openai>=1.1.10",
    "numpy>=2.2.6",
    "openai>=2.24.0",
    "pandas>=2.2.3",
    "sentence-transformers>=5.2.3",
]

一、什么是 RAG？

大语言模型（LLM）虽然能力强大，但存在两个根本性问题：

  • 知识截止：模型的训练数据有截止日期，无法回答最新发生的事情。
  • 幻觉问题：当模型不确定答案时，可能会"一本正经地胡说八道"。

RAG（Retrieval-Augmented Generation，检索增强生成） 正是为了解决这两个问题而提出的，核心思路是先检索，再生成。当用户提问时，系统先从外部知识库中检索出与问题相关的文档片段，再把这些片段连同问题一起发送给 LLM，让模型基于真实资料来回答。

用户提问 → 检索相关文档 → 将文档 + 问题一起发给 LLM → 生成有依据的回答

在这条链路中，检索的质量取决于文本向量化的质量，而检索的效率和管理能力则取决于向量数据库。因此本文的学习路线是：

Embedding（文本向量化）→ 向量数据库（存储与检索）→ 文档加载与分割 → 完整 RAG 链路

我们先从 Embedding 开始。

二、为什么需要 Embedding？

在传统的关键词搜索中，"我想吃多汁的烤肉"和"美味的烧烤"之间几乎没有词汇重叠，搜索引擎很难将它们关联起来。而 Embedding（文本嵌入） 的核心思想是：将文本映射到一个高维向量空间中，让语义相近的文本在空间中彼此靠近。

这就意味着，即使两段话遣词造句完全不同，只要它们表达的意思接近，向量之间的距离就会很小。语义搜索、推荐系统以及 RAG，都建立在这个基础之上。

接下来，我们从最简单的 OpenAI API 调用开始，一步步走向本地私有化部署，最终整合到 LangChain 生态中。

三、OpenAI Embedding 快速体验

3.1 直接调用 OpenAI API

如果你已经有了 OpenAI 的 API Key，最快的方式是直接使用 OpenAI 官方 SDK 生成文本向量。

from openai import OpenAI

from env_utils import OPENAI_API_KEY, OPENAI_BASE_URL

# 不用 LangChain，直接调用 OpenAI SDK
client = OpenAI(
    api_key=OPENAI_API_KEY,
    base_url=OPENAI_BASE_URL
)

text = "I like large language models."
resp = client.embeddings.create(
    model='text-embedding-3-large',
    dimensions=256,
    input=text
)

print(resp.data[0].embedding)   # 256 维浮点数向量
print(len(resp.data[0].embedding))  # 256

关键参数说明：

参数说明
model嵌入模型名称，text-embedding-3-large 是 OpenAI 最新的大规模嵌入模型
dimensions输出向量的维度，越高精度越高，但存储和计算成本也越大。这里设为 256 是为了缩短输出方便观察，实际项目中通常使用更高维度（如 2560）
input待向量化的文本，支持字符串或字符串列表

这种方式简洁直接，但每一次调用都是裸 API 操作，没法直接接入 LangChain 的链路。

3.2 通过 LangChain 调用 OpenAI Embedding

LangChain 提供了 OpenAIEmbeddings 封装类，让 Embedding 模型可以直接接入 LangChain 的链路中使用。

from langchain_openai import OpenAIEmbeddings

from env_utils import OPENAI_API_KEY, OPENAI_BASE_URL

openai_embedding = OpenAIEmbeddings(
    api_key=OPENAI_API_KEY,
    base_url=OPENAI_BASE_URL,
    model="text-embedding-3-large",
    dimensions=2560,
)

# 批量嵌入多个文档
resp = openai_embedding.embed_documents(
    ['I like large language models.',
     '今天的天气非常不错！'
     ]
)

print(resp[0])              # 第一个文档的向量
print(len(resp[0]))         # 2560

与裸 API 的区别：

  • embed_documents(texts) 用于批量嵌入文档，返回 list[list[float]]。
  • embed_query(text) 用于嵌入单条查询，返回 list[float]。
  • 它实现了 LangChain 的 Embeddings 接口，可以直接传入向量数据库、检索链等下游组件。

OpenAI 的模型虽然效果好，但每次调用都需要网络请求并产生费用。在对延迟敏感或数据不能外传的场景中，我们需要本地部署的嵌入模型。

四、本地私有化部署 Embedding 模型

4.1 使用 sentence-transformers 加载 Qwen3

阿里开源的 Qwen3-Embedding 系列支持中英双语，且有从 0.6B 到 4B 的多个规格，适合不同算力条件。

from sentence_transformers import SentenceTransformer

# model = SentenceTransformer("Qwen/Qwen3-Embedding-4B")  # 4B 版本，效果更好，采用 0.6B 方便本地快速演示效果
qwen3_embedding = SentenceTransformer("Qwen/Qwen3-Embedding-0.6B")

resp = qwen3_embedding.encode(
    ['I like large language models.',
     '今天的天气非常不错！'
     ]
)

print(resp[0])
print(len(resp[0]))

原理解析：

  • SentenceTransformer 会在首次运行时自动从 HuggingFace Hub 下载模型到本地缓存目录（可通过环境变量 HF_HOME 自定义路径）。
  • encode() 方法接收文本列表，返回 NumPy 数组，每行对应一个文本的向量表示。
  • 0.6B 版本在 MacBook 上即可运行，4B 版本需要更大显存。

4.2 使用 LangChain + HuggingFace 加载 BGE 模型

BGE（BAAI General Embedding）是智源研究院开源的嵌入模型，bge-small-zh-v1.5 小巧高效，非常适合中文语义检索场景。LangChain 提供了 HuggingFaceEmbeddings 对其进行封装。

from langchain_huggingface import HuggingFaceEmbeddings
import torch

# 自动检测可用设备：Apple MPS > CUDA > CPU
if torch.backends.mps.is_available():
    device = 'mps'
elif torch.cuda.is_available():
    device = 'cuda'
else:
    device = 'cpu'

model_name = "BAAI/bge-small-zh-v1.5"
model_kwargs = {'device': device}
encode_kwargs = {'normalize_embeddings': True}  # 归一化后可直接计算余弦相似度

# 第一次运行会自动下载模型，可通过 HF_HOME 环境变量指定缓存目录
bge_hf_embedding = HuggingFaceEmbeddings(
    model_name=model_name,
    model_kwargs=model_kwargs,
    encode_kwargs=encode_kwargs
)

resp = bge_hf_embedding.embed_documents(
    ['I like large language models.',
     '今天的天气非常不错！'
     ]
)

print(resp[0])
print(len(resp[0]))

关键参数说明：

参数说明
model_nameHuggingFace 模型标识，如 BAAI/bge-small-zh-v1.5
model_kwargs传给模型加载的参数，常用于指定运行设备
encode_kwargs传给 encode() 的参数，normalize_embeddings=True 归一化向量

归一化后，向量的余弦相似度等价于向量点积，计算更加高效。

到这里我们已经有了两条本地化路线：sentence-transformers 直接加载和 langchain-huggingface 封装加载。但前者返回的是 NumPy 数组，无法直接接入 LangChain 的向量数据库。接下来我们解决这个问题。

五、自定义 Embedding 与 LangChain 整合

当使用的嵌入模型不在 LangChain 官方支持列表中时（比如 Qwen3-Embedding），我们可以通过继承 Embeddings 基类来自定义一个适配器。

5.1 基础版本

最核心的工作是实现两个方法：embed_query 和 embed_documents。

from langchain_core.embeddings import Embeddings
from sentence_transformers import SentenceTransformer


class CustomQwen3Embeddings(Embeddings):
    """自定义一个 Qwen3 的 Embedding 和 LangChain 整合的类"""

    def __init__(self, model_name):
        self.qwen3_embedding = SentenceTransformer(model_name)

    def embed_query(self, text: str) -> list[float]:
        return self.embed_documents([text])[0]

    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        return self.qwen3_embedding.encode(texts).tolist()

5.2 生产级版本

在实际部署中，我们通常还需要配置设备映射、数据精度和分词器参数：

from langchain_core.embeddings import Embeddings
from sentence_transformers import SentenceTransformer


class CustomQwen3Embeddings(Embeddings):
    """自定义一个 Qwen3 的 Embedding 和 LangChain 整合的类"""

    def __init__(self, model_name):
        self.qwen3_embedding = SentenceTransformer(
            model_name,
            model_kwargs={
                "device_map": "auto",        # 自动选择可用设备
                "torch_dtype": "float16",    # 半精度推理，节省约一半显存
            },
            tokenizer_kwargs={
                "padding_side": "left",
                "trust_remote_code": True,   # Qwen 模型需要此参数
            },
        )

    def embed_query(self, text: str) -> list[float]:
        return self.embed_documents([text])[0]

    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        return self.qwen3_embedding.encode(texts).tolist()

设计要点：

  • device_map="auto"：自动选择 GPU/MPS/CPU，免去手动判断设备的逻辑。
  • torch_dtype="float16"：半精度浮点数推理，在 MPS 和 CUDA 上均可显著降低内存占用，且几乎不影响向量质量。
  • trust_remote_code=True：Qwen 系列模型在分词器中包含自定义代码，必须开启此选项才能正常加载。

封装完成后，CustomQwen3Embeddings 的用法就和 OpenAIEmbeddings 完全一样——可以直接传给向量数据库、检索链等 LangChain 组件。后续案例中我们将复用这个类。

六、Embedding 实战：语义搜索美食评论

Embedding 模型已经准备就绪，但它到底好不好用？我们先不急着引入向量数据库，而是用最原始的方式——手动计算余弦相似度——来验证语义检索的效果。等我们理解了底层原理，再引入向量数据库来替代这些手动操作。

本案例使用的数据集是 fine_food_reviews_1k.csv（Amazon 美食评论数据，共 1000 条），核心字段如下：

字段说明示例
Time评论时间戳1351123200
ProductId商品 IDB003XPF9BO
UserId用户 IDA3R7JR3FMEBXQB
Score评分（1~5）5
Summary评论摘要where does one start...and stop... with a treat like this
Text评论正文Wanted to save some to bring to my Chicago family but my North Carolina family ate all 4 boxes before I could pack. These are excellent...could serve to anyone

6.1 数据准备与向量化

import ast
import pandas as pd
import numpy as np
from langchain_huggingface import HuggingFaceEmbeddings

model_name = "BAAI/bge-small-zh-v1.5"
model_kwargs = {'device': 'mps'}
encode_kwargs = {'normalize_embeddings': True}

bge_hf_embedding = HuggingFaceEmbeddings(
    model_name=model_name,
    model_kwargs=model_kwargs,
    encode_kwargs=encode_kwargs
)


def text_2_embedding(text):
    """将单条文本转为向量"""
    resp = bge_hf_embedding.embed_documents([text])
    return resp[0]


def embedding_2_file(source_file, output_file):
    """读取原始美食评论数据，生成向量并保存到新文件"""
    # 步骤1：读取数据
    df = pd.read_csv(source_file, index_col=0)
    df = df[['Time', 'ProductId', 'UserId', 'Score', 'Summary', 'Text']]

    # 步骤2：清洗和合并数据——将摘要和正文合并为一个字段
    df['text_content'] = 'Summary: ' + df.Summary.str.strip() + "; Text: " + df.Text.str.strip()

    # 步骤3：调用 Embedding 模型逐条向量化，存入新文件
    df['embedding'] = df.text_content.apply(lambda x: text_2_embedding(x))
    df.to_csv(output_file)

数据流分析：

  1. 读取 CSV 文件，提取评论的 Summary（摘要）和 Text（正文）字段
  2. 合并为 text_content，格式如 "Summary: 很好吃; Text: 这款巧克力口感丝滑..."
  3. 对每条 text_content 调用嵌入模型生成向量
  4. 将向量追加为新列 embedding，写入 CSV 文件

6.2 余弦相似度语义检索

向量化完成后，我们实现基于余弦相似度的检索函数：

def cosine_distance(a, b):
    """计算两个向量的余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))


def search_text(input, embedding_file, top_n=3):
    """根据用户输入进行语义检索，返回最相似的前 top_n 条结果"""
    df_data = pd.read_csv(embedding_file)
    # 将字符串形式的向量还原为 Python 列表
    df_data['embedding_vector'] = df_data['embedding'].apply(ast.literal_eval)

    # 将用户输入也转为向量
    input_vector = text_2_embedding(input)

    # 计算用户输入与每条数据的余弦相似度
    df_data['similarity'] = df_data.embedding_vector.apply(
        lambda x: cosine_distance(x, input_vector)
    )

    # 按相似度降序排列，取前 top_n 条
    res = (
        df_data.sort_values('similarity', ascending=False)
        .head(top_n)
        .text_content.str.replace('Summary: ', "")
        .str.replace('; Text: ', ';')
    )

    for r in res:
        print(r)
        print('-' * 30)

# 第一步：生成向量文件（只需执行一次）
# embedding_2_file('../datas/fine_food_reviews_1k.csv', '../datas/output_embedding.csv')

# 第二步：语义检索
search_text('I like juicy barbecued meat.', '../datas/output_embedding.csv')

核心原理：

用户问题 → Embedding 模型 → 查询向量
                                ↓
美食评论数据 → Embedding 模型 → 文档向量矩阵
                                ↓
      逐一计算余弦相似度 → 排序 → Top-N 结果

这个案例展示了 Embedding 语义检索的底层原理。但实际开发中，我们不可能每次查询都遍历全部数据计算相似度——数据量一大，性能完全不可接受。我们需要一个专门管理向量的数据库，来高效地完成存储、索引和检索。

七、向量数据库选型

我们已经掌握了 Embedding，下一步是选择一个向量数据库来高效地存储和检索向量。市面上有多种方案可供选择：

特性FAISSChromaMilvus
开发方MetaChroma 社区Zilliz
定位高性能向量检索库轻量级嵌入式向量数据库分布式生产级向量数据库
持久化需手动序列化内置自动持久化原生支持
元数据过滤基础字典匹配支持丰富的查询运算符支持标量过滤 + 混合检索
分布式不支持不支持原生支持
适用场景本地开发、大规模单机检索本地开发、快速原型生产环境、海量数据
安装pip install faiss-cpupip install langchain-chroma需部署服务端

本文选择 FAISS 和 Chroma 进行实操演示——前者代表高性能路线，后者代表开发体验优先路线。两者都支持 LangChain 的 VectorStore 统一接口，切换成本很低。

八、FAISS 向量数据库

8.1 内存模式：快速入门

FAISS（Facebook AI Similarity Search）是 Meta 开源的高性能向量检索库。LangChain 对其进行了封装，让我们可以用统一的接口操作向量存储。

在 LangChain 中，所有要存入向量数据库的数据都需要封装为 Document 对象。每个 Document 包含两部分：

  • page_content：文本内容，会被 Embedding 模型向量化后存入索引，用于语义检索。
  • metadata：元数据字典，不参与向量化，但可以用于过滤检索（如按来源、时间等条件筛选）。
import faiss
from langchain_community.docstore import InMemoryDocstore
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document

from embeddings_demo.custom_embedding import CustomQwen3Embeddings

qwen_embedding = CustomQwen3Embeddings("Qwen/Qwen3-Embedding-0.6B")

# 1. 初始化 FAISS 向量数据库
# 先根据嵌入模型的输出维度创建索引
index = faiss.IndexFlatL2(len(qwen_embedding.embed_query('Hello world!')))
vector_store = FAISS(
    embedding_function=qwen_embedding,
    index=index,
    docstore=InMemoryDocstore(),
    index_to_docstore_id={}
)

# 2. 准备文档数据（Document = 文本内容 + 元数据）
document_1 = Document(
    page_content="今天早餐我吃了巧克力薄煎饼和炒蛋。",
    metadata={"source": "tweet", "time": "上午"},   # 元数据用于过滤检索
)

document_2 = Document(
    page_content="明天的天气预报是阴天多云，最高气温62华氏度。",
    metadata={"source": "news"},
)

document_3 = Document(
    page_content="正在用LangChain构建一个激动人心的新项目——快来看看吧！",
    metadata={"source": "tweet"},
)

# ... 省略 document_4 ~ document_9

document_10 = Document(
    page_content="我有种不好的预感，我要被删除了 :(",
    metadata={"source": "tweet"},
)

documents = [document_1, document_2, document_3, ..., document_10]
ids = ['id' + str(i + 1) for i in range(len(documents))]

# 3. 写入向量数据库
vector_store.add_documents(documents, ids=ids)

# 4. 语义检索
results = vector_store.similarity_search('有美食的内容吗', k=2)
for res in results:
    print(f"* {res.page_content} [{res.metadata}]")

关键概念：

概念说明
DocumentLangChain 的文档对象，包含 page_content（文本）和 metadata（元数据）
IndexFlatL2FAISS 的 L2 距离（欧氏距离）精确索引
InMemoryDocstore内存中的文档存储，用于通过 ID 查找原始文档
similarity_search语义检索，返回与查询最相似的 k 个文档

8.2 持久化：保存到磁盘

内存中的数据在程序退出后就会丢失。FAISS 支持将索引和文档存储序列化到本地磁盘：

# 写入磁盘（保存到 faiss_db 目录）
vector_store.save_local('../faiss_db')

8.3 从磁盘加载与高级检索

下次启动程序时，我们可以直接从磁盘加载已有的向量数据库，不需要重新向量化：

from langchain_community.vectorstores import FAISS
from embeddings_demo.custom_embedding import CustomQwen3Embeddings

qwen_embedding = CustomQwen3Embeddings("Qwen/Qwen3-Embedding-0.6B")

# 从磁盘加载
vector_store = FAISS.load_local(
    '../faiss_db',
    embeddings=qwen_embedding,
    allow_dangerous_deserialization=True
)

# 删除指定文档
vector_store.delete(ids=['id10'])

# 带分数 + 元数据过滤的检索
results = vector_store.similarity_search_with_score(
    '有美食的内容吗',
    k=4,
    filter={"source": 'tweet'}   # 只在 source=tweet 的文档中检索
)
for res, score in results:
    print(f"* [Score={score:3f}] {res.page_content} [{res.metadata}]")

FAISS 检索方法对比：

方法返回值特点
similarity_search(query, k)list[Document]最简洁，只返回文档
similarity_search_with_score(query, k)list[tuple[Document, float]]附带相似度分数
similarity_search_with_score(..., filter={})同上支持按元数据过滤

九、Chroma 向量数据库

了解了 FAISS 之后，我们再来看 Chroma。相比 FAISS 需要手动创建索引和文档存储，Chroma 的初始化更加简洁，而且内置了自动持久化功能。

from langchain_chroma import Chroma
from langchain_core.documents import Document

from embeddings_demo.custom_embedding import CustomQwen3Embeddings

qwen_embedding = CustomQwen3Embeddings("Qwen/Qwen3-Embedding-0.6B")

# 初始化 Chroma——自动持久化到指定目录
vector_store = Chroma(
    collection_name='t_news',            # 集合名称（类似数据库中的表名）
    embedding_function=qwen_embedding,
    persist_directory='../chroma_db'      # 持久化目录
)

# 准备和写入数据（与 FAISS 完全一致的 Document 格式）
documents = [document_1, document_2, ..., document_10]
ids = ['id' + str(i + 1) for i in range(len(documents))]

vector_store.add_documents(documents, ids=ids)

# 带分数 + 元数据过滤检索
results = vector_store.similarity_search_with_score(
    '有美食的内容吗',
    k=4,
    filter={"source": 'tweet'}
)
for res, score in results:
    print(f"* [Score={score:3f}] {res.page_content} [{res.metadata}]")

FAISS 和 Chroma 的上层 API 高度一致（都遵循 LangChain 的 VectorStore 接口），我们在两者之间切换时改动很小。开发阶段用 Chroma 比较省心（开箱即用），生产环境面对大规模数据时可以换成 FAISS。

十、Web 文档加载与文本分割

前面几章中，我们向向量数据库写入的都是手动构造的示例数据。但在真实的 RAG 场景中，数据源往往是在线网页、PDF 文档、数据库等。如何将这些外部数据转化为可以存入向量数据库的 Document 对象？LangChain 提供了丰富的 Document Loader 来解决这个问题。

10.1 递归加载网页文档

RecursiveUrlLoader 可以从一个入口 URL 出发，递归爬取关联页面，非常适合加载在线文档站。

import re
from bs4 import BeautifulSoup
from langchain_community.document_loaders import RecursiveUrlLoader


def bs4_extractor(html: str) -> str:
    """使用 BeautifulSoup 从 HTML 中提取纯文本"""
    soup = BeautifulSoup(html, "lxml")
    return re.sub(r"\n\n+", "\n\n", soup.text).strip()


loader = RecursiveUrlLoader(
    "https://docs.python.org/zh-cn/3.13/tutorial/controlflow.html",
    max_depth=2,         # 最大递归深度
    extractor=bs4_extractor
)

# 使用 lazy_load 逐页加载，避免一次性占用过多内存
pages = []
for doc in loader.lazy_load():
    pages.append(doc)

print(f"共加载 {len(pages)} 个页面")

10.2 使用 WebBaseLoader 加载指定页面

当只需要加载特定页面时，WebBaseLoader 更加轻量：

import bs4
from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader(
    web_path=('https://lilianweng.github.io/posts/2023-06-23-agent/',),
    bs_kwargs=dict(
        # 只解析特定 class 的 HTML 元素，过滤掉导航栏、侧边栏等噪声
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    )
)

docs_list = loader.load()

10.3 文本分割

网页内容通常很长，如果把整篇内容直接传给 LLM，既浪费 Token 也不够精准。我们需要把长文档切割成小块，再分别做 Embedding 存入向量数据库。

LangChain 提供了多种文本分割策略：

分割方式原理优点缺点
按字符数切割 (RecursiveCharacterTextSplitter)按固定字符数切块，支持重叠简单直观，适用面广可能在句子中间断开，破坏语义完整性
按段落/标题切割 (MarkdownHeaderTextSplitter 等)根据文档结构（标题、段落）切割保留文档结构，语义完整依赖文档本身有清晰的结构标记
语义切割 (SemanticChunker)通过 Embedding 相似度判断语义边界语义最完整，检索质量最高需要额外计算 Embedding，速度较慢

本文选用 RecursiveCharacterTextSplitter，原因是它不依赖文档格式，通用性最好，作为入门方案足够使用。它会按照字符数将文档切割为小块，并保留相邻块之间的重叠部分，避免上下文断裂。在实际生产环境中，建议根据文档类型选择更精细的切割策略（如对 Markdown 文档使用按标题切割，对长文使用语义切割）。

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,     # 每块最大字符数（通常在 500~2000 之间，需根据文档类型调整）
    chunk_overlap=200    # 相邻块之间的重叠字符数（一般为 chunk_size 的 10%~20%）
)

splits = text_splitter.split_documents(docs_list)
print(f"原始文档数: {len(docs_list)}, 分割后: {len(splits)} 个文本块")

分割策略示意：

原始文档（3000 字符）
├── 块1: 字符 0~999
├── 块2: 字符 800~1799    ← 与块1重叠 200 字符
├── 块3: 字符 1600~2599   ← 与块2重叠 200 字符
└── 块4: 字符 2400~2999   ← 与块3重叠 200 字符

重叠区域确保了上下文连续性——当检索命中某个文本块时，重叠部分能保留相邻段落的语境信息。

参数选择建议：chunk_size=1000 和 chunk_overlap=200 是社区中比较常用的默认值。chunk_size 太小会导致语义信息碎片化，太大则不够精准且浪费 Token；chunk_overlap 一般取 chunk_size 的 10%~20%。实际项目中应根据文档特点和 LLM 上下文窗口大小来调整。

十一、完整 RAG 链路：上下文感知的对话式检索增强生成

前面我们分别掌握了 Embedding、向量数据库、文档加载与分割。现在我们把它们串联起来，构建一个完整的上下文感知 RAG 对话系统。

本章会用到前几篇文章中介绍的 ChatPromptTemplate（提示词模板）、MessagesPlaceholder（历史消息占位符）和 RunnableWithMessageHistory（自动管理对话历史的 Runnable 包装器）等组件。如果你对这些概念不熟悉，建议先阅读前面的提示词模板和 LCEL 表达式语法两篇文章。

11.1 架构总览

                    ┌─────────────────────────────────┐
                    │         用户多轮提问              │
                    └──────────┬──────────────────────┘
                               ↓
                    ┌─────────────────────────────────┐
                    │   问题上下文化（结合聊天历史）      │
                    │   将"它怎么做的？"→ 独立问题       │
                    └──────────┬──────────────────────┘
                               ↓
                    ┌─────────────────────────────────┐
                    │   向量数据库检索（Top-K 文档块）    │
                    └──────────┬──────────────────────┘
                               ↓
                    ┌─────────────────────────────────┐
                    │   LLM 生成回答（基于检索上下文）    │
                    └──────────┬──────────────────────┘
                               ↓
                    ┌─────────────────────────────────┐
                    │         保存到对话历史             │
                    └─────────────────────────────────┘

11.2 第一步：构建向量知识库

这里选用 Chroma 作为向量数据库，因为它自带持久化功能，代码最简洁，适合教程演示。在生产环境中如需更高检索性能，可以替换为 FAISS 或 Milvus。Embedding 模型选用前面封装好的自定义 Qwen3 适配器，无需依赖外部 API，本地即可运行。

import bs4
from langchain_chroma import Chroma
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

from embeddings_demo.custom_embedding import CustomQwen3Embeddings

qwen_embedding = CustomQwen3Embeddings("Qwen/Qwen3-Embedding-0.6B")

# 初始化 Chroma 向量数据库
vector_store = Chroma(
    collection_name='t_agent_blog',
    embedding_function=qwen_embedding,
    persist_directory='../chroma_db'
)


def create_dense_db():
    """将网络上的 Agent 博客数据加载、分割、写入向量数据库"""
    # 1. 加载网页内容
    loader = WebBaseLoader(
        web_path=('https://lilianweng.github.io/posts/2023-06-23-agent/',),
        bs_kwargs=dict(
            parse_only=bs4.SoupStrainer(
                class_=("post-content", "post-title", "post-header")
            )
        )
    )
    docs_list = loader.load()

    # 2. 文本分割
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    splits = text_splitter.split_documents(docs_list)

    print('文本块数量:', len(splits))

    # 3. 写入向量数据库
    ids = ['id' + str(i + 1) for i in range(len(splits))]
    vector_store.add_documents(documents=splits, ids=ids)


# 首次运行执行一次即可
# create_dense_db()

11.3 第二步：问题上下文化

在多轮对话中，用户的后续提问常常包含代词指代（如"它怎么做的？"）。如果我们直接拿这种模糊的问题去向量数据库检索，效果会很差。因此我们需要一个上下文感知的检索器，它会先将用户问题结合聊天历史改写为一个独立的完整问题，再拿改写后的问题去检索。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chains.history_aware_retriever import create_history_aware_retriever

# 系统提示词：指导 LLM 将带上下文的问题改写为独立问题
contextualize_q_system_prompt = (
    "给定聊天历史和最新的用户问题（可能引用聊天历史中的上下文），"
    "将其重新表述为一个独立的问题（不需要聊天历史也能理解）。"
    "不要回答问题，只需在需要时重新表述问题，否则保持原样。"
)

contextualize_q_prompt = ChatPromptTemplate.from_messages([
    ("system", contextualize_q_system_prompt),
    MessagesPlaceholder("chat_history"),   # 聊天历史占位符
    ("human", "{input}"),
])

# 创建检索器（每次检索返回最相似的 2 个文本块）
# k 值越大，LLM 获得的上下文越丰富，但也会消耗更多 Token 并可能引入噪声
# 对于简洁问答场景，k=2~3 通常是一个合理的起点
retriever = vector_store.as_retriever(search_kwargs={'k': 2})

# 将检索器包装为上下文感知版本
history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_q_prompt
)

工作原理：

  • 当 chat_history 为空时，直接用原始问题检索。
  • 当 chat_history 非空时，先由 LLM 将问题改写（如"它怎么做的？" → “Task Decomposition 有哪些常见方法？”），再用改写后的问题检索。

11.4 第三步：构建 RAG 问答链

from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.retrieval import create_retrieval_chain

# RAG 系统提示词
system_prompt = (
    "你是一个问答任务助手。"
    "使用以下检索到的上下文来回答问题。"
    "如果不知道答案，就说你不知道。"
    "回答最多三句话，保持简洁。"
    "\n\n"
    "{context}"   # 由检索器自动填充的文档上下文
)

qa_prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])

# 创建文档处理链：将检索到的多个文档块"塞入"（stuff）系统提示词
# LangChain 提供了三种文档合并策略：stuff（全部塞入）、map-reduce（分别处理再汇总）、refine（逐步精炼）
# stuff 最简单直接，适合检索结果不多（如 2~5 个块）的场景；当文档块过多超出上下文窗口时，需考虑其他策略
question_chain = create_stuff_documents_chain(llm, qa_prompt)

# 创建 RAG 检索链：检索器 + 问答链
rag_chain = create_retrieval_chain(history_aware_retriever, question_chain)

11.5 第四步：接入对话历史

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables import RunnableWithMessageHistory

store = {}  # 用来保存历史消息，key 为会话 ID

def get_session_history(session_id: str):
    """根据会话 ID 获取对应的历史消息"""
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]


# 包装为带历史记录的 Runnable
conversational_rag_chain = RunnableWithMessageHistory(
    rag_chain,
    get_session_history,
    input_messages_key='input',
    history_messages_key='chat_history',
    output_messages_key='answer',
)

为什么用内存存储而不是数据库？ 这里使用 InMemoryChatMessageHistory（基于内存的字典）是为了降低示例的复杂度——不需要额外安装数据库，运行即可看到效果。它的缺点是程序重启后历史记录会丢失。在生产环境中，应替换为持久化方案，例如 SQLChatMessageHistory(session_id, 'sqlite:///history.db')，只需改动 get_session_history 函数的返回值即可，链路的其余部分完全不变。

11.6 第五步：多轮对话测试

# 第一轮：直接提问
resp1 = conversational_rag_chain.invoke(
    {"input": "What is Task Decomposition?"},
    config={"configurable": {"session_id": "abc123"}}
)
print(resp1['answer'])

# 第二轮：用代词指代上一轮的主题
resp2 = conversational_rag_chain.invoke(
    {"input": "What are common ways of doing it?"},   # "it" 指代 Task Decomposition
    config={"configurable": {"session_id": "abc123"}}
)
print(resp2['answer'])

为什么 session_id 写死为 "abc123"？ 因为这是一个脚本形式的示例，没有 Web 服务端或前端界面来分配用户会话。写死一个固定值是为了让两次 invoke 调用共享同一份聊天历史，从而验证"上下文感知"是否生效（第二轮的 “it” 能否正确指代第一轮提到的 Task Decomposition）。在实际的 Web 应用中，session_id 通常由后端为每个用户会话自动生成（如 uuid.uuid4()），不同用户的对话历史互不干扰。

数据流分析：

第一轮（无历史）：

"What is Task Decomposition?"
    → 聊天历史为空，直接检索
    → 向量数据库返回 2 个相关文本块
    → LLM 基于文本块生成回答
    → 保存问答到 chat_history

第二轮（有历史）：

"What are common ways of doing it?"
    → 结合 chat_history，LLM 改写为 "What are common ways of Task Decomposition?"
    → 用改写后的问题检索向量数据库
    → LLM 基于新的文本块 + 历史生成回答
    → 追加到 chat_history

十二、总结：核心组件速查表

组件核心能力典型应用场景
OpenAIEmbeddings调用 OpenAI 嵌入模型快速原型开发、高质量向量
HuggingFaceEmbeddings加载 HuggingFace 上的嵌入模型本地部署 BGE 等开源模型
SentenceTransformer通用 Embedding 加载库直接使用 Qwen3 等模型
自定义 Embeddings 子类将任意模型适配 LangChain 接口私有模型整合到 LangChain 生态
FAISS高性能向量索引与检索大规模数据、生产环境
Chroma自动持久化的向量数据库开发阶段、中小规模数据
Document统一的文档对象（文本 + 元数据）所有 LangChain 文档处理的基础结构
RecursiveUrlLoader递归加载网页文档在线文档站知识库构建
WebBaseLoader加载指定网页内容单页面数据提取
RecursiveCharacterTextSplitter按字符数切割文档长文档分块入库
create_history_aware_retriever上下文感知的检索器多轮对话中的精准检索
create_retrieval_chainRAG 检索链检索 + 生成的完整链路
RunnableWithMessageHistory自动管理对话历史带记忆的多轮 RAG 对话

写在最后

到这里，我们已经走通了 文本 → 向量 → 存储 → 检索 → 生成 这条完整的 RAG 链路。

当然，本文构建的 RAG 系统还有很大的优化空间：比如将 RAG 链路嵌入 Agent 中，让模型自主决定何时检索；或者升级为多模态 RAG，让系统不仅能检索文本，还能理解图片、音频等更丰富的输入形式。这些都是后续值得探索的方向。

结合前几篇文章中的提示词模板、LCEL 链式编排和多模态能力，我们已经积累了构建智能应用所需的核心技术。下一篇文章我们将学习 LangGraph，用状态图的方式编排 Agent 工作流，让 LLM 不再只是"一问一答"，而是能够自主规划、调用工具、循环推理。敬请期待。

如有疑问或建议，欢迎留言讨论！

# RAG
Logo
龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

cover

我用 NVIDIA SkillSpector 纯静态扫了一遍 GitHub 最火的 agent skills：能挖出多少注入？0 个

avatar 龙虾开发者社区
cover

周榜换血九成，Agent Skills 星数翻倍

avatar 龙虾开发者社区
cover

智能体从入门到精通：6个必学GitHub开源项目

avatar 龙虾开发者社区