这次我们来看一个 LangChain 与 Milvus 结合并使用 DQL 进行实战的项目。对于想要构建本地知识库、实现智能问答或文档检索的开发者来说LangChain 提供了强大的应用编排能力而 Milvus 作为高性能向量数据库是存储和检索向量化数据的核心。DQLDomain Query Language则是 Milvus 中用于执行复杂向量检索的查询语言。这个组合的重点不是概念多复杂而是能不能快速搭建一个可运行、可扩展的检索增强生成RAG系统。本文将带你从零开始完成一个基于 LangChain Milvus DQL 的实战项目。我们会重点关注环境搭建、数据准备、向量化存储、DQL 查询构建以及最终的问答应用集成。整个过程会涉及 Python 环境、Docker 部署、显存/内存占用观察以及 API 服务化思路。无论你是想验证技术可行性还是为实际项目做技术选型这篇文章都能提供一套可落地的操作指南。1. 核心能力速览在深入代码之前我们先快速了解这个技术栈的核心能力和门槛。能力项说明项目类型检索增强生成RAG应用原型 / 本地知识库系统核心组件LangChain应用框架、Milvus向量数据库、Embedding 模型如 text2vec、LLM如 OpenAI API 或本地模型主要功能文档切片与向量化、向量存储与检索、基于上下文的智能问答、支持复杂 DQL 过滤查询硬件门槛CPU/内存为主。Milvus 服务本身对 GPU 无硬性要求但 Embedding 模型推理可能需要 GPU 加速。纯 CPU 推理也可行速度较慢。显存占用取决于 Embedding 模型。例如使用text2vec系列小模型CPU 推理仅占用内存若使用bge-large等大模型并启用 GPU可能需要 2-4G 显存。LLM 部分若调用云端 API 则无本地显存压力。部署方式Milvus 推荐 Docker 部署LangChain 应用为 Python 脚本可通过命令行或 Flask/FastAPI 封装为 Web 服务。是否支持 API是。可轻松将核心检索与问答功能封装为 RESTful API 供其他系统调用。是否支持批量任务是。支持批量文档导入、批量向量化入库以及批量查询。适合场景企业知识库问答、法律/金融文档检索、代码库搜索、个人学习笔记管理、内容推荐系统原型开发。2. 适用场景与使用边界这个工具栈适合谁AI 应用开发者希望快速构建具备“记忆”能力的对话机器人或智能助手。数据分析师/研究者需要从大量非结构化文档论文、报告中精准查找信息。个人开发者或小团队寻求低成本、可私有化部署的知识管理解决方案。能解决什么问题“大海捞针”式检索从成千上万份文档中快速找到与用户问题语义最相关的片段。提升大模型回答的准确性与可信度为 LLM 提供来自可靠知识库的参考依据减少“幻觉”。实现对话上下文管理结合 Milvus可以存储和检索多轮对话的历史向量维持对话连贯性。不适合什么场景对延迟要求极苛刻的在线搜索如毫秒级网页搜索。向量检索虽快但叠加 Embedding 和 LLM 生成后整体链路延迟在秒级。完全结构化数据的查询。对于已有明确数据库 Schema 的查询直接使用 SQL 等传统数据库效率更高。无任何编程基础的纯终端用户。本项目需要一定的命令行和 Python 开发能力进行部署和调试。版权与合规边界文档数据确保你拥有所用文档的处理和存储权避免侵犯版权。Embedding/LLM 模型使用开源模型时遵守其对应许可证如 MIT, Apache-2.0。调用商用 API如 OpenAI时需注意其数据使用政策。隐私数据如果处理包含个人隐私或敏感信息的文档务必在本地部署 Milvus 和 Embedding 模型确保数据不出域。3. 环境准备与前置条件开始实战前请确保你的开发环境满足以下要求。操作系统推荐 Linux (Ubuntu 20.04/22.04, CentOS 7) 或 macOS。Windows 可通过 WSL2 获得最佳体验。本文演示环境为 Ubuntu 22.04。容器与运行时Docker Docker Compose用于部署 Milvus 服务。这是最简便的方式。# 检查 Docker 是否安装 docker --version docker-compose --version如果无法使用 Docker也可参考 Milvus 官方文档进行源码编译或下载二进制包但步骤会复杂很多。Python 环境Python 3.8 - 3.11。建议使用 3.9 或 3.10兼容性最广。pip包管理工具。建议使用venv或conda创建独立的虚拟环境。# 创建虚拟环境 python -m venv langchain-milvus-env # 激活环境 (Linux/macOS) source langchain-milvus-env/bin/activate # 激活环境 (Windows) .\langchain-milvus-env\Scripts\activate硬件资源内存至少 8GB。Milvus 运行和 Embedding 推理都会消耗内存文档量越大需求越高。磁盘空间预留 10GB 以上空间用于存储 Docker 镜像、Python 包、模型文件和向量数据。GPU可选如果使用较大的 Embedding 模型如BAAI/bge-large-zh-v1.5并希望加速需要支持 CUDA 的 NVIDIA GPU。CPU 亦可运行。网络需要能访问 PyPI 和 Docker Hub 以下载 Python 包和镜像。如果使用海外 LLM API如 OpenAI需确保网络连通性。4. 安装部署与启动方式我们将按照Milvus 服务 - Python 依赖 - 应用代码的顺序进行部署。4.1 启动 Milvus 服务Docker Compose 方式Milvus 提供了单机版的 Docker Compose 配置文件非常适合开发和测试。下载配置文件# 创建项目目录并进入 mkdir langchain-milvus-demo cd langchain-milvus-demo # 下载最新的 docker-compose.yml 文件 wget https://github.com/milvus-io/milvus/releases/download/v2.3.3/milvus-standalone-docker-compose.yml -O docker-compose.yml注意版本号v2.3.3可能更新请查阅 Milvus 发布页面 获取最新稳定版。启动服务sudo docker-compose up -d这个命令会在后台启动 Milvus 及其依赖的 etcd元数据存储和 MinIO对象存储。验证服务状态sudo docker-compose ps你应该看到三个服务milvus-standalone,etcd,minio的状态都是Up。 也可以检查 Milvus 的日志sudo docker-compose logs milvus-standalone看到Successfully loaded configuration等字样通常表示启动成功。关键信息记录服务地址localhost或127.0.0.1端口19530(gRPC 端口) 和9091(HTTP 端口用于管理)这些信息将在后续的 Python 连接中用到。4.2 安装 Python 依赖在激活的虚拟环境中安装必要的 Python 包。pip install langchain langchain-community pymilvus # 安装文本分割和 Embedding 相关 pip install langchain-text-splitters sentence-transformers # 安装用于演示的文档加载器以 Markdown 和文本文件为例 pip install unstructured # 如果需要 Web 界面或 API可以安装 FastAPI pip install fastapi uvicorn关键包说明langchain核心框架。langchain-community包含社区维护的众多集成工具如与 Milvus 的向量存储接口。pymilvusMilvus 的官方 Python SDK。sentence-transformers提供开源的 Sentence-BERT 模型用于生成文本向量。unstructured强大的文档解析库支持 PDF、Word、PPT、HTML 等多种格式。4.3 准备应用代码结构在项目目录下创建如下文件结构langchain-milvus-demo/ ├── docker-compose.yml # Milvus 服务配置 ├── requirements.txt # Python 依赖列表可选 ├── config.py # 配置文件 ├── data/ # 存放待处理的原始文档 │ └── your_document.md ├── main.py # 主程序文档加载、切分、向量化入库 ├── query.py # 查询与问答程序 └── api.py # 可选FastAPI 封装5. 功能测试与效果验证我们将分步测试整个流程文档处理 - 向量入库 - DQL 检索 - 问答生成。5.1 基础配置与连接测试首先创建config.py来集中管理配置。# config.py import os from sentence_transformers import SentenceTransformer # Milvus 连接配置 MILVUS_HOST localhost # 如果 Milvus 运行在其它机器请修改 MILVUS_PORT 19530 MILVUS_COLLECTION_NAME langchain_docs # 集合类似数据库的表名称 # Embedding 模型配置 # 选择一个合适的模型这里使用一个轻量级中英文模型 EMBEDDING_MODEL_NAME sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2 # 初始化模型首次运行会自动下载 EMBEDDING_MODEL SentenceTransformer(EMBEDDING_MODEL_NAME) # 向量维度取决于所选模型 EMBEDDING_DIMENSION EMBEDDING_MODEL.get_sentence_embedding_dimension() # 文本分割配置 CHUNK_SIZE 500 # 每个文本块的最大字符数 CHUNK_OVERLAP 50 # 块之间的重叠字符数保持上下文连贯 # LLM 配置示例使用 OpenAI也可替换为本地模型 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) # 建议从环境变量读取 OPENAI_API_BASE https://api.openai.com/v1 # 如有代理需修改 OPENAI_MODEL gpt-3.5-turbo然后编写main.py的第一部分测试 Milvus 连接并创建集合。# main.py (部分) from pymilvus import connections, utility, CollectionSchema, FieldSchema, DataType, Collection from config import MILVUS_HOST, MILVUS_PORT, MILVUS_COLLECTION_NAME, EMBEDDING_DIMENSION def connect_to_milvus(): 连接 Milvus 数据库 try: connections.connect(hostMILVUS_HOST, portMILVUS_PORT) print(f✅ 成功连接到 Milvus: {MILVUS_HOST}:{MILVUS_PORT}) # 检查服务状态 print(f Milvus 版本: {utility.get_server_version()}) except Exception as e: print(f❌ 连接 Milvus 失败: {e}) raise def create_collection_if_not_exists(): 如果集合不存在则创建它 if utility.has_collection(MILVUS_COLLECTION_NAME): print(f集合 {MILVUS_COLLECTION_NAME} 已存在。) return Collection(MILVUS_COLLECTION_NAME) # 1. 定义字段 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), # 存储原始文本 FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dimEMBEDDING_DIMENSION), FieldSchema(namesource, dtypeDataType.VARCHAR, max_length255), # 文档来源 FieldSchema(namechunk_index, dtypeDataType.INT64), # 块序号 ] # 2. 创建 Schema schema CollectionSchema(fields, descriptionLangChain 文档向量存储) # 3. 创建集合 collection Collection(nameMILVUS_COLLECTION_NAME, schemaschema) print(f✅ 集合 {MILVUS_COLLECTION_NAME} 创建成功。) # 4. 创建索引加速向量检索 index_params { metric_type: IP, # 内积相似度计算方式。也可用 L2 (欧氏距离) index_type: IVF_FLAT, # 一种经典索引类型适合中小规模数据集 params: {nlist: 128}, } collection.create_index(field_nameembedding, index_paramsindex_params) print(f✅ 在字段 embedding 上创建索引成功。) return collection if __name__ __main__: connect_to_milvus() collection create_collection_if_not_exists() print(基础连接与集合创建测试通过)运行python main.py如果看到成功连接的提示说明 Milvus 服务运行正常并且集合已就绪。5.2 文档加载、切分与向量化入库在main.py中添加文档处理函数。我们以处理data/目录下的 Markdown 文件为例。# main.py (续) import os from langchain_community.document_loaders import DirectoryLoader, UnstructuredMarkdownLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from config import CHUNK_SIZE, CHUNK_OVERLAP, EMBEDDING_MODEL from pymilvus import Collection def load_and_split_documents(data_dir./data): 加载并分割文档 # 使用 DirectoryLoader 加载指定目录下的所有 .md 文件 loader DirectoryLoader( data_dir, glob**/*.md, loader_clsUnstructuredMarkdownLoader, show_progressTrue ) documents loader.load() print(f 从 {data_dir} 加载了 {len(documents)} 个文档。) # 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_sizeCHUNK_SIZE, chunk_overlapCHUNK_OVERLAP, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(f 将文档切分为 {len(splits)} 个文本块。) return splits def generate_embeddings_and_insert(collection: Collection, splits): 生成向量并插入 Milvus texts [split.page_content for split in splits] sources [split.metadata.get(source, unknown) for split in splits] # 批量生成向量 (使用 sentence-transformers) print( 正在生成文本向量...) embeddings EMBEDDING_MODEL.encode(texts, show_progress_barTrue, normalize_embeddingsTrue) print(f✅ 已生成 {len(embeddings)} 个向量维度: {embeddings[0].shape}) # 准备插入数据 data [ texts, # text 字段 embeddings.tolist(), # embedding 字段 sources, # source 字段 list(range(len(texts))) # chunk_index 字段 ] # 插入数据 print( 正在插入数据到 Milvus...) insert_result collection.insert(data) print(f✅ 成功插入 {len(insert_result.primary_keys)} 条数据。) # 将数据从内存刷新到磁盘确保可查询 collection.flush() print( 数据已刷新。) return insert_result.primary_keys if __name__ __main__: # ... 之前的连接和创建集合代码 ... connect_to_milvus() collection create_collection_if_not_exists() # 加载、分割并插入文档 splits load_and_split_documents() if splits: primary_keys generate_embeddings_and_insert(collection, splits) print(f 文档向量化入库完成共处理 {len(primary_keys)} 个文本块。) else: print(⚠️ 未找到待处理的文档请检查 ./data 目录。)运行前请在./data目录下放置一些.md或.txt文件作为测试数据。运行后控制台会显示加载、切分、向量化和插入的进度。5.3 使用 DQL 进行向量检索DQL 是 Milvus 的查询语言功能强大。我们首先使用基础的相似性搜索然后演示加入过滤条件的 DQL 查询。创建query.py。# query.py from pymilvus import connections, Collection from config import MILVUS_HOST, MILVUS_PORT, MILVUS_COLLECTION_NAME, EMBEDDING_MODEL import numpy as np def basic_similarity_search(query_text, top_k3): 基础相似性搜索将问题向量化在 Milvus 中查找最相似的文本块 # 1. 连接并加载集合 connections.connect(hostMILVUS_HOST, portMILVUS_PORT) collection Collection(MILVUS_COLLECTION_NAME) collection.load() # 将集合数据加载到内存以加速搜索 # 2. 将查询文本向量化 query_embedding EMBEDDING_MODEL.encode([query_text], normalize_embeddingsTrue)[0] query_embedding query_embedding.tolist() # 3. 定义搜索参数 search_params { metric_type: IP, # 必须与创建索引时一致 params: {nprobe: 10}, # 搜索时探查的聚类数影响精度和速度 } # 4. 执行搜索 results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limittop_k, output_fields[text, source, chunk_index] # 指定要返回的字段 ) # 5. 解析结果 print(f\n 针对查询{query_text}) print(f 返回了 {len(results[0])} 个最相关结果) for i, hit in enumerate(results[0]): print(f\n--- 结果 {i1} (距离: {hit.distance:.4f}) ---) print(f 来源: {hit.entity.get(source)}) print(f 块索引: {hit.entity.get(chunk_index)}) print(f 内容: {hit.entity.get(text)[:200]}...) # 预览前200字符 return results def dql_search_with_filter(query_text, filter_expression, top_k3): 使用 DQL 进行带过滤条件的搜索 connections.connect(hostMILVUS_HOST, portMILVUS_PORT) collection Collection(MILVUS_COLLECTION_NAME) collection.load() query_embedding EMBEDDING_MODEL.encode([query_text], normalize_embeddingsTrue)[0].tolist() # 定义 DQL 表达式 # 例如source data/document1.md 或 chunk_index 10 dql_expr filter_expression search_params {metric_type: IP, params: {nprobe: 10}} # 关键使用 expr 参数传入 DQL 过滤表达式 results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limittop_k, exprdql_expr, # DQL 过滤表达式 output_fields[text, source, chunk_index] ) print(f\n DQL 过滤搜索{query_text} | 过滤条件: {filter_expression}) if results and results[0]: for i, hit in enumerate(results[0]): print(f\n--- 结果 {i1} (距离: {hit.distance:.4f}) ---) print(f 来源: {hit.entity.get(source)}) print(f 内容: {hit.entity.get(text)[:200]}...) else: print( ⚠️ 未找到符合过滤条件的匹配项。) return results if __name__ __main__: # 测试基础搜索 basic_similarity_search(LangChain 是什么, top_k2) # 测试带过滤的 DQL 搜索 (假设你的文档来源中有 data/intro.md) dql_search_with_filter( query_text如何安装 Milvus, filter_expressionsource data/intro.md, # 只从特定文档中搜索 top_k2 )运行python query.py你将看到系统首先进行普通的语义搜索然后执行一个增加了来源过滤的 DQL 搜索。这演示了如何将业务逻辑如按文档类型、日期、作者过滤与向量搜索结合。5.4 集成 LangChain 进行问答生成最后我们将检索到的上下文与 LLM 结合生成最终答案。这里以 OpenAI GPT 为例你也可以替换为其他 LangChain 支持的 LLM。# query.py (续) import os from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from config import OPENAI_API_KEY def rag_qa(query_text, top_k3): 检索增强生成先检索再让 LLM 基于检索结果回答 # 1. 检索相关上下文 search_results basic_similarity_search(query_text, top_ktop_k) if not search_results or not search_results[0]: return 抱歉在知识库中未找到相关信息。 # 2. 组装上下文 contexts [] for hit in search_results[0]: contexts.append(hit.entity.get(text)) combined_context \n\n---\n\n.join(contexts) # 3. 构建 Prompt prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个专业的助手请严格根据以下提供的上下文信息来回答问题。如果上下文没有提供足够的信息请直接说“根据已知信息无法回答该问题”。不要编造信息。), (human, 上下文信息\n{context}\n\n问题{question}) ]) # 4. 初始化 LLM 链 llm ChatOpenAI( openai_api_keyOPENAI_API_KEY, model_namegpt-3.5-turbo, temperature0.1 # 低温度答案更确定 ) chain prompt_template | llm | StrOutputParser() # 5. 调用链 answer chain.invoke({context: combined_context, question: query_text}) return answer if __name__ __main__: # 确保设置了 OPENAI_API_KEY 环境变量 if not OPENAI_API_KEY: print(请设置 OPENAI_API_KEY 环境变量。) else: question Milvus 的主要用途是什么 print(f\n 用户问题{question}) answer rag_qa(question, top_k2) print(f\n AI 回答\n{answer})运行这个脚本你会看到系统首先从 Milvus 中检索出与问题最相关的文本片段然后将这些片段作为上下文提交给 LLM 生成一个基于知识库的、引用来源的答案。6. 接口 API 与批量任务6.1 封装为 FastAPI 服务将核心功能封装成 API便于集成。创建api.py。# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import uvicorn from query import rag_qa, basic_similarity_search, dql_search_with_filter from main import load_and_split_documents, generate_embeddings_and_insert from pymilvus import connections, Collection from config import MILVUS_HOST, MILVUS_PORT, MILVUS_COLLECTION_NAME app FastAPI(titleLangChain-Milvus RAG API) class QueryRequest(BaseModel): question: str top_k: Optional[int] 3 filter_expr: Optional[str] None # DQL 过滤表达式 class IngestRequest(BaseModel): data_dir: str app.on_event(startup) def startup_event(): 启动时连接 Milvus try: connections.connect(hostMILVUS_HOST, portMILVUS_PORT) print(API 服务已连接至 Milvus。) except Exception as e: print(fAPI 服务连接 Milvus 失败: {e}) app.get(/) def read_root(): return {message: LangChain-Milvus RAG API 服务运行中} app.post(/query) def query_rag(request: QueryRequest): 问答接口 try: if request.filter_expr: # 使用带过滤的搜索演示实际RAG可能直接用基础搜索 # 这里简化处理实际可根据业务调整 answer rag_qa(request.question, top_krequest.top_k) else: answer rag_qa(request.question, top_krequest.top_k) return {question: request.question, answer: answer} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/search) def search_only(request: QueryRequest): 纯向量搜索接口返回原始片段 try: if request.filter_expr: results dql_search_with_filter(request.question, request.filter_expr, request.top_k) else: results basic_similarity_search(request.question, request.top_k) # 简化返回格式实际应结构化 return {query: request.question, results: results} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/ingest) def ingest_documents(request: IngestRequest): 批量文档入库接口 try: splits load_and_split_documents(request.data_dir) if not splits: return {message: 未找到文档, ingested: 0} connections.connect(hostMILVUS_HOST, portMILVUS_PORT) collection Collection(MILVUS_COLLECTION_NAME) collection.load() pks generate_embeddings_and_insert(collection, splits) return {message: 文档入库成功, ingested: len(pks)} except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)启动服务python api.py。然后你就可以通过http://localhost:8000/docs访问自动生成的 API 文档并进行测试。6.2 批量任务处理对于大量文档的定期更新可以编写脚本进行批处理。# batch_ingest.py import os import schedule import time from main import load_and_split_documents, generate_embeddings_and_insert, connect_to_milvus, create_collection_if_not_exists from pymilvus import Collection from config import MILVUS_COLLECTION_NAME def batch_ingest_job(data_directory): 批量入库任务 print(f[{time.strftime(%Y-%m-%d %H:%M:%S)}] 开始批量入库任务...) try: connect_to_milvus() collection Collection(MILVUS_COLLECTION_NAME) collection.load() splits load_and_split_documents(data_directory) if splits: pks generate_embeddings_and_insert(collection, splits) print(f[{time.strftime(%Y-%m-%d %H:%M:%S)}] 任务完成新增 {len(pks)} 条数据。) else: print(f[{time.strftime(%Y-%m-%d %H:%M:%S)}] 未发现新文档。) except Exception as e: print(f[{time.strftime(%Y-%m-%d %H:%M:%S)}] 任务失败: {e}) if __name__ __main__: DATA_DIR ./data # 监控的目录 # 立即执行一次 batch_ingest_job(DATA_DIR) # 可以配置定时任务例如每6小时执行一次 # schedule.every(6).hours.do(batch_ingest_job, DATA_DIR) # while True: # schedule.run_pending() # time.sleep(60)7. 资源占用与性能观察Milvus 服务资源占用使用 Docker Compose 启动后可以通过docker stats命令观察。单机版 Milvus 在空闲时内存占用约 1-2GB。插入和搜索时会上升。向量索引会占用额外内存和磁盘空间。IVF_FLAT索引相对节省内存但搜索速度比HNSW慢。Embedding 模型推理资源占用CPU 模式以paraphrase-multilingual-MiniLM-L12-v2为例推理一个句子约需 50-100ms内存占用约 500MB。GPU 模式将模型加载到 GPU如cuda:0可大幅加速批量编码。显存占用取决于模型参数量上述小模型约 500MB-1GB。性能优化建议索引选择开发测试用IVF_FLAT生产环境追求速度可用HNSW需要更多内存。批量操作插入向量和搜索时尽量使用批量接口减少网络往返。连接池在高并发 API 服务中使用pymilvus的连接池功能。分段加载如果集合非常大可以使用collection.load(_partition_names[...])分段加载数据到内存。如何监控Milvus 监控访问http://localhost:9091(Milvus 管理界面) 或使用 Prometheus Grafana。Python 进程使用top,htop或nvidia-smi(GPU) 观察资源使用。8. 常见问题与排查方法问题现象可能原因排查方式解决方案连接 Milvus 失败1. Milvus 服务未启动。2. 主机/端口错误。3. 防火墙/网络问题。1.docker-compose ps检查服务状态。2.telnet localhost 19530测试端口。3. 检查config.py中的MILVUS_HOST和MILVUS_PORT。1. 运行docker-compose up -d。2. 修改配置为正确地址。3. 检查防火墙设置。插入数据时报错Collection not loaded集合未加载到内存。在插入或搜索前确认执行了collection.load()。在操作集合前显式调用collection.load()。DQL 过滤查询无结果1. 过滤表达式语法错误。2. 字段名或值错误。3. 确实没有匹配数据。1. 检查 DQL 表达式字符串确保字段名与 Schema 一致。2. 使用collection.query(expr“...”)先测试过滤条件本身。1. 修正表达式注意字符串值用单引号。2. 确认待过滤字段已正确插入数据。Embedding 模型下载慢或失败网络问题或 Hugging Face 连接不稳定。观察下载进度条或错误信息。1. 使用国内镜像源。2. 手动下载模型文件到本地然后从本地路径加载。搜索速度慢1. 未创建索引。2. 索引参数 (nprobe) 设置过大。3. 数据量过大内存不足。1. 检查集合索引collection.index()。2. 降低nprobe值如从 128 降到 16。3. 观察系统内存和 swap 使用。1. 创建合适的索引。2. 调整search_params在精度和速度间权衡。3. 增加内存或使用支持磁盘的索引类型。LLM 回答不准确或“幻觉”1. 检索到的上下文不相关。2.top_k设置太小或太大。3. Prompt 设计不佳。1. 检查query.py中basic_similarity_search返回的片段是否相关。2. 调整top_k(如 3-5)。3. 优化系统 Prompt强调“根据上下文”。1. 优化 Embedding 模型或文本切分策略。2. 实验不同的top_k值。3. 改进 Prompt 工程增加 Few-shot 示例。API 服务启动后无法访问1. 端口被占用。2. 服务绑定到127.0.0.1而非0.0.0.0。3. 服务器防火墙。1.netstat -tlnp | grep 8000。2. 检查uvicorn.run的host参数。3. 检查云服务器安全组规则。1. 更换端口如port8001。2. 确保host0.0.0.0。3. 开放对应端口。9. 最佳实践与使用建议数据预处理是关键文本切分的质量直接影响检索效果。对于不同格式代码、论文、手册可能需要定制化的RecursiveCharacterTextSplitter分隔符。Embedding 模型选型中文场景可考虑BAAI/bge-large-zh-v1.5中英文混合用multilingual模型纯英文用all-MiniLM-L6-v2。在效果和速度间权衡。Milvus 集合管理为不同知识库创建不同集合。定期清理测试数据 (collection.drop())。生产环境考虑使用 Milvus 集群版。DQL 过滤的灵活运用将业务元数据如文档类别、创建时间、权限等级存入 Milvus 的标量字段利用 DQL 在向量检索前进行过滤能极大提升查询的精准度和安全性。版本控制与回滚对config.py、模型版本、Milvus 集合 Schema 进行版本控制。批量更新数据前先备份旧集合。安全性API 服务应增加认证如 API Key。用户上传文档进行向量化前需进行病毒扫描和内容审核。避免将包含敏感信息的 Prompt 和上下文记录到日志。效果评估建立一个小型测试集定期运行监控检索命中率和问答准确率作为迭代优化的依据。10. 总结与下一步这个 LangChain Milvus DQL 的实战项目展示了如何构建一个功能完整的本地知识库问答系统。最值得尝试的点在于它打通了从原始文档到智能问答的端到端流程并且通过 Milvus 的 DQL 功能你可以轻松实现复杂的多条件检索这是很多简单向量库不具备的能力。最先应该验证的功能基础流程跑通确保 Docker 启动 Milvus - 文档入库 - 简单问答整个链路畅通。DQL 过滤查询尝试用source、chunk_index等字段进行过滤感受其如何缩小搜索范围。更换 Embedding 模型换用不同的 Sentence Transformer 模型观察检索结果的质量变化。最容易踩的坑Milvus 连接问题确保 Docker 服务运行且端口正确。向量维度不匹配创建集合时指定的维度必须与 Embedding 模型输出的维度一致。DQL 表达式语法字符串比较要用单引号字段名要完全匹配。后续扩展方向前端界面使用 Gradio 或 Streamlit 快速搭建一个 Web 界面。多路召回与重排序结合关键词搜索如 BM25和向量搜索再用更精细的模型对结果重排序。对话历史将多轮对话的摘要或向量也存入 Milvus实现有记忆的对话。接入本地 LLM使用 Ollama、LM Studio 或 vLLM 部署本地大模型替换 OpenAI API实现完全私有化。性能监控与告警集成监控对 API 响应时间、Milvus 内存使用等进行告警。建议将本文代码作为起点根据你的具体数据和业务需求进行调整和优化。这套技术栈组合灵活、功能强大是构建企业级 RAG 应用的坚实基石。 基于LangChain与Milvus构建RAG系统：DQL实战与本地知识库问答
这次我们来看一个 LangChain 与 Milvus 结合并使用 DQL 进行实战的项目。对于想要构建本地知识库、实现智能问答或文档检索的开发者来说LangChain 提供了强大的应用编排能力而 Milvus 作为高性能向量数据库是存储和检索向量化数据的核心。DQLDomain Query Language则是 Milvus 中用于执行复杂向量检索的查询语言。这个组合的重点不是概念多复杂而是能不能快速搭建一个可运行、可扩展的检索增强生成RAG系统。本文将带你从零开始完成一个基于 LangChain Milvus DQL 的实战项目。我们会重点关注环境搭建、数据准备、向量化存储、DQL 查询构建以及最终的问答应用集成。整个过程会涉及 Python 环境、Docker 部署、显存/内存占用观察以及 API 服务化思路。无论你是想验证技术可行性还是为实际项目做技术选型这篇文章都能提供一套可落地的操作指南。1. 核心能力速览在深入代码之前我们先快速了解这个技术栈的核心能力和门槛。能力项说明项目类型检索增强生成RAG应用原型 / 本地知识库系统核心组件LangChain应用框架、Milvus向量数据库、Embedding 模型如 text2vec、LLM如 OpenAI API 或本地模型主要功能文档切片与向量化、向量存储与检索、基于上下文的智能问答、支持复杂 DQL 过滤查询硬件门槛CPU/内存为主。Milvus 服务本身对 GPU 无硬性要求但 Embedding 模型推理可能需要 GPU 加速。纯 CPU 推理也可行速度较慢。显存占用取决于 Embedding 模型。例如使用text2vec系列小模型CPU 推理仅占用内存若使用bge-large等大模型并启用 GPU可能需要 2-4G 显存。LLM 部分若调用云端 API 则无本地显存压力。部署方式Milvus 推荐 Docker 部署LangChain 应用为 Python 脚本可通过命令行或 Flask/FastAPI 封装为 Web 服务。是否支持 API是。可轻松将核心检索与问答功能封装为 RESTful API 供其他系统调用。是否支持批量任务是。支持批量文档导入、批量向量化入库以及批量查询。适合场景企业知识库问答、法律/金融文档检索、代码库搜索、个人学习笔记管理、内容推荐系统原型开发。2. 适用场景与使用边界这个工具栈适合谁AI 应用开发者希望快速构建具备“记忆”能力的对话机器人或智能助手。数据分析师/研究者需要从大量非结构化文档论文、报告中精准查找信息。个人开发者或小团队寻求低成本、可私有化部署的知识管理解决方案。能解决什么问题“大海捞针”式检索从成千上万份文档中快速找到与用户问题语义最相关的片段。提升大模型回答的准确性与可信度为 LLM 提供来自可靠知识库的参考依据减少“幻觉”。实现对话上下文管理结合 Milvus可以存储和检索多轮对话的历史向量维持对话连贯性。不适合什么场景对延迟要求极苛刻的在线搜索如毫秒级网页搜索。向量检索虽快但叠加 Embedding 和 LLM 生成后整体链路延迟在秒级。完全结构化数据的查询。对于已有明确数据库 Schema 的查询直接使用 SQL 等传统数据库效率更高。无任何编程基础的纯终端用户。本项目需要一定的命令行和 Python 开发能力进行部署和调试。版权与合规边界文档数据确保你拥有所用文档的处理和存储权避免侵犯版权。Embedding/LLM 模型使用开源模型时遵守其对应许可证如 MIT, Apache-2.0。调用商用 API如 OpenAI时需注意其数据使用政策。隐私数据如果处理包含个人隐私或敏感信息的文档务必在本地部署 Milvus 和 Embedding 模型确保数据不出域。3. 环境准备与前置条件开始实战前请确保你的开发环境满足以下要求。操作系统推荐 Linux (Ubuntu 20.04/22.04, CentOS 7) 或 macOS。Windows 可通过 WSL2 获得最佳体验。本文演示环境为 Ubuntu 22.04。容器与运行时Docker Docker Compose用于部署 Milvus 服务。这是最简便的方式。# 检查 Docker 是否安装 docker --version docker-compose --version如果无法使用 Docker也可参考 Milvus 官方文档进行源码编译或下载二进制包但步骤会复杂很多。Python 环境Python 3.8 - 3.11。建议使用 3.9 或 3.10兼容性最广。pip包管理工具。建议使用venv或conda创建独立的虚拟环境。# 创建虚拟环境 python -m venv langchain-milvus-env # 激活环境 (Linux/macOS) source langchain-milvus-env/bin/activate # 激活环境 (Windows) .\langchain-milvus-env\Scripts\activate硬件资源内存至少 8GB。Milvus 运行和 Embedding 推理都会消耗内存文档量越大需求越高。磁盘空间预留 10GB 以上空间用于存储 Docker 镜像、Python 包、模型文件和向量数据。GPU可选如果使用较大的 Embedding 模型如BAAI/bge-large-zh-v1.5并希望加速需要支持 CUDA 的 NVIDIA GPU。CPU 亦可运行。网络需要能访问 PyPI 和 Docker Hub 以下载 Python 包和镜像。如果使用海外 LLM API如 OpenAI需确保网络连通性。4. 安装部署与启动方式我们将按照Milvus 服务 - Python 依赖 - 应用代码的顺序进行部署。4.1 启动 Milvus 服务Docker Compose 方式Milvus 提供了单机版的 Docker Compose 配置文件非常适合开发和测试。下载配置文件# 创建项目目录并进入 mkdir langchain-milvus-demo cd langchain-milvus-demo # 下载最新的 docker-compose.yml 文件 wget https://github.com/milvus-io/milvus/releases/download/v2.3.3/milvus-standalone-docker-compose.yml -O docker-compose.yml注意版本号v2.3.3可能更新请查阅 Milvus 发布页面 获取最新稳定版。启动服务sudo docker-compose up -d这个命令会在后台启动 Milvus 及其依赖的 etcd元数据存储和 MinIO对象存储。验证服务状态sudo docker-compose ps你应该看到三个服务milvus-standalone,etcd,minio的状态都是Up。 也可以检查 Milvus 的日志sudo docker-compose logs milvus-standalone看到Successfully loaded configuration等字样通常表示启动成功。关键信息记录服务地址localhost或127.0.0.1端口19530(gRPC 端口) 和9091(HTTP 端口用于管理)这些信息将在后续的 Python 连接中用到。4.2 安装 Python 依赖在激活的虚拟环境中安装必要的 Python 包。pip install langchain langchain-community pymilvus # 安装文本分割和 Embedding 相关 pip install langchain-text-splitters sentence-transformers # 安装用于演示的文档加载器以 Markdown 和文本文件为例 pip install unstructured # 如果需要 Web 界面或 API可以安装 FastAPI pip install fastapi uvicorn关键包说明langchain核心框架。langchain-community包含社区维护的众多集成工具如与 Milvus 的向量存储接口。pymilvusMilvus 的官方 Python SDK。sentence-transformers提供开源的 Sentence-BERT 模型用于生成文本向量。unstructured强大的文档解析库支持 PDF、Word、PPT、HTML 等多种格式。4.3 准备应用代码结构在项目目录下创建如下文件结构langchain-milvus-demo/ ├── docker-compose.yml # Milvus 服务配置 ├── requirements.txt # Python 依赖列表可选 ├── config.py # 配置文件 ├── data/ # 存放待处理的原始文档 │ └── your_document.md ├── main.py # 主程序文档加载、切分、向量化入库 ├── query.py # 查询与问答程序 └── api.py # 可选FastAPI 封装5. 功能测试与效果验证我们将分步测试整个流程文档处理 - 向量入库 - DQL 检索 - 问答生成。5.1 基础配置与连接测试首先创建config.py来集中管理配置。# config.py import os from sentence_transformers import SentenceTransformer # Milvus 连接配置 MILVUS_HOST localhost # 如果 Milvus 运行在其它机器请修改 MILVUS_PORT 19530 MILVUS_COLLECTION_NAME langchain_docs # 集合类似数据库的表名称 # Embedding 模型配置 # 选择一个合适的模型这里使用一个轻量级中英文模型 EMBEDDING_MODEL_NAME sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2 # 初始化模型首次运行会自动下载 EMBEDDING_MODEL SentenceTransformer(EMBEDDING_MODEL_NAME) # 向量维度取决于所选模型 EMBEDDING_DIMENSION EMBEDDING_MODEL.get_sentence_embedding_dimension() # 文本分割配置 CHUNK_SIZE 500 # 每个文本块的最大字符数 CHUNK_OVERLAP 50 # 块之间的重叠字符数保持上下文连贯 # LLM 配置示例使用 OpenAI也可替换为本地模型 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) # 建议从环境变量读取 OPENAI_API_BASE https://api.openai.com/v1 # 如有代理需修改 OPENAI_MODEL gpt-3.5-turbo然后编写main.py的第一部分测试 Milvus 连接并创建集合。# main.py (部分) from pymilvus import connections, utility, CollectionSchema, FieldSchema, DataType, Collection from config import MILVUS_HOST, MILVUS_PORT, MILVUS_COLLECTION_NAME, EMBEDDING_DIMENSION def connect_to_milvus(): 连接 Milvus 数据库 try: connections.connect(hostMILVUS_HOST, portMILVUS_PORT) print(f✅ 成功连接到 Milvus: {MILVUS_HOST}:{MILVUS_PORT}) # 检查服务状态 print(f Milvus 版本: {utility.get_server_version()}) except Exception as e: print(f❌ 连接 Milvus 失败: {e}) raise def create_collection_if_not_exists(): 如果集合不存在则创建它 if utility.has_collection(MILVUS_COLLECTION_NAME): print(f集合 {MILVUS_COLLECTION_NAME} 已存在。) return Collection(MILVUS_COLLECTION_NAME) # 1. 定义字段 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), # 存储原始文本 FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dimEMBEDDING_DIMENSION), FieldSchema(namesource, dtypeDataType.VARCHAR, max_length255), # 文档来源 FieldSchema(namechunk_index, dtypeDataType.INT64), # 块序号 ] # 2. 创建 Schema schema CollectionSchema(fields, descriptionLangChain 文档向量存储) # 3. 创建集合 collection Collection(nameMILVUS_COLLECTION_NAME, schemaschema) print(f✅ 集合 {MILVUS_COLLECTION_NAME} 创建成功。) # 4. 创建索引加速向量检索 index_params { metric_type: IP, # 内积相似度计算方式。也可用 L2 (欧氏距离) index_type: IVF_FLAT, # 一种经典索引类型适合中小规模数据集 params: {nlist: 128}, } collection.create_index(field_nameembedding, index_paramsindex_params) print(f✅ 在字段 embedding 上创建索引成功。) return collection if __name__ __main__: connect_to_milvus() collection create_collection_if_not_exists() print(基础连接与集合创建测试通过)运行python main.py如果看到成功连接的提示说明 Milvus 服务运行正常并且集合已就绪。5.2 文档加载、切分与向量化入库在main.py中添加文档处理函数。我们以处理data/目录下的 Markdown 文件为例。# main.py (续) import os from langchain_community.document_loaders import DirectoryLoader, UnstructuredMarkdownLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from config import CHUNK_SIZE, CHUNK_OVERLAP, EMBEDDING_MODEL from pymilvus import Collection def load_and_split_documents(data_dir./data): 加载并分割文档 # 使用 DirectoryLoader 加载指定目录下的所有 .md 文件 loader DirectoryLoader( data_dir, glob**/*.md, loader_clsUnstructuredMarkdownLoader, show_progressTrue ) documents loader.load() print(f 从 {data_dir} 加载了 {len(documents)} 个文档。) # 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_sizeCHUNK_SIZE, chunk_overlapCHUNK_OVERLAP, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(f 将文档切分为 {len(splits)} 个文本块。) return splits def generate_embeddings_and_insert(collection: Collection, splits): 生成向量并插入 Milvus texts [split.page_content for split in splits] sources [split.metadata.get(source, unknown) for split in splits] # 批量生成向量 (使用 sentence-transformers) print( 正在生成文本向量...) embeddings EMBEDDING_MODEL.encode(texts, show_progress_barTrue, normalize_embeddingsTrue) print(f✅ 已生成 {len(embeddings)} 个向量维度: {embeddings[0].shape}) # 准备插入数据 data [ texts, # text 字段 embeddings.tolist(), # embedding 字段 sources, # source 字段 list(range(len(texts))) # chunk_index 字段 ] # 插入数据 print( 正在插入数据到 Milvus...) insert_result collection.insert(data) print(f✅ 成功插入 {len(insert_result.primary_keys)} 条数据。) # 将数据从内存刷新到磁盘确保可查询 collection.flush() print( 数据已刷新。) return insert_result.primary_keys if __name__ __main__: # ... 之前的连接和创建集合代码 ... connect_to_milvus() collection create_collection_if_not_exists() # 加载、分割并插入文档 splits load_and_split_documents() if splits: primary_keys generate_embeddings_and_insert(collection, splits) print(f 文档向量化入库完成共处理 {len(primary_keys)} 个文本块。) else: print(⚠️ 未找到待处理的文档请检查 ./data 目录。)运行前请在./data目录下放置一些.md或.txt文件作为测试数据。运行后控制台会显示加载、切分、向量化和插入的进度。5.3 使用 DQL 进行向量检索DQL 是 Milvus 的查询语言功能强大。我们首先使用基础的相似性搜索然后演示加入过滤条件的 DQL 查询。创建query.py。# query.py from pymilvus import connections, Collection from config import MILVUS_HOST, MILVUS_PORT, MILVUS_COLLECTION_NAME, EMBEDDING_MODEL import numpy as np def basic_similarity_search(query_text, top_k3): 基础相似性搜索将问题向量化在 Milvus 中查找最相似的文本块 # 1. 连接并加载集合 connections.connect(hostMILVUS_HOST, portMILVUS_PORT) collection Collection(MILVUS_COLLECTION_NAME) collection.load() # 将集合数据加载到内存以加速搜索 # 2. 将查询文本向量化 query_embedding EMBEDDING_MODEL.encode([query_text], normalize_embeddingsTrue)[0] query_embedding query_embedding.tolist() # 3. 定义搜索参数 search_params { metric_type: IP, # 必须与创建索引时一致 params: {nprobe: 10}, # 搜索时探查的聚类数影响精度和速度 } # 4. 执行搜索 results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limittop_k, output_fields[text, source, chunk_index] # 指定要返回的字段 ) # 5. 解析结果 print(f\n 针对查询{query_text}) print(f 返回了 {len(results[0])} 个最相关结果) for i, hit in enumerate(results[0]): print(f\n--- 结果 {i1} (距离: {hit.distance:.4f}) ---) print(f 来源: {hit.entity.get(source)}) print(f 块索引: {hit.entity.get(chunk_index)}) print(f 内容: {hit.entity.get(text)[:200]}...) # 预览前200字符 return results def dql_search_with_filter(query_text, filter_expression, top_k3): 使用 DQL 进行带过滤条件的搜索 connections.connect(hostMILVUS_HOST, portMILVUS_PORT) collection Collection(MILVUS_COLLECTION_NAME) collection.load() query_embedding EMBEDDING_MODEL.encode([query_text], normalize_embeddingsTrue)[0].tolist() # 定义 DQL 表达式 # 例如source data/document1.md 或 chunk_index 10 dql_expr filter_expression search_params {metric_type: IP, params: {nprobe: 10}} # 关键使用 expr 参数传入 DQL 过滤表达式 results collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limittop_k, exprdql_expr, # DQL 过滤表达式 output_fields[text, source, chunk_index] ) print(f\n DQL 过滤搜索{query_text} | 过滤条件: {filter_expression}) if results and results[0]: for i, hit in enumerate(results[0]): print(f\n--- 结果 {i1} (距离: {hit.distance:.4f}) ---) print(f 来源: {hit.entity.get(source)}) print(f 内容: {hit.entity.get(text)[:200]}...) else: print( ⚠️ 未找到符合过滤条件的匹配项。) return results if __name__ __main__: # 测试基础搜索 basic_similarity_search(LangChain 是什么, top_k2) # 测试带过滤的 DQL 搜索 (假设你的文档来源中有 data/intro.md) dql_search_with_filter( query_text如何安装 Milvus, filter_expressionsource data/intro.md, # 只从特定文档中搜索 top_k2 )运行python query.py你将看到系统首先进行普通的语义搜索然后执行一个增加了来源过滤的 DQL 搜索。这演示了如何将业务逻辑如按文档类型、日期、作者过滤与向量搜索结合。5.4 集成 LangChain 进行问答生成最后我们将检索到的上下文与 LLM 结合生成最终答案。这里以 OpenAI GPT 为例你也可以替换为其他 LangChain 支持的 LLM。# query.py (续) import os from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from config import OPENAI_API_KEY def rag_qa(query_text, top_k3): 检索增强生成先检索再让 LLM 基于检索结果回答 # 1. 检索相关上下文 search_results basic_similarity_search(query_text, top_ktop_k) if not search_results or not search_results[0]: return 抱歉在知识库中未找到相关信息。 # 2. 组装上下文 contexts [] for hit in search_results[0]: contexts.append(hit.entity.get(text)) combined_context \n\n---\n\n.join(contexts) # 3. 构建 Prompt prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个专业的助手请严格根据以下提供的上下文信息来回答问题。如果上下文没有提供足够的信息请直接说“根据已知信息无法回答该问题”。不要编造信息。), (human, 上下文信息\n{context}\n\n问题{question}) ]) # 4. 初始化 LLM 链 llm ChatOpenAI( openai_api_keyOPENAI_API_KEY, model_namegpt-3.5-turbo, temperature0.1 # 低温度答案更确定 ) chain prompt_template | llm | StrOutputParser() # 5. 调用链 answer chain.invoke({context: combined_context, question: query_text}) return answer if __name__ __main__: # 确保设置了 OPENAI_API_KEY 环境变量 if not OPENAI_API_KEY: print(请设置 OPENAI_API_KEY 环境变量。) else: question Milvus 的主要用途是什么 print(f\n 用户问题{question}) answer rag_qa(question, top_k2) print(f\n AI 回答\n{answer})运行这个脚本你会看到系统首先从 Milvus 中检索出与问题最相关的文本片段然后将这些片段作为上下文提交给 LLM 生成一个基于知识库的、引用来源的答案。6. 接口 API 与批量任务6.1 封装为 FastAPI 服务将核心功能封装成 API便于集成。创建api.py。# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import uvicorn from query import rag_qa, basic_similarity_search, dql_search_with_filter from main import load_and_split_documents, generate_embeddings_and_insert from pymilvus import connections, Collection from config import MILVUS_HOST, MILVUS_PORT, MILVUS_COLLECTION_NAME app FastAPI(titleLangChain-Milvus RAG API) class QueryRequest(BaseModel): question: str top_k: Optional[int] 3 filter_expr: Optional[str] None # DQL 过滤表达式 class IngestRequest(BaseModel): data_dir: str app.on_event(startup) def startup_event(): 启动时连接 Milvus try: connections.connect(hostMILVUS_HOST, portMILVUS_PORT) print(API 服务已连接至 Milvus。) except Exception as e: print(fAPI 服务连接 Milvus 失败: {e}) app.get(/) def read_root(): return {message: LangChain-Milvus RAG API 服务运行中} app.post(/query) def query_rag(request: QueryRequest): 问答接口 try: if request.filter_expr: # 使用带过滤的搜索演示实际RAG可能直接用基础搜索 # 这里简化处理实际可根据业务调整 answer rag_qa(request.question, top_krequest.top_k) else: answer rag_qa(request.question, top_krequest.top_k) return {question: request.question, answer: answer} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/search) def search_only(request: QueryRequest): 纯向量搜索接口返回原始片段 try: if request.filter_expr: results dql_search_with_filter(request.question, request.filter_expr, request.top_k) else: results basic_similarity_search(request.question, request.top_k) # 简化返回格式实际应结构化 return {query: request.question, results: results} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/ingest) def ingest_documents(request: IngestRequest): 批量文档入库接口 try: splits load_and_split_documents(request.data_dir) if not splits: return {message: 未找到文档, ingested: 0} connections.connect(hostMILVUS_HOST, portMILVUS_PORT) collection Collection(MILVUS_COLLECTION_NAME) collection.load() pks generate_embeddings_and_insert(collection, splits) return {message: 文档入库成功, ingested: len(pks)} except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)启动服务python api.py。然后你就可以通过http://localhost:8000/docs访问自动生成的 API 文档并进行测试。6.2 批量任务处理对于大量文档的定期更新可以编写脚本进行批处理。# batch_ingest.py import os import schedule import time from main import load_and_split_documents, generate_embeddings_and_insert, connect_to_milvus, create_collection_if_not_exists from pymilvus import Collection from config import MILVUS_COLLECTION_NAME def batch_ingest_job(data_directory): 批量入库任务 print(f[{time.strftime(%Y-%m-%d %H:%M:%S)}] 开始批量入库任务...) try: connect_to_milvus() collection Collection(MILVUS_COLLECTION_NAME) collection.load() splits load_and_split_documents(data_directory) if splits: pks generate_embeddings_and_insert(collection, splits) print(f[{time.strftime(%Y-%m-%d %H:%M:%S)}] 任务完成新增 {len(pks)} 条数据。) else: print(f[{time.strftime(%Y-%m-%d %H:%M:%S)}] 未发现新文档。) except Exception as e: print(f[{time.strftime(%Y-%m-%d %H:%M:%S)}] 任务失败: {e}) if __name__ __main__: DATA_DIR ./data # 监控的目录 # 立即执行一次 batch_ingest_job(DATA_DIR) # 可以配置定时任务例如每6小时执行一次 # schedule.every(6).hours.do(batch_ingest_job, DATA_DIR) # while True: # schedule.run_pending() # time.sleep(60)7. 资源占用与性能观察Milvus 服务资源占用使用 Docker Compose 启动后可以通过docker stats命令观察。单机版 Milvus 在空闲时内存占用约 1-2GB。插入和搜索时会上升。向量索引会占用额外内存和磁盘空间。IVF_FLAT索引相对节省内存但搜索速度比HNSW慢。Embedding 模型推理资源占用CPU 模式以paraphrase-multilingual-MiniLM-L12-v2为例推理一个句子约需 50-100ms内存占用约 500MB。GPU 模式将模型加载到 GPU如cuda:0可大幅加速批量编码。显存占用取决于模型参数量上述小模型约 500MB-1GB。性能优化建议索引选择开发测试用IVF_FLAT生产环境追求速度可用HNSW需要更多内存。批量操作插入向量和搜索时尽量使用批量接口减少网络往返。连接池在高并发 API 服务中使用pymilvus的连接池功能。分段加载如果集合非常大可以使用collection.load(_partition_names[...])分段加载数据到内存。如何监控Milvus 监控访问http://localhost:9091(Milvus 管理界面) 或使用 Prometheus Grafana。Python 进程使用top,htop或nvidia-smi(GPU) 观察资源使用。8. 常见问题与排查方法问题现象可能原因排查方式解决方案连接 Milvus 失败1. Milvus 服务未启动。2. 主机/端口错误。3. 防火墙/网络问题。1.docker-compose ps检查服务状态。2.telnet localhost 19530测试端口。3. 检查config.py中的MILVUS_HOST和MILVUS_PORT。1. 运行docker-compose up -d。2. 修改配置为正确地址。3. 检查防火墙设置。插入数据时报错Collection not loaded集合未加载到内存。在插入或搜索前确认执行了collection.load()。在操作集合前显式调用collection.load()。DQL 过滤查询无结果1. 过滤表达式语法错误。2. 字段名或值错误。3. 确实没有匹配数据。1. 检查 DQL 表达式字符串确保字段名与 Schema 一致。2. 使用collection.query(expr“...”)先测试过滤条件本身。1. 修正表达式注意字符串值用单引号。2. 确认待过滤字段已正确插入数据。Embedding 模型下载慢或失败网络问题或 Hugging Face 连接不稳定。观察下载进度条或错误信息。1. 使用国内镜像源。2. 手动下载模型文件到本地然后从本地路径加载。搜索速度慢1. 未创建索引。2. 索引参数 (nprobe) 设置过大。3. 数据量过大内存不足。1. 检查集合索引collection.index()。2. 降低nprobe值如从 128 降到 16。3. 观察系统内存和 swap 使用。1. 创建合适的索引。2. 调整search_params在精度和速度间权衡。3. 增加内存或使用支持磁盘的索引类型。LLM 回答不准确或“幻觉”1. 检索到的上下文不相关。2.top_k设置太小或太大。3. Prompt 设计不佳。1. 检查query.py中basic_similarity_search返回的片段是否相关。2. 调整top_k(如 3-5)。3. 优化系统 Prompt强调“根据上下文”。1. 优化 Embedding 模型或文本切分策略。2. 实验不同的top_k值。3. 改进 Prompt 工程增加 Few-shot 示例。API 服务启动后无法访问1. 端口被占用。2. 服务绑定到127.0.0.1而非0.0.0.0。3. 服务器防火墙。1.netstat -tlnp | grep 8000。2. 检查uvicorn.run的host参数。3. 检查云服务器安全组规则。1. 更换端口如port8001。2. 确保host0.0.0.0。3. 开放对应端口。9. 最佳实践与使用建议数据预处理是关键文本切分的质量直接影响检索效果。对于不同格式代码、论文、手册可能需要定制化的RecursiveCharacterTextSplitter分隔符。Embedding 模型选型中文场景可考虑BAAI/bge-large-zh-v1.5中英文混合用multilingual模型纯英文用all-MiniLM-L6-v2。在效果和速度间权衡。Milvus 集合管理为不同知识库创建不同集合。定期清理测试数据 (collection.drop())。生产环境考虑使用 Milvus 集群版。DQL 过滤的灵活运用将业务元数据如文档类别、创建时间、权限等级存入 Milvus 的标量字段利用 DQL 在向量检索前进行过滤能极大提升查询的精准度和安全性。版本控制与回滚对config.py、模型版本、Milvus 集合 Schema 进行版本控制。批量更新数据前先备份旧集合。安全性API 服务应增加认证如 API Key。用户上传文档进行向量化前需进行病毒扫描和内容审核。避免将包含敏感信息的 Prompt 和上下文记录到日志。效果评估建立一个小型测试集定期运行监控检索命中率和问答准确率作为迭代优化的依据。10. 总结与下一步这个 LangChain Milvus DQL 的实战项目展示了如何构建一个功能完整的本地知识库问答系统。最值得尝试的点在于它打通了从原始文档到智能问答的端到端流程并且通过 Milvus 的 DQL 功能你可以轻松实现复杂的多条件检索这是很多简单向量库不具备的能力。最先应该验证的功能基础流程跑通确保 Docker 启动 Milvus - 文档入库 - 简单问答整个链路畅通。DQL 过滤查询尝试用source、chunk_index等字段进行过滤感受其如何缩小搜索范围。更换 Embedding 模型换用不同的 Sentence Transformer 模型观察检索结果的质量变化。最容易踩的坑Milvus 连接问题确保 Docker 服务运行且端口正确。向量维度不匹配创建集合时指定的维度必须与 Embedding 模型输出的维度一致。DQL 表达式语法字符串比较要用单引号字段名要完全匹配。后续扩展方向前端界面使用 Gradio 或 Streamlit 快速搭建一个 Web 界面。多路召回与重排序结合关键词搜索如 BM25和向量搜索再用更精细的模型对结果重排序。对话历史将多轮对话的摘要或向量也存入 Milvus实现有记忆的对话。接入本地 LLM使用 Ollama、LM Studio 或 vLLM 部署本地大模型替换 OpenAI API实现完全私有化。性能监控与告警集成监控对 API 响应时间、Milvus 内存使用等进行告警。建议将本文代码作为起点根据你的具体数据和业务需求进行调整和优化。这套技术栈组合灵活、功能强大是构建企业级 RAG 应用的坚实基石。