目录1. 检索系统与检索器1.2 检索器在 LangChain 中的定位1.3 LangChain 检索器接口2配置参数as_retriever()2. RAG 案例从检索到生成的完整链路1. 检索系统与检索器1.1 什么是检索系统检索系统是一个从大规模、非结构化数据集合中自动、高效地查找、排序并返回相关信息的计算机系统。最典型的例子就是搜索引擎如 Google、百度其核心使命是在正确的时间以正确的方式将正确的信息传递给正确的人。在 RAG检索增强生成 时代检索系统的形态更加多样常见的有类型说明典型代表关系数据库结构化数据存储通过 SQL 查询MySQL, PostgreSQL词法搜索索引基于关键词匹配倒排索引Elasticsearch, Lucene向量数据库基于嵌入向量相似度搜索Redis, Pinecone, Milvus1.2 检索器在 LangChain 中的定位检索器Retriever 是检索系统中负责接收查询Query并返回相关文档列表的组件LangChain 为其定义了极其简洁的统一接口输入一个查询字符串query: str输出一个 LangChain Document 对象列表List[Document]这种抽象使得任何满足该接口的组件Runnable都可以作为检索器使用LangChain 提供了 chain 装饰器可以将一个普通函数快速转换为 Runnable# 场景 1底层是向量数据库语义检索 # 内部将Query转成向量在向量空间算余弦相似度 retriever vector_store.as_retriever() docs retriever.invoke(数据库表怎么设计) # 调用方式 # 场景 2底层是 SQL 数据库关系检索 # 内部将Query转成SQL语句去MySQL查表 chain def sql_retriever(query: str) - List[Document]: sql fSELECT * FROM knowledge WHERE content LIKE %{query}% rows db.run(sql) return [Document(page_contentstr(row)) for row in rows] docs sql_retriever.invoke(数据库表怎么设计) # 调用方式一模一样 # 场景 3底层是外部 APIHTTP 调用 # 内部向百度搜索引擎或内部API发HTTP请求 chain def api_retriever(query: str) - List[Document]: response requests.get(fhttps://my-search-api.com/search?q{query}) return [Document(page_contentitem[text]) for item in response.json()] docs api_retriever.invoke(数据库表怎么设计) # 调用方式依然一样1.3 LangChain 检索器接口1基本接口方法LangChain 检索器是一个Runnable对象LangChain 标准组件因此它支持以下核心方法方法描述返回值类型说明invoke(query)执行检索返回文档列表List[Document]最常用的同步调用直接返回匹配的文档集合batch([queries])批量检索List[List[Document]]并发执行多个查询。外层列表对应输入查询的顺序内层列表是每个查询对应的文档结果stream(query)流式输出Iterator[List[Document]]注意检索器不支持逐字流式它会一次性生成完整的List[Document]作为迭代器的唯一元素返回行为与invoke本质相同仅接口形式适配流式协议检索器本质上是同步阻塞操作因为检索过程如数据库查询、向量搜索通常是一次性完成的。此时的 stream 方法实际上与 invoke 无异不会分块输出from typing import List, Iterator from langchain_core.documents import Document # 1. invoke单进单出 docs: List[Document] retriever.invoke(什么是RAG) # 2. batch多进多出嵌套列表 queries [什么是RAG, 什么是检索器] results: List[List[Document]] retriever.batch(queries) # results[0] 是第一个问题的文档列表results[1] 是第二个问题的文档列表 # 3. stream返回迭代器但实际一次性吐出全部结果 stream_iter: Iterator[List[Document]] retriever.stream(什么是RAG) for docs in stream_iter: # 实际上只会循环一次 print(len(docs))2配置参数as_retriever()当我们使用向量存储的 as_retriever() 方法时可以将向量数据库转换为检索器并通过传入以下参数来精细控制检索行为参数可选值说明search_typesimilarity默认mmrsimilarity_score_threshold相似性算法类型-similarity普通相似性搜索-mmr最大边际相关性兼顾多样性-similarity_score_threshold设置分数阈值过滤search_kwargs字典常用键-k返回文档数量默认 4-fetch_kMMR 算法的候选文档数需大于k-score_threshold相似度阈值仅对similarity_score_threshold有效附加检索参数k 是在过滤后生效的分数范围取决于嵌入模型similarity_score_threshold 是准入下界大于等于NGramOverlapExampleSelector 中的参数 threshold 是剔除下界仅大于from langchain_openai import OpenAIEmbeddings from langchain_core.vectorstores import InMemoryVectorStore from langchain_core.documents import Document # ---------- 1. 准备数据 ---------- # 与之前相同的测试数据 documents [ Document(page_content苹果是一种蔷薇科的落叶乔木果实富含维生素, metadata{category: fruit}), Document(page_content香蕉是芭蕉科的巨型草本植物热带地区广泛种植, metadata{category: fruit}), Document(page_content梨属于蔷薇科苹果亚科果实多汁酸甜, metadata{category: fruit}), Document(page_content汽车通常由内燃机或电动机驱动有四个轮子, metadata{category: vehicle}), ] # 初始化嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-large) # ---------- 2. 初始化 InMemoryVectorStore 并添加文档 ---------- # 方式一先初始化再添加文档[reference:1][reference:2] vector_store InMemoryVectorStore(embeddings) vector_store.add_documents(documentsdocuments) # 或者也可以直接用 from_documents 一步完成 # vector_store InMemoryVectorStore.from_documents(documents, embeddings) query 水果 # ---------- 3. 模式一普通相似性搜索 (similarity) ---------- print(*40) print(模式similarity (普通相似度)) retriever_1 vector_store.as_retriever( search_typesimilarity, search_kwargs{k: 3} ) for idx, doc in enumerate(retriever_1.invoke(query)): print(f 结果{idx1}: {doc.page_content}) # ---------- 4. 模式二MMR (最大边际相关性) ---------- print(\n *40) print(模式mmr (兼顾多样性)) retriever_2 vector_store.as_retriever( search_typemmr, search_kwargs{ k: 2, fetch_k: 4 } ) for idx, doc in enumerate(retriever_2.invoke(query)): print(f 结果{idx1}: {doc.page_content}) # ---------- 5. 模式三相似度阈值过滤 (similarity_score_threshold) ---------- print(\n *40) print(模式similarity_score_threshold (阈值过滤)) retriever_3 vector_store.as_retriever( search_typesimilarity_score_threshold, search_kwargs{ k: 3, score_threshold: 0.25 } ) for idx, doc in enumerate(retriever_3.invoke(query)): print(f 结果{idx1}: {doc.page_content})2. RAG 案例从检索到生成的完整链路现在我们基于上面的检索器搭建一个完整的 RAG 问答链。流程如下接收用户问题question检索相关文档retriever → 返回 List[Document]格式化文档将多个文档拼接为上下文字符串构建提示词将问题和上下文填入模板调用 LLM 生成答案解析输出以字符串形式返回from langchain_openai import OpenAIEmbeddings, ChatOpenAI 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 # ---------- 初始化组件 ---------- # 聊天模型使用 gpt-4o-mini 兼顾速度与效果 model ChatOpenAI(modelgpt-4o-mini) # 嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-large) # Redis 配置 config RedisConfig( index_nameqa, redis_urlredis://192.168.100.238:6379, metadata_schema[ {name: category, type: tag}, {name: num, type: numeric}, ], ) vector_store RedisVectorStore(embeddings, configconfig) # 检索器返回前 4 个文档 retriever vector_store.as_retriever(search_kwargs{k: 4}) # ---------- 定义提示词模板 ---------- 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) # ---------- 构建 RAG 链 ---------- rag_chain ( { context: retriever | format_docs, # 检索 格式化 question: RunnablePassthrough() # 原样传递问题 } | prompt | model | StrOutputParser() ) # ---------- 执行并流式输出 ---------- for chunk in rag_chain.stream(数据库表怎么设计的): print(chunk, end|, flushTrue)外层花括号在 LCEL 管道中并非普通字典而是被自动转换为RunnableParallel可并行执行容器它允许将同一个用户输入同时分发到多个独立的分支进行处理最终再将各分支的输出结果合并成一个新的字典传递给管道中的下一步from langchain_core.runnables import RunnableParallel, RunnablePassthrough # 显式声明 RunnableParallel rag_chain ( RunnableParallel( contextretriever | format_docs, questionRunnablePassthrough() ) | prompt | model | StrOutputParser() )第一个键值对 context: retriever | format_docs 构建了一个检索与格式化的串联管道retriever 先根据用户问题从向量库中召回相关的文档列表随后 format_docs 函数将这些文档拼接成一个连续的上下文字符串为后续模型生成答案提供知识依据当|管道直接接普通函数时该函数必须满足“单参数、单返回值”的签名约束即只能接收一个参数用于接收上游的输出且必须有一个返回值作为下游的输入LangChain 会自动将其包装成RunnableLambda使其成为合法的管道节点第二个键值对 question:RunnablePassthrough() 利用 RunnablePassthrough 的透传特性不对输入做任何修改直接将用户的原始问题原封不动地传递下去确保提示词模板能够同时获得 context检索知识和 question原始提问这两个必要字段当这两路并行执行完毕后合并生成的字典包含 context 和 question会被送入提示词模板中填充占位符随后依次经过 LLM 模型生成回答最后通过输出解析器将结果格式化为字符串从而完成整个检索增强生成的准备与推理流程关于输出方式检索器由于需要一次性查询外部存储并等待全部结果返回属于同步阻塞操作因此本身不支持流式输出但后续的LLM 自回归生成阶段完全支持流式调用当使用 rag_chain.stream() 方法时最终答案会以逐块的形式实时推送给用户显著提升了交互的流畅度和响应体验