为什么需要向量数据库？

在 RAG 程序中。我们通常需要将外部知识库切分成片段，并答案。

Milvus 是专为 AI 应用设计的开源向量数据库，支持海量向量的高效存储和检索。这篇文章基于 Milvus 的 Python SDK和 LangChain 的嵌入模型接口，实战操作整个流程。怎么说呢，

. 环境准备

安装 Docker 并启动 Milvus

• 痛点一：Windows 原生不支持 Docker Desktop。需要先安装 WSL，步骤繁琐后容易出现“未启用 WSL”错误。从解决办法来看，先运行 wsl --install 并重启电脑。
• 痛点二：镜像下载慢或被墙。可配置国内镜像源，在 Docker Desktop → Settings → Docker Engine 中替换 "registry-mirrors"
• 步骤：
• 打开管理员 PowerShell：Win+X → A.
• 下载并执行官方脚本：

  Invoke-WebRequest https://raw.githubusercontent.com/milvus-io/milvus/refs/heads/master/scripts/standalone_embed.bat -OutFile standalone.bat
  .\standalone.bat start
  

• Mildus 默认监听 localhost:19530.
• 确认服务已启动后即可在 Docker 内部管理 Milvus。

安装 Python 依赖

pip install pymilvus langchain python-dotenv rich

获取嵌入模型 API Key

*请根据所选服务自行注册并获取 API Key，例如 OpenRouter、OpenAI 等。*

Create a file named .env:

OPENROUTER_API_KEY=your-api-key
OPENROUTER_BASE_URL=https://openrouter.ai/api/v1

. 建立连接并创建数据库 / Collection

建立客户端连接

from pymilvus import MilvusClient
client = MilvusClient
print) # 默认有 'default'

创建自定义数据库

db_name = "rag_demo"
if db_name not in client.list_databases:
client.create_database
client.use_database

创建 Collection

*痛点三：dimension 必须与嵌入模型输出一致，否则插入报错*。

collection_name = "docs"
if collection_name not in client.list_collections:
client.create_collection(
collection_name=collection_name,dimension=1024,# bge-m3 输出维度
metric_type="COSINE",# 推荐用于语义检索
enable_dynamic_field=True # 可动态添加字段
)
metadata = client.describe_collection
print

查看 Collection 元数据示例输出：

{
"fields":,"index_params":
}

. 加载嵌入模型并生成向量

加载环境变量与初始化 Embedding 模型

import os
from dotenv import load_dotenv
from langchain.embeddings import init_embeddings
load_dotenv
embedding_model = init_embeddings(
model="openai:bge-m3"。api_key=os.getenv,base_url=os.getenv
)

生成示例文本向量：

texts =
vectors = embedding_model.embed_documents
print} 条向量")

. 插入数据还有持久化操作

构造插入格式的数据列表

 data =,"text": texts,"source": "demo" } for i in range) ] insert_res = client.upsert print

• 注意： upsert 会插入新数据或更新已存在的 id。数据插入后不会立即写磁盘。可手动调用 flush 强制落盘：

   client.flush 

  使用 get_collection_stats 检查行数是否正确：

   stats = client.get_collection_stats print

. 查询与相似度检索

遍历所有数据——大数据场景推荐使用此方式

 iterator = client.query_iterator while True : rows = iterator.next if not rows : break for row in rows : print iterator.close 

通过主键批量查询示例：

 res = client.get print

相似度检索 - 使用者提问场景举例：

痛点四：如果没有正确设置 metric_type 或者 vector 未归一化，检索结果可能不准确。 建议使用 COSINE 并确保 Embedding 输出已归一化。

 query_text = “什么是向量数据库？” query_vector = embedding_model.embed_query search_res = client.search for hit in search_res: print

输出示例这方面。

示例 JSON：

{ “distance”:0.95,“entity”:{“id”:1,“text”:“Milvus 是一个适合 AI 应用的向量数据库。”} }

根据返回值排序即可得到最佳上下文片段。

. 清理资源

• 删除 Collection： client.dropcollection
• 删除 Database： client.dropdatabase

. 常见坑 & 提示

1. 维度必须严格匹配 * - 否则 upsert 会报错，请务必检查 Embedding 输出维度与 Collection dimension 一致。如使用 bge-m3，请设置 dimension 为 1024；如果更换其他模型请调整此值。
2. metrictype 推荐 COSINE * - 对于语义相似度检测更合适；若业务更偏距离计算，可改为 L2 或 IP，但需确保数据归一化。
3. enabledynamic_field=True * - 开启后可随意添加字段。无需提前定义 schema，极大提高灵活性。
4. 批处理操作 * - upsert、search 均支持批次可明显提高吞吐率；单条操作要注意网络延迟累计过高。
5. 生产环境建议自定义索引 * - 本例采用 FLAT 索引速度慢，可改为 IVF 或 HNSW 加速检索；具体参数参照官方文档，

---