最近在尝试构建一个基于大语言模型的智能问答系统时遇到了一个核心挑战如何高效地存储和检索海量的非结构化文本向量。单纯使用内存或简单的文件存储在数据量增长后查询速度急剧下降且难以管理。经过技术选型Milvus 这款高性能向量数据库进入了视野而 LangChain 则提供了与 LLM 集成的绝佳框架。本文将详细拆解如何将 LangChain 与 Milvus 结合并重点演示 DML数据操作语言的实战应用手把手带你搭建一个可运行的向量检索系统。本文适合有一定 Python 基础希望将 AI 应用落地的开发者。你将学到 LangChain 集成 Milvus 的完整流程掌握向量数据的插入、更新、删除和查询CRUD操作并理解背后的核心概念。1. 背景与核心概念在深入代码之前我们需要厘清几个关键概念这有助于理解整个技术栈的价值。LangChain是一个用于开发由语言模型驱动的应用程序的框架。它并非一个具体的模型而是一个“粘合剂”和“工具箱”旨在通过提供标准化的接口、组件和链式调用模式简化与大语言模型LLM交互的复杂性。其核心价值在于模块化让你可以轻松组合数据加载、向量化、记忆管理、工具调用等环节。Milvus是一款开源的向量数据库专为处理海量向量数据的存储和相似性搜索而设计。与传统的关系型数据库如 MySQL存储标量数据不同Milvus 能够高效地索引和检索高维向量例如由文本、图像、音频生成的嵌入向量。它支持多种索引类型如 IVF_FLAT, HNSW和度量方式如 L2 距离、内积、余弦相似度是实现 AI 应用“记忆”和“知识库”的关键基础设施。DML (Data Manipulation Language)即数据操作语言。在数据库领域它指的是对数据库中数据进行操作增、删、改、查的命令集合。在 Milvus 的语境下DML 操作主要包括插入 (Insert)将向量数据及其关联的元数据如原始文本、ID、类别存入集合。查询/搜索 (Search/Query)根据输入向量在集合中查找最相似的向量近似最近邻搜索ANN。删除 (Delete)根据条件如 ID删除集合中的实体。更新 (Upsert)更新已存在实体的数据Milvus 2.x 通过upsert操作实现即存在则更新不存在则插入。为什么是 LangChain MilvusLangChain 提供了VectorStore抽象层封装了与多种向量数据库交互的通用接口。Milvus 作为其官方支持的向量库之一能够提供生产级别的性能、可扩展性和可靠性。两者的结合使得开发者可以专注于业务逻辑而无需深入底层向量索引和检索的复杂细节。2. 环境准备与版本说明为了确保示例的可复现性以下是本次实战的环境配置。你的实际环境可能有所不同但核心步骤和代码逻辑是通用的。操作系统Ubuntu 20.04 LTS / macOS Monterey 或更高版本 / Windows 10/11 (建议使用 WSL2)。Python 版本3.8 或 3.9LangChain 和 pymilvus 对 3.10 也支持良好但 3.8 是较稳妥的选择。关键库及其版本langchain0.1.0(请注意LangChain 版本迭代较快核心接口相对稳定)langchain-community0.0.10(社区集成包包含 Milvus 等向量库支持)pymilvus2.3.0(Milvus 的 Python SDK)sentence-transformers2.2.2(用于生成文本嵌入向量)python-dotenv1.0.0(管理环境变量如 OpenAI API Key)Milvus 服务部署 你可以选择多种方式运行 MilvusDocker (推荐用于本地开发)使用docker-compose快速启动一个 Standalone单机模式的 Milvus 服务。安装包从官网下载对应系统的安装包。云服务使用 Zilliz CloudMilvus 的托管服务。本文将以Docker Compose方式为例因为它最快捷也最接近开发环境。3. 核心组件与原理拆解3.1 LangChain 的 VectorStore 抽象LangChain 定义了VectorStore基类所有向量数据库的集成如 Milvus, Pinecone, Chroma都需要实现其接口。主要方法包括from_documents: 从文档列表创建向量存储并插入数据。add_documents/add_texts: 向已有存储中添加新数据。similarity_search: 执行相似性搜索。similarity_search_with_score: 返回结果及相似度分数。as_retriever: 将其转换为一个检索器便于接入 LangChain 的链Chain。3.2 Milvus 的核心概念集合 (Collection)相当于关系型数据库中的表是存储向量和元数据的容器。分区 (Partition)集合的子集用于数据隔离和管理提升查询效率。实体 (Entity)集合中的一条记录包含一个向量字段和多个标量字段元数据。模式 (Schema)定义集合的结构包括字段名、数据类型如FloatVector,VarChar,Int64和是否为主键等属性。索引 (Index)为了加速向量搜索而创建的数据结构。创建索引是搜索前的必要步骤。3.3 工作流程一个典型的 LangChain Milvus 应用流程如下启动 Milvus 服务。连接 Milvus使用pymilvus建立连接。准备文本与嵌入模型使用嵌入模型如sentence-transformers或 OpenAItext-embedding-ada-002将文本转换为向量。定义模式并创建集合在 Milvus 中创建用于存储向量和文本的集合。插入数据 (Insert DML)将文本向量及其元数据插入集合。创建索引在向量字段上创建索引如IVF_FLAT,HNSW。加载集合将集合数据加载到内存以供搜索。执行搜索 (Search DML)输入查询文本将其向量化并在集合中搜索相似项。(可选) 更新与删除数据对已有数据进行管理。4. 完整实战案例构建电影摘要智能检索系统我们将构建一个系统存储一系列电影摘要然后通过自然语言问题进行检索例如“我想看一部关于黑客的科幻电影”。4.1 启动 Milvus 服务首先使用 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 healthcheck: test: [CMD, etcdctl, endpoint, health] interval: 30s timeout: 20s retries: 3 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 --console-address :9090 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.0 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 healthcheck: test: [CMD, curl, -f, http://localhost:9091/healthz] interval: 30s timeout: 20s retries: 3 networks: default: name: milvus在终端中进入该文件所在目录运行docker-compose up -d等待所有服务健康启动可通过docker-compose ps查看状态。Milvus 服务将在localhost:19530监听。4.2 创建 Python 项目并安装依赖创建一个新的项目目录并初始化虚拟环境。mkdir langchain-milvus-demo cd langchain-milvus-demo python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate安装必要的 Python 包pip install langchain langchain-community pymilvus sentence-transformers python-dotenv4.3 编写核心代码我们创建一个名为movie_retrieval.py的主文件。4.3.1 导入库与连接 Milvus# movie_retrieval.py import os from typing import List, Optional from pymilvus import connections, utility, Collection, CollectionSchema, FieldSchema, DataType from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import Milvus from langchain.schema import Document from langchain.text_splitter import CharacterTextSplitter # 1. 连接到 Milvus 服务 def connect_to_milvus(hostlocalhost, port19530): 建立与 Milvus 数据库的连接 try: connections.connect(aliasdefault, hosthost, portport) print(f✅ 成功连接到 Milvus: {host}:{port}) except Exception as e: print(f❌ 连接 Milvus 失败: {e}) raise # 执行连接 connect_to_milvus()4.3.2 准备数据与嵌入模型我们使用一个电影摘要的小数据集并选择sentence-transformers中的all-MiniLM-L6-v2模型它体积小且效果不错。# 2. 初始化嵌入模型 embedding_model HuggingFaceEmbeddings( model_namesentence-transformers/all-MiniLM-L6-v2, model_kwargs{device: cpu}, # 使用 GPU 可改为 cuda encode_kwargs{normalize_embeddings: True} # 归一化向量便于使用余弦相似度 ) # 3. 准备电影摘要数据 movie_data [ {title: The Matrix, genre: Sci-Fi, summary: A computer hacker learns from mysterious rebels about the true nature of his reality and his role in the war against its controllers.}, {title: Inception, genre: Sci-Fi, Thriller, summary: A thief who steals corporate secrets through the use of dream-sharing technology is given the inverse task of planting an idea into the mind of a C.E.O.}, {title: The Dark Knight, genre: Action, Crime, Drama, summary: When the menace known as the Joker wreaks havoc and chaos on the people of Gotham, Batman must accept one of the greatest psychological and physical tests of his ability to fight injustice.}, {title: Parasite, genre: Comedy, Drama, Thriller, summary: Greed and class discrimination threaten the newly formed symbiotic relationship between the wealthy Park family and the destitute Kim clan.}, {title: Interstellar, genre: Adventure, Drama, Sci-Fi, summary: A team of explorers travel through a wormhole in space in an attempt to ensure humanitys survival.}, {title: The Social Network, genre: Biography, Drama, summary: As Harvard student Mark Zuckerberg creates the social networking site that would become known as Facebook, he is sued by the twins who claimed he stole their idea, and by the co-founder who was later squeezed out of the business.}, ] # 将数据转换为 LangChain 的 Document 对象 documents [] for i, movie in enumerate(movie_data): # 将元数据标题、类型和内容摘要一起存储 doc Document( page_contentmovie[summary], metadata{title: movie[title], genre: movie[genre], id: i} ) documents.append(doc) print(f已创建 {len(documents)} 个文档对象。)4.3.3 使用 LangChain 集成 Milvus 进行插入 (Insert DML)这是最关键的步骤之一。我们使用Milvus.from_documents方法它会自动处理集合创建、向量化、数据插入等过程。# 4. 定义集合名称和连接参数 COLLECTION_NAME movie_collection # Milvus 集合名 CONNECTION_ARGS { host: localhost, port: 19530, collection_name: COLLECTION_NAME, drop_old: True, # 如果集合已存在则删除旧集合开发环境方便生产环境慎用 } # 5. 插入数据并创建向量存储 print(开始将文档向量化并插入 Milvus...) vector_store Milvus.from_documents( documentsdocuments, embeddingembedding_model, connection_argsCONNECTION_ARGS, # 以下参数用于定义集合的 Schema如果不指定LangChain 会使用默认 Schema # 默认 Schema 包含: id (主键), vector (向量), text (文本内容) # 我们的 metadata 中的 title 和 genre 会被自动存储为额外的标量字段。 ) print(✅ 文档插入完成)4.3.4 执行相似性搜索 (Search DML)现在我们可以用自然语言问题进行检索了。# 6. 执行相似性搜索 query A movie about hackers and artificial reality print(f\n搜索查询: {query}) results vector_store.similarity_search(query, k2) # k 表示返回最相似的前几条结果 print(\n 搜索结果 ) for i, doc in enumerate(results): print(f\n结果 {i1}:) print(f 电影标题: {doc.metadata[title]}) print(f 电影类型: {doc.metadata[genre]}) print(f 内容摘要: {doc.page_content}) print(f 元数据ID: {doc.metadata[id]})运行此部分代码你应该能看到与“黑客”、“现实”相关的《黑客帝国》被检索出来。4.3.5 进阶 DML 操作直接使用 pymilvus有时我们需要更精细的控制比如直接删除或更新某条数据。这就需要用到pymilvus的原生接口。首先获取我们刚刚通过 LangChain 创建的集合对象。# 7. 获取集合对象进行原生操作 from pymilvus import Collection collection Collection(COLLECTION_NAME) # 加载集合 collection.load() # 将集合加载到内存搜索前必须执行 print(f\n集合 {COLLECTION_NAME} 中的实体数量: {collection.num_entities})删除操作 (Delete DML) 假设我们想删除 ID 为 0 的电影即《黑客帝国》。# 删除操作 print(\n--- 执行删除操作 ---) # 在 Milvus 中删除需要通过布尔表达式指定条件。默认主键字段名是 id。 delete_expr id in [0] collection.delete(exprdelete_expr) print(f已执行删除表达式: {delete_expr}) # 删除后需要刷新一下确保更改生效对于Standalone删除是近实时 collection.flush() print(f删除后实体数量: {collection.num_entities})更新/插入操作 (Upsert DML) Milvus 2.x 使用upsert操作来更新数据。如果主键存在则更新不存在则插入。# 更新/插入操作 print(\n--- 执行 Upsert 操作 ---) # 准备要 upsert 的数据。数据必须与 Schema 字段顺序严格对应。 # 我们需要知道 Schema。可以通过 LangChain 创建的默认 Schema 或自行定义。 # 假设我们想更新 ID 为 1 的电影摘要或插入一条新记录。 # 首先获取 ID 为 1 的文档的向量这里为了演示我们重新生成一个向量。 new_summary A skilled extractor is offered a chance to have his criminal history erased in exchange for planting an idea in a targets subconscious. This is the updated summary. new_vector embedding_model.embed_query(new_summary) # 准备数据行。需要与创建集合时的字段顺序一致。 # 默认 LangChain Milvus 的 Schema 字段顺序可能是: [id, vector, text, title, genre, ...] # 更稳妥的方式是直接使用我们之前插入的数据结构或查询现有数据。 # 这里我们演示 upsert 一条新数据ID10。 new_id 10 new_title Inception Redux new_genre Sci-Fi, Mind-Bender new_text new_summary # 注意向量字段需要是二维数组即使只有一条数据。 data [ [new_id], # id 字段 (主键) [new_vector], # vector 字段 [new_text], # text 字段 (对应 page_content) [new_title], # title 字段 (来自 metadata) [new_genre], # genre 字段 (来自 metadata) ] # 执行 upsert collection.upsert(data) print(f已 Upsert 数据ID{new_id}) collection.flush() print(fUpsert 后实体数量: {collection.num_entities})复杂条件查询 除了相似性搜索Milvus 也支持基于标量字段的过滤查询。# 结合标量过滤的向量搜索 print(\n--- 执行带过滤的向量搜索 ---) search_params {metric_type: L2, params: {nprobe: 10}} # 搜索参数 # 创建一个过滤表达式只搜索类型为 Sci-Fi 的电影 filter_expr genre like %Sci-Fi% # 将查询文本向量化 query_vector embedding_model.embed_query(space travel and black holes) # 执行混合搜索 results collection.search( data[query_vector], anns_fieldvector, # 在哪个向量字段上搜索 paramsearch_params, limit3, exprfilter_expr, # 过滤表达式 output_fields[title, genre, text] # 指定返回哪些标量字段 ) print(f在类型包含Sci-Fi的电影中搜索‘space travel’:) for hits in results: for hit in hits: print(f - 电影: {hit.entity.get(title)}, 类型: {hit.entity.get(genre)}, 距离: {hit.distance:.4f}) # 距离越小越相似L2距离4.4 运行与验证在终端运行完整的脚本python movie_retrieval.py你应该能看到以下输出成功连接 Milvus。文档插入完成的提示。针对“hackers”的搜索返回《The Matrix》。删除操作后实体数量减少。Upsert 操作后实体数量增加。带过滤的搜索只返回科幻电影。4.5 结果说明通过以上步骤我们成功实现了一个完整的流程环境搭建使用 Docker 启动了 Milvus 服务。数据准备与向量化将电影摘要文本通过sentence-transformers模型转化为向量。数据插入利用 LangChain 的高级 APIMilvus.from_documents一键完成集合创建和数据插入。相似性搜索使用similarity_search实现了基于语义的检索。原生 DML 操作深入使用pymilvus进行了删除 (delete)、更新插入 (upsert) 和混合搜索展示了更细粒度的数据管理能力。这个系统现在可以作为一个简单的“电影知识库”后端轻松集成到聊天机器人或推荐系统中。5. 常见问题与排查思路在集成 LangChain 和 Milvus 时你可能会遇到以下典型问题问题现象常见原因解决思路pymilvus.exceptions.MilvusException: MilvusException: (code1, messagecollection not found)1. 集合名称拼写错误。2. 集合确实未创建。3. 连接到了错误的 Milvus 实例或数据库。1. 使用utility.list_collections()检查现有集合名。2. 确认from_documents或Collection初始化时使用的名称一致。3. 检查连接参数host, port是否正确。pymilvus.exceptions.MilvusException: MilvusException: (code1, messageunavailable: no available node)Milvus 服务未启动或连接失败。1. 运行docker-compose ps确认所有容器etcd, minio, standalone状态为Up。2. 检查防火墙是否开放了19530端口。3. 尝试用curl localhost:9091/healthz检查 Milvus 健康状态。搜索速度很慢1. 未创建向量索引。2. 索引类型或参数不适合数据规模和查询需求。3. 集合未加载到内存。1. 确认在搜索前已执行collection.load()。2. 使用collection.index()创建索引对于开发测试IVF_FLAT是平衡的选择。3. 通过 LangChain 创建时可传入index_params参数。插入数据时 metadata 字段丢失LangChain 的默认 Schema 可能只包含id,vector,text三个字段。在Milvus.from_documents中显式指定collection_description和schema_config或使用pymilvus先自定义 Schema 创建集合。相似性搜索结果不相关1. 嵌入模型不适合领域。2. 向量未归一化但使用了余弦相似度。3. 搜索参数如nprobe设置不当。1. 尝试更换嵌入模型如all-mpnet-base-v2效果更好但更慢。2. 确保嵌入时normalize_embeddingsTrue并使用metric_type“IP”内积进行搜索。3. 调整nprobe参数增大可提高召回率但降低速度。RuntimeError: Failed to infer the type of argument(LangChain 错误)LangChain 和pymilvus或langchain-community版本不兼容。锁定常用版本组合langchain0.1.0,langchain-community0.0.10,pymilvus2.3.0。使用pip list检查。通用排查步骤服务先行始终先确认 Milvus 服务是否健康运行。连接测试用最简单的pymilvus代码测试连接和基本操作。版本核对检查langchain,pymilvus,milvus服务器版本是否兼容。日志分析查看 Milvus 容器的日志 (docker-compose logs standalone) 获取详细错误信息。简化复现剥离 LangChain直接用pymilvus实现功能以确定问题是出在集成层还是底层。6. 最佳实践与工程建议将 LangChain 与 Milvus 用于生产环境时需要考虑以下几点集合与分区设计合理命名为集合设计清晰的命名规则如project_entity_envqa_documents_prod。使用分区如果数据有明显类别如用户、月份、产品线使用分区可以提高查询效率和管理便利性。在Milvus.from_documents中可通过partition_key_field参数指定。预定义 Schema对于生产环境建议先用pymilvus明确定义包含所有元数据字段的 Schema再传入 LangChain。这能避免字段类型推断错误。索引策略优化理解索引类型HNSW适用于高召回率、高维度的场景但内存占用大IVF_FLAT是精度和速度的平衡需要训练。在插入数据后创建索引先插入一定量的数据如 1 万条再创建索引这样索引会更准确。测试不同参数对nlist(IVF)、M/efConstruction(HNSW) 等参数进行基准测试找到适合你数据规模和 QPS 的最佳配置。数据插入批处理避免逐条插入。利用Milvus.add_documents或pymilvus的insert进行批量操作可以极大提升吞吐量。建议批处理大小在 100-1000 之间。连接与资源管理使用连接池pymilvus支持连接池在生产环境中应配置合理的连接数。及时释放资源长时间不用的集合在业务低峰期可以调用collection.release()将其从内存中释放需要时再load()。监控关注 Milvus 的内存、CPU 使用情况以及查询延迟、QPS 等指标。与 LangChain 集成的模式分离读写可以考虑创建两个 VectorStore 对象一个用于只读搜索频繁使用一个用于写操作偶尔使用。错误处理与重试在网络调用插入、搜索周围添加重试逻辑和异常捕获提高系统鲁棒性。元数据规范化确保Document的metadata字典中的值都是简单类型str, int, float复杂对象需要序列化。版本控制与迁移Milvus 版本升级可能带来不兼容的变更。在升级前务必在测试环境充分验证。对于重要的向量数据定期备份集合的 Schema 和数据可通过utility.export和utility.import工具。安全与权限生产环境切勿使用默认的minioadmin账号密码。为 Milvus 的 MinIO 和 Etcd 组件配置强密码。通过网络策略限制对 Milvus 端口19530的访问仅允许应用服务器访问。遵循这些实践可以帮你构建一个更稳定、高效且易于维护的向量检索系统。从简单的 Demo 到生产系统关键在于对细节的把握和对组件特性的深入理解。