【AI】十二.Retrievers检索器+RAG文档助手项目实战

小滴课堂讲师 2025年09月19日 ai大模型, aigc 预计阅读 13 分钟

#### LangChain检索器Retrievers案例实战

##### 什么是Retriever

* **统一接口**：标准化检索流程，无论数据来源如何，最终输出`Document`对象列表。

* **多源混合检索**：支持同时查询向量库、传统数据库和搜索引擎【提高召回率】

* **与VectorStore的关系**：Retriever不直接管理存储，依赖VectorStore（如FAISS、Chroma）实现向量化与检索。

* **RAG中的角色**：作为检索增强生成（RAG）流程的“数据入口”，为生成模型提供精准上下文

* 有多个实现：VectorStoreRetriever、MultiQueryRetriever、SelfQueryRetriever等

* 特点

* **模块化设计**：支持插件式扩展，可自定义检索算法（如混合搜索、重排序）。
  * **异步支持**：通过`async_get_relevant_documents`实现高并发场景下的高效检索。
  * **链式调用**：可与LangChain的其他组件（如Text Splitters、Memory）无缝集成。

```
  # from langchain_core.retrievers import BaseRetriever
  ```

* 补充知识点：召回率（Recall）信息检索和机器学习中衡量模型找全相关结果能力的核心指标

* 比如
    * 在文档检索中，如果有100篇相关文档，系统找出了80篇，那么召回率就是80%。
    * 召回率高意味着系统漏掉的少，但可能夹杂了不相关的结果，这时候准确率可能低。

##### Retriever常见类型之基础检索器 VectorStoreRetriever

* 基础使用参考案例

```python
#将文档嵌入为向量，通过相似度计算（如余弦相似度）检索
from langchain_community.vectorstores import FAISS
retriever = FAISS.from_documents(docs, embeddings).as_retriever(
    search_type="mmr",  # 最大边际相关性
    search_kwargs={"k": 5, "filter": {"category": "news"}}
)
```

* `as_retriever()` 方法介绍

* 将向量库实例转换为检索器对象，实现与 LangChain 链式调用（如 `RetrievalQA`）的无缝对接。

* 源码

```python
    def as_retriever(self, **kwargs: Any) -> VectorStoreRetriever:
            tags = kwargs.pop("tags", None) or [] + self._get_retriever_tags()
            return VectorStoreRetriever(vectorstore=self, tags=tags, **kwargs)
            """
            向量库实例转换为检索器对象，实现与 LangChain 链式调用
            """
    ```

* 关键参数详解

* search_type 搜索类型

```python
    # MMR 检索配置示例
    mmr_retriever = vector_store.as_retriever(
        search_type="mmr",
        search_kwargs={
            "k": 3,
            "fetch_k": 20,
            "lambda_mult": 0.5
        }
    )
    ```

* search_kwargs 参数详解

|      参数       |   类型   | 默认值 |        说明         |
    | :-------------: | :------: | :----: | :-----------------: |
    |       `k`       |   int    |   4    |    返回结果数量     |
    | `filter`/`expr` | str/dict |  None  |   元数据过滤条件    |
    |     `param`     |   dict   |  None  | Milvus 搜索参数配置 |

##### 综合案例实战

* 默认是similarity search，需要安装依赖包: `pip install langchain-milvus `

````python
  #!/usr/bin/env python3
  # -*- coding: utf-8 -*-
  from langchain_community.embeddings import DashScopeEmbeddings
  from langchain_milvus import Milvus
  from langchain_core.documents import Document
  
  import secret
  
  # 初始化模型
  embeddings = DashScopeEmbeddings(
      model="text-embedding-v4",  # 第二代通用模型
      max_retries=3,
      dashscope_api_key=secret.api_key,
  )
  
  document_1 = Document(
      page_content="LangChain支持多种数据库集成,小滴课堂的AI大课",
      metadata={"source": "xdclass.net/doc1"},
  )
  document_2 = Document(
      page_content="Milvus擅长处理向量搜索,老王的课程不错",
      metadata={"source": "xdclass.net/doc2"},
  )
  document_3 = Document(
      page_content="我要去学小滴课堂的架构大课",
      metadata={"source": "xdclass.net/doc3"},
  )
  document_4 = Document(
      page_content="今天天气不错，老王和老帆去按摩了",
      metadata={"source": "xdclass.net/doc4"},
  )
  documents = [document_1, document_2, document_3, document_4]
  vector_store = Milvus.from_documents(
      documents=documents,
      embedding=embeddings,
      collection_name="retriever_test1",
      connection_args={"uri": "http://124.25.25.26:19530", "user": "root", "password": "123456","db_name": "my_database"}
  )
  
  # 默认是 similarity search
  retriever = vector_store.as_retriever(search_kwargs={"k": 2})
  
  results = retriever.invoke("如何进行数据库操作")
  
  for result in results:
      print(f"内容 {result.page_content} 元数据 {result.metadata}")
  
  ````

* 可以调整为MMR检索

```
  retriever = vector_store.as_retriever(search_type="mmr",search_kwargs={"k": 2})
  ```

#### 大厂面试题-如何提升大模型召回率和实战

LLM大模型开发高频面试题：如何提升大模型召回率和准确率?

##### 需求背景

* 当原始查询不够明确时，或者当文档库中的内容使用不同的术语表达同一概念时
* 单一查询可能无法有效检案到所有相关内容; 
* 或者用户的问题可能有不同的表达方式，导致的检索结果不理想，
* 需要从多个角度切入才能找到最相关的文档片段。这种情况下，生成多个变体查询可以提高召回率，确保覆盖更多相关文档。

##### MultiQueryRetriever

* 通过生成多个相关查询来增强检索效果，解决单一查询可能不够全面或存在歧义的问题。

* 原理：

* **查询扩展技术**：通过LLM生成N个相关查询（如改写、扩展、翻译），合并结果去重，生成多个变体查询
  * **双重增强效果**：提升召回率（+25%↑）和准确率（+18%↑）的平衡

![1](D:/IdeaProjects/learn/ycloud-aipan/img/1-2885071.png)

* 用法

```python
  retriever = MultiQueryRetriever.from_llm(
      retriever=base_retriever,
      llm=ChatOpenAI()
  )
  ```

**典型问题场景**

- **术语差异问题**：用户提问使用"SSL证书" vs 文档中使用"TLS证书"
  - **表述模糊问题**："怎么备份数据库" vs "数据库容灾方案实施步骤"
  - **多语言混合**：中英文混杂查询（常见于技术文档检索）
  - **专业领域知识**：医学问诊中的症状不同描述方式

##### 案例实战

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_milvus import Milvus
from langchain_openai import ChatOpenAI

from langchain_community.document_loaders import TextLoader
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_text_splitters import RecursiveCharacterTextSplitter

import logging
import os
import secret

os.environ["USER_AGENT"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36 Edg/140.0.0.0"

# 设置日志记录的基本配置
logging.basicConfig()
# 设置多查询检索器的日志记录级别为INFO
logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)

# 使用TextLoader加载文本数据
loader = TextLoader("../xd-rag/data/qa.txt", encoding="utf-8")
# 加载数据到变量中
data = loader.load()

# 初始化文本分割器，将文本分割成小块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=10)
# 执行文本分割
splits = text_splitter.split_documents(data)

# 初始化模型
embedding = DashScopeEmbeddings(
    model="text-embedding-v2",  # 第二代通用模型
    max_retries=3,
    dashscope_api_key=secret.api_key,
)

# 初始化向量数据库
vector_store = Milvus.from_documents(
    documents=splits,
    embedding=embedding,
    collection_name="mulit_retriever2",
    connection_args={"uri": "http://12.45.42.25:19530", "user": "root", "password": "123456","db_name": "my_database"}
)

# 定义问题
question = "老王不知道为啥抽筋了"
# 初始化语言模型
llm = ChatOpenAI(
    model_name = "qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=secret.api_key,
    temperature=0.7
)

# 从语言模型中创建多查询检索器
retriever_from_llm = MultiQueryRetriever.from_llm(
    retriever=vector_store.as_retriever(), llm=llm
)

# 使用检索器执行问题检索
results = retriever_from_llm.invoke(question)
# 打印检索到的结果数量
len(results)

# 遍历并打印每个检索结果的内容和元数据
for result in results:
    print(f"内容 {result.page_content} 元数据 {result.metadata}")

# 以下是未使用的代码片段，已将其注释掉
# vector_store = Milvus(
#     embeddings,
#     connection_args={"uri": "http://47.119.128.20:19530"},
#     collection_name="mmr_test",
# )
# print(vector_store)
```

#### RAG综合项目实战-AI文档问答助手

##### 需求：在线文档的问答助手，方便查找相关手册和接口API

* 主要功能包括
  * 文档加载与切分、向量嵌入生成、向量存储与检索。
  * 基于检索增强生成（Retrieval-Augmented Generation, RAG）的问答。
* 技术选型：LangChain框架+Milvus向量数据库

##### 实现的功能

* 文档加载与切分
  * 使用`WebBaseLoader`从指定URL加载文档。
  * 使用`RecursiveCharacterTextSplitter`将加载的文档按照指定的块大小和重叠大小进行切分。

* 向量嵌入生成
  - 使用`DashScopeEmbeddings`生成文档切片的向量嵌入，模型为`text-embedding-v2`，支持最大重试次数为3次。

* 向量存储与检索

- 使用`Milvus`作为向量数据库，创建名为`doc_qa_db`的Collection。

- 将生成的向量嵌入存储到Milvus中，并支持相似性检索。

* 基于RAG的问答

- 初始化`ChatOpenAI`模型，使用`qwen-plus`作为LLM模型。

- 定义`PromptTemplate`，用于构建输入给LLM的提示信息。

- 构建RAG链，结合相似性检索结果和LLM生成回答。

* 编码实战

```python
  from langchain_community.document_loaders import WebBaseLoader
  from langchain.text_splitter import RecursiveCharacterTextSplitter
  from langchain_milvus import Milvus
  
  from langchain.schema.runnable import RunnablePassthrough
  from langchain.prompts import PromptTemplate
  from langchain_community.embeddings import DashScopeEmbeddings
  from langchain_openai import ChatOpenAI
  import os
  
  import secret
  
  os.environ["USER_AGENT"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36 Edg/140.0.0.0"
  
  # 设置Milvus Collection名称。
  COLLECTION_NAME = 'doc_qa_db'
  
  # 初始化WebBaseLoader加载指定URL的文档。
  loader = WebBaseLoader([
  'https://milvus.io/docs/overview.md',
  'https://milvus.io/docs/release_notes.md'
  ])
  
  # 加载文档。
  docs = loader.load()
  
  # 初始化RecursiveCharacterTextSplitter，用于切分文档。
  text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=0)
  
  # 使用LangChain将输入文档按照chunk_size切分。
  all_splits = text_splitter.split_documents(docs)
  
  # 初始化DashScopeEmbeddings，设置embedding模型为DashScope的text-embedding-v2。
  embeddings = DashScopeEmbeddings(
      model="text-embedding-v2",  # 第二代通用模型
      max_retries=3,
      dashscope_api_key=secret.api_key,
  )
  
  # 创建connection，为阿里云Milvus的访问域名。
  connection_args = {"uri": "http://127.0.0.1:19530", "user": "root", "password": "123456", "db_name": "my_database"}
  # 创建Collection。
  vector_store = Milvus(
      embedding_function=embeddings,
      connection_args=connection_args,
      collection_name=COLLECTION_NAME,
      drop_old=True,
  ).from_documents(
      all_splits,
      embedding=embeddings,
      collection_name=COLLECTION_NAME,
      connection_args=connection_args,
  )
  # 利用Milvus向量数据库进行相似性检索。
  query = "What are the main components of Milvus?"
  docs = vector_store.similarity_search(query)
  print(len(docs))
  ```

* 编码测试实战

```python
  from langchain_milvus import Milvus
  from langchain.schema.runnable import RunnablePassthrough
  from langchain.prompts import PromptTemplate
  from langchain_community.embeddings import DashScopeEmbeddings
  from langchain_openai import ChatOpenAI
  import os
  import secret
  
  os.environ["USER_AGENT"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36 Edg/140.0.0.0"
  # 设置Milvus Collection名称。
  COLLECTION_NAME = 'doc_qa_db'
  # 初始化DashScopeEmbeddings，设置embedding模型为DashScope的text-embedding-v2。
  embeddings = DashScopeEmbeddings(
      model="text-embedding-v2",  # 第二代通用模型
      max_retries=3,
      dashscope_api_key=secret.api_key,
  )
  # 创建connection，为阿里云Milvus的访问域名。
  connection_args = {"uri": "http://127.0.0.1:19530", "user": "root", "password": "123456", "db_name": "my_database"}
  vector_store = Milvus(
      embeddings,
      connection_args=connection_args,
      collection_name=COLLECTION_NAME,
  )
  # 初始化ChatOpenAI模型。
  llm = ChatOpenAI(
      model_name="qwen-plus",
      base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
      api_key=secret.api_key,
      temperature=0.7
  )
  # 将上述相似性检索的结果作为retriever，提出问题输入到LLM之后，获取检索增强之后的回答。
  retriever = vector_store.as_retriever()
  print(retriever)
  
  # 定义PromptTemplate，用于构建输入给LLM的prompt。
  template = """你是AI文档助手，使用以下上下文来回答最后的问题。
  如果你不知道答案，就说你不知道，不要试图编造答案。
  最多使用10句话，并尽可能简洁地回答。总是在答案的末尾说“谢谢你的提问！”.
  {context}
  问题: {question}
  """
  
  rag_prompt = PromptTemplate.from_template(template)
  
  # 构建Retrieval-Augmented Generation链。
  rag_chain = (
          {"context": retriever, "question": RunnablePassthrough()}
          | rag_prompt
          | llm
  )
  
  # 调用rag_chain回答问题。
  print(rag_chain.invoke("Milvus的版本"))
  ```

* 应用场景

* 教育领域：可用于备课笔记、课程内容总结等场景。
  * 企业知识库：帮助企业快速构建基于内部文档的知识问答系统。
  * 技术支持：提供技术文档的智能检索与问答服务。

* 扩展方向

* 支持更多类型的文档加载器（如PDF、Word等）。
  * 增加多语言支持。
  * 优化向量嵌入生成与检索效率

* 大家的疑惑点（下一章讲解）

```python
  # 构建Retrieval-Augmented Generation链。
  rag_chain = (
      {"context": retriever, "question": RunnablePassthrough()}
      | rag_prompt
      | llm
  )
  ```

* 在 `rag_chain` 的定义中，`{"context": retriever, "question": RunnablePassthrough()}` 创建了一个输入字典。
  * `context` 的值来自 `retriever`，它将使用向量存储检索与问题相关的文档片段。
  * `question`键的值通过 `RunnablePassthrough()` 直接传递，用户输入的问题会被透传到后续的处理步骤。
  * 输入字典随后会被传递给 `rag_prompt`，构建最终的提示（prompt）被传递给 `llm`（语言模型），生成最终的回答
  * 总结：
    * 用户输入的问题会同时传给`retriever`和`RunnablePassthrough()`
    * `retriever`完成检索后，会自动把结果赋值给`context`。
    * 检索结果`context`和用户输入`question`一并传给提示模板`prompt_template`。

###

评论区