如果你正在构建一个基于大语言模型LLM的智能应用比如一个能回答专业问题的知识库助手那么你很可能已经遇到了一个核心难题如何让 LLM 高效、准确地从你的私有数据中获取信息简单地将所有文档喂给 LLM 不仅成本高昂、速度慢更关键的是LLM 的上下文窗口有限无法处理海量数据。这时“检索增强生成”RAG技术成为了标准答案。而 RAG 的核心在于一个能快速从海量向量中找出最相似项的“搜索引擎”——向量数据库。在众多向量数据库中Milvus 以其高性能、可扩展性和丰富的功能脱颖而出。但很多开发者在初步接触后往往止步于简单的相似性搜索。当你的业务逻辑变得复杂比如需要根据文档类型、作者、时间等多维度条件进行过滤或者需要对搜索结果进行排序、聚合时仅靠基础的search接口就显得力不从心了。这正是DQLMilvus Data Query Language大显身手的地方。它让你能够像使用 SQL 查询关系型数据库一样灵活、精确地查询 Milvus 中的向量和标量数据。本文将带你深入实战解决一个具体问题如何结合 LangChain 与 Milvus DQL构建一个支持复杂过滤与混合查询的智能检索系统。读完本文你将能清晰地掌握从环境搭建、数据准备、DQL 语句构建到与 LangChain 集成的全流程并避开集成过程中的常见陷阱。1. 这篇文章真正要解决的问题很多关于 LangChain Milvus 的教程演示的往往是“开箱即用”的最简流程加载文档、切分、向量化、存入 Milvus、然后进行语义搜索。这就像只学会了汽车的“前进”挡位。但在真实项目中需求远不止于此。例如你的知识库包含技术文档、用户手册、会议纪要等多种类型你只想在“技术文档”中搜索。你想查找“最近三个月发布的”、“由某位专家撰写的”、且与问题相关的文档。你需要先根据关键词过滤出一批文档再在这批文档中进行向量相似度排序。这些需求都涉及到对元数据标量数据的过滤和复杂查询。如果只用简单的相似度搜索你不得不先取出大量可能不相关的向量然后在应用层进行过滤效率低下且不精确。本文要解决的核心问题就是如何利用 Milvus 强大的 DQL在 LangChain 框架内实现高效的“向量相似度搜索 标量属性过滤”混合查询。这能让你构建的 RAG 应用更加智能和精准。我们将通过一个完整的项目示例演示如何为一家“虚拟科技公司”构建一个能按部门、文档类型、时间等条件进行智能问答的知识库系统。2. 基础概念与核心原理在深入代码之前我们需要统一几个关键概念这能帮助你理解后续每一步的设计意图。2.1 LangChain 与 Vector StoresLangChain 是一个用于开发由 LLM 驱动的应用程序的框架。它提供了“链”Chains、“代理”Agents等高级抽象而Vector Stores是其生态中用于存储和检索向量嵌入Embeddings的组件标准接口。LangChain 支持多种向量数据库如 Milvus, Pinecone, Chroma通过统一的接口调用降低了切换底层的成本。2.2 Milvus 与数据组织Milvus 是一个云原生的向量数据库专为处理海量向量数据而设计。集合Collection相当于关系数据库中的表是存储数据的基本单位。实体Entity相当于表中的一行记录。每个实体包含一个主键、一个向量字段和多个标量字段属性。分区Partition集合的物理子集用于数据隔离和管理常用于按业务维度如部门、时间划分数据提升查询效率。2.3 DQL (Data Query Language)DQL 是 Milvus 的查询语言语法上借鉴了 SQL用于执行复杂的查询操作。与简单的searchAPI 相比DQL 的核心能力在于标量过滤Scalar Filtering使用inand,or等运算符对非向量字段进行条件筛选。混合查询Hybrid Search将向量相似度搜索与标量过滤结合在一个查询中。丰富的结果处理支持对结果进行排序ORDER BY、限制LIMIT、输出字段选择等。2.4 工作流程我们的目标流程如下图所示概念性描述文档处理原始文本经过切分Text Splitter变成一个个文本块Chunk。向量化每个文本块通过嵌入模型Embedding Model转化为一个高维向量并提取或赋予其元数据如来源、类型、日期。数据入库将向量和元数据作为一个整体实体写入 Milvus 集合。混合查询用户提问时先将问题转化为向量然后构造一个 DQL 语句该语句同时包含“在向量空间找相似”和“按元数据条件过滤”。生成答案将检索到的最相关文本块作为上下文与用户问题一起提交给 LLM生成最终答案。理解了这个流程我们就知道代码需要围绕“如何让 LangChain 的检索器Retriever使用我们自定义的 DQL 语句”来展开。3. 环境准备与前置条件为了复现本文的实战项目你需要准备好以下环境。建议使用 Python 3.8-3.11 版本。3.1 安装核心库通过 pip 安装必要的 Python 包。建议使用虚拟环境。# 安装 LangChain 及其相关组件 pip install langchain langchain-community langchain-openai # 安装 Milvus Python SDK (pymilvus) 和 LangChain 的 Milvus 集成包 pip install pymilvus langchain-milvus # 安装文本嵌入模型相关库这里以 OpenAI 和 Sentence Transformers 为例 pip install openai sentence-transformers # 安装环境变量管理库可选但推荐 pip install python-dotenv3.2 启动 Milvus 服务你需要一个运行中的 Milvus 实例。对于本地开发和测试使用 Docker 启动 Standalone单机模式是最快捷的方式。# 拉取最新的 Milvus Standalone 镜像 docker pull milvusdb/milvus:latest # 运行 Milvus Standalone 容器 docker run -d \ --name milvus-standalone \ -p 19530:19530 \ -p 9091:9091 \ -v ~/milvus/data:/var/lib/milvus \ -v ~/milvus/conf:/etc/milvus \ milvusdb/milvus:latest说明-p 19530:19530: 暴露 Milvus 的服务端口。-p 9091:9091: 暴露 Milvus 的管理端口用于 Attu 等管理工具。-v ...: 将数据卷和配置卷挂载到本地防止容器删除后数据丢失。运行后可以通过docker ps检查容器状态确保其正常运行。3.3 配置访问密钥如使用云端模型如果你使用 OpenAI 的嵌入模型需要设置 API Key。强烈建议将其存储在环境变量中而不是硬编码在代码里。# 在项目根目录创建 .env 文件 echo OPENAI_API_KEYyour_openai_api_key_here .env对应的 Python 代码中可以通过os.getenv(“OPENAI_API_KEY”)读取。4. 项目结构与核心流程拆解我们将构建一个名为tech_doc_qa的项目。其核心流程可以拆解为以下四个步骤每一步我们都将详细实现数据建模与集合创建在 Milvus 中设计并创建一个符合我们业务需求的集合。文档加载与向量化入库将本地知识文档处理成向量和元数据并批量插入 Milvus。构建支持 DQL 的自定义检索器这是本文的关键我们将扩展 LangChain 的基类使其能执行复杂的 DQL 查询。集成问答链进行测试将自定义检索器接入 LangChain 的 QA 链完成端到端的智能问答。5. 数据建模与 Milvus 集合创建首先我们需要定义数据的结构。假设我们的科技公司文档有以下属性id: 主键自增或UUIDvector: 文本块的向量表示例如 768 维text: 文本块内容本身source: 文档来源如 “employee_handbook.pdf”doc_type: 文档类型如 “tech_doc”, “meeting_minutes”, “user_guide”department: 所属部门如 “engineering”, “product”, “hr”created_at: 文档创建时间戳我们在 Python 中连接 Milvus 并创建这个集合。# file: create_collection.py from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility import time # 1. 连接到 Milvus 服务 connections.connect(hostlocalhost, port19530) # 2. 定义集合名称 collection_name tech_company_docs dim 768 # 向量维度根据你选择的嵌入模型确定例如 sentence-transformers 的 all-MiniLM-L6-v2 模型是 384 维 # 3. 如果集合已存在则删除仅用于演示生产环境慎用 if utility.has_collection(collection_name): utility.drop_collection(collection_name) # 4. 定义字段模式 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dimdim), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), # 存储文本内容 FieldSchema(namesource, dtypeDataType.VARCHAR, max_length255), FieldSchema(namedoc_type, dtypeDataType.VARCHAR, max_length100), FieldSchema(namedepartment, dtypeDataType.VARCHAR, max_length100), FieldSchema(namecreated_at, dtypeDataType.INT64), # 使用时间戳存储 ] # 5. 定义集合模式 schema CollectionSchema(fieldsfields, descriptionTech company internal documents with metadata) # 6. 创建集合 collection Collection(namecollection_name, schemaschema) # 7. 为向量字段和常用过滤字段创建索引以加速搜索 index_params { index_type: IVF_FLAT, # 一种高效的向量索引类型适合中等规模数据集 metric_type: L2, # 距离度量方式L2 欧氏距离 params: {nlist: 128}, # 聚类中心数值越大精度越高但速度越慢 } collection.create_index(field_namevector, index_paramsindex_params) # 为标量字段创建索引可加速过滤查询 collection.create_index(field_namedoc_type, index_params{index_type: TRIE}) collection.create_index(field_namedepartment, index_params{index_type: TRIE}) collection.create_index(field_namecreated_at, index_params{index_type: STL_SORT}) print(fCollection {collection_name} created successfully with indexes.) print(Remember to load the collection before searching:) print(fcollection.load())关键点解释auto_idTrue让 Milvus 自动生成递增的 ID简化插入操作。IVF_FLAT索引在准确性和性能之间取得良好平衡的索引适合开发测试。生产环境可根据数据量选择HNSW等。为doc_type,department等字段创建TRIE索引能为,in这类等值过滤条件大幅提速。创建索引后需要执行collection.load()将集合和数据加载到内存查询才能生效。我们将在检索器部分处理。6. 文档加载、处理与向量化入库接下来我们模拟一些文档数据并将其处理、向量化后存入刚创建的集合。这里我们使用sentence-transformers本地模型生成向量避免调用外部 API。# file: ingest_documents.py from pymilvus import connections, Collection from langchain.text_splitter import RecursiveCharacterTextSplitter from sentence_transformers import SentenceTransformer import time import json # 1. 连接 Milvus connections.connect(hostlocalhost, port19530) collection Collection(tech_company_docs) # 2. 初始化文本分割器和嵌入模型 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个文本块的最大字符数 chunk_overlap50 # 块之间的重叠字符数保持上下文连贯 ) # 选择嵌入模型all-MiniLM-L6-v2 是一个轻量且效果不错的通用模型 embedding_model SentenceTransformer(all-MiniLM-L6-v2) # 注意该模型维度为384需要与创建集合时的 dim 参数一致。请将 create_collection.py 中的 dim 改为 384。 # 3. 模拟文档数据 documents [ { content: # 工程部后端服务部署规范 V2.1 本文档规定了所有微服务上线至生产环境的标准化流程。 核心要求包括1. 必须通过 CI/CD 流水线。2. 健康检查接口需返回 200。3. 日志必须聚合到中央 ELK。 适用部门工程部。文档类型技术规范。 , source: deploy_guide_v2.1.md, doc_type: tech_doc, department: engineering, created_at: 1672502400, # 2023-01-01 }, { content: Q4 产品规划会议纪要 时间2023-10-15。参会人Alice(产品), Bob(工程), Carol(设计)。 主要决议1. 优先开发智能报表模块。2. 用户体验优化延期至下季度。 下一步工程部需在两周内给出技术方案评审。 , source: q4_planning_meeting.txt, doc_type: meeting_minutes, department: product, created_at: 1697328000, # 2023-10-15 }, { content: HR 系统新员工使用指南 欢迎加入公司本指南将帮助你快速熟悉 HR 系统。 内容包括如何提交请假申请、查看薪资单、更新个人信息。 如有问题请联系人力资源部。 , source: hr_system_guide.pdf, doc_type: user_guide, department: hr, created_at: 1688169600, # 2023-07-01 }, # ... 可以添加更多模拟文档 ] # 4. 处理每个文档切分、向量化、准备插入数据 all_texts [] all_metadatas [] all_vectors [] for doc in documents: # 分割文本 chunks text_splitter.split_text(doc[content]) for chunk in chunks: all_texts.append(chunk) # 为每个 chunk 复制文档的元数据 metadata { source: doc[source], doc_type: doc[doc_type], department: doc[department], created_at: doc[created_at], } all_metadatas.append(metadata) # 生成向量嵌入 vector embedding_model.encode(chunk).tolist() # 转换为 list all_vectors.append(vector) # 5. 准备批量插入的数据字段顺序需与集合定义一致 entities [ all_vectors, # vector 字段 all_texts, # text 字段 [meta[source] for meta in all_metadatas], # source 字段 [meta[doc_type] for meta in all_metadatas], # doc_type 字段 [meta[department] for meta in all_metadatas], # department 字段 [meta[created_at] for meta in all_metadatas], # created_at 字段 ] # 6. 插入数据 insert_result collection.insert(entities) print(fInserted {len(all_texts)} text chunks into Milvus.) print(fInserted PKs: {insert_result.primary_keys[:5]}...) # 打印前5个主键 # 7. 确保数据持久化并加载到内存使可搜索 collection.flush() # 将内存中的数据写入磁盘 collection.load() # 将集合加载到内存 print(Data flushed and collection loaded. Ready for search.)关键点解释RecursiveCharacterTextSplitter是 LangChain 中常用的分割器能较好地按字符递归分割保持段落和句子的完整性。我们使用本地 SentenceTransformer 模型生成向量这避免了网络延迟和 API 成本适合离线或内部环境。确保其维度与集合定义的dim一致。collection.flush()确保插入操作被持久化。collection.load()将数据加载到内存这是执行搜索前的必要步骤。7. 构建支持 DQL 的自定义检索器这是本文最核心的部分。LangChain 内置的Milvus检索器可能不支持复杂的 DQL 表达式过滤。我们需要自定义一个检索器它能够接收一个过滤条件字符串并将其整合到 Milvus 的查询中。我们将创建一个继承自langchain.vectorstores.VectorStore的类并实现_similarity_search_with_score等方法。# file: custom_milvus_retriever.py from typing import Any, Dict, List, Optional, Tuple from langchain.vectorstores import VectorStore from langchain.schema import Document from pymilvus import Collection, connections import numpy as np class CustomMilvusVectorStore(VectorStore): 支持复杂 DQL 过滤条件的自定义 Milvus VectorStore. def __init__( self, collection_name: str, embedding_function: Any, # 例如 SentenceTransformer.encode 函数 connection_args: Dict[str, Any] {host: localhost, port: 19530}, **kwargs: Any, ): 初始化。 Args: collection_name: Milvus 集合名称。 embedding_function: 用于将文本转换为向量的函数。 connection_args: 连接 Milvus 的参数。 self.collection_name collection_name self.embedding_function embedding_function self.connection_args connection_args # 建立连接并获取集合对象 connections.connect(**self.connection_args) self.collection Collection(self.collection_name) # 确保集合已加载到内存 if not self.collection.has_index(): raise ValueError(fCollection {collection_name} has no index. Please create index first.) self.collection.load() # 获取集合的向量字段名假设只有一个向量字段 self.vector_field None for field in self.collection.schema.fields: if field.dtype.name FLOAT_VECTOR: self.vector_field field.name break if self.vector_field is None: raise ValueError(No vector field found in the collection schema.) # 获取主键字段名 self.primary_field None for field in self.collection.schema.fields: if field.is_primary: self.primary_field field.name break # 文本字段名我们假设名为 text self.text_field text super().__init__(**kwargs) def _build_search_params(self, expr: Optional[str] None) - Dict: 构建搜索参数支持 DQL 过滤表达式。 search_params { metric_type: L2, # 与创建索引时一致 params: {nprobe: 10}, # 搜索时探查的聚类中心数影响速度和精度 } # 如果提供了过滤表达式加入搜索参数 if expr: search_params[expr] expr return search_params def similarity_search_with_score_by_vector( self, embedding: List[float], k: int 4, expr: Optional[str] None, **kwargs: Any, ) - List[Tuple[Document, float]]: 根据向量进行相似度搜索并返回文档及其分数。 Args: embedding: 查询向量。 k: 返回的最相似结果数量。 expr: Milvus DQL 过滤表达式字符串。例如: doc_type tech_doc and department engineering Returns: 包含 (Document, similarity_score) 的列表。 search_params self._build_search_params(expr) # 执行搜索 results self.collection.search( data[embedding], anns_fieldself.vector_field, paramsearch_params, limitk, output_fields[self.text_field, source, doc_type, department, created_at], # 指定需要返回的字段 **kwargs, ) # 解析结果 ret [] for hits in results: for hit in hits: # hit.entity 包含了 output_fields 中指定的字段 metadata { source: hit.entity.get(source), doc_type: hit.entity.get(doc_type), department: hit.entity.get(department), created_at: hit.entity.get(created_at), id: hit.id, # 主键 } doc Document( page_contenthit.entity.get(self.text_field), metadatametadata ) # Milvus 的 score 是距离对于 L2 距离越小越相似。我们将其转换为相似度分数可选。 # 这里直接使用距离的负数使其越大表示越相似更符合常规认知。 similarity_score -hit.distance ret.append((doc, similarity_score)) return ret def similarity_search_with_score( self, query: str, k: int 4, expr: Optional[str] None, **kwargs: Any, ) - List[Tuple[Document, float]]: 根据查询文本进行相似度搜索。 # 将查询文本转换为向量 query_embedding self.embedding_function(query) if isinstance(query_embedding, np.ndarray): query_embedding query_embedding.tolist() return self.similarity_search_with_score_by_vector(query_embedding, kk, exprexpr, **kwargs) def similarity_search( self, query: str, k: int 4, expr: Optional[str] None, **kwargs: Any, ) - List[Document]: 根据查询文本进行相似度搜索只返回文档。 docs_and_scores self.similarity_search_with_score(query, kk, exprexpr, **kwargs) return [doc for doc, _ in docs_and_scores] def as_retriever(self, search_kwargs: Optional[Dict] None): 返回一个检索器对象可以传入自定义搜索参数如 expr。 from langchain.vectorstores.base import VectorStoreRetriever return VectorStoreRetriever(vectorstoreself, search_kwargssearch_kwargs or {}) # 必须实现 add_texts 方法为了继承 VectorStore这里简化处理 def add_texts(self, texts, metadatasNone, **kwargs): raise NotImplementedError(This custom class is for search only. Use ingest_documents.py for ingestion.) classmethod def from_texts(cls, texts, embedding, metadatasNone, **kwargs): raise NotImplementedError(This custom class is for search only. Use ingest_documents.py for ingestion.)核心逻辑解释初始化连接 Milvus加载集合并自动识别集合中的向量字段、主键字段和文本字段。_build_search_params方法构建搜索参数。最关键的一步是如果调用者传入了exprDQL 表达式就将其加入到搜索参数中。Milvus 的search接口的expr参数正是用于标量过滤。similarity_search_with_score_by_vector方法这是执行混合查询的核心。它接收一个向量、返回数量k和过滤表达式expr调用collection.search并传入expr。返回的结果会自动应用过滤条件。结果解析将 Milvus 返回的实体Entity转换为 LangChain 的Document对象并携带所有元数据。as_retriever方法返回一个检索器允许我们通过search_kwargs动态传入expr等参数这在与 LangChain Chain 集成时非常方便。8. 集成问答链与复杂查询实战现在我们将自定义的检索器接入 LangChain 的检索问答链RetrievalQA并执行几个具有复杂过滤条件的查询。# file: query_with_dql.py from custom_milvus_retriever import CustomMilvusVectorStore from sentence_transformers import SentenceTransformer from langchain.chains import RetrievalQA from langchain.llms import OpenAI # 或者使用 ChatOpenAI from langchain.prompts import PromptTemplate import os from dotenv import load_dotenv # 加载环境变量用于 OpenAI API Key load_dotenv() # 1. 初始化嵌入模型和向量库 embedding_model SentenceTransformer(all-MiniLM-L6-v2) def encode_function(text: str) - List[float]: return embedding_model.encode(text).tolist() vector_store CustomMilvusVectorStore( collection_nametech_company_docs, embedding_functionencode_function, connection_args{host: localhost, port: 19530} ) # 2. 创建支持 DQL 过滤的检索器 # 示例1检索器默认不带过滤 base_retriever vector_store.as_retriever(search_kwargs{k: 3}) print(--- 基础检索器创建成功 ---) # 3. 初始化 LLM这里使用 OpenAI GPT-3.5-turbo你需要设置 OPENAI_API_KEY from langchain_openai import ChatOpenAI llm ChatOpenAI(model_namegpt-3.5-turbo, temperature0) # 4. 构建一个简单的问答链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 将检索到的所有文档内容“塞”进上下文 retrieverbase_retriever, return_source_documentsTrue, # 返回源文档便于调试 verboseTrue, # 打印详细日志 ) # 5. 执行不同场景的查询 print(\n 场景 1: 无过滤普通语义搜索 ) query1 如何部署服务 result1 qa_chain.invoke({query: query1}) print(f问题: {query1}) print(f答案: {result1[result][:200]}...) # 打印前200字符 print(f参考来源: {[doc.metadata.get(source) for doc in result1[source_documents]]}) print(\n 场景 2: 只搜索工程部的技术文档 ) # 关键创建带有过滤表达式的检索器 engineering_tech_retriever vector_store.as_retriever( search_kwargs{ k: 3, expr: doc_type tech_doc and department engineering } ) qa_chain_engineering RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverengineering_tech_retriever, return_source_documentsTrue, ) query2 部署服务有什么规范 result2 qa_chain_engineering.invoke({query: query2}) print(f问题: {query2}) print(f答案: {result2[result][:200]}...) print(f参考来源: {[doc.metadata.get(source) for doc in result2[source_documents]]}) print(f文档类型: {[doc.metadata.get(doc_type) for doc in result2[source_documents]]}) print(\n 场景 3: 搜索产品部门最近的会议纪要时间过滤) # 假设“最近”指创建时间大于某个时间戳例如 2023年10月1日 (1696118400) recent_product_retriever vector_store.as_retriever( search_kwargs{ k: 3, expr: department product and doc_type meeting_minutes and created_at 1696118400 } ) # 也可以直接使用向量库的搜索方法进行更灵活的调试 print(直接使用向量库搜索进行调试:) docs vector_store.similarity_search( query产品规划, k2, exprdepartment product and doc_type meeting_minutes ) for i, doc in enumerate(docs): print(f 结果{i1}: {doc.page_content[:100]}... | 来源: {doc.metadata.get(source)}) print(\n 场景 4: 使用 in 操作符进行多值过滤 ) # 查询属于“工程部”或“产品部”的文档 multi_dept_retriever vector_store.as_retriever( search_kwargs{ k: 4, expr: department in [engineering, product] } ) qa_chain_multi RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievermulti_dept_retriever, ) query4 各部门接下来有什么计划 result4 qa_chain_multi.invoke({query: query4}) print(f问题: {query4}) print(f答案: {result4[result][:300]}...)运行与验证运行此脚本你将看到针对不同过滤条件系统返回了不同的答案和来源。例如对于“部署服务有什么规范”这个问题当使用工程部技术文档过滤时它应该只从《工程部后端服务部署规范》中寻找答案而不会混杂 HR 指南的内容。这证明了我们的 DQL 过滤是有效的。9. 常见问题与排查思路在实际集成中你可能会遇到以下问题问题现象可能原因排查方式解决方案连接 Milvus 失败Milvus 服务未启动端口错误网络问题。1.docker ps检查容器状态。2.telnet localhost 19530测试端口连通性。3. 检查连接参数host和port。确保 Milvus 容器正常运行并使用正确的连接参数。搜索时返回空结果集合未加载 (load()); 过滤表达式语法错误数据未插入成功。1. 检查是否执行了collection.load()。2. 在 Milvus 客户端如 Attu中直接执行 DQL 验证表达式。3. 检查插入操作是否成功并flush()。1. 确保搜索前调用load()。2. 简化表达式逐步调试。3. 确认数据插入流程无误。DQL 过滤表达式不生效表达式字段名错误字段类型不匹配运算符使用错误。1. 使用collection.schema打印字段名和类型。2. 对于字符串字段值需用单引号value。3. 检查 Milvus 官方文档确认 DQL 语法。1. 确保字段名与集合定义完全一致。2. 整数字段不要加引号字符串必须加。3. 使用and/or连接条件时注意括号。向量维度不匹配创建集合时定义的dim与嵌入模型输出的维度不一致。打印嵌入模型输出向量的长度len(embedding)。修改create_collection.py中的dim参数使其与模型维度一致并重建集合。检索速度慢数据量增大后未调整索引参数过滤字段未建索引。1. 使用collection.get_index_stats()查看索引情况。2. 对常用过滤字段创建二级索引如 TRIE。1. 调整nlist创建时和nprobe搜索时参数。2. 为过滤字段创建合适的索引。LangChain 链报错expr参数无效使用的 LangChain 内置 Milvus 集成版本可能不支持expr。检查是否使用了本文的自定义CustomMilvusVectorStore类。务必使用我们自定义的类它显式处理了expr参数。10. 最佳实践与工程建议将 LangChain 与 Milvus DQL 投入生产环境还需要注意以下几点10.1 索引策略优化向量索引对于千万级以上向量考虑使用HNSW或DISKANN索引以获得更好的查询性能。IVF_FLAT适合百万级数据。标量索引对频繁用于过滤的标量字段如doc_type,department,created_at务必创建索引TRIE用于字符串STL_SORT用于数值/时间。分区管理如果数据有明显的冷热区分或业务维度如按年份、地区使用 Milvus 的分区功能。查询时指定分区可以极大提升速度。10.2 查询性能调优nprobe参数在搜索参数params中设置。它代表搜索时探查的聚类中心数。值越大精度越高但速度越慢。需要在精度和延迟之间做权衡测试。分批查询如果一次需要查询大量向量使用batch_search接口。结果集大小合理设置limit即k值。在 RAG 中通常 3-5 个相关块已足够 LLM 生成优质答案获取过多反而可能引入噪声。10.3 应用层设计检索器封装在业务系统中不要每次都新建检索器。可以设计一个工厂类或配置中心根据不同的查询场景如不同部门、不同文档类型返回预配置了不同expr的检索器。表达式安全如果过滤条件来自用户输入例如前端下拉框务必进行严格的校验和转义防止 DQL 注入攻击。最佳实践是使用参数化查询如果 Milvus SDK 支持或在应用层构建安全的表达式白名单。错误处理与降级对 Milvus 查询做好异常捕获如超时、连接失败。可以考虑设置备用检索方案如回退到关键词搜索或缓存常用查询结果。10.4 数据维护元数据一致性确保插入的元数据格式规范、值域明确。例如doc_type字段的值应来自一个预定义的枚举列表。定时加载在 Milvus 集群中新插入的数据需要手动或定时调用load()才能被搜索到。生产环境需要规划数据加载策略如低峰期定时加载。版本管理当嵌入模型升级导致向量维度或语义空间变化时需要重建整个向量库。应有对应的数据迁移和版本回滚方案。通过本文的实战你已经掌握了利用 Milvus DQL 增强 LangChain 应用检索能力的核心方法。这不仅仅是语法上的学习更是设计思想的转变将向量数据库视为一个支持复杂查询的智能存储层而不仅仅是一个简单的相似度匹配工具。接下来你可以尝试将过滤条件动态化与用户身份、会话上下文结合构建出更精细、更个性化的 RAG 应用。 LangChain与Milvus DQL实战：构建支持复杂过滤的RAG智能检索系统
如果你正在构建一个基于大语言模型LLM的智能应用比如一个能回答专业问题的知识库助手那么你很可能已经遇到了一个核心难题如何让 LLM 高效、准确地从你的私有数据中获取信息简单地将所有文档喂给 LLM 不仅成本高昂、速度慢更关键的是LLM 的上下文窗口有限无法处理海量数据。这时“检索增强生成”RAG技术成为了标准答案。而 RAG 的核心在于一个能快速从海量向量中找出最相似项的“搜索引擎”——向量数据库。在众多向量数据库中Milvus 以其高性能、可扩展性和丰富的功能脱颖而出。但很多开发者在初步接触后往往止步于简单的相似性搜索。当你的业务逻辑变得复杂比如需要根据文档类型、作者、时间等多维度条件进行过滤或者需要对搜索结果进行排序、聚合时仅靠基础的search接口就显得力不从心了。这正是DQLMilvus Data Query Language大显身手的地方。它让你能够像使用 SQL 查询关系型数据库一样灵活、精确地查询 Milvus 中的向量和标量数据。本文将带你深入实战解决一个具体问题如何结合 LangChain 与 Milvus DQL构建一个支持复杂过滤与混合查询的智能检索系统。读完本文你将能清晰地掌握从环境搭建、数据准备、DQL 语句构建到与 LangChain 集成的全流程并避开集成过程中的常见陷阱。1. 这篇文章真正要解决的问题很多关于 LangChain Milvus 的教程演示的往往是“开箱即用”的最简流程加载文档、切分、向量化、存入 Milvus、然后进行语义搜索。这就像只学会了汽车的“前进”挡位。但在真实项目中需求远不止于此。例如你的知识库包含技术文档、用户手册、会议纪要等多种类型你只想在“技术文档”中搜索。你想查找“最近三个月发布的”、“由某位专家撰写的”、且与问题相关的文档。你需要先根据关键词过滤出一批文档再在这批文档中进行向量相似度排序。这些需求都涉及到对元数据标量数据的过滤和复杂查询。如果只用简单的相似度搜索你不得不先取出大量可能不相关的向量然后在应用层进行过滤效率低下且不精确。本文要解决的核心问题就是如何利用 Milvus 强大的 DQL在 LangChain 框架内实现高效的“向量相似度搜索 标量属性过滤”混合查询。这能让你构建的 RAG 应用更加智能和精准。我们将通过一个完整的项目示例演示如何为一家“虚拟科技公司”构建一个能按部门、文档类型、时间等条件进行智能问答的知识库系统。2. 基础概念与核心原理在深入代码之前我们需要统一几个关键概念这能帮助你理解后续每一步的设计意图。2.1 LangChain 与 Vector StoresLangChain 是一个用于开发由 LLM 驱动的应用程序的框架。它提供了“链”Chains、“代理”Agents等高级抽象而Vector Stores是其生态中用于存储和检索向量嵌入Embeddings的组件标准接口。LangChain 支持多种向量数据库如 Milvus, Pinecone, Chroma通过统一的接口调用降低了切换底层的成本。2.2 Milvus 与数据组织Milvus 是一个云原生的向量数据库专为处理海量向量数据而设计。集合Collection相当于关系数据库中的表是存储数据的基本单位。实体Entity相当于表中的一行记录。每个实体包含一个主键、一个向量字段和多个标量字段属性。分区Partition集合的物理子集用于数据隔离和管理常用于按业务维度如部门、时间划分数据提升查询效率。2.3 DQL (Data Query Language)DQL 是 Milvus 的查询语言语法上借鉴了 SQL用于执行复杂的查询操作。与简单的searchAPI 相比DQL 的核心能力在于标量过滤Scalar Filtering使用inand,or等运算符对非向量字段进行条件筛选。混合查询Hybrid Search将向量相似度搜索与标量过滤结合在一个查询中。丰富的结果处理支持对结果进行排序ORDER BY、限制LIMIT、输出字段选择等。2.4 工作流程我们的目标流程如下图所示概念性描述文档处理原始文本经过切分Text Splitter变成一个个文本块Chunk。向量化每个文本块通过嵌入模型Embedding Model转化为一个高维向量并提取或赋予其元数据如来源、类型、日期。数据入库将向量和元数据作为一个整体实体写入 Milvus 集合。混合查询用户提问时先将问题转化为向量然后构造一个 DQL 语句该语句同时包含“在向量空间找相似”和“按元数据条件过滤”。生成答案将检索到的最相关文本块作为上下文与用户问题一起提交给 LLM生成最终答案。理解了这个流程我们就知道代码需要围绕“如何让 LangChain 的检索器Retriever使用我们自定义的 DQL 语句”来展开。3. 环境准备与前置条件为了复现本文的实战项目你需要准备好以下环境。建议使用 Python 3.8-3.11 版本。3.1 安装核心库通过 pip 安装必要的 Python 包。建议使用虚拟环境。# 安装 LangChain 及其相关组件 pip install langchain langchain-community langchain-openai # 安装 Milvus Python SDK (pymilvus) 和 LangChain 的 Milvus 集成包 pip install pymilvus langchain-milvus # 安装文本嵌入模型相关库这里以 OpenAI 和 Sentence Transformers 为例 pip install openai sentence-transformers # 安装环境变量管理库可选但推荐 pip install python-dotenv3.2 启动 Milvus 服务你需要一个运行中的 Milvus 实例。对于本地开发和测试使用 Docker 启动 Standalone单机模式是最快捷的方式。# 拉取最新的 Milvus Standalone 镜像 docker pull milvusdb/milvus:latest # 运行 Milvus Standalone 容器 docker run -d \ --name milvus-standalone \ -p 19530:19530 \ -p 9091:9091 \ -v ~/milvus/data:/var/lib/milvus \ -v ~/milvus/conf:/etc/milvus \ milvusdb/milvus:latest说明-p 19530:19530: 暴露 Milvus 的服务端口。-p 9091:9091: 暴露 Milvus 的管理端口用于 Attu 等管理工具。-v ...: 将数据卷和配置卷挂载到本地防止容器删除后数据丢失。运行后可以通过docker ps检查容器状态确保其正常运行。3.3 配置访问密钥如使用云端模型如果你使用 OpenAI 的嵌入模型需要设置 API Key。强烈建议将其存储在环境变量中而不是硬编码在代码里。# 在项目根目录创建 .env 文件 echo OPENAI_API_KEYyour_openai_api_key_here .env对应的 Python 代码中可以通过os.getenv(“OPENAI_API_KEY”)读取。4. 项目结构与核心流程拆解我们将构建一个名为tech_doc_qa的项目。其核心流程可以拆解为以下四个步骤每一步我们都将详细实现数据建模与集合创建在 Milvus 中设计并创建一个符合我们业务需求的集合。文档加载与向量化入库将本地知识文档处理成向量和元数据并批量插入 Milvus。构建支持 DQL 的自定义检索器这是本文的关键我们将扩展 LangChain 的基类使其能执行复杂的 DQL 查询。集成问答链进行测试将自定义检索器接入 LangChain 的 QA 链完成端到端的智能问答。5. 数据建模与 Milvus 集合创建首先我们需要定义数据的结构。假设我们的科技公司文档有以下属性id: 主键自增或UUIDvector: 文本块的向量表示例如 768 维text: 文本块内容本身source: 文档来源如 “employee_handbook.pdf”doc_type: 文档类型如 “tech_doc”, “meeting_minutes”, “user_guide”department: 所属部门如 “engineering”, “product”, “hr”created_at: 文档创建时间戳我们在 Python 中连接 Milvus 并创建这个集合。# file: create_collection.py from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility import time # 1. 连接到 Milvus 服务 connections.connect(hostlocalhost, port19530) # 2. 定义集合名称 collection_name tech_company_docs dim 768 # 向量维度根据你选择的嵌入模型确定例如 sentence-transformers 的 all-MiniLM-L6-v2 模型是 384 维 # 3. 如果集合已存在则删除仅用于演示生产环境慎用 if utility.has_collection(collection_name): utility.drop_collection(collection_name) # 4. 定义字段模式 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dimdim), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), # 存储文本内容 FieldSchema(namesource, dtypeDataType.VARCHAR, max_length255), FieldSchema(namedoc_type, dtypeDataType.VARCHAR, max_length100), FieldSchema(namedepartment, dtypeDataType.VARCHAR, max_length100), FieldSchema(namecreated_at, dtypeDataType.INT64), # 使用时间戳存储 ] # 5. 定义集合模式 schema CollectionSchema(fieldsfields, descriptionTech company internal documents with metadata) # 6. 创建集合 collection Collection(namecollection_name, schemaschema) # 7. 为向量字段和常用过滤字段创建索引以加速搜索 index_params { index_type: IVF_FLAT, # 一种高效的向量索引类型适合中等规模数据集 metric_type: L2, # 距离度量方式L2 欧氏距离 params: {nlist: 128}, # 聚类中心数值越大精度越高但速度越慢 } collection.create_index(field_namevector, index_paramsindex_params) # 为标量字段创建索引可加速过滤查询 collection.create_index(field_namedoc_type, index_params{index_type: TRIE}) collection.create_index(field_namedepartment, index_params{index_type: TRIE}) collection.create_index(field_namecreated_at, index_params{index_type: STL_SORT}) print(fCollection {collection_name} created successfully with indexes.) print(Remember to load the collection before searching:) print(fcollection.load())关键点解释auto_idTrue让 Milvus 自动生成递增的 ID简化插入操作。IVF_FLAT索引在准确性和性能之间取得良好平衡的索引适合开发测试。生产环境可根据数据量选择HNSW等。为doc_type,department等字段创建TRIE索引能为,in这类等值过滤条件大幅提速。创建索引后需要执行collection.load()将集合和数据加载到内存查询才能生效。我们将在检索器部分处理。6. 文档加载、处理与向量化入库接下来我们模拟一些文档数据并将其处理、向量化后存入刚创建的集合。这里我们使用sentence-transformers本地模型生成向量避免调用外部 API。# file: ingest_documents.py from pymilvus import connections, Collection from langchain.text_splitter import RecursiveCharacterTextSplitter from sentence_transformers import SentenceTransformer import time import json # 1. 连接 Milvus connections.connect(hostlocalhost, port19530) collection Collection(tech_company_docs) # 2. 初始化文本分割器和嵌入模型 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个文本块的最大字符数 chunk_overlap50 # 块之间的重叠字符数保持上下文连贯 ) # 选择嵌入模型all-MiniLM-L6-v2 是一个轻量且效果不错的通用模型 embedding_model SentenceTransformer(all-MiniLM-L6-v2) # 注意该模型维度为384需要与创建集合时的 dim 参数一致。请将 create_collection.py 中的 dim 改为 384。 # 3. 模拟文档数据 documents [ { content: # 工程部后端服务部署规范 V2.1 本文档规定了所有微服务上线至生产环境的标准化流程。 核心要求包括1. 必须通过 CI/CD 流水线。2. 健康检查接口需返回 200。3. 日志必须聚合到中央 ELK。 适用部门工程部。文档类型技术规范。 , source: deploy_guide_v2.1.md, doc_type: tech_doc, department: engineering, created_at: 1672502400, # 2023-01-01 }, { content: Q4 产品规划会议纪要 时间2023-10-15。参会人Alice(产品), Bob(工程), Carol(设计)。 主要决议1. 优先开发智能报表模块。2. 用户体验优化延期至下季度。 下一步工程部需在两周内给出技术方案评审。 , source: q4_planning_meeting.txt, doc_type: meeting_minutes, department: product, created_at: 1697328000, # 2023-10-15 }, { content: HR 系统新员工使用指南 欢迎加入公司本指南将帮助你快速熟悉 HR 系统。 内容包括如何提交请假申请、查看薪资单、更新个人信息。 如有问题请联系人力资源部。 , source: hr_system_guide.pdf, doc_type: user_guide, department: hr, created_at: 1688169600, # 2023-07-01 }, # ... 可以添加更多模拟文档 ] # 4. 处理每个文档切分、向量化、准备插入数据 all_texts [] all_metadatas [] all_vectors [] for doc in documents: # 分割文本 chunks text_splitter.split_text(doc[content]) for chunk in chunks: all_texts.append(chunk) # 为每个 chunk 复制文档的元数据 metadata { source: doc[source], doc_type: doc[doc_type], department: doc[department], created_at: doc[created_at], } all_metadatas.append(metadata) # 生成向量嵌入 vector embedding_model.encode(chunk).tolist() # 转换为 list all_vectors.append(vector) # 5. 准备批量插入的数据字段顺序需与集合定义一致 entities [ all_vectors, # vector 字段 all_texts, # text 字段 [meta[source] for meta in all_metadatas], # source 字段 [meta[doc_type] for meta in all_metadatas], # doc_type 字段 [meta[department] for meta in all_metadatas], # department 字段 [meta[created_at] for meta in all_metadatas], # created_at 字段 ] # 6. 插入数据 insert_result collection.insert(entities) print(fInserted {len(all_texts)} text chunks into Milvus.) print(fInserted PKs: {insert_result.primary_keys[:5]}...) # 打印前5个主键 # 7. 确保数据持久化并加载到内存使可搜索 collection.flush() # 将内存中的数据写入磁盘 collection.load() # 将集合加载到内存 print(Data flushed and collection loaded. Ready for search.)关键点解释RecursiveCharacterTextSplitter是 LangChain 中常用的分割器能较好地按字符递归分割保持段落和句子的完整性。我们使用本地 SentenceTransformer 模型生成向量这避免了网络延迟和 API 成本适合离线或内部环境。确保其维度与集合定义的dim一致。collection.flush()确保插入操作被持久化。collection.load()将数据加载到内存这是执行搜索前的必要步骤。7. 构建支持 DQL 的自定义检索器这是本文最核心的部分。LangChain 内置的Milvus检索器可能不支持复杂的 DQL 表达式过滤。我们需要自定义一个检索器它能够接收一个过滤条件字符串并将其整合到 Milvus 的查询中。我们将创建一个继承自langchain.vectorstores.VectorStore的类并实现_similarity_search_with_score等方法。# file: custom_milvus_retriever.py from typing import Any, Dict, List, Optional, Tuple from langchain.vectorstores import VectorStore from langchain.schema import Document from pymilvus import Collection, connections import numpy as np class CustomMilvusVectorStore(VectorStore): 支持复杂 DQL 过滤条件的自定义 Milvus VectorStore. def __init__( self, collection_name: str, embedding_function: Any, # 例如 SentenceTransformer.encode 函数 connection_args: Dict[str, Any] {host: localhost, port: 19530}, **kwargs: Any, ): 初始化。 Args: collection_name: Milvus 集合名称。 embedding_function: 用于将文本转换为向量的函数。 connection_args: 连接 Milvus 的参数。 self.collection_name collection_name self.embedding_function embedding_function self.connection_args connection_args # 建立连接并获取集合对象 connections.connect(**self.connection_args) self.collection Collection(self.collection_name) # 确保集合已加载到内存 if not self.collection.has_index(): raise ValueError(fCollection {collection_name} has no index. Please create index first.) self.collection.load() # 获取集合的向量字段名假设只有一个向量字段 self.vector_field None for field in self.collection.schema.fields: if field.dtype.name FLOAT_VECTOR: self.vector_field field.name break if self.vector_field is None: raise ValueError(No vector field found in the collection schema.) # 获取主键字段名 self.primary_field None for field in self.collection.schema.fields: if field.is_primary: self.primary_field field.name break # 文本字段名我们假设名为 text self.text_field text super().__init__(**kwargs) def _build_search_params(self, expr: Optional[str] None) - Dict: 构建搜索参数支持 DQL 过滤表达式。 search_params { metric_type: L2, # 与创建索引时一致 params: {nprobe: 10}, # 搜索时探查的聚类中心数影响速度和精度 } # 如果提供了过滤表达式加入搜索参数 if expr: search_params[expr] expr return search_params def similarity_search_with_score_by_vector( self, embedding: List[float], k: int 4, expr: Optional[str] None, **kwargs: Any, ) - List[Tuple[Document, float]]: 根据向量进行相似度搜索并返回文档及其分数。 Args: embedding: 查询向量。 k: 返回的最相似结果数量。 expr: Milvus DQL 过滤表达式字符串。例如: doc_type tech_doc and department engineering Returns: 包含 (Document, similarity_score) 的列表。 search_params self._build_search_params(expr) # 执行搜索 results self.collection.search( data[embedding], anns_fieldself.vector_field, paramsearch_params, limitk, output_fields[self.text_field, source, doc_type, department, created_at], # 指定需要返回的字段 **kwargs, ) # 解析结果 ret [] for hits in results: for hit in hits: # hit.entity 包含了 output_fields 中指定的字段 metadata { source: hit.entity.get(source), doc_type: hit.entity.get(doc_type), department: hit.entity.get(department), created_at: hit.entity.get(created_at), id: hit.id, # 主键 } doc Document( page_contenthit.entity.get(self.text_field), metadatametadata ) # Milvus 的 score 是距离对于 L2 距离越小越相似。我们将其转换为相似度分数可选。 # 这里直接使用距离的负数使其越大表示越相似更符合常规认知。 similarity_score -hit.distance ret.append((doc, similarity_score)) return ret def similarity_search_with_score( self, query: str, k: int 4, expr: Optional[str] None, **kwargs: Any, ) - List[Tuple[Document, float]]: 根据查询文本进行相似度搜索。 # 将查询文本转换为向量 query_embedding self.embedding_function(query) if isinstance(query_embedding, np.ndarray): query_embedding query_embedding.tolist() return self.similarity_search_with_score_by_vector(query_embedding, kk, exprexpr, **kwargs) def similarity_search( self, query: str, k: int 4, expr: Optional[str] None, **kwargs: Any, ) - List[Document]: 根据查询文本进行相似度搜索只返回文档。 docs_and_scores self.similarity_search_with_score(query, kk, exprexpr, **kwargs) return [doc for doc, _ in docs_and_scores] def as_retriever(self, search_kwargs: Optional[Dict] None): 返回一个检索器对象可以传入自定义搜索参数如 expr。 from langchain.vectorstores.base import VectorStoreRetriever return VectorStoreRetriever(vectorstoreself, search_kwargssearch_kwargs or {}) # 必须实现 add_texts 方法为了继承 VectorStore这里简化处理 def add_texts(self, texts, metadatasNone, **kwargs): raise NotImplementedError(This custom class is for search only. Use ingest_documents.py for ingestion.) classmethod def from_texts(cls, texts, embedding, metadatasNone, **kwargs): raise NotImplementedError(This custom class is for search only. Use ingest_documents.py for ingestion.)核心逻辑解释初始化连接 Milvus加载集合并自动识别集合中的向量字段、主键字段和文本字段。_build_search_params方法构建搜索参数。最关键的一步是如果调用者传入了exprDQL 表达式就将其加入到搜索参数中。Milvus 的search接口的expr参数正是用于标量过滤。similarity_search_with_score_by_vector方法这是执行混合查询的核心。它接收一个向量、返回数量k和过滤表达式expr调用collection.search并传入expr。返回的结果会自动应用过滤条件。结果解析将 Milvus 返回的实体Entity转换为 LangChain 的Document对象并携带所有元数据。as_retriever方法返回一个检索器允许我们通过search_kwargs动态传入expr等参数这在与 LangChain Chain 集成时非常方便。8. 集成问答链与复杂查询实战现在我们将自定义的检索器接入 LangChain 的检索问答链RetrievalQA并执行几个具有复杂过滤条件的查询。# file: query_with_dql.py from custom_milvus_retriever import CustomMilvusVectorStore from sentence_transformers import SentenceTransformer from langchain.chains import RetrievalQA from langchain.llms import OpenAI # 或者使用 ChatOpenAI from langchain.prompts import PromptTemplate import os from dotenv import load_dotenv # 加载环境变量用于 OpenAI API Key load_dotenv() # 1. 初始化嵌入模型和向量库 embedding_model SentenceTransformer(all-MiniLM-L6-v2) def encode_function(text: str) - List[float]: return embedding_model.encode(text).tolist() vector_store CustomMilvusVectorStore( collection_nametech_company_docs, embedding_functionencode_function, connection_args{host: localhost, port: 19530} ) # 2. 创建支持 DQL 过滤的检索器 # 示例1检索器默认不带过滤 base_retriever vector_store.as_retriever(search_kwargs{k: 3}) print(--- 基础检索器创建成功 ---) # 3. 初始化 LLM这里使用 OpenAI GPT-3.5-turbo你需要设置 OPENAI_API_KEY from langchain_openai import ChatOpenAI llm ChatOpenAI(model_namegpt-3.5-turbo, temperature0) # 4. 构建一个简单的问答链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 将检索到的所有文档内容“塞”进上下文 retrieverbase_retriever, return_source_documentsTrue, # 返回源文档便于调试 verboseTrue, # 打印详细日志 ) # 5. 执行不同场景的查询 print(\n 场景 1: 无过滤普通语义搜索 ) query1 如何部署服务 result1 qa_chain.invoke({query: query1}) print(f问题: {query1}) print(f答案: {result1[result][:200]}...) # 打印前200字符 print(f参考来源: {[doc.metadata.get(source) for doc in result1[source_documents]]}) print(\n 场景 2: 只搜索工程部的技术文档 ) # 关键创建带有过滤表达式的检索器 engineering_tech_retriever vector_store.as_retriever( search_kwargs{ k: 3, expr: doc_type tech_doc and department engineering } ) qa_chain_engineering RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverengineering_tech_retriever, return_source_documentsTrue, ) query2 部署服务有什么规范 result2 qa_chain_engineering.invoke({query: query2}) print(f问题: {query2}) print(f答案: {result2[result][:200]}...) print(f参考来源: {[doc.metadata.get(source) for doc in result2[source_documents]]}) print(f文档类型: {[doc.metadata.get(doc_type) for doc in result2[source_documents]]}) print(\n 场景 3: 搜索产品部门最近的会议纪要时间过滤) # 假设“最近”指创建时间大于某个时间戳例如 2023年10月1日 (1696118400) recent_product_retriever vector_store.as_retriever( search_kwargs{ k: 3, expr: department product and doc_type meeting_minutes and created_at 1696118400 } ) # 也可以直接使用向量库的搜索方法进行更灵活的调试 print(直接使用向量库搜索进行调试:) docs vector_store.similarity_search( query产品规划, k2, exprdepartment product and doc_type meeting_minutes ) for i, doc in enumerate(docs): print(f 结果{i1}: {doc.page_content[:100]}... | 来源: {doc.metadata.get(source)}) print(\n 场景 4: 使用 in 操作符进行多值过滤 ) # 查询属于“工程部”或“产品部”的文档 multi_dept_retriever vector_store.as_retriever( search_kwargs{ k: 4, expr: department in [engineering, product] } ) qa_chain_multi RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievermulti_dept_retriever, ) query4 各部门接下来有什么计划 result4 qa_chain_multi.invoke({query: query4}) print(f问题: {query4}) print(f答案: {result4[result][:300]}...)运行与验证运行此脚本你将看到针对不同过滤条件系统返回了不同的答案和来源。例如对于“部署服务有什么规范”这个问题当使用工程部技术文档过滤时它应该只从《工程部后端服务部署规范》中寻找答案而不会混杂 HR 指南的内容。这证明了我们的 DQL 过滤是有效的。9. 常见问题与排查思路在实际集成中你可能会遇到以下问题问题现象可能原因排查方式解决方案连接 Milvus 失败Milvus 服务未启动端口错误网络问题。1.docker ps检查容器状态。2.telnet localhost 19530测试端口连通性。3. 检查连接参数host和port。确保 Milvus 容器正常运行并使用正确的连接参数。搜索时返回空结果集合未加载 (load()); 过滤表达式语法错误数据未插入成功。1. 检查是否执行了collection.load()。2. 在 Milvus 客户端如 Attu中直接执行 DQL 验证表达式。3. 检查插入操作是否成功并flush()。1. 确保搜索前调用load()。2. 简化表达式逐步调试。3. 确认数据插入流程无误。DQL 过滤表达式不生效表达式字段名错误字段类型不匹配运算符使用错误。1. 使用collection.schema打印字段名和类型。2. 对于字符串字段值需用单引号value。3. 检查 Milvus 官方文档确认 DQL 语法。1. 确保字段名与集合定义完全一致。2. 整数字段不要加引号字符串必须加。3. 使用and/or连接条件时注意括号。向量维度不匹配创建集合时定义的dim与嵌入模型输出的维度不一致。打印嵌入模型输出向量的长度len(embedding)。修改create_collection.py中的dim参数使其与模型维度一致并重建集合。检索速度慢数据量增大后未调整索引参数过滤字段未建索引。1. 使用collection.get_index_stats()查看索引情况。2. 对常用过滤字段创建二级索引如 TRIE。1. 调整nlist创建时和nprobe搜索时参数。2. 为过滤字段创建合适的索引。LangChain 链报错expr参数无效使用的 LangChain 内置 Milvus 集成版本可能不支持expr。检查是否使用了本文的自定义CustomMilvusVectorStore类。务必使用我们自定义的类它显式处理了expr参数。10. 最佳实践与工程建议将 LangChain 与 Milvus DQL 投入生产环境还需要注意以下几点10.1 索引策略优化向量索引对于千万级以上向量考虑使用HNSW或DISKANN索引以获得更好的查询性能。IVF_FLAT适合百万级数据。标量索引对频繁用于过滤的标量字段如doc_type,department,created_at务必创建索引TRIE用于字符串STL_SORT用于数值/时间。分区管理如果数据有明显的冷热区分或业务维度如按年份、地区使用 Milvus 的分区功能。查询时指定分区可以极大提升速度。10.2 查询性能调优nprobe参数在搜索参数params中设置。它代表搜索时探查的聚类中心数。值越大精度越高但速度越慢。需要在精度和延迟之间做权衡测试。分批查询如果一次需要查询大量向量使用batch_search接口。结果集大小合理设置limit即k值。在 RAG 中通常 3-5 个相关块已足够 LLM 生成优质答案获取过多反而可能引入噪声。10.3 应用层设计检索器封装在业务系统中不要每次都新建检索器。可以设计一个工厂类或配置中心根据不同的查询场景如不同部门、不同文档类型返回预配置了不同expr的检索器。表达式安全如果过滤条件来自用户输入例如前端下拉框务必进行严格的校验和转义防止 DQL 注入攻击。最佳实践是使用参数化查询如果 Milvus SDK 支持或在应用层构建安全的表达式白名单。错误处理与降级对 Milvus 查询做好异常捕获如超时、连接失败。可以考虑设置备用检索方案如回退到关键词搜索或缓存常用查询结果。10.4 数据维护元数据一致性确保插入的元数据格式规范、值域明确。例如doc_type字段的值应来自一个预定义的枚举列表。定时加载在 Milvus 集群中新插入的数据需要手动或定时调用load()才能被搜索到。生产环境需要规划数据加载策略如低峰期定时加载。版本管理当嵌入模型升级导致向量维度或语义空间变化时需要重建整个向量库。应有对应的数据迁移和版本回滚方案。通过本文的实战你已经掌握了利用 Milvus DQL 增强 LangChain 应用检索能力的核心方法。这不仅仅是语法上的学习更是设计思想的转变将向量数据库视为一个支持复杂查询的智能存储层而不仅仅是一个简单的相似度匹配工具。接下来你可以尝试将过滤条件动态化与用户身份、会话上下文结合构建出更精细、更个性化的 RAG 应用。