在实际项目中当我们需要将海量的非结构化数据如文档、图片、音频转化为可查询、可分析的知识时向量数据库结合大语言模型LLM的检索增强生成RAG架构已成为主流方案。Milvus 作为一款高性能、云原生的开源向量数据库因其出色的扩展性和易用性常被选为向量存储的核心组件。而 LangChain 作为一个强大的 LLM 应用开发框架提供了连接 LLM、向量库和各种工具的统一接口。然而仅仅将 LangChain 的VectorStore接口与 Milvus 连接起来有时并不能满足复杂的业务查询需求例如需要根据元数据metadata进行精确过滤、对向量和标量进行混合查询、或者执行复杂的聚合分析。这时直接使用 Milvus 的数据查询语言DQL就变得至关重要。本文面向已经了解 LangChain 和 Milvus 基础用法但希望突破框架限制实现更精细、更高效数据查询的开发者。我们将深入探讨如何在 LangChain 项目中绕过其高层抽象直接集成和使用 Milvus DQL 来完成复杂的向量检索任务。你将学习到如何配置环境、建立连接、编写 DQL 表达式并将查询结果无缝集成回 LangChain 的 Agent 或 Chain 中构建一个既灵活又强大的智能检索系统。1. 理解 Milvus DQL 在 LangChain 项目中的定位在典型的 LangChain RAG 流程中我们通常使用Milvus类来自langchain.vectorstores来存储文档的向量嵌入Embeddings并通过其similarity_search方法进行语义检索。这个方法内部封装了 Milvus 的搜索逻辑对于简单的“根据文本找相似”场景已经足够。但它将查询的复杂性隐藏了起来。1.1 为什么需要直接使用 DQL当你的需求超越简单的相似度搜索时高层抽象就可能成为限制。以下场景是直接使用 DQL 的典型动机混合查询Hybrid Search你需要同时考虑向量的语义相似度和标量字段的过滤条件。例如“查找与‘机器学习’相关且发布于2023年以后、作者是‘张三’的论文”。这需要将向量相似度搜索与结构化字段的布尔过滤and,or,等结合起来。复杂元数据过滤Milvus 支持为每个向量点存储丰富的 JSON 格式元数据。DQL 允许你使用json_contains、array_contains等函数深入查询这些动态结构。例如查询所有“标签”数组中包含“AI”且“状态”字段为“已发布”的文档。聚合与分析除了检索 Top-K 相似项你可能还需要知道某个条件下的数据分布如统计不同类别文档的数量、计算向量的中心点等。这需要用到count、group by等聚合操作。性能调优与高级参数直接控制搜索参数如nprobe搜索的聚类单元数量、metric_type距离计算方式如 L2、IP、offset和limit分页可以针对特定数据和硬件进行精细优化以达到最佳的查询延迟和召回率。1.2 LangChain 抽象与 DQL 原语的关系可以把 LangChain 的VectorStore接口看作一个“自动挡汽车”它提供了便捷的similarity_search“油门”和“刹车”。而 Milvus DQL 则是“手动挡”让你直接操控“离合器”、“换挡杆”和“发动机转速”。在复杂的路况业务场景下“手动挡”能带来更高的操控性和效率。我们的目标不是抛弃 LangChain而是在 LangChain 的应用框架内在需要精细控制的地方巧妙地换入“手动模式”——即直接调用 Milvus 的 Python SDK 来执行 DQL然后将获取到的原始数据向量ID、距离、元数据转换回 LangChain 期望的Document对象格式供后续的 Chain 或 Agent 处理。2. 环境准备与项目依赖配置在开始编写 DQL 代码之前必须确保你的开发环境拥有正确版本的 Milvus 和 LangChain。版本不匹配是后续一切问题的根源。2.1 系统与 Milvus 部署要求Milvus 可以以单机Standalone或集群Cluster模式部署。对于开发和测试单机模式完全足够。操作系统Linux (Ubuntu 18.04 CentOS 7) macOS Windows (WSL2 推荐)。Docker Docker Compose这是部署 Milvus Standalone 最简便的方式。确保已安装。CPU支持 AVX 指令集现代 CPU 一般都支持。内存至少 8GB建议 16GB 以上。磁盘空间预留 20GB 以上空间。2.2 通过 Docker Compose 启动 Milvus Standalone这是最推荐的方式。在项目根目录或一个专用目录下创建docker-compose.yml文件。version: 3.5 services: etcd: container_name: milvus-etcd image: quay.io/coreos/etcd:v3.5.5 environment: - ETCD_AUTO_COMPACTION_MODErevision - ETCD_AUTO_COMPACTION_RETENTION1000 - ETCD_QUOTA_BACKEND_BYTES4294967296 - ETCD_SNAPSHOT_COUNT50000 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd command: etcd -advertise-client-urlshttp://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd minio: container_name: milvus-minio image: minio/minio:RELEASE.2023-03-20T20-16-18Z environment: MINIO_ACCESS_KEY: minioadmin MINIO_SECRET_KEY: minioadmin volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/minio:/minio_data command: minio server /minio_data healthcheck: test: [CMD, curl, -f, http://localhost:9000/minio/health/live] interval: 30s timeout: 20s retries: 3 standalone: container_name: milvus-standalone image: milvusdb/milvus:v2.3.3 command: [milvus, run, standalone] environment: ETCD_ENDPOINTS: etcd:2379 MINIO_ADDRESS: minio:9000 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus ports: - 19530:19530 - 9091:9091 depends_on: - etcd - minio networks: default: name: milvus保存文件后在终端执行docker-compose up -d等待所有容器启动完成。你可以通过docker-compose ps检查状态或访问http://localhost:9091查看 Milvus 的图形化管理界面Attu。2.3 Python 环境与依赖包安装创建一个新的 Python 虚拟环境如conda create -n milvus-langchain python3.9然后安装核心依赖。pip install pymilvus2.3.0 # Milvus Python SDK版本尽量与服务器一致 pip install langchain0.1.0 # LangChain 核心 pip install langchain-openai0.0.5 # 如果你使用 OpenAI 的 Embeddings 和 LLM pip install openai # OpenAI Python 客户端 # 可选用于文本分割和加载 pip install langchain-text-splitters pip install langchain-community # 社区贡献的组件关键版本对齐pymilvus的版本应与 Milvus 服务端版本本例为 v2.3.3保持大版本一致否则可能出现兼容性问题。LangChain 版本迭代较快文中版本为示例实际安装时请查阅官方文档确认稳定版本。3. 项目初始化连接 Milvus 与准备数据在编写 DQL 之前我们需要先建立与 Milvus 的连接并创建一个包含向量和丰富元数据的集合Collection作为我们后续查询的“靶场”。3.1 建立 Milvus 连接创建一个 Python 脚本例如init_milvus.py。from pymilvus import connections, utility, FieldSchema, CollectionSchema, DataType, Collection import numpy as np # 1. 连接到 Milvus 服务端 # 默认端口 19530 如果部署在其他机器或端口请修改 host 和 port connections.connect(aliasdefault, hostlocalhost, port19530) # 2. 检查连接是否成功 print(utility.get_server_version()) # 输出类似 (2, 3, 3, 2.3.3) # 3. 定义集合的字段 Schema # 假设我们存储技术博客每个文档有ID、向量、文本内容、作者、标签、发布日期和阅读量 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim768), # 假设使用 768 维向量 FieldSchema(namecontent, dtypeDataType.VARCHAR, max_length65535), FieldSchema(nameauthor, dtypeDataType.VARCHAR, max_length255), FieldSchema(nametags, dtypeDataType.ARRAY, element_typeDataType.VARCHAR, max_capacity50, max_length100), FieldSchema(namepublish_date, dtypeDataType.INT64), # 使用时间戳存储 FieldSchema(nameview_count, dtypeDataType.INT64), ] # 4. 定义集合 Schema并指定向量索引的字段 schema CollectionSchema(fieldsfields, descriptionA collection of technical blog posts) collection_name tech_blogs # 5. 如果集合已存在则删除仅用于演示生产环境慎用 if utility.has_collection(collection_name): utility.drop_collection(collection_name) # 6. 创建集合 collection Collection(namecollection_name, schemaschema) print(fCollection {collection_name} created successfully.) # 7. 为向量字段创建索引这是高效相似性搜索的前提 index_params { index_type: IVF_FLAT, # 一种基于量化的索引适合平衡速度和精度 metric_type: L2, # 使用欧氏距离 (L2) params: {nlist: 128}, # 聚类中心数值越大精度越高但速度越慢 } collection.create_index(field_nameembedding, index_paramsindex_params) print(Index on embedding field created successfully.) # 8. 加载集合到内存准备接收数据或进行搜索 collection.load() print(fCollection {collection_name} loaded.)运行此脚本确保没有报错。这为你创建了一个名为tech_blogs的集合并建立了向量索引。3.2 插入示例数据接下来我们插入一些模拟数据。在实际项目中这些数据通常来自你的文档处理流水线。# 接上面的代码或新建一个脚本 import random import time from datetime import datetime # 生成模拟数据 num_entities 1000 # 插入1000条数据 dim 768 # 模拟向量数据 (随机生成实际应使用 Embedding 模型生成) embeddings [[random.random() for _ in range(dim)] for _ in range(num_entities)] # 模拟文本内容 sample_contents [ 深入理解 LangChain 的 Agent 执行机制, Milvus 向量数据库的安装与性能调优指南, 使用 Flask 构建基于 RAG 的智能问答系统, 对比 LangChain 与 LangGraph 在复杂工作流中的应用, Python 异步编程在 AI 应用中的实践, Docker 容器化部署机器学习模型的完整流程, 如何利用 Embedding 技术提升搜索相关性, CentOS 7 系统下常见服务配置与排错, ] authors [Alice, Bob, Charlie, Diana] all_tags [AI, Database, Tutorial, DevOps, Python, MachineLearning, Web, System] # 构造插入数据 data [ embeddings, # 对应 embedding 字段 [random.choice(sample_contents) for _ in range(num_entities)], # content [random.choice(authors) for _ in range(num_entities)], # author [[random.choice(all_tags) for _ in range(random.randint(1, 3))] for _ in range(num_entities)], # tags 每个文档1-3个随机标签 [int(time.time()) - random.randint(0, 365*24*60*60) for _ in range(num_entities)], # publish_date 过去一年内的随机时间戳 [random.randint(100, 100000) for _ in range(num_entities)], # view_count ] # 注意id 字段是自增的所以我们不需要提供。插入的数据顺序必须与 fields 定义一致除了 id。 # 对应关系: [embedding, content, author, tags, publish_date, view_count] collection.insert(data) # 刷新数据确保插入的数据立即可见对于小批量插入flush 不是必须但建议 collection.flush() print(fInserted {collection.num_entities} entities into {collection_name}.)现在你的 Milvus 数据库中已经有了一个包含向量和多样化元数据的测试集合。4. 掌握 Milvus DQL 核心语法与实战DQL 主要通过collection.search()和collection.query()方法执行。search()用于基于向量的相似性搜索query()用于基于标量字段的精确查询。两者都可以结合丰富的表达式Expr进行过滤。4.1 基础向量相似性搜索 (search)这是最核心的操作。我们首先构造一个查询向量。# 假设我们有一个查询文本 “如何学习 LangChain” # 在实际项目中你需要用相同的 Embedding 模型将查询文本转化为向量 # 这里我们随机生成一个查询向量作为示例 query_embedding [random.random() for _ in range(dim)] # 定义搜索参数 search_params { metric_type: L2, # 必须与索引的 metric_type 一致 params: {nprobe: 10}, # 搜索的聚类单元数影响速度和精度 } # 执行搜索 results collection.search( data[query_embedding], # 查询向量列表支持批量查询 anns_fieldembedding, # 要搜索的向量字段名 paramsearch_params, # 搜索参数 limit5, # 返回最相似的 K 个结果 exprNone, # 过滤表达式暂时为空 output_fields[content, author, tags, view_count], # 指定返回的元数据字段 consistency_levelStrong # 一致性级别 ) # 解析结果 for i, hits in enumerate(results): print(fSearch results for query #{i}:) for hit in hits: print(f ID: {hit.id}, Distance: {hit.distance:.4f}) print(f Content: {hit.entity.get(content)}) print(f Author: {hit.entity.get(author)}) print(f Tags: {hit.entity.get(tags)}) print(f Views: {hit.entity.get(view_count)}) print(- * 20)4.2 引入过滤表达式 (Expr)Expr 是 DQL 的灵魂它允许你在搜索或查询时添加复杂的过滤条件。表达式是一个字符串遵循类 SQL 的语法。场景一过滤作者和发布时间# 查找作者是 “Alice” 且在最近30天内发布的文档 import time current_timestamp int(time.time()) thirty_days_ago current_timestamp - 30 * 24 * 60 * 60 expr fauthor Alice and publish_date {thirty_days_ago} results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limit5, exprexpr, # 应用过滤表达式 output_fields[content, author, publish_date], )场景二过滤数组字段标签# 查找标签中包含 “AI” 或 “Python” 的文档 # json_contains 用于检查数组是否包含某个元素 expr json_contains(tags, AI) or json_contains(tags, Python) results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limit5, exprexpr, output_fields[content, tags], )场景三混合条件与范围查询# 查找阅读量超过1000且标签包含“Tutorial”且不是“Bob”写的文档 expr view_count 1000 and json_contains(tags, Tutorial) and author ! Bob results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limit5, exprexpr, output_fields[content, author, view_count, tags], )4.3 纯标量查询 (query)当你不需要向量相似度只想根据元数据查询时使用query更高效。# 查询所有作者是 “Charlie” 的文档按发布时间倒序排列取前10条 expr author Charlie query_results collection.query( exprexpr, output_fields[id, content, author, publish_date], limit10, # 排序order_by 字段 asc(升序) 或 desc(降序) order_by[(publish_date, desc)] ) for result in query_results: print(fID: {result[id]}, Content: {result[content][:50]}..., Date: {datetime.fromtimestamp(result[publish_date]).strftime(%Y-%m-%d)})5. 将 DQL 集成到 LangChain 工作流中直接使用 DQL 获取到原始数据后我们需要将其“翻译”回 LangChain 的Document对象以便后续的RetrievalQA、ConversationalRetrievalChain或Agent使用。5.1 构建一个自定义的 LangChain RetrieverRetriever是 LangChain 中用于获取相关文档的抽象接口。我们可以创建一个自定义的 Retriever内部使用我们编写的 DQL 进行搜索。from typing import List, Optional, Any, Dict from pymilvus import Collection from langchain.schema import BaseRetriever, Document from langchain.embeddings.base import Embeddings import numpy as np class MilvusDQLRetriever(BaseRetriever): 一个使用 Milvus DQL 进行混合检索的自定义 Retriever。 def __init__( self, collection: Collection, embedding_model: Embeddings, search_params: Dict[str, Any], filter_expr: Optional[str] None, top_k: int 5, output_fields: Optional[List[str]] None ): self.collection collection self.embedding_model embedding_model self.search_params search_params self.default_filter_expr filter_expr self.top_k top_k # 默认输出字段确保包含文本内容字段 self.default_output_fields output_fields or [id, content, author, tags] def _get_relevant_documents(self, query: str, *, filter_expr: Optional[str] None) - List[Document]: 核心检索方法。 # 1. 将查询文本转换为向量 query_embedding self.embedding_model.embed_query(query) # 2. 确定过滤表达式优先使用传入的否则用默认的 expr_to_use filter_expr if filter_expr is not None else self.default_filter_expr # 3. 执行 Milvus DQL 搜索 search_results self.collection.search( data[query_embedding], anns_fieldembedding, paramself.search_params, limitself.top_k, exprexpr_to_use, output_fieldsself.default_output_fields, ) # 4. 将结果转换为 LangChain Document 对象 docs [] # search_results 是一个列表每个元素对应一个查询向量我们只有一个 for hits in search_results: for hit in hits: # 构建 Document 的元数据 metadata {field: hit.entity.get(field) for field in self.default_output_fields if field ! content} metadata[id] hit.id metadata[distance] hit.distance # 相似度距离 # 假设 content 字段存储了文档的原始文本 page_content hit.entity.get(content, ) doc Document(page_contentpage_content, metadatametadata) docs.append(doc) return docs async def _aget_relevant_documents(self, query: str) - List[Document]: # 异步版本可以根据需要实现 raise NotImplementedError(Async retrieval not implemented for this retriever.)5.2 在 LangChain Chain 中使用自定义 Retriever现在我们可以像使用任何其他 LangChain Retriever 一样使用它。from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate # 1. 初始化 Embedding 模型和 LLM # 注意这里的 Embedding 模型必须与插入数据时使用的模型一致 embeddings OpenAIEmbeddings(modeltext-embedding-3-small, openai_api_keyyour-api-key) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyyour-api-key) # 2. 确保集合已加载 tech_blogs_collection Collection(tech_blogs) tech_blogs_collection.load() # 3. 创建我们的自定义 Retriever search_params {metric_type: L2, params: {nprobe: 10}} retriever MilvusDQLRetriever( collectiontech_blogs_collection, embedding_modelembeddings, search_paramssearch_params, filter_exprview_count 500, # 默认只检索阅读量大于500的优质文章 top_k3, output_fields[id, content, author, tags, view_count] ) # 4. 创建一个简单的问答链 prompt_template 基于以下上下文片段回答最后的问题。如果你不知道答案就说你不知道不要编造答案。 上下文 {context} 问题{question} 有帮助的答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 将检索到的所有文档内容“塞”进上下文 retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue # 返回源文档用于调试 ) # 5. 进行查询 question 有哪些关于 LangChain 的教程 result qa_chain.invoke({query: question}) print(问题, question) print(答案, result[result]) print(\n--- 检索到的源文档 ---) for i, doc in enumerate(result[source_documents]): print(f\n文档 {i1}:) print(f 内容摘要: {doc.page_content[:200]}...) print(f 元数据: {doc.metadata})通过这种方式我们成功地将复杂的 Milvus DQL 查询能力注入到了 LangChain 的流程中。你可以在调用qa_chain时动态地修改filter_expr实现高度定制化的检索。6. 高级技巧与性能优化6.1 动态表达式构建在实际应用中过滤条件往往由用户输入或业务逻辑动态生成。你需要安全地构建表达式字符串。def build_filter_expr(author: Optional[str] None, min_views: Optional[int] None, required_tags: List[str] None) - Optional[str]: 安全地构建 Milvus 过滤表达式。 expr_parts [] if author: # 注意字符串值需要用双引号包裹 expr_parts.append(fauthor {author}) if min_views is not None: expr_parts.append(fview_count {min_views}) if required_tags: tag_exprs [fjson_contains(tags, {tag}) for tag in required_tags] expr_parts.append(f({ and .join(tag_exprs)})) # 要求包含所有指定标签 if not expr_parts: return None return and .join(expr_parts) # 使用示例 expr build_filter_expr(authorAlice, min_views1000, required_tags[AI, Tutorial]) print(expr) # 输出 author Alice and view_count 1000 and (json_contains(tags, AI) and json_contains(tags, Tutorial))6.2 搜索参数调优search_params对查询性能和精度有巨大影响。metric_type必须与创建索引时一致。L2欧氏距离和IP内积是最常用的。对于 OpenAI 的text-embedding-ada-002等模型通常使用余弦相似度而 Milvus 的IP在向量标准化后等价于余弦相似度。nprobe在IVF类索引中它代表搜索的聚类中心数量。值越大搜索越精确但耗时越长。通常需要在精度和速度之间权衡通过实验确定。对于百万级数据nprobe在 10 到 256 之间是常见范围。ef或search_k在HNSW索引中类似的参数控制搜索的广度。建议对不同参数进行基准测试记录查询延迟和召回率。6.3 分页与游标查询对于大量结果的查询需要使用offset和limit或者 Milvus 2.3 支持的游标Cursor进行分页。# 使用 offset 和 limit 进行简单分页可能慢适合小数据量 expr author Bob page_size 20 page_num 2 offset (page_num - 1) * page_size results collection.query( exprexpr, output_fields[id, content], limitpage_size, offsetoffset, # 指定偏移量 order_by[(publish_date, desc)] ) # 对于大数据集使用游标更高效需要 Milvus 2.3 并启用该功能 # 首次查询 res collection.query_iterator( exprexpr, batch_sizepage_size, output_fields[id, content], order_by[(publish_date, desc)] ) first_batch res.next() # 获取游标用于下一次查询 cursor res.cursor res.close() # 后续查询使用游标 res2 collection.query_iterator( exprexpr, batch_sizepage_size, output_fields[id, content], cursorcursor, order_by[(publish_date, desc)] ) second_batch res2.next()7. 常见问题排查与最佳实践7.1 常见错误与解决方案问题现象可能原因检查与解决步骤连接失败 (ConnectError)Milvus 服务未启动网络或端口不通版本不兼容。1. 运行docker-compose ps检查容器状态。2. 使用telnet localhost 19530测试端口。3. 确认pymilvus与 Milvus 服务器版本兼容。集合不存在 (CollectionNotExistException)集合名称拼写错误集合未被创建。1. 使用utility.list_collections()查看所有集合。2. 检查创建集合的代码是否成功执行。搜索时返回空结果过滤表达式Expr太严格查询向量与数据分布差异大索引未构建或未加载。1. 先不加expr进行搜索看是否有结果。2. 检查expr语法是否正确字段名和值类型是否匹配。3. 确认集合已load()。4. 确认查询向量与插入向量的维度dim一致。查询性能慢数据量过大nprobe等参数设置过高未建立合适索引硬件资源不足。1. 为标量过滤字段如publish_date,author创建二级索引 (create_index)。2. 降低nprobe值。3. 检查服务器 CPU、内存和磁盘 I/O。json_contains过滤数组不生效数组字段定义或数据格式问题表达式语法错误。1. 确认字段定义为ARRAY类型。2. 确认插入的数据是列表格式如[AI, Python]。3. 表达式应为json_contains(tags, AI)注意字段名和值的大小写。返回字段缺失或为Noneoutput_fields未指定或指定错误该字段在某些条目中确实为null。1. 在output_fields中明确列出需要的字段名。2. 使用collection.query并指定output_fields检查原始数据。7.2 生产环境最佳实践连接管理不要为每次查询创建新连接。使用连接池或在应用生命周期内保持一个稳定的连接。对于 Web 服务可以在启动时连接并加载集合。异常处理对search和query操作进行完善的异常捕获try-except记录日志并设计降级策略如返回空结果或使用缓存。监控与指标监控 Milvus 集群的健康状态CPU、内存、磁盘、查询延迟QPS、P99 latency和错误率。Milvus 提供了 Prometheus 指标接口。数据一致性根据业务需求选择合适的一致性级别Strong、Bounded、Eventually。强一致性影响性能最终一致性提供更高吞吐。索引策略根据数据规模和查询模式选择合适的向量索引如IVF_FLAT,IVF_SQ8,HNSW和标量字段索引。在插入大量数据前创建索引比之后创建更高效。配置外置将 Milvus 连接地址、端口、集合名、搜索参数等写入配置文件如config.yaml或环境变量避免硬编码。版本控制严格锁定pymilvus和 Milvus 服务器版本并在升级前在测试环境充分验证。7.3 下一步扩展方向掌握了基础的 DQL 集成后你可以进一步探索与 LangChain Agent 结合让 Agent 根据用户问题动态生成过滤表达式实现更智能的检索。例如用户问“张三最近写的 AI 文章”Agent 解析出“作者张三”和“标签包含 AI”并调用你的自定义 Retriever。多向量检索一个文档包含多个向量如摘要向量、段落向量。探索 Milvus 的多向量字段或父子集合Collection-Partition来支持此类场景。混合检索Hybrid Search分数融合将向量相似度分数距离与元数据相关性分数如关键词匹配度、热度分数进行加权融合得到最终排序。使用 Milvus 的 RESTful 或 gRPC 接口在非 Python 环境或微服务架构中通过 HTTP/gRPC 直接调用 Milvus 的搜索 API。通过将 Milvus DQL 的强大查询能力与 LangChain 的灵活应用框架相结合你可以构建出远超简单语义搜索的复杂、高效且可靠的智能检索系统。关键在于理解两者的边界在合适的层面进行交互从而充分发挥各自的技术优势。
在实际项目中当我们需要将海量的非结构化数据如文档、图片、音频转化为可查询、可分析的知识时向量数据库结合大语言模型LLM的检索增强生成RAG架构已成为主流方案。Milvus 作为一款高性能、云原生的开源向量数据库因其出色的扩展性和易用性常被选为向量存储的核心组件。而 LangChain 作为一个强大的 LLM 应用开发框架提供了连接 LLM、向量库和各种工具的统一接口。然而仅仅将 LangChain 的VectorStore接口与 Milvus 连接起来有时并不能满足复杂的业务查询需求例如需要根据元数据metadata进行精确过滤、对向量和标量进行混合查询、或者执行复杂的聚合分析。这时直接使用 Milvus 的数据查询语言DQL就变得至关重要。本文面向已经了解 LangChain 和 Milvus 基础用法但希望突破框架限制实现更精细、更高效数据查询的开发者。我们将深入探讨如何在 LangChain 项目中绕过其高层抽象直接集成和使用 Milvus DQL 来完成复杂的向量检索任务。你将学习到如何配置环境、建立连接、编写 DQL 表达式并将查询结果无缝集成回 LangChain 的 Agent 或 Chain 中构建一个既灵活又强大的智能检索系统。1. 理解 Milvus DQL 在 LangChain 项目中的定位在典型的 LangChain RAG 流程中我们通常使用Milvus类来自langchain.vectorstores来存储文档的向量嵌入Embeddings并通过其similarity_search方法进行语义检索。这个方法内部封装了 Milvus 的搜索逻辑对于简单的“根据文本找相似”场景已经足够。但它将查询的复杂性隐藏了起来。1.1 为什么需要直接使用 DQL当你的需求超越简单的相似度搜索时高层抽象就可能成为限制。以下场景是直接使用 DQL 的典型动机混合查询Hybrid Search你需要同时考虑向量的语义相似度和标量字段的过滤条件。例如“查找与‘机器学习’相关且发布于2023年以后、作者是‘张三’的论文”。这需要将向量相似度搜索与结构化字段的布尔过滤and,or,等结合起来。复杂元数据过滤Milvus 支持为每个向量点存储丰富的 JSON 格式元数据。DQL 允许你使用json_contains、array_contains等函数深入查询这些动态结构。例如查询所有“标签”数组中包含“AI”且“状态”字段为“已发布”的文档。聚合与分析除了检索 Top-K 相似项你可能还需要知道某个条件下的数据分布如统计不同类别文档的数量、计算向量的中心点等。这需要用到count、group by等聚合操作。性能调优与高级参数直接控制搜索参数如nprobe搜索的聚类单元数量、metric_type距离计算方式如 L2、IP、offset和limit分页可以针对特定数据和硬件进行精细优化以达到最佳的查询延迟和召回率。1.2 LangChain 抽象与 DQL 原语的关系可以把 LangChain 的VectorStore接口看作一个“自动挡汽车”它提供了便捷的similarity_search“油门”和“刹车”。而 Milvus DQL 则是“手动挡”让你直接操控“离合器”、“换挡杆”和“发动机转速”。在复杂的路况业务场景下“手动挡”能带来更高的操控性和效率。我们的目标不是抛弃 LangChain而是在 LangChain 的应用框架内在需要精细控制的地方巧妙地换入“手动模式”——即直接调用 Milvus 的 Python SDK 来执行 DQL然后将获取到的原始数据向量ID、距离、元数据转换回 LangChain 期望的Document对象格式供后续的 Chain 或 Agent 处理。2. 环境准备与项目依赖配置在开始编写 DQL 代码之前必须确保你的开发环境拥有正确版本的 Milvus 和 LangChain。版本不匹配是后续一切问题的根源。2.1 系统与 Milvus 部署要求Milvus 可以以单机Standalone或集群Cluster模式部署。对于开发和测试单机模式完全足够。操作系统Linux (Ubuntu 18.04 CentOS 7) macOS Windows (WSL2 推荐)。Docker Docker Compose这是部署 Milvus Standalone 最简便的方式。确保已安装。CPU支持 AVX 指令集现代 CPU 一般都支持。内存至少 8GB建议 16GB 以上。磁盘空间预留 20GB 以上空间。2.2 通过 Docker Compose 启动 Milvus Standalone这是最推荐的方式。在项目根目录或一个专用目录下创建docker-compose.yml文件。version: 3.5 services: etcd: container_name: milvus-etcd image: quay.io/coreos/etcd:v3.5.5 environment: - ETCD_AUTO_COMPACTION_MODErevision - ETCD_AUTO_COMPACTION_RETENTION1000 - ETCD_QUOTA_BACKEND_BYTES4294967296 - ETCD_SNAPSHOT_COUNT50000 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd command: etcd -advertise-client-urlshttp://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd minio: container_name: milvus-minio image: minio/minio:RELEASE.2023-03-20T20-16-18Z environment: MINIO_ACCESS_KEY: minioadmin MINIO_SECRET_KEY: minioadmin volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/minio:/minio_data command: minio server /minio_data healthcheck: test: [CMD, curl, -f, http://localhost:9000/minio/health/live] interval: 30s timeout: 20s retries: 3 standalone: container_name: milvus-standalone image: milvusdb/milvus:v2.3.3 command: [milvus, run, standalone] environment: ETCD_ENDPOINTS: etcd:2379 MINIO_ADDRESS: minio:9000 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus ports: - 19530:19530 - 9091:9091 depends_on: - etcd - minio networks: default: name: milvus保存文件后在终端执行docker-compose up -d等待所有容器启动完成。你可以通过docker-compose ps检查状态或访问http://localhost:9091查看 Milvus 的图形化管理界面Attu。2.3 Python 环境与依赖包安装创建一个新的 Python 虚拟环境如conda create -n milvus-langchain python3.9然后安装核心依赖。pip install pymilvus2.3.0 # Milvus Python SDK版本尽量与服务器一致 pip install langchain0.1.0 # LangChain 核心 pip install langchain-openai0.0.5 # 如果你使用 OpenAI 的 Embeddings 和 LLM pip install openai # OpenAI Python 客户端 # 可选用于文本分割和加载 pip install langchain-text-splitters pip install langchain-community # 社区贡献的组件关键版本对齐pymilvus的版本应与 Milvus 服务端版本本例为 v2.3.3保持大版本一致否则可能出现兼容性问题。LangChain 版本迭代较快文中版本为示例实际安装时请查阅官方文档确认稳定版本。3. 项目初始化连接 Milvus 与准备数据在编写 DQL 之前我们需要先建立与 Milvus 的连接并创建一个包含向量和丰富元数据的集合Collection作为我们后续查询的“靶场”。3.1 建立 Milvus 连接创建一个 Python 脚本例如init_milvus.py。from pymilvus import connections, utility, FieldSchema, CollectionSchema, DataType, Collection import numpy as np # 1. 连接到 Milvus 服务端 # 默认端口 19530 如果部署在其他机器或端口请修改 host 和 port connections.connect(aliasdefault, hostlocalhost, port19530) # 2. 检查连接是否成功 print(utility.get_server_version()) # 输出类似 (2, 3, 3, 2.3.3) # 3. 定义集合的字段 Schema # 假设我们存储技术博客每个文档有ID、向量、文本内容、作者、标签、发布日期和阅读量 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim768), # 假设使用 768 维向量 FieldSchema(namecontent, dtypeDataType.VARCHAR, max_length65535), FieldSchema(nameauthor, dtypeDataType.VARCHAR, max_length255), FieldSchema(nametags, dtypeDataType.ARRAY, element_typeDataType.VARCHAR, max_capacity50, max_length100), FieldSchema(namepublish_date, dtypeDataType.INT64), # 使用时间戳存储 FieldSchema(nameview_count, dtypeDataType.INT64), ] # 4. 定义集合 Schema并指定向量索引的字段 schema CollectionSchema(fieldsfields, descriptionA collection of technical blog posts) collection_name tech_blogs # 5. 如果集合已存在则删除仅用于演示生产环境慎用 if utility.has_collection(collection_name): utility.drop_collection(collection_name) # 6. 创建集合 collection Collection(namecollection_name, schemaschema) print(fCollection {collection_name} created successfully.) # 7. 为向量字段创建索引这是高效相似性搜索的前提 index_params { index_type: IVF_FLAT, # 一种基于量化的索引适合平衡速度和精度 metric_type: L2, # 使用欧氏距离 (L2) params: {nlist: 128}, # 聚类中心数值越大精度越高但速度越慢 } collection.create_index(field_nameembedding, index_paramsindex_params) print(Index on embedding field created successfully.) # 8. 加载集合到内存准备接收数据或进行搜索 collection.load() print(fCollection {collection_name} loaded.)运行此脚本确保没有报错。这为你创建了一个名为tech_blogs的集合并建立了向量索引。3.2 插入示例数据接下来我们插入一些模拟数据。在实际项目中这些数据通常来自你的文档处理流水线。# 接上面的代码或新建一个脚本 import random import time from datetime import datetime # 生成模拟数据 num_entities 1000 # 插入1000条数据 dim 768 # 模拟向量数据 (随机生成实际应使用 Embedding 模型生成) embeddings [[random.random() for _ in range(dim)] for _ in range(num_entities)] # 模拟文本内容 sample_contents [ 深入理解 LangChain 的 Agent 执行机制, Milvus 向量数据库的安装与性能调优指南, 使用 Flask 构建基于 RAG 的智能问答系统, 对比 LangChain 与 LangGraph 在复杂工作流中的应用, Python 异步编程在 AI 应用中的实践, Docker 容器化部署机器学习模型的完整流程, 如何利用 Embedding 技术提升搜索相关性, CentOS 7 系统下常见服务配置与排错, ] authors [Alice, Bob, Charlie, Diana] all_tags [AI, Database, Tutorial, DevOps, Python, MachineLearning, Web, System] # 构造插入数据 data [ embeddings, # 对应 embedding 字段 [random.choice(sample_contents) for _ in range(num_entities)], # content [random.choice(authors) for _ in range(num_entities)], # author [[random.choice(all_tags) for _ in range(random.randint(1, 3))] for _ in range(num_entities)], # tags 每个文档1-3个随机标签 [int(time.time()) - random.randint(0, 365*24*60*60) for _ in range(num_entities)], # publish_date 过去一年内的随机时间戳 [random.randint(100, 100000) for _ in range(num_entities)], # view_count ] # 注意id 字段是自增的所以我们不需要提供。插入的数据顺序必须与 fields 定义一致除了 id。 # 对应关系: [embedding, content, author, tags, publish_date, view_count] collection.insert(data) # 刷新数据确保插入的数据立即可见对于小批量插入flush 不是必须但建议 collection.flush() print(fInserted {collection.num_entities} entities into {collection_name}.)现在你的 Milvus 数据库中已经有了一个包含向量和多样化元数据的测试集合。4. 掌握 Milvus DQL 核心语法与实战DQL 主要通过collection.search()和collection.query()方法执行。search()用于基于向量的相似性搜索query()用于基于标量字段的精确查询。两者都可以结合丰富的表达式Expr进行过滤。4.1 基础向量相似性搜索 (search)这是最核心的操作。我们首先构造一个查询向量。# 假设我们有一个查询文本 “如何学习 LangChain” # 在实际项目中你需要用相同的 Embedding 模型将查询文本转化为向量 # 这里我们随机生成一个查询向量作为示例 query_embedding [random.random() for _ in range(dim)] # 定义搜索参数 search_params { metric_type: L2, # 必须与索引的 metric_type 一致 params: {nprobe: 10}, # 搜索的聚类单元数影响速度和精度 } # 执行搜索 results collection.search( data[query_embedding], # 查询向量列表支持批量查询 anns_fieldembedding, # 要搜索的向量字段名 paramsearch_params, # 搜索参数 limit5, # 返回最相似的 K 个结果 exprNone, # 过滤表达式暂时为空 output_fields[content, author, tags, view_count], # 指定返回的元数据字段 consistency_levelStrong # 一致性级别 ) # 解析结果 for i, hits in enumerate(results): print(fSearch results for query #{i}:) for hit in hits: print(f ID: {hit.id}, Distance: {hit.distance:.4f}) print(f Content: {hit.entity.get(content)}) print(f Author: {hit.entity.get(author)}) print(f Tags: {hit.entity.get(tags)}) print(f Views: {hit.entity.get(view_count)}) print(- * 20)4.2 引入过滤表达式 (Expr)Expr 是 DQL 的灵魂它允许你在搜索或查询时添加复杂的过滤条件。表达式是一个字符串遵循类 SQL 的语法。场景一过滤作者和发布时间# 查找作者是 “Alice” 且在最近30天内发布的文档 import time current_timestamp int(time.time()) thirty_days_ago current_timestamp - 30 * 24 * 60 * 60 expr fauthor Alice and publish_date {thirty_days_ago} results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limit5, exprexpr, # 应用过滤表达式 output_fields[content, author, publish_date], )场景二过滤数组字段标签# 查找标签中包含 “AI” 或 “Python” 的文档 # json_contains 用于检查数组是否包含某个元素 expr json_contains(tags, AI) or json_contains(tags, Python) results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limit5, exprexpr, output_fields[content, tags], )场景三混合条件与范围查询# 查找阅读量超过1000且标签包含“Tutorial”且不是“Bob”写的文档 expr view_count 1000 and json_contains(tags, Tutorial) and author ! Bob results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limit5, exprexpr, output_fields[content, author, view_count, tags], )4.3 纯标量查询 (query)当你不需要向量相似度只想根据元数据查询时使用query更高效。# 查询所有作者是 “Charlie” 的文档按发布时间倒序排列取前10条 expr author Charlie query_results collection.query( exprexpr, output_fields[id, content, author, publish_date], limit10, # 排序order_by 字段 asc(升序) 或 desc(降序) order_by[(publish_date, desc)] ) for result in query_results: print(fID: {result[id]}, Content: {result[content][:50]}..., Date: {datetime.fromtimestamp(result[publish_date]).strftime(%Y-%m-%d)})5. 将 DQL 集成到 LangChain 工作流中直接使用 DQL 获取到原始数据后我们需要将其“翻译”回 LangChain 的Document对象以便后续的RetrievalQA、ConversationalRetrievalChain或Agent使用。5.1 构建一个自定义的 LangChain RetrieverRetriever是 LangChain 中用于获取相关文档的抽象接口。我们可以创建一个自定义的 Retriever内部使用我们编写的 DQL 进行搜索。from typing import List, Optional, Any, Dict from pymilvus import Collection from langchain.schema import BaseRetriever, Document from langchain.embeddings.base import Embeddings import numpy as np class MilvusDQLRetriever(BaseRetriever): 一个使用 Milvus DQL 进行混合检索的自定义 Retriever。 def __init__( self, collection: Collection, embedding_model: Embeddings, search_params: Dict[str, Any], filter_expr: Optional[str] None, top_k: int 5, output_fields: Optional[List[str]] None ): self.collection collection self.embedding_model embedding_model self.search_params search_params self.default_filter_expr filter_expr self.top_k top_k # 默认输出字段确保包含文本内容字段 self.default_output_fields output_fields or [id, content, author, tags] def _get_relevant_documents(self, query: str, *, filter_expr: Optional[str] None) - List[Document]: 核心检索方法。 # 1. 将查询文本转换为向量 query_embedding self.embedding_model.embed_query(query) # 2. 确定过滤表达式优先使用传入的否则用默认的 expr_to_use filter_expr if filter_expr is not None else self.default_filter_expr # 3. 执行 Milvus DQL 搜索 search_results self.collection.search( data[query_embedding], anns_fieldembedding, paramself.search_params, limitself.top_k, exprexpr_to_use, output_fieldsself.default_output_fields, ) # 4. 将结果转换为 LangChain Document 对象 docs [] # search_results 是一个列表每个元素对应一个查询向量我们只有一个 for hits in search_results: for hit in hits: # 构建 Document 的元数据 metadata {field: hit.entity.get(field) for field in self.default_output_fields if field ! content} metadata[id] hit.id metadata[distance] hit.distance # 相似度距离 # 假设 content 字段存储了文档的原始文本 page_content hit.entity.get(content, ) doc Document(page_contentpage_content, metadatametadata) docs.append(doc) return docs async def _aget_relevant_documents(self, query: str) - List[Document]: # 异步版本可以根据需要实现 raise NotImplementedError(Async retrieval not implemented for this retriever.)5.2 在 LangChain Chain 中使用自定义 Retriever现在我们可以像使用任何其他 LangChain Retriever 一样使用它。from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate # 1. 初始化 Embedding 模型和 LLM # 注意这里的 Embedding 模型必须与插入数据时使用的模型一致 embeddings OpenAIEmbeddings(modeltext-embedding-3-small, openai_api_keyyour-api-key) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyyour-api-key) # 2. 确保集合已加载 tech_blogs_collection Collection(tech_blogs) tech_blogs_collection.load() # 3. 创建我们的自定义 Retriever search_params {metric_type: L2, params: {nprobe: 10}} retriever MilvusDQLRetriever( collectiontech_blogs_collection, embedding_modelembeddings, search_paramssearch_params, filter_exprview_count 500, # 默认只检索阅读量大于500的优质文章 top_k3, output_fields[id, content, author, tags, view_count] ) # 4. 创建一个简单的问答链 prompt_template 基于以下上下文片段回答最后的问题。如果你不知道答案就说你不知道不要编造答案。 上下文 {context} 问题{question} 有帮助的答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 将检索到的所有文档内容“塞”进上下文 retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue # 返回源文档用于调试 ) # 5. 进行查询 question 有哪些关于 LangChain 的教程 result qa_chain.invoke({query: question}) print(问题, question) print(答案, result[result]) print(\n--- 检索到的源文档 ---) for i, doc in enumerate(result[source_documents]): print(f\n文档 {i1}:) print(f 内容摘要: {doc.page_content[:200]}...) print(f 元数据: {doc.metadata})通过这种方式我们成功地将复杂的 Milvus DQL 查询能力注入到了 LangChain 的流程中。你可以在调用qa_chain时动态地修改filter_expr实现高度定制化的检索。6. 高级技巧与性能优化6.1 动态表达式构建在实际应用中过滤条件往往由用户输入或业务逻辑动态生成。你需要安全地构建表达式字符串。def build_filter_expr(author: Optional[str] None, min_views: Optional[int] None, required_tags: List[str] None) - Optional[str]: 安全地构建 Milvus 过滤表达式。 expr_parts [] if author: # 注意字符串值需要用双引号包裹 expr_parts.append(fauthor {author}) if min_views is not None: expr_parts.append(fview_count {min_views}) if required_tags: tag_exprs [fjson_contains(tags, {tag}) for tag in required_tags] expr_parts.append(f({ and .join(tag_exprs)})) # 要求包含所有指定标签 if not expr_parts: return None return and .join(expr_parts) # 使用示例 expr build_filter_expr(authorAlice, min_views1000, required_tags[AI, Tutorial]) print(expr) # 输出 author Alice and view_count 1000 and (json_contains(tags, AI) and json_contains(tags, Tutorial))6.2 搜索参数调优search_params对查询性能和精度有巨大影响。metric_type必须与创建索引时一致。L2欧氏距离和IP内积是最常用的。对于 OpenAI 的text-embedding-ada-002等模型通常使用余弦相似度而 Milvus 的IP在向量标准化后等价于余弦相似度。nprobe在IVF类索引中它代表搜索的聚类中心数量。值越大搜索越精确但耗时越长。通常需要在精度和速度之间权衡通过实验确定。对于百万级数据nprobe在 10 到 256 之间是常见范围。ef或search_k在HNSW索引中类似的参数控制搜索的广度。建议对不同参数进行基准测试记录查询延迟和召回率。6.3 分页与游标查询对于大量结果的查询需要使用offset和limit或者 Milvus 2.3 支持的游标Cursor进行分页。# 使用 offset 和 limit 进行简单分页可能慢适合小数据量 expr author Bob page_size 20 page_num 2 offset (page_num - 1) * page_size results collection.query( exprexpr, output_fields[id, content], limitpage_size, offsetoffset, # 指定偏移量 order_by[(publish_date, desc)] ) # 对于大数据集使用游标更高效需要 Milvus 2.3 并启用该功能 # 首次查询 res collection.query_iterator( exprexpr, batch_sizepage_size, output_fields[id, content], order_by[(publish_date, desc)] ) first_batch res.next() # 获取游标用于下一次查询 cursor res.cursor res.close() # 后续查询使用游标 res2 collection.query_iterator( exprexpr, batch_sizepage_size, output_fields[id, content], cursorcursor, order_by[(publish_date, desc)] ) second_batch res2.next()7. 常见问题排查与最佳实践7.1 常见错误与解决方案问题现象可能原因检查与解决步骤连接失败 (ConnectError)Milvus 服务未启动网络或端口不通版本不兼容。1. 运行docker-compose ps检查容器状态。2. 使用telnet localhost 19530测试端口。3. 确认pymilvus与 Milvus 服务器版本兼容。集合不存在 (CollectionNotExistException)集合名称拼写错误集合未被创建。1. 使用utility.list_collections()查看所有集合。2. 检查创建集合的代码是否成功执行。搜索时返回空结果过滤表达式Expr太严格查询向量与数据分布差异大索引未构建或未加载。1. 先不加expr进行搜索看是否有结果。2. 检查expr语法是否正确字段名和值类型是否匹配。3. 确认集合已load()。4. 确认查询向量与插入向量的维度dim一致。查询性能慢数据量过大nprobe等参数设置过高未建立合适索引硬件资源不足。1. 为标量过滤字段如publish_date,author创建二级索引 (create_index)。2. 降低nprobe值。3. 检查服务器 CPU、内存和磁盘 I/O。json_contains过滤数组不生效数组字段定义或数据格式问题表达式语法错误。1. 确认字段定义为ARRAY类型。2. 确认插入的数据是列表格式如[AI, Python]。3. 表达式应为json_contains(tags, AI)注意字段名和值的大小写。返回字段缺失或为Noneoutput_fields未指定或指定错误该字段在某些条目中确实为null。1. 在output_fields中明确列出需要的字段名。2. 使用collection.query并指定output_fields检查原始数据。7.2 生产环境最佳实践连接管理不要为每次查询创建新连接。使用连接池或在应用生命周期内保持一个稳定的连接。对于 Web 服务可以在启动时连接并加载集合。异常处理对search和query操作进行完善的异常捕获try-except记录日志并设计降级策略如返回空结果或使用缓存。监控与指标监控 Milvus 集群的健康状态CPU、内存、磁盘、查询延迟QPS、P99 latency和错误率。Milvus 提供了 Prometheus 指标接口。数据一致性根据业务需求选择合适的一致性级别Strong、Bounded、Eventually。强一致性影响性能最终一致性提供更高吞吐。索引策略根据数据规模和查询模式选择合适的向量索引如IVF_FLAT,IVF_SQ8,HNSW和标量字段索引。在插入大量数据前创建索引比之后创建更高效。配置外置将 Milvus 连接地址、端口、集合名、搜索参数等写入配置文件如config.yaml或环境变量避免硬编码。版本控制严格锁定pymilvus和 Milvus 服务器版本并在升级前在测试环境充分验证。7.3 下一步扩展方向掌握了基础的 DQL 集成后你可以进一步探索与 LangChain Agent 结合让 Agent 根据用户问题动态生成过滤表达式实现更智能的检索。例如用户问“张三最近写的 AI 文章”Agent 解析出“作者张三”和“标签包含 AI”并调用你的自定义 Retriever。多向量检索一个文档包含多个向量如摘要向量、段落向量。探索 Milvus 的多向量字段或父子集合Collection-Partition来支持此类场景。混合检索Hybrid Search分数融合将向量相似度分数距离与元数据相关性分数如关键词匹配度、热度分数进行加权融合得到最终排序。使用 Milvus 的 RESTful 或 gRPC 接口在非 Python 环境或微服务架构中通过 HTTP/gRPC 直接调用 Milvus 的搜索 API。通过将 Milvus DQL 的强大查询能力与 LangChain 的灵活应用框架相结合你可以构建出远超简单语义搜索的复杂、高效且可靠的智能检索系统。关键在于理解两者的边界在合适的层面进行交互从而充分发挥各自的技术优势。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
