🌈个人主页：秦jh__https://blog.csdn.net/qinjh_?spm=1010.2135.3001.5343 🔥 系列专栏：https://blog.csdn.net/qinjh_/category_13137010.html
目录
检索器（Retrievers）
相关概念
检索系统
检索器
使用向量数据库作为检索器
基本使用
使用 @chain 创建“检索器”
RAG 案例
前言
💬 hello! 各位铁子们大家好哇。
今日更新了LangChain相关内容 🎉 欢迎大家关注🔍点赞👍收藏⭐️留言📝
检索器（Retrievers）
相关概念
检索系统
检索系统（Information Retrieval System, IR System）是一个为了满足用户信息需求，从大规模、非 结构化的数据集合中，自动、高效地查找、排序并返回相关信息的计算机系统。
它的核心任务是：在正确的时间，以正确的方式，将正确的信息传递给正确的人。最常见的例子就 是：搜索引擎（如Google、百度）。
随着大型语言模型的流行，检索系统已成为人工智能应用（例如 RAG）的重要组成部分。且存在多种 【不同类型】的检索系统，包括：
- 关系数据库
关系数据库是许多应用程序中使用的结构化数据存储的基本类型。数据存储在行（记录）和列（属 性）中，可以通过 SQL（结构化查询语言）进行高效的查询和作。关系数据库擅长维护数据完整性、 支持复杂查询以及处理不同数据实体之间的关系。
- 词法搜索索引
许多搜索引擎基于将查询中的单词与每个文档中的单词进行匹配，这种方法称为词法检索。即一个单 词经常出现在用户的查询和特定文档中，那么这个文档可能是一个很好的匹配。这通常使用【倒排索 引】实现。
- 向量数据库
向量存储不使用字频，而是使用【嵌入模型】将文档转换为高维向量表示。这允许使用余弦相似度等 数学运算对嵌入向量进行有效的相似性搜索。
检索器
检索器是检索系统中的一个核心组件，它接收来自用户接口的查询（Query），检索出包含查询关键词 的候选文档集合。
我们可以使用上面提到的任何检索系统实现方式创建检索器！如关系数据库、向量数据库等。由于其 重要性和多样性，LangChain 提供了一个统一的接口来与不同类型的检索系统进行交互。LangChain 的检索器接口非常简单：
例如，使用关系数据库的检索系统，检索器可以将问题转换为 SQL 语句，并执行查询，最后将查询结 果响应用户。
下面，我们来演示如何使用向量存储作为检索器。
使用向量数据库作为检索器
基本使用
向量存储是索引和检索非结构化数据的一种强大而有效的方法。可以通过调用向量数据库的 as_retriever 方法，将向量存储用作检索器。在这里我们使用 Redis 向量存储。
# 嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# Redis 配置
config = RedisConfig(
index_name="qa", # 定义索引名
redis_url="redis://192.168.100.238:6379",
metadata_schema=[
{"name": "category", "type": "tag"}, # 添加索引字段：分类
{"name": "num", "type": "numeric"}, # 添加索引字段：编号
]
)
# 初始化 Redis 向量存储实例（建立了索引结构）
vector_store = RedisVectorStore(
embeddings=embeddings,
config=config,
)
# 检索器依赖向量库(Runnable)
retriever = vector_store.as_retriever()
search_docs = retriever.invoke("项目介绍")
for doc in search_docs:
print("*" * 30)
print(doc.page_content)
LangChain 检索器是一个 Runnable 的对象 ，它是 LangChain 组件的标准接口。这意味着它有一些常 用方法，包括 invoke ，用于与其交互。默认情况下，向量存储检索器使用相似性搜索。
as_retriever 方法也支持我们传递相关参数修改搜索结果，如：
- search_type ：设置相似算法，包括： “similarity” （默 认）、 “mmr” 或 “similarity_score_threshold” （相似性分数阈值）
- search_kwargs ：
- k ：限制检索器返回的文档 k 数量。
- retriever = vector_store.as_retriever(search_kwargs={"k": 2})
- fetch_k ：要传递给 MMR 算法的文档量。
- retriever = vector_store.as_retriever(
search_type="mmr",
search_kwargs={
"k": 2,
"fetch_k": 10
}
)
- retriever = vector_store.as_retriever(
- k ：限制检索器返回的文档 k 数量。
要注意的是， Retrievers 检索器虽然是 Runnable 对象，但其不提供任何流式处理，因为它本身 通常是同步的、阻塞的操作。如下所示：
chunks = []
for chunk in retriever.stream("数据库表怎么设计的？"):
chunks.append(chunk)
print(chunk, end="|", flush=True)
调用可以发现，它是一次性返回，与 invoke 调用无异。
使用 @chain 创建“检索器”
除了使用 as_retriever 方法，我们还可以自行创建一个“检索器”。回想一下检索器的特点：
综上所述，我们可以：
# 使用@chain，定义检索器函数，当作具有Runnable的"检索器"使用
@chain
def retriever(query: str) -> List[Document]:
return vector_store.similarity_search(query=query ,k=2)
search_docs = retriever.invoke("项目介绍")
for doc in search_docs:
print("*" * 30)
print(doc.page_content)
上面定义了一个函数，使用 @chain 修饰，该修饰可以使其成为 Runnable 函数，且满足检索器输入 输出的要求。在函数中，我们依旧使用向量数据库的相似性搜索方法，这样灵活性也更高，想要进行 元数据筛选也更方便。
注意，这并不是真正的检索器，检索器是一个 Runnable 对象，而我们定义的只是一个函数，具备其 特点罢了。
RAG 案例
RAG 是当前大语言模型应用的核心模式。当用户向 LLM 提问时，系统首先使用嵌入模型在知识库中进 行语义搜索，找到最相关的内容，然后将这些内容和问题一起交给 LLM 来生成答案。这极大地提高了 答案的准确性和时效性。
由于 LangChain 检索器是一个 Runnable 的对象，我们便可以方便的使用 链 完成相关的调用。下面 我们来完成一个最简单的 RAG 案例，将会完成：
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_redis import RedisConfig, RedisVectorStore
# 构建链：完成RAG能力
# 定义组件，构建链
# 嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# 聊天模型
model = ChatOpenAI(model="gpt-4o-mini")
# 1. 先从知识库中检索
# Redis 配置
config = RedisConfig(
index_name="qa", # 定义索引名
redis_url="redis://192.168.100.238:6379",
metadata_schema=[
{"name": "category", "type": "tag"}, # 添加索引字段：分类
{"name": "num", "type": "numeric"}, # 添加索引字段：编号
]
)
# Redis 向量库
vector_store = RedisVectorStore(
embeddings=embeddings,
config=config,
)
# 检索器
retriever = vector_store.as_retriever()
# 2. 将检索结果+查询语句 构建为提示词（提示词模板）
# 提示词模板
prompt = ChatPromptTemplate.from_messages(
[
(
"human",
"""你是负责回答问题的助手。使用一下检索到的上下文片段来回答问题。如果你不知道答案，就说不知道答案。最多回复三句话的结果，回答要简明扼要
Question:{question}
Context:{context}
Answer:"""
)
]
)
# 将检索出来的文档转换成文本传递给提示词模板
def format_docs(docs):
return "\\n\\n".join(doc.page_content for doc in docs)
# 3. 将消息发送给 LLM（实例化消息，交由链完成）
# 定义链，执行时需要 question
# 检索器 + format_docs， question （同时传递）
# prompt
# model
# 输出解析器
chain = (
# 检索器 + format_docs 分支1
# question 分支2: RunnablePassthrough() 在链中透传输入数据，保持原始问题不变，直接传递给后续步骤
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser() # 输出解析器
)
# 执行流：
# 输入问题："项目介绍"
# 并行执行两个分支，提高效率
# 输出结果：
# {
# "context":"xxxx"
# "question":"项目介绍"
# }
# 4. 打印字符串结果 （流式）
while True:
# 获取用户输入
question = input("\\n请输入您的问题（输入'退出'或'quit'结束程序）: ").strip()
# 检查是否退出
if question.lower() in ["退出", "quit"]:
print("程序已结束，再见！")
break
# 检查输入是否为空
if not question:
print("问题不能为空，请重新输入。")
continue
# 执行链，流式输出
print("回答: ", end="", flush=True)
for chunk in chain.stream(question):
print(chunk, end="", flush=True)
print() # 换行
上述代码中， RunnablePassthrough 我们之前还没有见过，简单来说， RunnablePassthrough 是一个“伪”Runnable，它的主要作用是在链（Chain）中透明地传递 输入数据，而不做任何修改。
当我们需要将原始输入和另一个处理过程的输出一起传递给下一个步骤时，就需要 RunnablePassthrough 。就例如代码中，我们需要将【Query】与【通过检索出来的文档转换的 字符串】同时发送给提示词模板。
上面讲过检索器不提供任何流式处理。但这里可以使用流式输出。我们可以把像 RAG 这样的链的执行 过程想象成两个阶段：
- 阶段一：准备阶段（同步、阻塞）：接收输入、检索文档、构建提示
- 阶段二：生成阶段（可流式）：调用LLM
流式输出的是最终答案，而非中间过程。
