这次我们来看一个 LangChain 与 Milvus 向量数据库结合并应用 DML数据操作语言进行实战的项目。对于想要构建本地知识库、实现智能问答或进行高效语义搜索的开发者来说这是一个非常核心且实用的技术栈。它解决的核心问题是如何将非结构化的文本数据如文档、网页内容转化为向量存入专业的向量数据库并利用 LangChain 的智能体Agent能力进行灵活的查询和操作。最值得关注的几个点第一这套方案完全可以在本地或私有化环境中部署对显存没有直接要求主要依赖 CPU 和内存但 GPU 可以加速 Embedding 模型推理。第二它不仅仅是简单的存储和检索通过 LangChain 的 Agent 和 DML 操作可以实现动态的数据更新、条件删除等复杂管理任务。第三它提供了清晰的接口可以轻松集成到 Flask 等 Web 框架中构建成可用的服务。本文会带你从零开始完成 Milvus 的安装、LangChain 环境的配置并重点演示如何通过代码连接两者实现数据的插入Insert、搜索Search、更新Update和删除Delete这一完整的 DML 操作闭环。无论你是想学习 LangChain 的实战应用还是希望为你的项目增加一个强大的向量检索后端这篇文章都能提供直接的参考。1. 核心能力速览能力项说明技术栈核心LangChain (应用框架) Milvus (向量数据库) Embedding 模型 (如 text2vec, OpenAI)主要功能文本向量化存储、高相似度语义搜索、通过 Agent 执行复杂的数据库 DML 操作增删改查硬件门槛无强制 GPU 要求。CPU 即可运行 Milvus 和基础 Embedding 模型。使用 GPU 或高性能 Embedding 模型可提升向量化速度。内存要求运行 Milvus 服务需要至少 2GB 可用内存实际根据数据量而定。部署方式Milvus 支持 Docker 一键部署或手动安装LangChain 为 Python 库通过 pip 安装。是否支持 API是。Milvus 提供 gRPC/HTTP 接口LangChain 可封装成 REST API 服务如使用 Flask/FastAPI。是否支持批量任务是。支持批量文本的向量化与入库以及批量查询。适合场景构建本地知识库、智能问答系统、文档内容检索、相似内容推荐、Agent 工具调用持久化存储。2. 适用场景与使用边界这个技术组合非常适合以下几类开发者和场景全栈或后端开发者希望在自己的应用中集成智能语义搜索功能而不完全依赖昂贵的云端 AI 服务。AI 应用创业者/团队需要快速构建基于私有数据的原型或产品如企业内部的智能客服、知识管理平台。LangChain 学习者和研究者希望超越简单的 Chain 使用深入理解如何将 LangChain Agent 与外部工具数据库进行深度集成。它能解决的具体问题包括海量文档的即时检索将公司手册、产品文档向量化后员工可以用自然语言快速找到相关内容。智能问答结合 LLM大语言模型实现基于给定知识库的精准问答减少幻觉。内容去重与推荐通过向量相似度判断文章、商品描述的相似性实现去重或关联推荐。需要注意的使用边界非实时高频写入Milvus 适合以搜索为主的场景对于每秒数万次写入的高频 OLTP 场景并非最佳选择。数据规模与硬件平衡亿级向量数据需要规划好集群部署和硬件资源CPU、内存、磁盘单机 Standalone 模式更适合千万级以下数据量学习和测试。语义理解依赖 Embedding 模型搜索效果的好坏很大程度上取决于选择的 Embedding 模型的质量。需要针对中文、专业领域等场景选择合适的模型。合规与版权如果处理的文档数据涉及商业机密、个人隐私或受版权保护的内容务必确保数据使用的合法授权并在部署时做好网络访问控制。3. 环境准备与前置条件在开始代码实战前需要准备好以下环境。这里以 Linux/macOS 或 WSL2 环境为例进行说明。操作系统: Ubuntu 20.04/22.04, CentOS 7, macOS, 或 Windows 10/11 with WSL2。推荐使用 Linux 环境。Python: 版本 3.8 - 3.11。建议使用 3.9 或 3.10兼容性最广。包管理工具:pip最新版。容器环境 (推荐): Docker 和 Docker Compose。这是启动 Milvus 最简便的方式。网络: 确保能正常访问 Python PyPI 仓库和 Docker Hub 以下载依赖和镜像。硬件检查清单:[ ] CPU: 4 核以上更佳。[ ] 内存: 至少 4GB建议 8GB 以上。[ ] 磁盘: 预留 10GB 以上空间用于安装和存储数据。[ ] GPU (可选): 如果使用本地 Embedding 模型且追求速度可准备 NVIDIA GPU 并安装对应 CUDA。4. 安装部署与启动方式我们将分两步走先部署 Milvus 向量数据库服务再配置 Python 的 LangChain 环境。4.1 部署 Milvus (Docker Compose 方式)这是最快上手的方案。确保已安装 Docker 和 Docker Compose。下载配置文件 在终端中创建一个工作目录并下载 Milvus Standalone 的docker-compose.yml文件。mkdir milvus-langchain-demo cd milvus-langchain-demo wget https://github.com/milvus-io/milvus/releases/download/v2.4.0/milvus-standalone-docker-compose.yml -O docker-compose.yml注意版本号v2.4.0可能更新请查阅 Milvus 官方发布页 获取最新稳定版。启动 Milvus 服务 使用 Docker Compose 启动所有相关容器包括 Milvus、Etcd 和 MinIO。sudo docker-compose up -d验证服务状态 运行以下命令检查容器是否正常运行。sudo docker-compose ps你应该看到名为milvus-standalone、etcd、minio的容器状态均为Up。 还可以通过curl检查 Milvus 的健康状态curl http://localhost:9091/healthz返回{status:OK}即表示服务健康。连接信息 Milvus 服务启动后默认的 gRPC 端口为19530HTTP 端口为9091。后续 LangChain 将通过localhost:19530进行连接。4.2 配置 Python 与 LangChain 环境创建并激活 Python 虚拟环境强烈推荐python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows安装核心 Python 包 我们将安装 LangChain、Milvus 的 Python SDKpymilvus、一个本地 Embedding 模型以text2vec为例以及必要的工具链。pip install langchain langchain-community pymilvus sentence-transformerssentence-transformers提供了高质量的本地 Embedding 模型。5. 功能测试与效果验证现在进入实战环节。我们将创建一个 Python 脚本完成从连接数据库、创建集合、插入数据、执行搜索到进行 DML 操作的全流程。5.1 基础连接与集合创建首先我们编写一个脚本milvus_langchain_demo.py建立与 Milvus 的连接并定义一个存储文章片段的集合。from pymilvus import connections, utility, FieldSchema, CollectionSchema, DataType, Collection from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import Milvus from langchain.schema import Document import numpy as np # 1. 连接到 Milvus 服务 connections.connect(aliasdefault, hostlocalhost, port19530) print(Connected to Milvus successfully.) # 2. 定义 Embedding 模型 (使用本地模型无需API Key) embedding_model HuggingFaceEmbeddings(model_nameshibing624/text2vec-base-chinese) # 这是一个不错的中文Embedding模型也可以换成 moka-ai/m3e-base 等 # 3. 准备测试数据 documents [ Document(page_contentLangChain 是一个用于开发由大语言模型驱动的应用程序的框架。, metadata{source: doc1, page: 1}), Document(page_contentMilvus 是一个开源的向量数据库专为海量向量数据的相似性搜索而设计。, metadata{source: doc2, page: 1}), Document(page_contentDML 是数据操作语言包括 INSERT、UPDATE、DELETE 等操作。, metadata{source: doc3, page: 1}), Document(page_content向量检索是人工智能应用中的核心技术用于寻找相似的内容。, metadata{source: doc4, page: 1}), ] # 4. 使用 LangChain 的 Milvus 集成模块创建向量库并插入数据 # 注意如果集合已存在Milvus.from_documents 会尝试插入数据但可能因schema冲突失败。 # 首次运行或需要清空重来时可以先手动删除已存在的集合。 collection_name langchain_demo_collection if utility.has_collection(collection_name): print(fCollection {collection_name} exists, dropping it.) utility.drop_collection(collection_name) # 此步骤会自动创建集合并基于 embedding_model 将文档内容转换为向量存入 Milvus vector_store Milvus.from_documents( documents, embedding_model, connection_args{host: localhost, port: 19530}, collection_namecollection_name, ) print(fDocuments inserted into collection {collection_name}.)运行此脚本如果没有报错并看到连接成功和插入成功的提示说明基础环境打通了。5.2 语义搜索测试接下来我们在同一个脚本的后续部分或在新的脚本中测试搜索功能。# 接上面的代码... # 5. 进行相似性搜索 print(\n--- Testing Similarity Search ---) query 什么是向量数据库 search_results vector_store.similarity_search(query, k2) # k 表示返回最相似的前k个结果 for i, doc in enumerate(search_results): print(fResult {i1}:) print(f Content: {doc.page_content}) print(f Metadata: {doc.metadata}) print(- * 50) # 6. 使用 LangChain Retriever 进行搜索 (这是更常见的集成方式) print(\n--- Testing Retriever ---) retriever vector_store.as_retriever(search_kwargs{k: 3}) retriever_results retriever.get_relevant_documents(如何进行数据插入操作) for doc in retriever_results: print(f- {doc.page_content[:80]}...)预期输出当你查询“什么是向量数据库”时系统应该能返回与“Milvus 是一个开源的向量数据库...”高度相关的内容。这验证了向量存储和检索的核心功能是正常的。5.3 DML 实战更新与删除这才是本次实战的重点。我们将直接使用pymilvusSDK 来演示如何对已存入的数据进行更新Update和删除Delete模拟数据管理的场景。首先我们需要知道要操作的数据条目的唯一 ID在 Milvus 中通常是int64类型的id字段。我们可以通过一次搜索来获取目标条目的 ID。# 接上面的代码... # 7. DML 操作实战先查询出要操作的数据的 ID print(\n--- DML Operations: Preparing ---) # 获取底层的 Milvus Collection 对象 collection Collection(collection_name) collection.load() # 将集合加载到内存以进行搜索 # 进行一次搜索获取结果及其对应的 id search_params {metric_type: L2, params: {nprobe: 10}} query_vector embedding_model.embed_query(DML 是数据操作语言) # 将查询文本转为向量 results collection.search( data[query_vector], anns_fieldembedding, # 假设 LangChain 创建的向量字段名为 embedding paramsearch_params, limit1, output_fields[id, content] # 指定要返回的字段content 是存储原始文本的字段名 ) if results and results[0]: target_id results[0][0].id print(fFound target document with id: {target_id}, content: {results[0][0].entity.get(content)}) # 7.1 UPDATE 操作更新该条数据的元数据metadata print(\n--- Performing UPDATE ---) # 假设我们要更新 source 这个 metadata 字段 # 在 Milvus 中更新是通过 upsert 操作实现的需要提供主键和要更新的字段 # 注意LangChain 默认的 schema 可能将元数据存储在单独的字段或打包成一个 JSON 字段。 # 为了简化演示我们假设有一个 meta 字段存储了所有 metadata。 # 更稳妥的做法是直接使用 pymilvus 的 query 和 upsert。 # 这里演示一个概念性操作 # 先通过 id 查询出现有数据 expr fid {target_id} query_result collection.query(expr, output_fields[id, content, metadata]) if query_result: old_meta query_result[0].get(metadata, {}) print(fOld metadata: {old_meta}) new_meta old_meta.copy() new_meta[source] updated_doc3 # 修改 source new_meta[updated] True # 添加新字段 # 构建 upsert 数据 (需要完整的向量这里需要重新获取或查询出来) # 由于操作较复杂实际生产中建议在应用层维护向量和元数据的映射关系。 print((Note: Full vector update requires fetching the original embedding, skipped for brevity.)) print(Metadata update logic demonstrated.) # 7.2 DELETE 操作根据表达式删除数据 print(\n--- Performing DELETE ---) delete_expr fid {target_id} print(fDeleting document with expression: {delete_expr}) # 执行删除 (生产环境请谨慎操作) # collection.delete(delete_expr) # print(Delete command issued (commented out for safety).) print((Delete operation is commented out to prevent accidental data loss.)) print(To execute, uncomment the collection.delete(delete_expr) line.) else: print(No target document found for DML operations.)关键点说明更新Update在 Milvus 中更新本质上是upsert插入或更新。你需要提供主键id和所有要写入的字段包括向量字段。如果只更新元数据必须同时提供原有的向量值否则向量会被覆盖或置空。因此最佳实践是在应用层维护好数据的最新状态然后执行upsert。删除Delete支持通过布尔表达式删除数据如id in [1, 2, 3]或metadata[source] \obsolete\。删除操作不可逆务必先做好查询确认。字段名需要根据 LangChain 创建集合时实际定义的字段名来操作。通常向量字段名为embedding文本字段名为content元数据可能被展平或存储在单独的字段中。6. 接口 API 与批量任务将上述能力封装成 API 服务是常见的需求。这里给出一个使用 Flask 构建简易 REST API 的示例支持批量文档入库和查询。6.1 Flask API 服务示例创建一个app.py文件from flask import Flask, request, jsonify from pymilvus import connections from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import Milvus from langchain.schema import Document from langchain.text_splitter import RecursiveCharacterTextSplitter import logging app Flask(__name__) # 初始化全局组件 connections.connect(aliasdefault, hostlocalhost, port19530) embedding_model HuggingFaceEmbeddings(model_nameshibing624/text2vec-base-chinese) COLLECTION_NAME api_demo_collection def get_vector_store(): 获取或创建向量存储对象 return Milvus( embedding_functionembedding_model, connection_args{host: localhost, port: 19530}, collection_nameCOLLECTION_NAME, ) app.route(/ingest, methods[POST]) def ingest_documents(): 批量摄入文档 data request.json if not data or documents not in data: return jsonify({error: Missing documents array in request body}), 400 raw_texts data[documents] # 对长文本进行分割 text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) all_splits [] for text in raw_texts: splits text_splitter.split_text(text) all_splits.extend([Document(page_contents) for s in splits]) try: vector_store get_vector_store() # 注意from_documents 会创建集合如果已存在且schema不匹配会报错。 # 更健壮的做法是检查集合是否存在然后使用 add_documents。 # 这里为演示简化。 if len(all_splits) 0: # 假设集合已存在使用 add_documents vector_store.add_documents(all_splits) return jsonify({message: fSuccessfully ingested {len(all_splits)} text chunks.}), 200 except Exception as e: logging.error(fIngestion error: {e}) return jsonify({error: str(e)}), 500 app.route(/search, methods[GET]) def search(): 语义搜索 query request.args.get(q, ) k int(request.args.get(k, 3)) if not query: return jsonify({error: Missing query parameter q}), 400 try: vector_store get_vector_store() results vector_store.similarity_search(query, kk) output [ { content: doc.page_content, metadata: doc.metadata, # 可以添加相似度分数需要调用 similarity_search_with_score } for doc in results ] return jsonify({query: query, results: output}), 200 except Exception as e: logging.error(fSearch error: {e}) return jsonify({error: str(e)}), 500 if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)6.2 批量任务处理建议对于大规模数据入库建议队列化处理使用 Celery、RQ 或数据库任务表将入库请求放入队列异步执行避免 API 请求超时。分批次提交不要一次性提交数十万条数据。使用add_documents时可以每 100-1000 条数据提交一次。错误重试与日志在批量任务中捕获每批数据的异常记录日志并设计重试机制。进度监控为长时间运行的批量任务提供进度查询接口。7. 资源占用与性能观察这套技术栈的性能瓶颈通常出现在两个地方Embedding 模型推理和 Milvus 向量搜索。Embedding 阶段CPU 模式使用sentence-transformers在 CPU 上推理处理速度约为每秒几十到几百个句子取决于模型和文本长度。监控 CPU 使用率。GPU 模式如果安装了 CUDA 版本的 PyTorch 并将模型加载到 GPU推理速度可提升数倍至数十倍。使用nvidia-smi命令观察 GPU 显存占用和利用率。一个中等规模的 Embedding 模型如m3e-base在 GPU 上可能占用 1-2GB 显存。Milvus 搜索阶段内存Milvus 在搜索时需要将索引和数据加载到内存。观察docker stats中 Milvus 容器的内存使用量它会随着集合数据量增长。磁盘 I/O大量数据插入和索引构建会导致磁盘写入。确保使用 SSD 硬盘以获得更好的体验。搜索参数nprobe等搜索参数影响搜索速度和精度。值越大精度越高但速度越慢。需要在业务场景中权衡。性能优化小贴士对于静态知识库可以在数据全部入库后在 Milvus 中建立索引如IVF_FLAT,HNSW以大幅加速搜索。使用连接池管理 Milvus 连接避免频繁创建断开连接的开销。考虑使用更轻量级的 Embedding 模型如all-MiniLM-L6-v2在速度和精度间取得平衡。8. 常见问题与排查方法问题现象可能原因排查方式解决方案连接 Milvus 失败 (pymilvus.exceptions.MilvusException)1. Milvus 服务未启动。2. 主机/端口错误。3. 防火墙或网络策略阻止。1. 运行docker-compose ps检查容器状态。2. 尝试telnet localhost 19530(Linux) 或Test-NetConnection localhost -Port 19530(PowerShell)。3. 检查 Milvus 容器日志docker-compose logs milvus-standalone。1. 使用docker-compose up -d启动服务。2. 确认连接代码中的host和port与 docker-compose.yml 中定义的一致。3. 关闭防火墙或添加规则。Milvus.from_documents报错提示集合已存在或 Schema 冲突同名集合已存在且字段定义如向量维度与当前 Embedding 模型不匹配。1. 使用utility.list_collections()查看现有集合。2. 使用utility.drop_collection(collection_name)删除旧集合数据会丢失。在初始化前先判断并处理已有集合if utility.has_collection(name): utility.drop_collection(name)或使用不同的collection_name。搜索返回结果不相关或为空1. Embedding 模型不适合当前文本领域如用英文模型处理中文。2. 插入的数据未成功生成向量或向量维度错误。3. 搜索参数k或nprobe设置不当。1. 检查 Embedding 模型名称。2. 查询集合统计信息collection.num_entities。3. 尝试用一条已知数据的内容进行搜索。1. 更换合适的 Embedding 模型如中文任务用text2vec,m3e。2. 确保插入流程无报错。3. 调整搜索参数增大k或nprobe。批量插入时速度慢或内存溢出1. 一次性提交数据量过大。2. 未使用 GPU 进行 Embedding 推理。1. 监控进程内存使用情况。2. 观察 CPU/GPU 使用率。1. 将大数据集分批次如每批 500 条插入。2. 考虑使用 GPU 并安装对应版本的 PyTorch。LangChain 的add_documents后搜索不到新数据新增数据后Milvus 中的索引可能未自动更新或者数据未持久化。1. 检查插入操作是否返回成功且无异常。2. 在插入后执行collection.flush()确保数据落盘。3. 确认搜索前已调用collection.load()将数据加载到内存。在插入操作后显式调用vector_store._collection.flush()和vector_store._collection.load()如果直接使用 pymilvus 对象。API 服务请求超时1. Embedding 或搜索操作耗时过长。2. Flask 默认是同步单线程。1. 查看服务端日志定位耗时操作。2. 使用异步框架如 FastAPI或增加 Worker 数量如 Gunicorn。1. 优化 Embedding 模型或 Milvus 索引。2. 将耗时操作如批量入库改为异步任务。9. 最佳实践与使用建议环境隔离始终使用 Python 虚拟环境venv, conda来管理项目依赖避免包冲突。配置管理将 Milvus 连接地址、端口、集合名、Embedding 模型名等配置项抽取到环境变量或配置文件中如.env,config.yaml。数据预处理入库前对文本进行清洗、分段chunking。合理的分段策略能极大提升检索质量。RecursiveCharacterTextSplitter是个不错的起点。索引优化对于生产环境在数据稳定后根据数据规模和查询模式在 Milvus 中创建合适的索引如HNSW并在查询时使用对应的搜索参数。错误处理与日志在所有数据库操作和 API 端点中加入完善的异常捕获和日志记录便于排查问题。版本控制记录 Milvus、pymilvus、LangChain 等关键组件的版本号升级时注意兼容性。安全与权限如果服务对外暴露务必为 Milvus 设置密码认证并对 API 接口实施访问控制如 API Key、JWT 令牌。资源监控对 Milvus 服务的 CPU、内存、磁盘使用率进行监控设置告警阈值。10. 总结与下一步通过本次实战我们完成了 LangChain 与 Milvus 向量数据库的整合并深入到了 DML 数据操作层面。这套组合的核心价值在于它将强大的语义理解能力通过 Embedding 模型与高效的向量检索能力通过 Milvus结合并通过 LangChain 提供了灵活的应用层编程接口。最值得尝试的点是快速构建一个可用的本地知识库检索原型。你只需要准备一批文本数据运行本文的代码就能立刻拥有一个具备语义搜索能力的后端。最先应该验证的功能是搜索准确度。换用不同的 Embedding 模型调整文本分段大小观察对搜索结果的影响找到最适合你数据特征的配置。最容易踩的坑是集合 Schema 的管理和向量维度的一致性。确保每次连接时使用的 Embedding 模型是同一个否则向量维度对不上会导致搜索失败或结果异常。后续可以探索的方向集成 LLM将检索到的文档片段作为上下文输入给 ChatGPT、通义千问等大模型构建一个真正的 RAG检索增强生成问答系统。使用 Agent利用 LangChain Agent 的能力让 LLM 自主决定何时调用 Milvus 进行检索完成更复杂的任务。升级架构将 Standalone 模式的 Milvus 升级为分布式集群以支持更大规模的数据和更高的并发请求。前端界面使用 Gradio、Streamlit 或 Vue/React 构建一个简单的前端界面让非开发者也能方便地使用检索和问答功能。建议将本文的代码作为基础模板收藏在需要构建智能检索相关功能时可以快速复用和修改。