如何在LangChain链中动态更新Pinecone检索器的search_kwargs过滤条件

1. 先让LLM返回结构化结果（推荐） #

首先确保你的generate_need能返回包含nature字段的结构化数据，这样提取字段更稳定。可以用Pydantic输出解析器实现：

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

# 定义结构化输出的模型
class UserDemand(BaseModel):
    nature: str = Field(description="用户需求的属性，比如'办公'、'休闲'、'商务'")
    # 其他需要的字段可以在这里补充

# 初始化解析器
parser = PydanticOutputParser(pydantic_object=UserDemand)

# 把解析器加到你的generate_need链中
generate_need = prompt | llm | parser

2. 编写动态更新检索器Filter的逻辑 #

写一个函数，接收LLM生成的结构化需求，给每个检索器的search_kwargs["filter"]追加nature条件，然后执行检索：

from langchain_core.runnables import RunnableLambda

def apply_nature_filter(inputs):
    # inputs包含用户查询和LLM生成的结构化需求
    user_query = inputs["query"]
    user_demand = inputs["need"]
    
    # 假设你的检索器存在一个字典里，比如{"retriever_a": pinecone_ret_a, "retriever_b": pinecone_ret_b}
    retrieval_results = {}
    for ret_name, retriever in your_retriever_dict.items():
        # 复制原始filter，避免修改检索器的默认配置
        updated_filter = retriever.search_kwargs["filter"].copy()
        updated_filter["nature"] = user_demand.nature
        
        # 用更新后的filter执行检索
        retrieval_results[ret_name] = retriever.invoke(
            user_query, 
            search_kwargs={"filter": updated_filter}
        )
    return retrieval_results

3. 重构完整推荐链 #

调整链的执行顺序：先生成用户需求，再用需求动态更新检索器Filter并并行检索，最后执行后续的推荐生成步骤：

from langchain_core.runnables import RunnablePassthrough, RunnableParallel

recommendation_chain = (
    # 接收原始查询
    {"query": RunnablePassthrough()}
    # 第一步：生成结构化用户需求，同时保留原始查询
    | RunnableParallel(
        query=RunnablePassthrough(),
        need=generate_need
    )
    # 第二步：动态更新Filter并执行并行检索
    | RunnableLambda(apply_nature_filter)
    # 第三步：后续的推荐结果生成逻辑（比如把检索结果传给LLM生成最终推荐）
    | final_recommendation_generator
)

简化替代方案（直接在并行检索分支传参） #

如果你的检索器数量不多，也可以直接在RunnableParallel的每个检索分支里动态传入更新后的Filter，不用单独写函数：

recommendation_chain = (
    {"query": RunnablePassthrough()}
    | RunnableParallel(
        need=generate_need,
        query=RunnablePassthrough()
    )
    | RunnableParallel(
        ret_a=lambda x: retriever_a.invoke(
            x["query"],
            search_kwargs={
                "filter": {**retriever_a.search_kwargs["filter"], "nature": x["need"].nature}
            }
        ),
        ret_b=lambda x: retriever_b.invoke(
            x["query"],
            search_kwargs={
                "filter": {**retriever_b.search_kwargs["filter"], "nature": x["need"].nature}
            }
        )
    )
    | final_recommendation_generator
)

关键注意事项 #

• 如果暂时没法用结构化输出，也可以用正则表达式或者轻量LLM调用从LLM的文本输出中抽取nature字段，但结构化输出的可靠性更高。
• 必须复制原始Filter再修改，不能直接修改检索器的默认search_kwargs，否则会影响其他场景下的检索调用。
• 确保nature的取值和Pinecone向量库中存储的字段值格式完全匹配（比如大小写、枚举值一致性），否则会导致检索结果为空或不准确。

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