👋 欢迎阅读
📑 目录
一、前置讲解
在进入正文前，先花 10 秒了解本篇会反复用到的核心概念，带着印象去读正文，学习效率更高。
🧠 前置知识一：嵌入与向量
把文字/图片翻译成数字（向量），是语义搜索、RAG 的基础。
📏 前置知识二：向量维度与距离度量
维度决定能捕捉多细的语义；余弦相似度是度量语义相似的主力。
🗄️ 前置知识三：向量存储（Vector Stores）
专门做高维向量相似性检索的存储介质，解决 MySQL 不擅长的语义搜索。
🔍 前置知识四：检索器（Retriever）
输入查询字符串、输出文档列表的 Runnable，是 RAG 的取数组件。
🔗 前置知识五：RAG 链
retriever → prompt → model → parser，检索 + 生成一气呵成。
二、文本向量与嵌入
2.1 嵌入与嵌入模型
计算机天生擅长处理数字，但不理解文字、图片的含义。嵌入（Embedding） 的核心思想就是将人类世界的符号（如单词、句子、产品、用户、图片）转换为计算机能够理解的数值形式（即向量，本质上是一个数字列表），并且要求这种转换能够保留原始符号的语义和关系。我们可以把它想象成一个翻译过程，把人类语言"翻译"成计算机的"数学语言"。
生成式模型 vs 表示型模型：
| 模型 | 目标 |
|---|---|
| 大语言模型（生成式） | 理解输入并生成新的文本（回答问题、写文章），内部也使用嵌入，但目标是"创造" |
| 嵌入模型（表示型） | 不生成文本，而是为输入创建富含语义的数值表示（向量） |
常见的嵌入模型如 DeepSeek 的 deepseek-embedding-v2、DeepSeek-Embedding、bge-large 等。
2.2 什么是向量？
嵌入的结果就是一个向量，它本质上是一个数字列表（一维数组），例如：[0.023, 0.487, -0.129, ..., 0.325]。有两个关键概念需要了解：
向量维度
嵌入结果得到的列表长度固定，称为向量的"维度"。例如 deepseek 的 DeepSeek-Embedding 模型会生成一个 1024 维的向量。维度越高，通常能捕捉更细微的语义信息，但也需要更多的计算和存储资源。
向量空间
想象一个无限延伸、拥有无数个维度的宇宙，这个宇宙就是一个向量空间：
| 世界 | 点的表示 |
|---|---|
| 三维世界 | (x, y, z)，如 (2, 5, -1) |
| 机器学习高维空间 | (0.1, 0.7, -0.2, ..., 0.02)，包含数百或数千个数字 |
在这个空间里，每个点（向量）都能代表一个概念——一个单词、一句话、一张图片、一个用户、一部电影。
向量空间的威力：我们可以用数学来度量语义，通过计算两个向量之间的距离或相似度来实现：
| 度量方式 | 原理 | 特点 |
|---|---|---|
| 欧氏距离 | 高中几何的两点间直线距离 | 距离越短越相似 |
| 余弦相似度 | 忽略向量绝对长度，只看方向差异 | "方向"代表含义，语义相似度更常用 |
因此，在捕捉语义相似性上，余弦相似度是更常用的度量方式。维度越高，越能捕捉极其细微和复杂的语义差别（如"高兴"与"喜悦"的区别）。这能解决传统数据库（如 MySQL）不擅长的基于内容的相似性搜索问题，而非精确匹配查询。
2.3 Embeddings 嵌入模型类
在 LangChain 中，有很多的嵌入模型提供方。使用不同的模型提供方，需要安装为其各自对应的包：
| 提供方 | 安装包 |
|---|---|
| OpenAI | pip install -U langchain-openai |
| DeepSeek | pip install -U langchain-deepseek |
| Ollama | pip install -U langchain-ollama |
| Google Gemini | pip install -U langchain-google-genai |
更多见官方集成文档。
定义嵌入模型（DeepSeekEmbeddings）
# 安装：pip install -U langchain-deepseek
from langchain_deepseek import DeepSeekEmbeddings
embeddings = DeepSeekEmbeddings(
model="deepseek-embedding-v2", # DeepSeek 嵌入模型
# api_key 从环境变量 DEEPSEEK_API_KEY 自动读取，base_url 默认为 api.deepseek.com
)基础 Embeddings 类设计了两个核心方法：
| 方法 | 语义 | 用途 |
|---|---|---|
.embed_documents() | "索引" | 批量嵌入文档文本，离线、批量预处理，供后续搜索 |
.embed_query() | "搜索" | 嵌入单个查询文本，在线、实时，用于检索 |
为什么设计成两个方法？ 因为某些嵌入模型提供商会针对"被搜索的文档"和"搜索查询本身"采用不同的优化策略和模型——即使底层是同一个模型，也可能进行不同的预处理（如添加不同的指令前缀），以获得更好的搜索效果。
嵌入文档列表（embed_documents）
from langchain_deepseek import DeepSeekEmbeddings
from langchain_community.document_loaders import UnstructuredMarkdownLoader
from langchain_text_splitters import CharacterTextSplitter
# 1. 加载 Markdown 文档
markdown_path = "QT初识（1）.md"
loader = UnstructuredMarkdownLoader(markdown_path)
data = loader.load()
# 2. 文本分割器（按 token 长度切分）
text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
encoding_name="cl100k_base", chunk_size=200, chunk_overlap=50
)
# 3. 分割文档
documents = text_splitter.split_documents(data)
# 4. 定义嵌入模型
embeddings = DeepSeekEmbeddings(model="deepseek-embedding-v2")
# 5. 嵌入文档，生成向量列表（需提取文档内容为字符串列表）
texts = [doc.page_content for doc in documents]
documents_vector = embeddings.embed_documents(texts)
# 6. 打印结果
print(f"文档数量为：{len(documents)}，生成了{len(documents_vector)}个向量的列表")
print(f"第一个文档向量维度：{len(documents_vector[0])}")运行结果：
Created a chunk of size 916, which is longer than the specified 200 文档数量为：96，生成了96个向量的列表 第一个文档向量维度：1024
嵌入单个查询（embed_query）
from langchain_deepseek import DeepSeekEmbeddings
# 1. 定义嵌入模型
embeddings = DeepSeekEmbeddings(model="deepseek-embedding-v2")
# 2. 嵌入单个查询
query_vector = embeddings.embed_query("项目中遇到了哪些挑战？如何解决？")
# 3. 打印结果
print(f"向量维度：{len(query_vector)}")
print(f"向量前五个数值为：{query_vector[:5]}")运行结果：
向量维度：1024 向量前五个数值为：[-0.01306734886020422, -0.004718094132840633, -0.005446741823107004, ...]
三、向量存储（Vector Stores）
3.1 向量数据库介绍
在 LangChain 中，实际上并不需要我们手动调用嵌入模型去生成向量，再手动去比较向量。在 RAG 知识地图中，存在一个 Vector Stores（向量存储）——向量被管理在专门的向量存储介质（如向量数据库）中。向量存储的核心任务是解决传统数据库（如 MySQL）不擅长的问题：基于内容的相似性搜索，而非基于精确匹配的查询。
为什么需要专门的向量数据库？ 一篇文档转换成 1536 个浮点数的向量，一百万篇文档就是约 1.5 GB 纯向量数据，现实中可能达到千万甚至亿级。向量数据库提供了专门用于高效存储、管理和检索高维向量的能力。
向量数据库的核心机制：
| 机制 | 说明 |
|---|---|
| 专门的索引（灵魂） | 不用暴力搜索，预先构建索引结构。如近似最近邻（ANN）搜索：牺牲一点点精度换取极致速度，通过聚类/分层/压缩把搜索范围缩小到"几个最可能的候选集"（像图书馆按分类找书） |
| CRUD 操作 | 支持增删改查，动态更新向量数据 |
| 元数据过滤（关键） | 先按元数据（时间/作者/类别）过滤，再在缩小范围内做向量搜索，提升准确性和效率 |
| 可扩展性与持久化 | 支持分布式部署处理海量数据，数据持久化避免断电丢失 |
| 集成方便 | 提供 gRPC、RESTful 等 API，LangChain 可轻松集成 |
常见的向量数据库有 Chroma、Weaviate、Pinecone、Qdrant、Milvus 等。LangChain 与它们集成，让开发者无需手动处理向量生成、存储和比较的复杂性。
3.2 内存存储（InMemoryVectorStore）
我们将使用 LangChain 的 InMemoryVectorStore 来实现向量的内存存储（程序结束即消失，不持久化）。
① 初始化
代码示例：
from langchain_deepseek import DeepSeekEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
# 1. 定义嵌入模型
embeddings = DeepSeekEmbeddings(model="deepseek-embedding-v2")
# 2. 初始化内存向量存储（存/查都用它转向量）
vector_store = InMemoryVectorStore(embedding=embeddings)② 添加文档（add_documents）
添加文档时，该方法会为文档编排索引。这也是我们之前提到的数据检索两大步骤：编制索引（从源摄取数据并编索引）+ 检索和生成（接受查询、检索相关数据、传给模型）。
代码示例：
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.documents import Document
from langchain_deepseek import DeepSeekEmbeddings
embeddings = DeepSeekEmbeddings() # 默认就是 deepseek-embedding-v2
vector_store = InMemoryVectorStore(embedding=embeddings)
# 手工创建 Document 对象（带 id + metadata）
doc1 = Document(id="1", page_content="机器学习是人工智能的一个子集", metadata={"source": "wiki", "topic": "AI"})
doc2 = Document(id="2", page_content="深度学习使用多层神经网络", metadata={"source": "wiki", "topic": "AI"})
documents = [doc1, doc2]
# 手动指定 id（不传则自动生成随机 id）
vector_store.add_documents(documents=documents, ids=["1", "2"])③ 获取文档（get_by_ids）
代码示例：
docs_back = vector_store.get_by_ids(["1", "2"])
for d in docs_back:
print(d.id, "->", d.page_content)④ 删除文档（delete）
删除传入索引列表对应的文档；不传入索引列表则全量删除。
vector_store.delete(ids=["1"]) # 删除 id="1" 的文档3.3 相似性搜索
想要获取相似性搜索的结果（嵌入查询 → 查找相似文档 → 返回文档列表），可以使用 similarity_search 方法。重点：InMemoryVectorStore 是根据余弦相似度来捕捉语义的。
代码示例：
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_deepseek import DeepSeekEmbeddings
from langchain_core.documents import Document
embeddings = DeepSeekEmbeddings()
vector_store = InMemoryVectorStore(embedding=embeddings)
# 添加文档
docs = [
Document(page_content="猫是可爱的动物", metadata={"type": "animal"}),
Document(page_content="狗是人类的朋友", metadata={"type": "animal"}),
Document(page_content="今天天气晴朗", metadata={"type": "weather"}),
]
vector_store.add_documents(docs)
# 相似性搜索（k=返回数量）
results = vector_store.similarity_search("宠物", k=2)
for doc in results:
print(doc.page_content) # 输出"狗是人类的朋友" 和 "猫是可爱的动物"运行结果：
狗是人类的朋友 猫是可爱的动物
💡 为什么搜出来的是"相似的文档"而不是答案？ 恭喜你，已经具备了构建智能应用的基础逻辑思维——把"找到的最相关内容"和"问题"一起交给 LLM 生成答案，就能极大提高答案的准确性和时效性。以上流程，称为检索增强生成（RAG）。
3.4 元数据过滤
每个 Document 对象都包含：id（唯一标识符）、page_content（字符串文本）、metadata（任意元数据）。虽然向量数据库能搜索所有嵌入文档，但现实场景中，我们还希望先按元数据过滤缩小搜索范围。
代码示例：
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.documents import Document
from langchain_deepseek import DeepSeekEmbeddings
# 1. 初始化向量存储
embeddings = DeepSeekEmbeddings()
vector_store = InMemoryVectorStore(embedding=embeddings)
# 2. 添加带 metadata 的文档（含 source 字段，供过滤）
vector_store.add_documents([
Document(page_content="数据库表怎么设计？主键和外键", metadata={"source": "hahaha"}),
Document(page_content="数据表如何建索引优化查询", metadata={"source": "hahaha"}),
Document(page_content="机器学习模型训练流程", metadata={"source": "wiki"}),
])
# 3. 定义过滤函数（返回 bool）
def _filter_function(doc: Document) -> bool:
return doc.metadata.get("source") == "hahaha"
# 4. 相似性搜索：query + k + filter（先过滤再搜）
search_docs = vector_store.similarity_search(
query="数据库表怎么设计的？", k=2, filter=_filter_function)
for doc in search_docs:
print("*" * 30)
print(doc.page_content)运行结果：
****************************** 数据库表怎么设计？主键和外键 ****************************** 数据表如何建索引优化查询
filter 参数接收一个过滤函数（返回布尔值），先根据文档元数据过滤出符合条件的文档，再进行向量搜索。
3.5 Redis 向量存储
我们还可以使用 Redis 来存储向量。它速度快、拥有庞大的客户端库生态。从本质上讲，Redis 是键值型 NoSQL 数据库，还提供了搜索和查询功能（RediSearch），使其以缓存的速度充当向量数据库。
核心概念：
| 概念 | 说明 |
|---|---|
| RediSearch | Redis 官方的高性能搜索与全文索引引擎模块，支持分词搜索，无需部署 ES 等大型组件 |
| Index（索引） | RediSearch 中的查询目录：独立数据结构，建立在多个 Redis Key 之上，存指针 + 字段索引信息 |
| Index Fields（索引字段） | 创建索引时指定的字段，定义索引的"蓝图" |
| metadata schema | 描述元数据结构的声明格式（name = 字段名，type = 字段类型） |
Index Fields 的四种类型：
| 类型 | 用途 |
|---|---|
| TAG | 精确匹配的分类/标签，可过滤、分组 |
| NUMERIC | 整数/浮点数，可数值范围查询、排序 |
| TEXT | 全文搜索的字符串，支持分词、模糊匹配 |
| GEO | 经纬度坐标，用于地理空间查询 |
初始化（RedisVectorStore）：
from langchain_deepseek import DeepSeekEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore
# 1. 定义嵌入模型
embeddings = DeepSeekEmbeddings(model="deepseek-embedding-v2")
# 2. 配置 Redis 客户端
config = RedisConfig(
index_name="qa", # 索引名
redis_url="redis://192.168.100.238:6379",
metadata_schema=[ # 元数据字段声明（Index Fields）
{"name": "category", "type": "tag"},
{"name": "num", "type": "numeric"},
],
)
# 3. Redis 存储初始化（需要真连上 Redis 才能运行）
vector_store = RedisVectorStore(embeddings, config=config)RedisConfig 关键参数：
| 参数 | 说明 |
|---|---|
| index_name | 索引名称，默认生成 ULID 唯一标识符 |
| key_prefix | Redis Key 的前缀，默认 index_name |
| redis_url | Redis 实例 URL，默认 redis://localhost:6379 |
| metadata_schema | 元数据字段 Schema，设置后有助于元数据过滤 |
四种搜索方法：
| 方法 | 查询方式 | 返回 |
|---|---|---|
| similarity_search | 查询文本 | 文档列表 |
| similarity_search_by_vector | 查询向量 | 文档列表 |
| similarity_search_with_score | 查询文本 | 文档 + 相似度分数 |
| similarity_search_with_score_by_vector | 查询向量 | 文档 + 相似度分数 |
用文本查询 = 内部先把文本转向量再检索；用向量查询 = 适合已预计算查询向量的场景，避免重复编码。
元数据过滤（Redis 过滤表达式）：
代码示例：
from langchain_deepseek import DeepSeekEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore
from redisvl.query.filter import Tag
embeddings = DeepSeekEmbeddings(model="deepseek-embedding-v2")
config = RedisConfig(
index_name="qa",
redis_url="redis://192.168.100.238:6379",
metadata_schema=[
{"name": "category", "type": "tag"},
{"name": "num", "type": "numeric"},
{"name": "source", "type": "tag"}, # 供下面过滤用
],
)
vector_store = RedisVectorStore(embeddings, config=config)
# Tag 过滤表达式：限定 source 字段
filter_condition = Tag("source") == "hahahaha"
scored_results = vector_store.similarity_search_with_score(
query="数据库表怎么设计的？",
k=2,
filter=filter_condition,
)
for doc, score in scored_results:
print("*" * 30)
print(f"Content: {doc.page_content[:100]}...")
print(f"Metadata: {doc.metadata}")
print(f"Score: {score}")3.6 最大边际相关性搜索（MMR）
最大边际相关性（MMR） 是一种重排序算法，利用语义相似性作为基础工具，从候选集中选出一组既能代表查询主题又彼此多样化的结果。
| 方式 | 比喻 | 目标 |
|---|---|---|
| 语义相似性 | 面试官为每个应聘者与职位要求打分 | 单点匹配度 |
| MMR | 团队经理组建团队（技术达标 + 技能互补） | 一组"精华"结果 |
MMR 使用场景： 推荐系统（相关但不同类型，避免"信息茧房"）、文档摘要（覆盖主旨且不重复）、RAG（检索大量相关文档后用 MMR 去重和多样化筛选，提升答案质量、减少幻觉）。
使用 MMR 搜索，需要使用 max_marginal_relevance_search 方法：
| 参数 | 说明 |
|---|---|
| query | 查询文本 |
| k | 最终返回的文档数量 |
| fetch_k | 初始检索的候选文档数（需 ≥ k） |
| lambda_mult | 相关性-多样性权衡系数（0=最大多样性，1=最大相似性，默认 0.5） |
代码示例（典型用法）：
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_deepseek import DeepSeekEmbeddings
embeddings = DeepSeekEmbeddings()
vector_store = InMemoryVectorStore(embedding=embeddings)
# 假设已添加文档 ...
# MMR 搜索
results = vector_store.max_marginal_relevance_search(
query="人工智能应用",
k=3,
fetch_k=10,
lambda_mult=0.6
)
for doc in results:
print(doc.page_content)3.7 Pinecone 向量存储
Pinecone 是为机器学习应用量身打造的生产级向量数据库服务，适用于高维向量数据的高效存储、索引与查询。它是一个全托管的向量数据库平台——负责所有后端维护、扩展、更新和监控，让用户专注应用开发。
Pinecone 官网：The vector database to build knowledgeable AI | Pinecone（需科学上网）
环境设置：
-
注册新用户，选择个人免费版（可直接 Skip 跳过账户信息）
-
注册成功生成默认 API Key（妥善保存）
-
设置环境变量
PINECONE_API_KEY -
安装：
pip install -qU pinecone langchain-pinecone
使用步骤：
| 步骤 | 说明 |
|---|---|
| 创建客户端 | pc = Pinecone(api_key=...) |
| 创建索引 | pc.create_index(name=..., dimension=1024, metric="cosine", spec=ServerlessSpec(...)) |
| 获取索引 | index = pc.Index(index_name) |
| 定义向量存储 | vector_store = PineconeVectorStore(index=index, embedding=embeddings) |
代码示例：
from pinecone import Pinecone, ServerlessSpec
from langchain_pinecone import PineconeVectorStore
from langchain_deepseek import DeepSeekEmbeddings
pc = Pinecone(api_key="your-api-key")
# 创建新索引（维度需与嵌入模型一致）
pc.create_index(
name="my-index",
dimension=1024,
metric="cosine",
spec=ServerlessSpec(cloud="aws", region="us-east-1"),
)
# 定义嵌入模型
embeddings = DeepSeekEmbeddings(model="deepseek-embedding-v2")
# 获取索引 + 定义向量存储
index = pc.Index("my-index")
vector_store = PineconeVectorStore(index=index, embedding=embeddings)create_index 关键参数：
| 参数 | 说明 |
|---|---|
| name | 索引的唯一名称 |
| dimension | 向量维度，必须与嵌入向量维度匹配 |
| metric | 相似度度量：cosine / dotproduct / euclidean |
| spec | 部署规格：ServerlessSpec（无服务器）或 PodSpec |
四、检索器（Retrievers）
4.1 检索系统与检索器
检索系统（IR System） 是一个从大规模、非结构化数据集合中，自动、高效查找、排序并返回相关信息的计算机系统——在正确的时间，以正确的方式，将正确的信息传递给正确的人（如搜索引擎）。随着大语言模型的流行，检索系统已成为 RAG 的重要组成部分。
检索系统的三种类型：
| 类型 | 原理 | 擅长 |
|---|---|---|
| 关系数据库 | 数据存行/列，用 SQL 查询 | 维护完整性、复杂查询、实体关系 |
| 词法搜索索引 | 匹配查询与文档中的单词（倒排索引） | 精确词匹配检索 |
| 向量数据库 | 嵌入模型转高维向量，算余弦相似度 | 语义相似性搜索 |
检索器（Retriever） 是检索系统的核心组件：接收查询（Query），检索出候选文档集合。LangChain 提供统一接口：
输入：查询字符串 → 输出：文档列表（Document）
例如，使用关系数据库的检索系统，检索器可以将问题转换为 SQL 语句并执行查询，最后将结果响应用户。
4.2 使用向量数据库作为检索器（as_retriever）
向量存储是索引和检索非结构化数据的强大方法。可以通过调用向量数据库的 as_retriever 方法，将向量存储用作检索器。
as_retriever 常用参数：
| 参数 | 说明 |
|---|---|
| search_type | 相似度算法："similarity"（默认）/ "mmr" / "similarity_score_threshold" |
| search_kwargs.k | 限制返回的文档数量 |
| search_kwargs.fetch_k | 传递给 MMR 算法的候选文档数 |
代码示例：
# 仅限制返回文档数
retriever = vector_store.as_retriever(search_kwargs={"k": 2})
# 使用 MMR 算法
retriever = vector_store.as_retriever(
search_type="mmr",
search_kwargs={"k": 2, "fetch_k": 10},
)完整实例（Redis 向量存储 → 检索器）：
from langchain_deepseek import DeepSeekEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore
embeddings = DeepSeekEmbeddings(model="deepseek-embedding-v2")
config = RedisConfig(
index_name="qa",
redis_url="redis://192.168.100.238:6379",
metadata_schema=[
{"name": "category", "type": "tag"},
{"name": "num", "type": "numeric"},
],
)
vector_store = RedisVectorStore(embeddings, config=config)
# 把向量库变成"检索器"（Runnable）
retriever = vector_store.as_retriever()
# invoke(query)：传入查询文本，内部转向量 → 找最相似 → 返回文档列表
docs = retriever.invoke("数据库表怎么设计的？")
for doc in docs:
print("*" * 30)
print(doc.page_content[:30])运行结果：
****************************** 数据库表怎么设计？主键和外键 ****************************** 数据表如何建索引优化查询
注意：Retrievers 是 Runnable 对象，但不提供任何流式处理——因为它通常是同步的、阻塞的操作。
4.3 使用 @chain 创建"检索器"
除了使用 as_retriever，我们还可以用 @chain 装饰器自行创建检索器——把一个普通 Python 函数变成 LangChain 的 Runnable（输入查询字符串，输出文档列表）。
代码示例：
from langchain_deepseek import DeepSeekEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore
from langchain_core.runnables import chain
from typing import List
from langchain_core.documents import Document
# ……（定义嵌入模型 / 配置 Redis / 初始化向量存储）……
# @chain：把普通函数变成 Runnable
@chain
def retriever(query: str) -> List[Document]:
"""把查询文本做相似搜索，返回最相关的 k 个文档（这里 k=2）。"""
return vector_store.similarity_search(query, k=2)
# 调用检索器
docs = retriever.invoke("数据库表怎么设计的？")
for doc in docs:
print("*" * 30)
print(doc.page_content[:30])运行结果：
****************************** 数据库表怎么设计？主键和外键 ****************************** 数据库表如何设计索引优化查询
4.4 RAG 案例
由于检索器是 Runnable 对象，可以方便地使用链完成 RAG 调用。最简单的 RAG 案例步骤：
-
根据 Query 搜索最相关的 4 篇文档
-
将相关文档转换为字符串
-
将 Query 与文档字符串发送给聊天模型
-
聊天模型依据输出解析器格式输出内容
代码示例：
from langchain_deepseek import DeepSeekEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain.chat_models import init_chat_model
# 1. 定义聊天模型（适配 DeepSeek）
model = init_chat_model(model="deepseek-chat", model_provider="deepseek")
# 2. 定义嵌入模型
embeddings = DeepSeekEmbeddings(model="deepseek-embedding-v2")
# 3. 配置 Redis 客户端
config = RedisConfig(
index_name="qa",
redis_url="redis://192.168.100.238:6379",
metadata_schema=[
{"name": "category", "type": "tag"},
{"name": "num", "type": "numeric"},
],
)
# 4. 定义向量存储 + 检索器
vector_store = RedisVectorStore(embeddings, config=config)
retriever = vector_store.as_retriever()
# 5. 定义提示词模板（{question} 和 {context} 两个占位符）
prompt = ChatPromptTemplate.from_messages(
[
(
"human",
"""你是负责回答问题的助手。使用以下检索到的上下文片段来回答问题。如果你不知道答案，就说你不知道。最多只用三句话，回答要简明扼要。
Question: {question}
Context: {context}
Answer:""",
),
]
)
# 6. 将文档转换为字符串
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
# 7. 定义 RAG 链
# context = 从向量库检索文档并格式化；question = 原样透传用户问题
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)
# 8. 执行链，流式输出
for chunk in rag_chain.stream("数据库表怎么设计的？"):
print(chunk, end="|", flush=True)运行结果（节选）：
|数据库|表|设计|可以|通过|水平|分|表|来|优化|性能|，|特别|是|当|单|表|数据|量|过大|...
RunnablePassthrough 是什么？ 它是一个"伪" Runnable，主要作用是在链（Chain） 中透明地传递输入数据，不做任何修改。当我们需要将原始输入和另一个处理过程的输出一起传递给下一个步骤时，就需要它——例如代码中把 Query 与检索出的文档字符串同时发送给提示词模板。
五、复盘（附答案）
💡 思考什么是嵌入（Embedding）？为什么说"嵌入模型是表示型模型"？
-
欧氏距离和余弦相似度有什么区别？为什么语义相似度更常用余弦？
-
embed_documents 和 embed_query 有什么区别？为什么设计成两个方法？
-
向量数据库解决了传统数据库（MySQL）的什么问题？它有哪些核心机制？
-
RAG 链是如何串起来的？RunnablePassthrough 的作用是什么？
📝 答案
-
嵌入把人类符号（单词/句子/图片/用户/产品）转成保留语义关系的数值向量。大语言模型是生成式模型（目标是"创造"文本）；嵌入模型是表示型模型——不生成文本，只为输入创建富含语义的数值表示（向量）。
-
欧氏距离是两点间直线距离（距离越短越相似），受向量长度影响；余弦相似度忽略绝对长度，只看方向（"方向"代表含义，"长度"往往只代表文本长短），因此更常用于度量语义相似性。
-
embed_documents 语义是"索引"：批量嵌入文档，离线预处理供后续搜索；embed_query 语义是"搜索"：在线嵌入单个查询用于检索。分开设计是因为部分提供商对"文档"和"查询"采用不同优化策略或预处理（如不同的指令前缀）。
-
解决基于内容的相似性搜索问题（MySQL 只擅长精确匹配）。核心机制：专门的索引（ANN 近似最近邻，牺牲一点精度换速度）、CRUD、元数据过滤（先缩小范围再搜索）、可扩展性与持久化、集成方便。
-
RAG 链：
{"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | model | parser——先并行准备 context（检索+格式化）和 question（透传），再填模板、交给模型、解析输出。RunnablePassthrough 是透明传递输入的"伪 Runnable"，用于把原始输入和另一处理过程的输出一起传给下一步。
🎯 闭幕
从"嵌入与向量"到"向量存储"，从"内存/Redis/Pinecone"到"检索器与 RAG 链"——这一篇帮你把 LangChain 的向量检索链路走了一遍。
如果本文对你有帮助，欢迎：
👍 点赞 ｜ ⭐ 收藏 ｜ 👤 关注作者 ｜ 💬 留言交流你的疑问或补充
你的每一次互动都是我继续更新的动力，我们下一篇见！🚀
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/2601_96587588/article/details/164404125
