Langchain 0.218：如何在Agent运行时动态过滤向量数据库？

核心思路 #

不用嵌套双链式结构，通过自定义检索逻辑+Agent工具绑定就能实现需求，核心分两步：1. 从用户输入中提取目标文件名；2. 基于文件名动态修改检索器的search_kwargs（设置过滤条件+将k值设为对应文件的chunk总数）。

具体实现步骤 #

1. 实现文件名提取逻辑 #

用极简的LLM链完成文件名提取，只返回文件名或默认标识，避免多余输出：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 提取文件名的链
extract_filename_prompt = ChatPromptTemplate.from_messages([
    ("system", "仅从用户问题中提取涉及的文件名，无明确文件名时返回'None'，不要额外解释。"),
    ("human", "{input}")
])
filename_extractor = extract_filename_prompt | ChatOpenAI(model="gpt-3.5-turbo") | StrOutputParser()

2. 封装动态过滤的检索器 #

基于文件名生成带过滤条件和指定k值的检索器，假设你已维护好file_chunk_count映射表（文件名→对应chunk数量）：

from langchain_community.vectorstores import Chroma
from langchain_core.vectorstores import VectorStoreRetriever

# 预先维护的文件-chunk数量映射
file_chunk_count = {
    "project_plan.pdf": 12,
    "financial_report.xlsx": 18
    # 补充你的文件映射
}

def get_filtered_retriever(filename: str, vectorstore: Chroma) -> VectorStoreRetriever:
    if filename == "None":
        # 无指定文件名时的默认检索配置
        return vectorstore.as_retriever(search_kwargs={"k": 4})
    # 设置过滤条件（假设你的vectorstore元数据包含"filename"字段）
    filter_dict = {"filename": filename}
    # 从映射表取对应chunk数量作为k，确保获取全量chunk
    k = file_chunk_count.get(filename, 4)
    return vectorstore.as_retriever(search_kwargs={"filter": filter_dict, "k": k})

3. 整合到Agent工作流 #

将上述逻辑封装成Agent可调用的工具，替代原始检索器：

from langchain_core.tools import tool
from langchain.agents import create_openai_tools_agent, AgentExecutor

# 初始化你的Chroma向量库
vectorstore = Chroma(persist_directory="./your_chroma_db", embedding_function=your_embedding_model)

# 封装检索工具
@tool
def retrieve_specific_file_content(input_text: str) -> str:
    """根据用户输入中的文件名，提取对应文件的全量chunk内容"""
    filename = filename_extractor.invoke({"input": input_text})
    retriever = get_filtered_retriever(filename, vectorstore)
    docs = retriever.get_relevant_documents(input_text)
    # 将文档内容拼接为字符串返回给Agent
    return "\n\n".join([doc.page_content for doc in docs])

# 构建Agent
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是专业问答助手，需要检索文件内容时必须调用retrieve_specific_file_content工具。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

llm = ChatOpenAI(model="gpt-3.5-turbo")
tools = [retrieve_specific_file_content]
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 测试执行
agent_executor.invoke({"input": "请提取project_plan.pdf里的项目里程碑内容"})

为什么Self-Query Retrieval不适用？ #

Self-Query Retrieval的核心是基于用户查询自动生成语义化过滤条件，它无法强制获取指定文件的全量chunk，更适合模糊语义检索场景。你的需求是精准定位单文件全量内容，直接手动控制search_kwargs的filter和k值更高效。

简化方案：自定义Retriever子类 #

如果不想单独封装工具，可以直接写一个动态检索器子类，内置文件名提取逻辑：

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

class DynamicFileRetriever(BaseRetriever):
    vectorstore: Chroma
    filename_extractor: callable
    file_chunk_count: dict

    def _get_relevant_documents(self, query: str) -> List[Document]:
        filename = self.filename_extractor.invoke({"input": query})
        filter_dict = {"filename": filename} if filename != "None" else {}
        k = self.file_chunk_count.get(filename, 4)
        return self.vectorstore.similarity_search(query, filter=filter_dict, k=k)

# 初始化自定义检索器
dynamic_retriever = DynamicFileRetriever(
    vectorstore=vectorstore,
    filename_extractor=filename_extractor,
    file_chunk_count=file_chunk_count
)

# 后续可直接将该检索器接入QA链，再把QA链包装为Agent工具

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