如何在查询分析中处理多个检索器

有时，查询分析技术允许选择要使用的检索器。要使用此功能，您需要添加一些逻辑来选择要执行的检索器。我们将展示一个简单的示例（使用模拟数据）来说明如何做到这一点。

设置

安装依赖

%pip install -qU langchain langchain-community langchain-openai langchain-chroma

Note: you may need to restart the kernel to use updated packages.

设置环境变量

在此示例中，我们将使用 OpenAI：

import getpass
import os

if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass.getpass()

# Optional, uncomment to trace runs with LangSmith. Sign up here: https://smith.langchain.com.
# os.environ["LANGSMITH_TRACING"] = "true"
# os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

创建索引

我们将围绕虚假信息创建一个向量存储。

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

texts = ["Harrison worked at Kensho"]
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_texts(texts, embeddings, collection_name="harrison")
retriever_harrison = vectorstore.as_retriever(search_kwargs={"k": 1})

texts = ["Ankush worked at Facebook"]
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_texts(texts, embeddings, collection_name="ankush")
retriever_ankush = vectorstore.as_retriever(search_kwargs={"k": 1})

API Reference:Chroma | OpenAIEmbeddings | RecursiveCharacterTextSplitter

查询分析

我们将使用函数调用来构建输出。我们将允许它返回多个查询。

from typing import List, Optional

from pydantic import BaseModel, Field


class Search(BaseModel):
    """Search for information about a person."""

    query: str = Field(
        ...,
        description="Query to look up",
    )
    person: str = Field(
        ...,
        description="Person to look things up for. Should be `HARRISON` or `ANKUSH`.",
    )

from langchain_core.output_parsers.openai_tools import PydanticToolsParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI

output_parser = PydanticToolsParser(tools=[Search])

system = """You have the ability to issue search queries to get information to help answer user information."""
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "{question}"),
    ]
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(Search)
query_analyzer = {"question": RunnablePassthrough()} | prompt | structured_llm

API Reference:PydanticToolsParser | ChatPromptTemplate | RunnablePassthrough | ChatOpenAI

我们可以看到，这允许在检索器之间进行路由

query_analyzer.invoke("where did Harrison Work")

Search(query='work history', person='HARRISON')

query_analyzer.invoke("where did ankush Work")

Search(query='work history', person='ANKUSH')

通过查询分析进行检索

那么，我们该如何将它包含在链中呢？我们只需要一些简单的逻辑来选择检索器，并将搜索查询传递进去。

from langchain_core.runnables import chain

API Reference:chain

retrievers = {
    "HARRISON": retriever_harrison,
    "ANKUSH": retriever_ankush,
}

@chain
def custom_chain(question):
    response = query_analyzer.invoke(question)
    retriever = retrievers[response.person]
    return retriever.invoke(response.query)

custom_chain.invoke("where did Harrison Work")

[Document(page_content='Harrison worked at Kensho')]

custom_chain.invoke("where did ankush Work")

[Document(page_content='Ankush worked at Facebook')]