企业官网建设流程全解析

如果你正在构建一个基于大语言模型（LLM）的智能应用，比如一个能回答专业问题的知识库助手，那么你很可能已经遇到了一个核心难题：如何让 LLM 高效、准确地从你的私有数据中获取信息？

简单地将所有文档喂给 LLM 不仅成本高昂、速度慢，更关键的是，LLM 的上下文窗口有限，无法处理海量数据。这时，“检索增强生成”（RAG）技术成为了标准答案。而 RAG 的核心，在于一个能快速从海量向量中找出最相似项的“搜索引擎”——向量数据库。

在众多向量数据库中，Milvus 以其高性能、可扩展性和丰富的功能脱颖而出。但很多开发者在初步接触后，往往止步于简单的相似性搜索。当你的业务逻辑变得复杂，比如需要根据文档类型、作者、时间等多维度条件进行过滤，或者需要对搜索结果进行排序、聚合时，仅靠基础的search接口就显得力不从心了。

这正是DQL（Milvus 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）：使用>，<，==，in，and,or等运算符对非向量字段进行条件筛选。
• 混合查询（Hybrid Search）：将向量相似度搜索与标量过滤结合在一个查询中。
• 丰富的结果处理：支持对结果进行排序（ORDER BY）、限制（LIMIT）、输出字段选择等。

2.4 工作流程我们的目标流程如下图所示（概念性描述）：

1. 文档处理：原始文本经过切分（Text Splitter），变成一个个文本块（Chunk）。
2. 向量化：每个文本块通过嵌入模型（Embedding Model）转化为一个高维向量，并提取或赋予其元数据（如来源、类型、日期）。
3. 数据入库：将向量和元数据作为一个整体实体，写入 Milvus 集合。
4. 混合查询：用户提问时，先将问题转化为向量，然后构造一个 DQL 语句，该语句同时包含“在向量空间找相似”和“按元数据条件过滤”。
5. 生成答案：将检索到的最相关文本块作为上下文，与用户问题一起提交给 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-dotenv

3.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_KEY=your_openai_api_key_here" > .env

对应的 Python 代码中可以通过os.getenv(“OPENAI_API_KEY”)读取。

4. 项目结构与核心流程拆解

我们将构建一个名为tech_doc_qa的项目。其核心流程可以拆解为以下四个步骤，每一步我们都将详细实现：

1. 数据建模与集合创建：在 Milvus 中设计并创建一个符合我们业务需求的集合。
2. 文档加载与向量化入库：将本地知识文档处理成向量和元数据，并批量插入 Milvus。
3. 构建支持 DQL 的自定义检索器：这是本文的关键，我们将扩展 LangChain 的基类，使其能执行复杂的 DQL 查询。
4. 集成问答链进行测试：将自定义检索器接入 LangChain 的 QA 链，完成端到端的智能问答。

5. 数据建模与 Milvus 集合创建

首先，我们需要定义数据的结构。假设我们的科技公司文档有以下属性：

• id: 主键（自增或UUID）
• vector: 文本块的向量表示（例如 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(host='localhost', port='19530') # 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(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=dim), FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535), # 存储文本内容 FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=255), FieldSchema(name="doc_type", dtype=DataType.VARCHAR, max_length=100), FieldSchema(name="department", dtype=DataType.VARCHAR, max_length=100), FieldSchema(name="created_at", dtype=DataType.INT64), # 使用时间戳存储 ] # 5. 定义集合模式 schema = CollectionSchema(fields=fields, description="Tech company internal documents with metadata") # 6. 创建集合 collection = Collection(name=collection_name, schema=schema) # 7. 为向量字段和常用过滤字段创建索引，以加速搜索 index_params = { "index_type": "IVF_FLAT", # 一种高效的向量索引类型，适合中等规模数据集 "metric_type": "L2", # 距离度量方式，L2 欧氏距离 "params": {"nlist": 128}, # 聚类中心数，值越大精度越高但速度越慢 } collection.create_index(field_name="vector", index_params=index_params) # 为标量字段创建索引（可加速过滤查询） collection.create_index(field_name="doc_type", index_params={"index_type": "TRIE"}) collection.create_index(field_name="department", index_params={"index_type": "TRIE"}) collection.create_index(field_name="created_at", index_params={"index_type": "STL_SORT"}) print(f"Collection '{collection_name}' created successfully with indexes.") print("Remember to load the collection before searching:") print(f"collection.load()")

关键点解释：

• auto_id=True：让 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(host='localhost', port='19530') collection = Collection("tech_company_docs") # 2. 初始化文本分割器和嵌入模型 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个文本块的最大字符数 chunk_overlap=50 # 块之间的重叠字符数，保持上下文连贯 ) # 选择嵌入模型，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(f"Inserted {len(all_texts)} text chunks into Milvus.") print(f"Inserted 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(f"Collection {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_field=self.vector_field, param=search_params, limit=k, 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_content=hit.entity.get(self.text_field), metadata=metadata ) # 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, k=k, expr=expr, **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, k=k, expr=expr, **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(vectorstore=self, search_kwargs=search_kwargs or {}) # 必须实现 add_texts 方法（为了继承 VectorStore），这里简化处理 def add_texts(self, texts, metadatas=None, **kwargs): raise NotImplementedError("This custom class is for search only. Use ingest_documents.py for ingestion.") @classmethod def from_texts(cls, texts, embedding, metadatas=None, **kwargs): raise NotImplementedError("This custom class is for search only. Use ingest_documents.py for ingestion.")

核心逻辑解释：

1. 初始化：连接 Milvus，加载集合，并自动识别集合中的向量字段、主键字段和文本字段。
2. _build_search_params方法：构建搜索参数。最关键的一步是，如果调用者传入了expr（DQL 表达式），就将其加入到搜索参数中。Milvus 的search接口的expr参数正是用于标量过滤。
3. similarity_search_with_score_by_vector方法：这是执行混合查询的核心。它接收一个向量、返回数量k和过滤表达式expr，调用collection.search并传入expr。返回的结果会自动应用过滤条件。
4. 结果解析：将 Milvus 返回的实体（Entity）转换为 LangChain 的Document对象，并携带所有元数据。
5. 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_name="tech_company_docs", embedding_function=encode_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_name="gpt-3.5-turbo", temperature=0) # 4. 构建一个简单的问答链 qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 将检索到的所有文档内容“塞”进上下文 retriever=base_retriever, return_source_documents=True, # 返回源文档，便于调试 verbose=True, # 打印详细日志 ) # 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( llm=llm, chain_type="stuff", retriever=engineering_tech_retriever, return_source_documents=True, ) 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="产品规划", k=2, expr="department == 'product' and doc_type == 'meeting_minutes'" ) for i, doc in enumerate(docs): print(f" 结果{i+1}: {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( llm=llm, chain_type="stuff", retriever=multi_dept_retriever, ) query4 = "各部门接下来有什么计划？" result4 = qa_chain_multi.invoke({"query": query4}) print(f"问题: {query4}") print(f"答案: {result4['result'][:300]}...")

运行与验证：运行此脚本，你将看到针对不同过滤条件，系统返回了不同的答案和来源。例如，对于“部署服务有什么规范？”这个问题，当使用工程部技术文档过滤时，它应该只从《工程部后端服务部署规范》中寻找答案，而不会混杂 HR 指南的内容。这证明了我们的 DQL 过滤是有效的。

9. 常见问题与排查思路

在实际集成中，你可能会遇到以下问题：

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 应用。