这次我们来看一个 LangChain 与 Milvus 结合并使用 DQL 进行查询的实战项目。对于想要构建本地知识库、实现智能问答或文档检索的开发者来说如何高效、精准地从向量数据库中检索信息是关键。LangChain 提供了强大的框架而 Milvus 作为高性能向量数据库其原生的查询语言 DQL 能带来更灵活、更底层的控制能力。本文将带你快速了解这个技术栈的核心价值它不是一个简单的概念演示而是一个能跑起来的、可复用的工程实践。我们会重点关注如何搭建环境、启动服务、编写 DQL 查询以及如何将检索结果无缝集成到 LangChain 的链中。无论你是想验证 Milvus 的检索性能还是希望为你的 AI 应用注入一个强大的“记忆体”这篇文章都能提供清晰的路径。1. 核心能力速览能力项说明项目类型LangChain 框架与 Milvus 向量数据库的集成实战核心技术使用 Milvus 的 DQL 进行向量相似性检索并集成到 LangChain 的 RetrievalQA 链中主要功能文档加载与切分、文本向量化、向量入库、DQL 查询构建、问答链集成硬件门槛依赖 Milvus 服务可本地 Docker 部署对客户端机器无特殊 GPU 要求启动方式Milvus 需通过 Docker 或二进制启动LangChain 应用为 Python 脚本启动接口能力通过 Milvus Python SDK 执行 DQL通过 LangChain 提供封装后的检索接口批量任务支持批量文档导入和向量化入库适合场景构建本地知识库问答系统、实现基于私有数据的精准检索、学习 LangChain 与向量数据库的高级集成2. 适用场景与使用边界这个实战项目主要面向两类开发者AI 应用开发者希望为自己的聊天机器人、智能助手增加基于私有文档的问答能力需要一套从文档处理到检索回答的完整方案。数据平台工程师需要评估或应用 Milvus 在真实业务场景中的检索性能与灵活性特别是希望绕过高级封装、直接使用 DQL 进行精细化查询控制的场景。它能解决什么问题知识碎片化将分散的文档PDF、Word、TXT转化为结构化的向量知识库。检索不精准通过向量相似度检索找到与问题语义最相关的文档片段而非关键词匹配。流程自动化将文档加载、切分、向量化、入库、检索、生成回答串联成一个自动化管道。不适合什么场景极小规模数据如果只有几十条文档使用简单的本地缓存或数据库全文检索可能更轻量。对延迟极度敏感虽然 Milvus 性能优异但涉及网络通信、向量计算延迟高于内存查询。缺乏编程基础本项目需要一定的 Python 和 Docker 操作能力。合规与边界提醒处理文档时请确保你拥有文档的使用权尊重知识产权与隐私。构建的知识库系统应在内部或授权范围内使用避免泄露敏感信息。生成的回答基于检索内容需注意内容准确性校验避免传播错误信息。3. 环境准备与前置条件在开始编码之前需要准备好以下环境。这是项目能成功运行的基础。3.1 操作系统推荐 Linux (Ubuntu 20.04) 或 macOS Windows 建议使用 WSL2。确保有稳定的网络环境以下载 Docker 镜像和 Python 包。3.2 Milvus 服务端这是本项目的核心依赖。Milvus 可以以多种方式部署对于本地开发和测试Docker Compose是最简单的方式。Docker Docker Compose确保已安装。可通过docker --version和docker-compose --version检查。磁盘空间预留至少 2 GB 空间用于 Milvus 镜像和存储。端口Milvus 默认占用19530(服务端口) 和9091(管理端口)确保它们未被占用。3.3 Python 客户端环境LangChain 应用将运行在此环境中。Python 版本 3.8 或 3.9 兼容性最好建议使用 3.9。包管理工具 使用pip或conda。关键 Python 包我们将安装langchain,pymilvus,sentence-transformers等。具体版本在下一节给出。4. 安装部署与启动方式4.1 启动 Milvus 服务单机版首先获取 Milvus 的 Docker Compose 配置文件并启动服务。# 1. 下载 docker-compose.yml 配置文件 wget https://github.com/milvus-io/milvus/releases/download/v2.3.3/milvus-standalone-docker-compose.yml -O docker-compose.yml # 2. 启动 Milvus 服务 sudo docker-compose up -d # 3. 检查服务状态 sudo docker-compose ps当看到milvus-standalone和etcd、minio等容器状态均为Up时表示启动成功。4.2 准备 Python 客户端环境创建一个新的 Python 虚拟环境并安装依赖。# 创建并激活虚拟环境以 venv 为例 python -m venv milvus_langchain_env source milvus_langchain_env/bin/activate # Linux/macOS # milvus_langchain_env\Scripts\activate # Windows # 安装核心依赖 pip install langchain0.0.350 pip install pymilvus2.3.3 pip install sentence-transformers pip install langchain-community # 社区提供的更多文档加载器 pip install pypdf # 用于加载PDF文档 pip install python-dotenv # 管理环境变量可选4.3 验证 Milvus 连接在编写主程序前先写一个简单的脚本测试是否能连接到 Milvus。# test_connection.py from pymilvus import connections, utility # 连接到本地 Milvus 服务 connections.connect(host127.0.0.1, port19530) # 检查连接是否成功 try: print(fMilvus 版本: {utility.get_server_version()}) print(连接成功) except Exception as e: print(f连接失败: {e})运行python test_connection.py如果输出 Milvus 版本号则环境准备就绪。5. 功能测试与效果验证我们将分步构建一个完整的流程文档处理 - 向量化入库 - DQL 查询 - 问答集成。5.1 文档加载与文本切分首先准备你的知识文档例如一个 PDF 文件并将其加载、切分成适合处理的文本片段。# step1_doc_processing.py from langchain.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # 1. 加载文档 loader PyPDFLoader(./your_knowledge_base.pdf) # 替换为你的PDF路径 documents loader.load() # 2. 切分文本 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个片段的字符数 chunk_overlap50, # 片段间的重叠字符数 separators[\n\n, \n, 。, , , , , 、, ] ) split_docs text_splitter.split_documents(documents) print(f原始文档页数: {len(documents)}) print(f切分后文本块数量: {len(split_docs)}) print(f第一块内容预览: {split_docs[0].page_content[:200]}...)5.2 向量化与 Milvus 集合创建使用 Sentence Transformer 模型将文本转化为向量并在 Milvus 中创建对应的集合Collection和索引。# step2_embedding_and_milvus.py from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import Milvus from pymilvus import CollectionSchema, FieldSchema, DataType, Collection, utility # 1. 初始化嵌入模型 embedding_model HuggingFaceEmbeddings( model_namesentence-transformers/paraphrase-multilingual-MiniLM-L12-v2 ) # 2. 定义 Milvus 集合的 Schema # 注意这里我们手动定义以便后续使用 DQL fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim384), # 维度需与模型匹配 FieldSchema(namesource, dtypeDataType.VARCHAR, max_length255), ] schema CollectionSchema(fields, descriptionLangChain知识库) # 3. 连接 Milvus 并创建集合 collection_name langchain_demo if utility.has_collection(collection_name): utility.drop_collection(collection_name) # 测试时先删除旧的 collection Collection(namecollection_name, schemaschema) # 4. 创建索引加速向量搜索 index_params { metric_type: L2, index_type: IVF_FLAT, params: {nlist: 128} } collection.create_index(field_nameembedding, index_paramsindex_params) collection.load() # 将集合加载到内存 print(f集合 {collection_name} 创建并加载成功。)5.3 使用 LangChain 的 VectorStore 接口插入数据这是最便捷的入库方式LangChain 帮我们处理了向量生成和批量插入。# step3_insert_data.py from langchain.vectorstores import Milvus # 使用之前初始化的 embedding_model 和 split_docs # 连接参数 vector_store Milvus.from_documents( documentssplit_docs, embeddingembedding_model, collection_namelangchain_demo, connection_args{host: 127.0.0.1, port: 19530}, ) print(文档向量已成功插入 Milvus。)5.4 核心使用 DQL 进行向量检索现在我们绕过 LangChain 的高级封装直接使用 Milvus 的 DQL 来执行查询体验更底层的控制力。# step4_dql_search.py from pymilvus import Collection, connections import numpy as np # 1. 连接并加载集合 connections.connect(host127.0.0.1, port19530) collection Collection(langchain_demo) collection.load() # 2. 准备查询向量将问题文本向量化 query_text LangChain 如何连接向量数据库 # 注意这里需要调用 embedding_model 来编码但为了演示 DQL我们假设已得到向量 # 实际中你需要用同样的模型将 query_text 编码 from langchain.embeddings import HuggingFaceEmbeddings embedding_model HuggingFaceEmbeddings(model_namesentence-transformers/paraphrase-multilingual-MiniLM-L12-v2) query_vector embedding_model.embed_query(query_text) # 3. 构建 DQL 查询表达式 search_params { metric_type: L2, params: {nprobe: 10}, # 搜索参数影响精度和速度 } # 4. 执行 DQL 查询 results collection.search( data[query_vector], # 查询向量 anns_fieldembedding, # 搜索的向量字段 paramsearch_params, limit3, # 返回最相似的3条 output_fields[text, source], # 指定返回的标量字段 exprNone # 可在此添加过滤表达式如 source manual.pdf ) # 5. 解析结果 for hits in results: for hit in hits: print(fID: {hit.id}, 距离: {hit.distance:.4f}) print(f文本: {hit.entity.get(text)[:150]}...) print(f来源: {hit.entity.get(source)}) print(- * 50)这段代码展示了 DQL 的核心collection.search()。你可以通过expr参数添加复杂的过滤条件这是相比简单相似度搜索更强大的地方。5.5 集成到 LangChain 的 RetrievalQA 链最后我们将 Milvus 检索器包装成 LangChain 的 Retriever并接入一个问答链实现“提问-检索-生成答案”的完整流程。# step5_qa_chain.py from langchain.vectorstores import Milvus from langchain.embeddings import HuggingFaceEmbeddings from langchain.chains import RetrievalQA from langchain.llms import OpenAI # 或用 ChatGLM、通义千问等本地模型 import os # 0. 设置 LLM (这里以 OpenAI 为例需配置 API_KEY) os.environ[OPENAI_API_KEY] your-api-key-here llm OpenAI(model_namegpt-3.5-turbo-instruct, temperature0) # 1. 初始化嵌入模型和向量库连接 embedding_model HuggingFaceEmbeddings(model_namesentence-transformers/paraphrase-multilingual-MiniLM-L12-v2) vector_store Milvus( embedding_functionembedding_model, collection_namelangchain_demo, connection_args{host: 127.0.0.1, port: 19530}, ) # 2. 将 VectorStore 转换为 Retriever retriever vector_store.as_retriever(search_kwargs{k: 3}) # 检索3个片段 # 3. 创建 RetrievalQA 链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 将检索到的所有文本“塞”给 LLM retrieverretriever, return_source_documentsTrue # 返回参考来源 ) # 4. 提问 query 请解释一下 LangChain 中 Retriever 的作用是什么 result qa_chain({query: query}) print(f问题: {query}) print(f答案: {result[result]}) print(\n参考来源:) for doc in result[source_documents]: print(f- {doc.page_content[:100]}... (来源: {doc.metadata.get(source, N/A)}))运行此脚本你将得到一个由 LLM 生成的、基于你知识库内容的答案并附上了检索到的参考文档片段。6. 接口 API 与批量任务6.1 封装为简易 API 服务我们可以用 FastAPI 将上面的 QA 链包装成一个 HTTP API 服务供其他系统调用。# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from langchain.vectorstores import Milvus from langchain.embeddings import HuggingFaceEmbeddings from langchain.chains import RetrievalQA from langchain.llms import OpenAI import os app FastAPI(titleLangChain-Milvus QA API) # 初始化全局启动时加载一次 embedding_model HuggingFaceEmbeddings(model_namesentence-transformers/paraphrase-multilingual-MiniLM-L12-v2) vector_store Milvus( embedding_functionembedding_model, collection_namelangchain_demo, connection_args{host: 127.0.0.1, port: 19530}, ) retriever vector_store.as_retriever(search_kwargs{k: 3}) os.environ[OPENAI_API_KEY] your-api-key-here llm OpenAI(model_namegpt-3.5-turbo-instruct, temperature0) qa_chain RetrievalQA.from_chain_type(llmllm, chain_typestuff, retrieverretriever, return_source_documentsTrue) class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str sources: list[str] app.post(/ask, response_modelQueryResponse) async def ask_question(request: QueryRequest): try: result qa_chain({query: request.question}) sources [doc.page_content[:200] for doc in result[source_documents]] return QueryResponse(answerresult[result], sourcessources) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务python app.py。然后可以使用 curl 或 Postman 测试curl -X POST http://127.0.0.1:8000/ask \ -H Content-Type: application/json \ -d {question: Milvus 是什么类型的数据库}6.2 批量文档处理任务对于大量文档需要设计一个健壮的批量处理流程。# batch_ingest.py import os from langchain.document_loaders import DirectoryLoader, PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.vectorstores import Milvus from langchain.embeddings import HuggingFaceEmbeddings import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def batch_ingest_documents(docs_dir: str, collection_name: str): 批量摄入指定目录下的所有文档 # 1. 加载目录下所有支持格式的文档 loader DirectoryLoader( docs_dir, glob**/*.pdf, # 可以添加更多格式如 **/*.txt loader_clsPyPDFLoader, show_progressTrue ) raw_documents loader.load() logger.info(f从 {docs_dir} 加载了 {len(raw_documents)} 个文档。) # 2. 切分文本 text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) all_splits text_splitter.split_documents(raw_documents) logger.info(f切分为 {len(all_splits)} 个文本块。) # 3. 初始化向量模型和存储 embedding_model HuggingFaceEmbeddings(model_namesentence-transformers/paraphrase-multilingual-MiniLM-L12-v2) # 4. 批量插入 Milvus (LangChain 内部会处理分批) vector_store Milvus.from_documents( documentsall_splits, embeddingembedding_model, collection_namecollection_name, connection_args{host: 127.0.0.1, port: 19530}, ) logger.info(f批量文档已成功导入 Milvus 集合 {collection_name}。) return vector_store if __name__ __main__: # 指定你的文档目录 DOCS_DIRECTORY ./my_knowledge_docs COLLECTION_NAME batch_knowledge_base batch_ingest_documents(DOCS_DIRECTORY, COLLECTION_NAME)7. 资源占用与性能观察7.1 Milvus 服务资源占用启动单机版 Milvus 后可以通过docker stats命令观察容器资源使用情况。通常milvus-standalone容器会占用数百 MB 内存。向量搜索时的 CPU 和内存消耗会随数据量和并发请求增加。7.2 客户端 Python 进程资源内存主要消耗在加载嵌入模型如 Sentence Transformer和缓存文档文本。sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2模型加载后约占用 300-500 MB 内存。CPU/GPU文本向量化embedding是计算密集型任务。该模型在 CPU 上运行较慢如果有 CUDA 环境它会自动使用 GPU 加速显著提升批量处理速度。网络 I/O客户端与 Milvus 服务通过 gRPC 通信查询延迟主要受网络和 Milvus 搜索性能影响。性能优化建议索引调优在collection.create_index时根据数据规模调整nlist参数。数据量越大nlist可适当增大但会占用更多内存。搜索参数search_params中的nprobe值影响搜索速度和精度。值越大精度越高速度越慢。通常在 10-100 之间调整。批量插入使用from_documents或手动分批插入数据避免单条插入的巨大开销。连接池在高并发场景下配置pymilvus的连接池参数。缓存 Retriever/Chain在 Web 服务中将初始化好的retriever和qa_chain对象设为全局变量避免每次请求重复加载。8. 常见问题与排查方法问题现象可能原因排查方式解决方案连接 Milvus 失败 (Cannot connect to Milvus server)1. Milvus 服务未启动2. 主机/端口错误3. 防火墙阻止1.docker-compose ps检查容器状态2.telnet 127.0.0.1 19530测试端口3. 检查客户端连接代码的host和port1. 执行docker-compose up -d启动服务2. 修正连接参数为正确的 IP 和端口3. 关闭防火墙或添加规则插入数据时报错Collection not loaded创建的集合没有执行load()操作检查代码在创建索引后是否调用了collection.load()在插入数据或搜索前确保执行collection.load()DQL 查询结果为空或不准1. 查询向量与入库向量模型不一致2. 索引未构建或参数不合理3. 过滤表达式 (expr) 太严格1. 确认embedding模型是否相同2. 检查是否创建了索引 (utility.list_indexes)3. 简化expr或设为None测试1. 统一使用同一个嵌入模型2. 重新创建合适的索引并load3. 逐步调试过滤条件LangChain 的from_documents速度慢1. 单条插入2. 嵌入模型在 CPU 上运行1. 观察日志是否在逐条处理2. 检查 CUDA 是否可用 (torch.cuda.is_available())1. LangChain 内部是批量的可检查网络2. 安装torch的 CUDA 版本确保模型使用 GPU问答链答案质量差1. 检索到的文本块不相关2. 文本切分不合理太大或太小3. LLM 本身能力或提示词问题1. 检查 DQL 检索出的文本是否与问题相关2. 调整chunk_size和chunk_overlap3. 直接测试 LLM 对纯文本的总结能力1. 优化检索参数 (k,search_params)2. 优化文本切分策略3. 更换更强的 LLM 或优化提示工程sentence-transformers模型下载失败网络问题检查错误信息通常是连接超时1. 使用国内镜像源2. 手动下载模型文件到本地指定cache_folder参数9. 最佳实践与使用建议项目结构规范化将配置如 Milvus 地址、集合名、模型名、工具函数文档处理、检索、应用代码API、主程序分离提高可维护性。配置外部化使用.env文件或配置类管理敏感信息如 API Key和可变参数如 chunk_size, 端口号。日志记录为关键步骤文档加载、切分、插入、查询添加日志便于监控和调试。异常处理与重试在网络请求如调用 Milvus、OpenAI API时添加重试机制和超时设置。数据版本管理当知识库更新时可以考虑创建新的集合如collection_v2而不是直接覆盖旧集合便于回滚和 A/B 测试。效果评估构建一个测试集定期评估检索的召回率、准确率以及问答链答案的满意度持续迭代优化。安全与权限如果部署为对外服务务必为 Milvus 和你的 API 服务配置认证和授权避免未授权访问。资源清理定期检查并清理 Milvus 中不再使用的集合释放磁盘和内存空间。10. 总结与下一步这个 LangChain Milvus DQL 的实战项目打通了从原始文档到智能问答的关键链路。其核心价值在于提供了灵活性既可以使用 LangChain 的高级 API 快速搭建原型又能通过 Milvus 的 DQL 进行精细化的检索控制满足生产级应用的需求。最值得尝试的点DQL 的过滤能力在向量相似度搜索的基础上用expr参数进行属性过滤是实现精准检索的利器。完整的本地化流水线从文档到答案所有环节均可部署在本地或私有环境保障数据安全。最先应该验证的功能 建议你先从step4_dql_search.py开始手动执行一次 DQL 查询感受直接操作向量数据库的粒度。然后运行step5_qa_chain.py体验端到端的问答效果。最容易踩的坑环境不一致确保开发、测试、生产环境的 Milvus 版本、Python 包版本、嵌入模型保持一致。向量维度不匹配创建集合时定义的向量维度必须与嵌入模型输出的维度完全相同。集合未加载在执行任何操作前别忘了collection.load()。后续扩展方向混合检索结合 DQL 的向量搜索和标量过滤实现基于元数据如日期、作者的混合查询。多路召回与重排序使用多个检索器如关键词向量进行初筛再用更精细的模型对结果重排序。接入本地大模型将 OpenAI 替换为 ChatGLM、Qwen 等本地部署的 LLM实现完全私有化的知识问答系统。实现 WebUI使用 Gradio 或 Streamlit 快速构建一个交互式界面方便非技术人员使用。建议将本文中的代码作为基础模板收藏在构建自己的知识库系统时根据实际数据规模和业务需求进行调整和优化。