LangChain中检索器动态路由的参数适配问题求助

以下是三种可行的解决方案，均能解决输入字典与Retriever期望查询字符串的适配问题：

方案一：用RunnableLambda拆分输入并路由检索器 #

通过RunnableLambda提取输入字典中的question和product_id，根据产品ID调用对应检索器，再将结果合并回输入字典供后续组件使用。

from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain.vectorstores.elasticsearch import ElasticsearchRetriever

# 初始化各产品对应的Elastic检索器
retriever_a = ElasticsearchRetriever.from_es_params(
    index="product_a_index", es_url="http://your-elastic-url:9200"
)
retriever_b = ElasticsearchRetriever.from_es_params(
    index="product_b_index", es_url="http://your-elastic-url:9200"
)
retriever_c = ElasticsearchRetriever.from_es_params(
    index="product_c_index", es_url="http://your-elastic-url:9200"
)

# 定义检索器路由逻辑
def route_to_retriever(input_dict):
    product_id = input_dict["product_id"]
    question = input_dict["question"]
    if product_id == "A":
        return retriever_a.invoke(question)
    elif product_id == "B":
        return retriever_b.invoke(question)
    elif product_id == "C":
        return retriever_c.invoke(question)
    else:
        raise ValueError(f"未知产品ID: {product_id}")

# 构建单链
chain = (
    RunnablePassthrough.assign(documents=RunnableLambda(route_to_retriever))
    | prompt  # 你的Prompt模板，需包含question和documents变量
    | llm     # 你的LLM实例
)

# 调用示例
result = chain.invoke({"question": "产品A的核心功能是什么？", "product_id": "A"})

方案二：自定义动态路由检索器类 #

继承LangChain的BaseRetriever接口，封装路由逻辑，让自定义检索器直接支持字典格式输入，更符合组件化设计规范。

from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from typing import List

class DynamicProductRetriever(BaseRetriever):
    def __init__(self, retriever_map: dict):
        self.retriever_map = retriever_map  # 格式：{"A": retriever_a, "B": retriever_b, ...}
        super().__init__()

    def _get_relevant_documents(self, query: dict, **kwargs) -> List[Document]:
        product_id = query["product_id"]
        question = query["question"]
        if product_id not in self.retriever_map:
            raise ValueError(f"未知产品ID: {product_id}")
        return self.retriever_map[product_id].get_relevant_documents(question, **kwargs)

# 初始化自定义检索器
dynamic_retriever = DynamicProductRetriever(
    retriever_map={"A": retriever_a, "B": retriever_b, "C": retriever_c}
)

# 构建链
chain = (
    RunnablePassthrough.assign(documents=dynamic_retriever)
    | prompt
    | llm
)

# 调用示例
result = chain.invoke({"question": "产品B的售后政策？", "product_id": "B"})

方案三：用RunnableBranch实现分支路由 #

利用LangChain的RunnableBranch组件，根据product_id直接分支到对应检索逻辑，代码更简洁直观。

from langchain_core.runnables import RunnableBranch

# 定义分支路由规则
retriever_branch = RunnableBranch(
    (lambda x: x["product_id"] == "A", lambda x: retriever_a.invoke(x["question"])),
    (lambda x: x["product_id"] == "B", lambda x: retriever_b.invoke(x["question"])),
    (lambda x: x["product_id"] == "C", lambda x: retriever_c.invoke(x["question"])),
    lambda x: ValueError(f"未知产品ID: {x['product_id']}")
)

# 构建链
chain = (
    RunnablePassthrough.assign(documents=retriever_branch)
    | prompt
    | llm
)

# 调用示例
result = chain.invoke({"question": "产品C的定价？", "product_id": "C"})

内容的提问来源于stack exchange，提问作者Assaf