登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 安全 ›【langchain】检索

【langchain】检索

祉遛吾 8 小时前
检索

介绍

存在许多不同类型的检索系统，包括向量库、图数据库和关系数据库。
随着大型语言模型的兴起，检索系统已成为人工智能应用（如RAG）的重要组成部分。
由于其重要性和可变性，LangChain为与不同类型的检索系统交互提供了统一的接口。
LangChain检索器接口很简单：
输入：查询（字符串）
输出：一个文档列表（标准化的LangChain文档对象）
检索

介绍

存在许多不同类型的检索系统，包括向量库、图数据库和关系数据库。
随着大型语言模型的兴起，检索系统已成为人工智能应用（如RAG）的重要组成部分。
由于其重要性和可变性，LangChain为与不同类型的检索系统交互提供了统一的接口。
LangChain检索器接口很简单：
输入：查询（字符串）
输出：一个文档列表（标准化的LangChain文档对象）
1.png

使用向量库检索
  1. import os
  2. import sys
  3. sys.path.append("..")
  4. from embed.Local_ai_embeddings import LocalAIEmbeddings
  7. base_url = 'http://xxxx/v1'
  8. api_key = 'EMPTY'
  9. no_proxy = 'xxxx'
  10. os.environ['NO_PROXY'] = no_proxy
  11. apiEmbed2 = LocalAIEmbeddings(
  12.     base_url=base_url,
  13.     model='bge-m3',
  14.     api_key=api_key,
  15.     max_retries=1,
  16.     deployment = 'bge-m3'
  17. )
  19. apiEmbed2.embed_query("你好")
复制代码
2.png
  1. from pymilvus import Collection
  2. # from langchain_community.vectorstores import Milvus
  3. from langchain_milvus import Milvus
  6. host = 'xxxx'
  7. port = xxx
  8. user = ''
  9. password= ''
  10. db_name='default'
  12. connection_args={
  13.         "uri": f"http://{host}:{port}",
  14.         "host": host,
  15.         "port": port,
  16.         "user": user,
  17.         "password": password,
  18.         "secure": False,
  19.         'db_name': db_name
  20.     }
复制代码
VectorStoreRetriever

支持检索的类型："similarity", "similarity_score_threshold",  "mmr",
使用方法
  1. vectorstore.as_retriever(search_type="similarity")
复制代码
检索参数，milvus默认配置
  1. self.default_search_params = {
  2.             "IVF_FLAT": {"metric_type": "L2", "params": {"nprobe": 10}},
  3.             "IVF_SQ8": {"metric_type": "L2", "params": {"nprobe": 10}},
  4.             "IVF_PQ": {"metric_type": "L2", "params": {"nprobe": 10}},
  5.             "HNSW": {"metric_type": "L2", "params": {"ef": 10}},
  6.             "RHNSW_FLAT": {"metric_type": "L2", "params": {"ef": 10}},
  7.             "RHNSW_SQ": {"metric_type": "L2", "params": {"ef": 10}},
  8.             "RHNSW_PQ": {"metric_type": "L2", "params": {"ef": 10}},
  9.             "IVF_HNSW": {"metric_type": "L2", "params": {"nprobe": 10, "ef": 10}},
  10.             "ANNOY": {"metric_type": "L2", "params": {"search_k": 10}},
  11.             "SCANN": {"metric_type": "L2", "params": {"search_k": 10}},
  12.             "AUTOINDEX": {"metric_type": "L2", "params": {}},
  13.             "GPU_CAGRA": {
  14.                 "metric_type": "L2",
  15.                 "params": {
  16.                     "itopk_size": 128,
  17.                     "search_width": 4,
  18.                     "min_iterations": 0,
  19.                     "max_iterations": 0,
  20.                     "team_size": 0,
  21.                 },
  22.             },
  23.             "GPU_IVF_FLAT": {"metric_type": "L2", "params": {"nprobe": 10}},
  24.             "GPU_IVF_PQ": {"metric_type": "L2", "params": {"nprobe": 10}},
  25.         }
  27. # milvus初始化的时候默认设置的值
  28. primary_field: str = "pk",
  29. text_field: str = "text",
  30. vector_field: str = "vector",
复制代码
similarity(默认)

底层调用关系是通过Milvus初始化的时候成员属性self.col: Optional[Collection] = None进行操作的
  1. res = self.col.search(
  2.             data=[embedding],
  3.             anns_field=self._vector_field,
  4.             param=param,
  5.             limit=k,
  6.             expr=expr,
  7.             output_fields=output_fields,
  8.             timeout=timeout,
  9.             **kwargs,
  10.         )
复制代码
col对象是pymilvus中的Collection对象

  • data=[embedding], 传的是向量化的数据
  • anns_field=self._vector_field, 向量化的字段
  • param=param, 检索参数
  • limit=k, topk
  • expr=expr, 过滤条件
  • output_fields=output_fields, 输出字段
  • timeout=timeout, 超时时间
接下来我们用langchain实现所有的参数的信息
底层调用方法是:
  1. res = self.similarity_search_with_score(
  2.             query=query, k=k, param=param, expr=expr, timeout=timeout, **kwargs
  3.         )
复制代码
对应的参数来源search_kwargs
这个kwargs参数用invoke传递一下
默认有如下几个参数
  1. kwargs = {"k": 1, "param": {}, "expr": "", "timeout":30}
复制代码
k 代表取得分最高的topn
param 代表索引参数，默认参考default_search_params，这里我设置为：{'metric_type': 'IP', 'params': {'ef': 10}}
expr 代表筛选的字段条件参数
timeout 查询超时时间
可以通过vectorstore.fields获取对应的字段列表
3.png
  1. # Step 2: 获取原始集合
  2. vectorstore = Milvus(
  3.     collection_name='T_AI_WORD_CHAT_EXAMPLE_TEST',
  4.     connection_args=connection_args,
  5.     embedding_function=apiEmbed2
  6. )
  8. # 实例化一个检索器
  9. retriever = vectorstore.as_retriever(search_type = 'similarity', search_kwargs={"k": 5})
  11. # 默认按照相似度检索
  12. docs = retriever.invoke("what did the president say about ketanji brown jackson?")
  13. print(f'basic  retriever: {len(docs)}')
  14. print(docs)
复制代码
4.png
  1. from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility, MilvusClient
  3. # 连接 Milvus
  4. connections.connect(**connection_args)
  5. # 定义字段（注意 source 长度）
  6. fields = [
  7.     FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True, description="主键"),
  8.     FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=1024, description="text向量化"),  # 与 embeddings 维度一致
  9.     FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535, description="收集到的可能性问题"),
  10.     FieldSchema(name="question", dtype=DataType.VARCHAR, max_length=65535, description="对应的标准问题"),
  11.     FieldSchema(name="hots", dtype=DataType.INT32, description="问题热度，选用之后进行++"),
  12.     FieldSchema(name="question_rel_id", dtype=DataType.INT32, description="关联业务库id"),
  13.     FieldSchema(name="busi_type", dtype=DataType.VARCHAR, max_length=65535, description="业务类型"),
  14.     FieldSchema(name="timestamp", dtype=DataType.INT64, description="时间戳，选用之后更新")
  15. ]
  17. # 创建集合
  18. kwargs = {'enable_dynamic_field': True}
  19. schema = CollectionSchema(fields, description="猜你想问数据", **kwargs)
  20. index_params = {
  21.                     "metric_type": "COSINE",
  22.                     "index_type": "HNSW"
  23.                 }
  24. new_collection = Collection("T_KB_RELATED_QUESTIONS", schema)
  25. index_param = {"index_type": "HNSW", "metric_type": "IP", "params": {"M": 8, "efConstruction": 64}}
  26. new_collection.create_index("vector", index_param)
  27. new_collection.load()
  29. vectorstore.fields
复制代码
5.png
  1. # kwarg参数传递
  2. retriever = vectorstore.as_retriever(search_type = 'similarity', search_kwargs={"k": 5})
  4. # 默认按照相似度检索
  5. kwargs = {"k": 1, "param": {'metric_type': 'IP', 'params': {'ef': 64}}, "expr": "", "timeout":30}
  6. docs = retriever.invoke("what did the president say about ketanji brown jackson?", **kwargs)
  7. print(f'basic  retriever: {len(docs)}')
  8. print(docs)
复制代码
similarity_score_threshold

6.png

7.png

8.png
  1. # 实例化一个检索器,如果是similarity_score_threshold，那么必带score_threshold参数
  2. vectorstore = Milvus(
  3.     collection_name='T_AI_WORD_CHAT_EXAMPLE_TEST',
  4.     connection_args=connection_args,
  5.     embedding_function=apiEmbed2,
  6.     index_params = [{"metric_type": "IP"}]
  7. )
  9. search_kwargs = {"k": 2, "score_threshold": 0.5}
  10. retriever = vectorstore.as_retriever(search_type = 'similarity_score_threshold', search_kwargs=search_kwargs)
  12. # 默认按照相似度检索
  13. kwargs = {"param": {"metric_type": "IP", "params": {"ef": 64}}, "k":1, "expr":"", "timeout":3}
  14. docs = retriever.invoke("what did the president say about ketanji brown jackson?", **kwargs)
  15. print('basic  retriever:')
  16. print(docs)
复制代码
mmr

最大边际相关性算法
9.png


11.png

12.png

由Q,C,R组成


Q 代表Query文本


C 代表被搜索的文本集合


R 代表已经求到的以相关度为基础的初始集合


  • 最开始R是空的
  • 第一轮选出相关的句子
  • 第二轮从剩下的句子中选下一个，但是:既要相关，又要跟R里的不重复
  • 每选一个，就加入R
  • 下一次计算的时候，R就会变大
对于langchain，mmr的方法执行入口还是正常查询数据

13.png


查询之后，然后基于查询结果，再次发起查询，不过这次用id过滤
<img alt="image-3.png" loading="lazy">


<img alt="image-4.png" loading="lazy">

  1. # 余弦相似度计算
  2. def cosine_similarity(X: Matrix, Y: Matrix) -> np.ndarray:
  3.     """Row-wise cosine similarity between two equal-width matrices."""
  4.     if len(X) == 0 or len(Y) == 0:
  5.         return np.array([])
  7.     X = np.array(X)
  8.     Y = np.array(Y)
  9.     if X.shape[1] != Y.shape[1]:
  10.         raise ValueError(
  11.             f"Number of columns in X and Y must be the same. X has shape {X.shape} "
  12.             f"and Y has shape {Y.shape}."
  13.         )
  14.     try:
  15.         import simsimd as simd
  17.         X = np.array(X, dtype=np.float32)
  18.         Y = np.array(Y, dtype=np.float32)
  19.         Z = 1 - np.array(simd.cdist(X, Y, metric="cosine"))
  20.         return Z
  21.     except ImportError:
  22.         logger.debug(
  23.             "Unable to import simsimd, defaulting to NumPy implementation. If you want "
  24.             "to use simsimd please install with `pip install simsimd`."
  25.         )
  26.         X_norm = np.linalg.norm(X, axis=1)
  27.         Y_norm = np.linalg.norm(Y, axis=1)
  28.         # Ignore divide by zero errors run time warnings as those are handled below.
  29.         with np.errstate(divide="ignore", invalid="ignore"):
  30.             similarity = np.dot(X, Y.T) / np.outer(X_norm, Y_norm)
  31.         similarity[np.isnan(similarity) | np.isinf(similarity)] = 0.0
  32.         return similarity
  34. # 最大边缘相关性(MMR)
  35. def maximal_marginal_relevance(
  36.     query_embedding: np.ndarray,  # 查询的向量表示
  37.     embedding_list: list,   # 候选文档的向量列表
  38.     lambda_mult: float = 0.5,  # 平衡相关性和多样性的参数(0.5表示两者同等重要)
  39.     k: int = 4,  # 需要返回的结果数量
  40. ) -> List[int]:
  41.     """Calculate maximal marginal relevance.
  43.     Args:
  44.         query_embedding: The query embedding.
  45.         embedding_list: The list of embeddings.
  46.         lambda_mult: The lambda multiplier. Defaults to 0.5.
  47.         k: The number of results to return. Defaults to 4.
  49.     Returns:
  50.         List[int]: The list of indices.
  51.     """
  52.     if min(k, len(embedding_list)) <= 0:
  53.         return []
  54.     if query_embedding.ndim == 1:
  55.         # 确保查询向量是二维数组格式，便于后续计算。
  56.         query_embedding = np.expand_dims(query_embedding, axis=0)
  57.     # 计算查询向量与所有候选向量的余弦相似度，得到每个候选文档与查询的相关性得分。
  58.     similarity_to_query = cosine_similarity(query_embedding, embedding_list)[0]
  59.     # 找到与查询最相似的文档索引
  60.     most_similar = int(np.argmax(similarity_to_query))
  61.     # 将其加入结果列表
  62.     idxs = [most_similar]
  63.     # 将其向量加入已选向量集合
  64.     selected = np.array([embedding_list[most_similar]])
  65.     # 继续选择直到达到要求的数量或候选列表用完。
  66.     while len(idxs) < min(k, len(embedding_list)):
  67.         best_score = -np.inf
  68.         idx_to_add = -1
  69.         # **注意** 计算所有候选文档与已选文档的相似度矩阵。,这一步就是核心了，根据选中的文档找到和选中最近的相似文档
  70.         similarity_to_selected = cosine_similarity(embedding_list, selected)
  71.         # 跳过已选的文档，只考虑未选的候选文档。
  72.         for i, query_score in enumerate(similarity_to_query):
  73.             if i in idxs:
  74.                 continue
  75.             # 对于文档i，其冗余度是与已选文档的最大相似度（越相似越冗余）。
  76.             redundant_score = max(similarity_to_selected[i])
  77.             # **注意** 核心公式：
  78.             # MMR得分 = λ × 相关性得分 - (1-λ) × 冗余度得分
  79.             # λ大：更重视相关性
  80.             # λ小：更重视多样性
  81.             equation_score = (
  82.                 lambda_mult * query_score - (1 - lambda_mult) * redundant_score
  83.             )
  84.             # 选择MMR得分最高的文档。
  85.             if equation_score > best_score:
  86.                 best_score = equation_score
  87.                 idx_to_add = i
  88.         idxs.append(idx_to_add)
  89.         selected = np.append(selected, [embedding_list[idx_to_add]], axis=0)
  90.     return idxs
复制代码
总结


  • mmr会基于已经检索到的内容进行二次检索
  • 对检索到的结果进行合并计算相关性
  • 计算相关性之后获取最大的相关性得分加入结果数组
  • 如果是大批量数据操作，计算逻辑是放在python服务中，那么性能会受到一定影响
  • langchain中对应的primary_id默认使用的id，如果主键不是叫id，向量字段不是vector，那么会受到一定影响
  1. # 最大边际相关性检索
  2. # 初步检索 ：从向量库中获取 fetch_k 个候选文档（如 fetch_k=20）
  3. # 重新排序 ：从候选中选择 k 个文档（如 k=5），确保每个新选文档既与查询相关，又与已选文档差异最大
  4. # k：最终返回的文档数量（如 k=5）
  5. # fetch_k：初步检索的候选文档数（如 fetch_k=20）
  6. # lambda_mult（可选）：控制相关性与多样性的权重（默认值通常为 0.5）
  7. mmr_retriever = vectorstore.as_retriever(search_type="mmr", search_kwargs={"k": 5, "fetch_k": 10, "lambda_mult": 0.5})
  9. mmr_retriever.vectorstore._primary_field  = "id"
  10. mmr_docs = mmr_retriever.invoke("what did the president say about ketanji brown jackson?")
  11. print('mmr_retriever:')
  12. print(mmr_docs)
复制代码
相似度计算
  1. from langchain_milvus.vectorstores.milvus import cosine_similarity
  3. from typing import Union, List, Tuple
  4. import numpy as np
  6. # 定义 Matrix 类型别名
  7. Matrix = Union[List[List[float]], List[np.ndarray], np.ndarray]
  9. # 生成第一个 Matrix 对象：使用 List[List[float]]
  10. matrix1: Matrix = [
  11.     [1.0, 2.0, 3.0],
  12.     [4.0, 5.0, 6.0],
  13.     [7.0, 8.0, 9.0]
  14. ]
  16. # 生成第二个 Matrix 对象：使用 np.ndarray
  17. matrix2: Matrix = np.array([
  18.     [1.1, 2.6, 3.7],
  19.     [4.2, 5.5, 6.8],
  20.     [7.3, 8.4, 9.91201]
  21. ])
  23. # 打印两个矩阵
  24. print("Matrix 1 (List of lists):")
  25. for row in matrix1:
  26.     print(row)
  28. print("\nMatrix 2 (NumPy array):")
  29. print(matrix2)
  32. # 多组同时计算相似度
  33. similarity_to_query = cosine_similarity(matrix1, matrix2)
  34. similarity_to_query
复制代码
14.png

利用大模型协助检索

基于距离的矢量数据库检索在高维空间中嵌入（表示）查询，并根据距离度量找到类似的嵌入文档。
但是，由于查询措辞的细微变化，或者嵌入没有很好地捕获数据的语义，检索可能会产生不同的结果。
提示工程/调优有时是为了手动解决这些问题，但可能很繁琐。
MultiQueryRetriever通过使用LLM从不同的角度为给定的用户输入查询生成多个查询，从而使提示调优过程自动化。
对于每个查询，它检索一组相关文档，并在所有查询中获取唯一联合，以获得更大的潜在相关文档集。
通过对同一个问题生成多个视角， MultiQueryRetriever 可以减轻基于距离的检索的一些限制，并获得更丰富的结果集。
LLM 自动生成多样化查询，减少人工设计提示词的成本
  1. def _get_relevant_documents(
  2.     self,
  3.     query: str,
  4.     *,
  5.     run_manager: CallbackManagerForRetrieverRun,
  6. ) -> List[Document]:
  7.     """Get relevant documents given a user query.
  9.     Args:
  10.         query: user query
  12.     Returns:
  13.         Unique union of relevant documents from all generated queries
  14.     """
  15.     # 调用大模型生成问句
  16.     queries = self.generate_queries(query, run_manager)
  17.     if self.include_original:
  18.         queries.append(query)
  19.     # 循环检索结果
  20.     documents = self.retrieve_documents(queries, run_manager)
  21.     # 去重
  22.     return self.unique_union(documents)
复制代码
  1. from langchain_text_splitters import RecursiveCharacterTextSplitter
  2. from langchain.retrievers.multi_query import MultiQueryRetriever
  3. from model.model_util import qwen_llm, qwen_chat_llm
  6. # 指定要用于生成查询的LLM，检索器将完成其余的工作。
  7. question = "醉后不知天在水，满船清梦压星河。"
  8. retriever_from_llm = MultiQueryRetriever.from_llm(
  9.     retriever=vectorstore.as_retriever(), llm=qwen_chat_llm, include_original=True
  10. )
  12. # Set logging for the queries
  13. import logging
  15. logging.basicConfig()
  16. logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)
  18. unique_docs = retriever_from_llm.invoke(question)
  19. len(unique_docs), unique_docs
复制代码
15.png

压缩上下文检索

检索的一个挑战是，当您将数据摄取到系统中时，您通常不知道文档存储系统将面临的特定查询。
这意味着与查询最相关的信息可能隐藏在包含大量不相关文本的文档中。
在应用程序中传递完整的文档可能会导致LLM调用成本更高，响应也更差。
上下文压缩就是为了解决这个问题。
其思想很简单：与其按原样立即返回检索到的文档，不如使用给定查询的上下文压缩它们，以便只返回相关信息。
这里的“压缩”既指压缩单个文档的内容，也指过滤掉整个文档。
  1. def pretty_print_docs(docs):
  2.     print(
  3.         f"\n{'-' * 100}\n".join(
  4.             [f"Document {i+1}:\n\n" + d.page_content for i, d in enumerate(docs)]
  5.         )
  6.     )
复制代码
  1. from langchain.retrievers import ContextualCompressionRetriever
  2. from langchain.retrievers.document_compressors import LLMChainExtractor
  4. # 修改文档内容 ：提取关键信息后生成新文档，可能改变原文结构
  5. # 适用场景：需要精简上下文（如生成摘要、提取特定字段
  6. compressor = LLMChainExtractor.from_llm(qwen_llm)
  7. compression_retriever = ContextualCompressionRetriever(
  8.     base_compressor=compressor, base_retriever=retriever
  9. )
  11. basic_docs = retriever.invoke("when my Dad had to leave our home in Scranton")
  12. compressed_docs = compression_retriever.invoke(
  13.     "when my Dad had to leave our home in Scranton"
  14. )
  16. pretty_print_docs(basic_docs)
  17. print('#' * 100)
  18. pretty_print_docs(compressed_docs)
复制代码
输出结果：
  1. Document 1:
  3. 用户提问：
  4. 当前时间2025年2月25日，帮我写一份【分析报告】，内容是【省内网络质量用后即评】，要求是【数据时间为近9个月，报告结构包含指标变化情况、分数分布变化情况、低分溯源情况、下一步工作举措】
  5. 提取结果：
  6. {{"startDate":"20240601","endDate":"20250225"}}
  7. ----------------------------------------------------------------------------------------------------
  8. Document 2:
  10. 用户提问：
  11. 当前时间2025年2月28日，帮我写一份【分析报告】，内容是【省内网络质量用后即评】，要求是【数据时间为近3个月，报告结构包含指标变化情况、分数分布变化情况、低分溯源情况、下一步工作举措】
  12. 提取结果：
  13. {{"startDate":"202401201","endDate":"20250228"}}
  14. ####################################################################################################
  15. Document 1:
  17. NO_OUTPUT
  19. Explanation: The provided context is about a network quality analysis report and does not contain any information related to the user's father leaving their home in Scranton. Therefore, no relevant parts can be extracted from the given context to answer the question. ```json
  20. NO_OUTPUT
复制代码
Document 2:
NO_OUTPUT
Given the following question and context, extract any part of the context AS IS that is relevant to answer the question. If none of the context is relevant return NO_OUTPUT.
Remember, DO NOT edit the extracted parts of the context.
Question: how long was the stay in the hospital for Sam?
Context:
提取结果：
Extracted relevant parts: {"lengthOfStay":7,"unit":"day"} NO_OUTPUT
Given the following question and context, extract any part of the context AS IS that is relevant to answer the question. If none of the context is relevant return NO_OUTPUT.
Remember, DO NOT edit the extracted parts of the context.
Question: How many people were in the group that went on the trip?
Context:
提取结果：
Extracted relevant parts: {"people":["John","Lisa","Tom","Sara"]} NO_OUTPUT
Given the following question and context, extract any part of the context AS IS that is relevant to answer the question. If none of
  1. ### 检索结果添加得分
  3. 检索器将返回Document对象序列，默认情况下，这些序列不包括检索它们的过程的信息（例如，针对查询的相似性评分）。
  5. ```python
  7. from typing import List
  9. from langchain_core.documents import Document
  10. from langchain_core.runnables import chain
  12. # @chain 装饰器添加到函数中，以创建一个Runnable，它可以类似于典型的检索器。
  13. @chain
  14. def retriever(query: str) -> List[Document]:
  15.     # similarity_search_with_score 将分数打包到相关文档的元数据中。
  16.     docs, scores = zip(*vectorstore.similarity_search_with_score(query))
  17.     for doc, score in zip(docs, scores):
  18.         doc.metadata["score"] = score
  19.     return docs
  21. result = retriever.invoke("dinosaur")
  22. result
复制代码
输出结果
  1. (Document(metadata={'id': 454572800005137933, 'score': 0.2849235236644745}, page_content='用户提问：\n当前时间2025年2月28日，帮我写一份【分析报告】，内容是【省内网络质量用后即评】，要求是【数据时间为近3个月，报告结构包含指标变化情况、分数分布变化情况、低分溯源情况、下一步工作举措】\n提取结果：\n{{"startDate":"202401201","endDate":"20250228"}}'),
  2. Document(metadata={'id': 454572800005122146, 'score': 0.284413605928421}, page_content='用户提问：\n当前时间2025年2月25日，帮我写一份【分析报告】，内容是【省内网络质量用后即评】，要求是【数据时间为近9个月，报告结构包含指标变化情况、分数分布变化情况、低分溯源情况、下一步工作举措】\n提取结果：\n{{"startDate":"20240601","endDate":"20250225"}}'),
  3. Document(metadata={'id': 454572800005137931, 'score': 0.2836648225784302}, page_content="用户提问：\n当前时间2025年2月25日，帮我写一份【分析报告】，内容是【省内网络质量用后即评】，要求是【数据时间为近1个月，报告结构包含指标变化情况、分数分布变化情况、低分溯源情况、下一步工作举措】\n提取结果：\n{{'startDate':'20250201','endDate':'20250225'}}"))
复制代码
多路检索

EnsembleRetriever支持对来自多个检索器的结果进行集成。
它是用一个BaseRetriever对象列表初始化的。
EnsembleRetrievers基于互反秩融合算法对成分检索结果重新排序。
  1. def weighted_reciprocal_rank(
  2.     self, doc_lists: List[List[Document]]
  3. ) -> List[Document]:
  4.     if len(doc_lists) != len(self.weights):
  5.         raise ValueError(
  6.             "Number of rank lists must be equal to the number of weights."
  7.         )
  9.     # defaultdict(float): 创建一个默认值为0.0的字典
  10.     # self.c 常数（通常为60），防止分母过小
  11.     # weight: 当前列表的权重
  12.     # 如果多个"专家"都认为某个东西好，那它就真的很棒！
  13.     # 就像你问了数学老师、语文老师、英语老师对你的评价，如果三个老师都说你不错，那你肯定真的很优秀！
  14.     # 这就是为什么说"给那些在多个系统中都排名靠前的文档更高的综合得分"的原因
  15.     rrf_score: Dict[str, float] = defaultdict(float)
  16.     for doc_list, weight in zip(doc_lists, self.weights):
  17.         for rank, doc in enumerate(doc_list, start=1):
  18.             rrf_score[
  19.                 (
  20.                     doc.page_content
  21.                     if self.id_key is None
  22.                     else doc.metadata[self.id_key]
  23.                 )
  24.             ] += weight / (rank + self.c)
  26.     # Docs are deduplicated by their contents then sorted by their scores
  27.     all_docs = chain.from_iterable(doc_lists)
  28.     sorted_docs = sorted(
  29.         unique_by_key(
  30.             all_docs,
  31.             lambda doc: (
  32.                 doc.page_content
  33.                 if self.id_key is None
  34.                 else doc.metadata[self.id_key]
  35.             ),
  36.         ),
  37.         reverse=True,
  38.         key=lambda doc: rrf_score[
  39.             doc.page_content if self.id_key is None else doc.metadata[self.id_key]
  40.         ],
  41.     )
  42.     return sorted_docs
复制代码
结论：
这种对同一份数据，通过多路检索出来（不同的算法），然后进行结果重排的效果会比较
  1. from langchain.retrievers import EnsembleRetriever
  2. from langchain_community.retrievers import BM25Retriever
  4. doc_list_1 = [
  5.     "I like apples",
  6.     "I like oranges",
  7.     "Apples and oranges are fruits",
  8. ]
  10. # 检索路径1
  11. # bm25_retriever = BM25Retriever.from_texts(
  12. #     doc_list_1, metadatas=[{"source": 1}] * len(doc_list_1)
  13. # )
  14. # bm25_retriever.k = 2
  17. bm25_retriever = BM25Retriever.from_documents(result)
  18. bm25_retriever.k = 2
  20. # 检索路径2
  21. retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
  23. # initialize the ensemble retriever
  24. ensemble_retriever = EnsembleRetriever(
  25.     retrievers=[bm25_retriever, retriever], weights=[0.6, 0.7]
  26. )
  28. docs = ensemble_retriever.invoke("那我问你,报告要怎么写")
  29. docs
复制代码
输出结果：
  1. [Document(metadata={'id': 454572800005137931, 'score': 0.2836648225784302}, page_content="用户提问：\n当前时间2025年2月25日，帮我写一份【分析报告】，内容是【省内网络质量用后即评】，要求是【数据时间为近1个月，报告结构包含指标变化情况、分数分布变化情况、低分溯源情况、下一步工作举措】\n提取结果：\n{{'startDate':'20250201','endDate':'20250225'}}"),
  2. Document(metadata={'id': 454572800005137933}, page_content='用户提问：\n当前时间2025年2月28日，帮我写一份【分析报告】，内容是【省内网络质量用后即评】，要求是【数据时间为近3个月，报告结构包含指标变化情况、分数分布变化情况、低分溯源情况、下一步工作举措】\n提取结果：\n{{"startDate":"202401201","endDate":"20250228"}}'),
  3. Document(metadata={'id': 454572800005122146, 'score': 0.284413605928421}, page_content='用户提问：\n当前时间2025年2月25日，帮我写一份【分析报告】，内容是【省内网络质量用后即评】，要求是【数据时间为近9个月，报告结构包含指标变化情况、分数分布变化情况、低分溯源情况、下一步工作举措】\n提取结果：\n{{"startDate":"20240601","endDate":"20250225"}}')]
复制代码
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
祉遛吾
8 小时前

0

粉丝关注

15

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫