Langchain系列[11]检索器 MultiQueryRetriever

目标

了解 MultiQueryRetriever 的使用场景和用法。

检索可能会因查询用词的微小变化或在嵌入不能很好地捕获数据语义时产生不同的结果。

多查询检索器（MultiQueryRetriever）通过使用大型语言模型（LLM）来自动化提示调优过程。对于给定的用户输入查询，它生成多个查询，每个查询都从不同的角度生成。

对于每个查询，它检索一组相关的文档，并取所有查询的独特并集，以获得一组可能相关的更大文档集合。通过为同一问题生成多个视角，多查询检索器可能能够克服基于距离的检索的一些局限性，并获取更丰富的结果集。

MultiQueryRetriever 基本思想：

基于某个问题，从不同角度提出新问题，每个新问题从数据库中召回文本

关键提示词

You are an AI language model assistant. Your task is 
    to generate 3 different versions of the given user 
    question to retrieve relevant documents from a vector  database. 
    By generating multiple perspectives on the user question, 
    your goal is to help the user overcome some of the limitations 
    of distance-based similarity search. Provide these alternative 
    questions separated by newlines. Original question:

langchain: 只需要传入 chatModel 和检索器

# 千帆 bge_large_en
embeddings_model = QianfanEmbeddingsEndpoint(model="bge_large_en", endpoint="bge_large_en")

# Chroma 数据库
# 文档向量化 放入数据库
vectorstore = Chroma(persist_directory="D:\\LLM\\my_projects\\chroma_db", embedding_function=embeddings_model)

# Azure Openai
os.environ["AZURE_OPENAI_API_KEY"] = os.getenv('MY_AZURE_OPENAI_API_KEY')
os.environ["AZURE_OPENAI_ENDPOINT"] = os.getenv('MY_AZURE_OPENAI_ENDPOINT')
DEPLOYMENT_NAME_GPT3P5 = os.getenv('MY_DEPLOYMENT_NAME_GPT3P5')
chat = AzureChatOpenAI(
    openai_api_version="2023-05-15",
    azure_deployment=DEPLOYMENT_NAME_GPT3P5,
    temperature=0
)

retriever_from_llm = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(), llm=chat
)

question = "What are the approaches to Task Decomposition?"
unique_docs = retriever_from_llm.get_relevant_documents(query=question)
len(unique_docs)

![image-20240627110636492]([Vol][11]检索器 MultiQueryRetriever.assets/image-20240627110636492.png)

自己实现：MultiQueryRetriever: 单一问题扩展

先获得问题，再查找

prompt_template = """
  You are an AI language model assistant. Your task is 
      to generate 3 different versions of the given user 
      question to retrieve relevant documents from a vector  database. 
      By generating multiple perspectives on the user question, 
      your goal is to help the user overcome some of the limitations 
      of distance-based similarity search. Provide these alternative 
      questions separated by newlines. Original question: {question}
    """
    prompt = PromptTemplate.from_template(prompt_template)

    add_q_chain = (
        {"question": lambda x: x["question"]}
        | prompt
        | chat
        | LineListOutputParser()
    )
    # 获取问题
    question_list = add_q_chain.invoke({"question":question})

    # 数据库 retriever
    retriever = vectorstore.as_retriever()
    docs = []
    for question in question_list:
        tmp_doc = retriever.get_relevant_documents(question)
        if len(tmp_doc) > 0:
            docs += tmp_doc
    print('done')

自己实现：复杂问题拆分

问题：

什么是任务分解的方法？您能提供不同的任务分解方法或策略吗？有哪些方法可以用来将任务分解成更小的子任务？如何使用不同的技术或方法来执行任务分解？

prompt_template = """
        你是一个问题分析助手，如果用户的问题过于复杂，你需要将问题拆解成几个问题。
        只返回问题。
        原始问题: {question}
        """
    prompt = PromptTemplate.from_template(prompt_template)

    decompose_q_chain = (
            {"question": lambda x: x["question"]}
            | prompt
            | chat
            | LineListOutputParser()
    )

    question = """
    什么是任务分解的方法？您能提供不同的任务分解方法或策略吗？
    有哪些方法可以用来将任务分解成更小的子任务？如何使用不同的技术或方法来执行任务分解？
    """

    # 分解问题
    question_list = decompose_q_chain.invoke({"question": question})

    # 数据库 retriever
    retriever = vectorstore.as_retriever()
    docs = []
    for question in question_list:
        tmp_doc = retriever.get_relevant_documents(question)
        if len(tmp_doc) > 0:
            docs += tmp_doc
    print('done')

自动判断是否需要拆解或者新增

【1】是不是复杂问题？

【2】复杂问题→拆解

【3】单一问题→提出新问题

  # 判断问题类型 选择分支
  prompt_template = """
      你是一个问题分析助手，如果问题非常复杂，可以拆解出多个问题，返回yes,否则返回No
      原始问题: {question}
      """
  prompt = PromptTemplate.from_template(prompt_template)

  check_q_chain = (
          {"question": lambda x: x["question"]}
          | prompt
          | chat
          | StrOutputParser()
  )  

def route(info):
      if "yes" in info["need_decompose"].lower():
          return decompose_q_chain
      else:
          return add_q_chain


  full_chain = {"need_decompose": check_q_chain, "question": lambda x: x["question"]} | RunnableLambda(route)

  # 简单问题 question = '什么是任务分解的方法？'
  # 复杂问题
  question = """
  什么是任务分解的方法？您能提供不同的任务分解方法或策略吗？
  有哪些方法可以用来将任务分解成更小的子任务？如何使用不同的技术或方法来执行任务分解？
  """

  res = full_chain.invoke({"question": question})
  print(res)

完整代码

import os
from uuid import uuid4

import numpy as np
from langchain.chains.llm import LLMChain
from langchain.retrievers.multi_query import LineListOutputParser, MultiQueryRetriever
from langchain_community.chat_models.azure_openai import AzureChatOpenAI
from langchain_community.chat_models.baidu_qianfan_endpoint import QianfanChatEndpoint
from langchain_community.document_loaders.web_base import WebBaseLoader

from langchain_community.embeddings import HuggingFaceEmbeddings, QianfanEmbeddingsEndpoint
from langchain_community.retrievers import ElasticSearchBM25Retriever
from langchain_community.vectorstores.chroma import Chroma
from langchain_community.vectorstores.elasticsearch import ElasticsearchStore
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from typing import List

from langchain.chains import LLMChain
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pydantic import BaseModel, Field

# 设置一个随机ID
unique_id = uuid4().hex[0:8]

# 设置一个Log 名称
os.environ["LANGCHAIN_PROJECT"] = f" [问题类型判断] 复杂问题  - {unique_id}"

# 设置 生成Langsmith 轨迹
os.environ["LANGCHAIN_TRACING_V2"] = 'true'

# 填写你的 API KEY
os.environ["LANGCHAIN_API_KEY"] = os.getenv('MY_LANGCHAIN_API_KEY')

if __name__ == '__main__':
    # 远程百度调用
    os.environ["QIANFAN_ACCESS_KEY"] = os.getenv('MY_QIANFAN_ACCESS_KEY')
    os.environ["QIANFAN_SECRET_KEY"] = os.getenv('MY_QIANFAN_SECRET_KEY')

    # 千帆 文心一言4, 会出错
    # chat = QianfanChatEndpoint(
    #     model="ERNIE-Bot-4"
    # )

    # Azure Openai
    os.environ["AZURE_OPENAI_API_KEY"] = os.getenv('MY_AZURE_OPENAI_API_KEY')
    os.environ["AZURE_OPENAI_ENDPOINT"] = os.getenv('MY_AZURE_OPENAI_ENDPOINT')
    DEPLOYMENT_NAME_GPT3P5 = os.getenv('MY_DEPLOYMENT_NAME_GPT3P5')
    chat = AzureChatOpenAI(
        openai_api_version="2023-05-15",
        azure_deployment=DEPLOYMENT_NAME_GPT3P5,
        temperature=0
    )

    # 千帆 bge_large_en
    embeddings_model = QianfanEmbeddingsEndpoint(model="bge_large_en", endpoint="bge_large_en")

    # Chroma 数据库
    # 文档向量化 放入数据库
    vectorstore = Chroma(persist_directory="D:\\LLM\\my_projects\\chroma_db", embedding_function=embeddings_model)

    # 解析并载入url
    # loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
    # data = loader.load()
    #
    # # 分割文本
    # text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)
    # splits = text_splitter.split_documents(data)

    # 写入数据库
    # vectorstore.add_documents(splits)

    retriever_from_llm = MultiQueryRetriever.from_llm(
        retriever=vectorstore.as_retriever(), llm=chat
    )

    question = "What are the approaches to Task Decomposition?"
    unique_docs = retriever_from_llm.get_relevant_documents(query=question)
    len(unique_docs)


    # 拆解 Langchain ，自己实现这个功能
    prompt_template = """
    You are an AI language model assistant. Your task is to generate five
        different versions of the given user question to retrieve relevant documents from a vector
        database. By generating multiple perspectives on the user question, your goal is to help
        the user overcome some of the limitations of the distance-based similarity search.
        Provide these alternative questions separated by newlines.
        Original question: {question}
    """
    prompt = PromptTemplate.from_template(prompt_template)

    add_q_chain = (
        {"question": lambda x: x["question"]}
        | prompt
        | chat
        | LineListOutputParser()
    )
    # 获取问题
    question_list = add_q_chain.invoke({"question":question})

    # 数据库 retriever
    retriever = vectorstore.as_retriever()
    docs = []
    for question in question_list:
        tmp_doc = retriever.get_relevant_documents(question)
        if len(tmp_doc) > 0:
            docs += tmp_doc
    print('done')

    # 实现一个问题的 拆解
    prompt_template = """
        你是一个问题分析助手，如果用户的问题过于复杂，你需要将问题拆解成几个问题。
        只返回问题。
        原始问题: {question}
        """
    prompt = PromptTemplate.from_template(prompt_template)

    decompose_q_chain = (
            {"question": lambda x: x["question"]}
            | prompt
            | chat
            | LineListOutputParser()
    )

    question = """
    什么是任务分解的方法？您能提供不同的任务分解方法或策略吗？
    有哪些方法可以用来将任务分解成更小的子任务？如何使用不同的技术或方法来执行任务分解？
    """

    # 分解问题
    question_list = decompose_q_chain.invoke({"question": question})

    # 数据库 retriever
    retriever = vectorstore.as_retriever()
    docs = []
    for question in question_list:
        tmp_doc = retriever.get_relevant_documents(question)
        if len(tmp_doc) > 0:
            docs += tmp_doc
    print('done')


    # 判断问题类型 选择分支
    prompt_template = """
        你是一个问题分析助手，如果问题非常复杂，可以拆解出多个问题，返回yes,否则返回No
        原始问题: {question}
        """
    prompt = PromptTemplate.from_template(prompt_template)

    check_q_chain = (
            {"question": lambda x: x["question"]}
            | prompt
            | chat
            | StrOutputParser()
    )
    question = '今天星期几？'
    res = check_q_chain.invoke({"question": question})
    pass

    def route(info):
        if "yes" in info["need_decompose"].lower():
            return decompose_q_chain
        else:
            return add_q_chain


    full_chain = {"need_decompose": check_q_chain, "question": lambda x: x["question"]} | RunnableLambda(route)

    # 简单问题 question = '什么是任务分解的方法？'
    # 复杂问题
    question = """
    什么是任务分解的方法？您能提供不同的任务分解方法或策略吗？
    有哪些方法可以用来将任务分解成更小的子任务？如何使用不同的技术或方法来执行任务分解？
    """

    res = full_chain.invoke({"question": question})
    print(res)