milvus向量数据库使用尝试

一.背景

在大语言模型（LLM）、计算机视觉、推荐系统等人工智能应用落地过程中，非结构化数据（文本、图片、音频、视频）的相似性检索成为核心需求 —— 这类数据需先通过模型转化为高维向量，再通过向量相似性计算实现快速匹配。Milvus 作为开源的分布式向量数据库，专为高维向量的存储、索引与相似性检索设计，能够解决传统关系型数据库、键值数据库在向量处理上的性能瓶颈。企业与开发者开展 “Milvus 向量数据库使用尝试” 的需求，源于传统数据存储方式在向量处理场景的痛点，以及对 “高效、可扩展、低成本” 实现非结构化数据检索的核心诉求。

1.传统数据存储方式处理向量的核心痛点

随着 AI 应用的普及，高维向量数据（如文本的 Embedding 向量、图片的特征向量）的规模呈指数级增长，传统数据存储方式在处理这类数据时，暴露出难以逾越的短板：

1. 向量检索性能极差，无法支撑大规模场景关系型数据库（MySQL/PostgreSQL）、键值数据库（Redis）均未针对向量相似性检索做优化：若要从百万级、千万级向量中检索与目标向量最相似的 Top-K 结果，需对所有向量进行全量遍历计算相似度（如余弦相似度、欧氏距离），时间复杂度为 O (n)。例如，在百万级 768 维向量数据中，全量遍历检索需数秒甚至数十秒，完全无法满足实时应用（如智能问答、商品推荐）的毫秒级响应要求。
2. 不支持高效向量索引，资源消耗过大传统数据库无法构建针对高维向量的专用索引（如 IVF_FLAT、HNSW、ANNOY），只能将向量以二进制或数组形式存储，检索时不仅需要消耗大量 CPU 算力进行相似度计算，还会占用海量内存加载向量数据，导致集群资源紧张，且检索效率随向量规模增长呈线性下降。
3. 分布式扩展能力不足，适配海量数据难当向量数据规模达到亿级甚至十亿级时，单节点数据库无法承载数据存储与检索压力，而传统数据库的分布式扩展多针对结构化数据设计，无法实现向量数据的分片存储、并行检索与负载均衡。例如，Redis 集群虽能分片存储向量，但无法跨节点协同完成相似性检索，最终仍需聚合所有节点数据进行全量计算。
4. 功能单一，无法适配 AI 应用的复杂需求AI 应用不仅需要向量相似性检索，还需结合结构化数据进行过滤（如 “检索与‘手机’相关的文本向量，且发布时间在 2025 年”）、向量更新 / 删除、多向量组合检索等操作。传统数据库仅能单独处理结构化数据或向量数据，无法实现两者的联合查询，需额外编写代码进行数据关联，开发成本高且性能低下。
5. 与 AI 生态集成性差，开发门槛高传统数据库缺乏与主流 Embedding 模型（如 OpenAI Embedding、BERT、CLIP）、AI 框架（LangChain、LlamaIndex）的原生集成能力，开发者需手动完成向量生成、存储、检索的全流程代码开发，还需自行处理向量数据的序列化、分片、容错，非专业人员难以快速上手。

2.Milvus 向量数据库使用尝试的核心价值

Milvus 作为专为向量检索设计的数据库，从底层架构上解决了传统存储方式的痛点。开发者与企业开展 “使用尝试”，本质是验证其在向量存储、检索、扩展等方面的能力，探索将其融入 AI 应用的可行性，核心价值体现在：

1. 极致的向量检索性能，支撑实时应用Milvus 支持多种高性能向量索引算法，可将向量检索的时间复杂度从 O (n) 降低至 O (log n) 或近似常数级。例如，采用 HNSW 索引时，千万级 768 维向量的 Top-K 检索响应时间可控制在 10ms 以内，完全满足智能问答、实时推荐等场景的实时性要求。即使是亿级向量数据，通过分布式检索也能保持毫秒级响应。
2. 专用向量索引与优化，降低资源消耗Milvus 内置了针对不同场景的向量索引（如 IVF_FLAT 适用于高精度检索，HNSW 适用于高速度检索，SCANN 适用于超大规模向量），并对索引构建、相似度计算进行了底层优化（如利用 GPU 加速、向量化计算）。相比全量遍历，索引检索可减少 90% 以上的 CPU 与内存消耗，大幅降低集群部署成本。
3. 分布式架构，适配海量向量数据Milvus 采用分片存储、并行检索的分布式架构，支持向量数据的水平扩展：可通过增加节点实现存储容量与检索性能的线性提升，轻松承载亿级、十亿级甚至百亿级向量数据。同时，Milvus 支持数据的动态扩容与缩容，无需中断服务，适配企业向量数据的快速增长。
4. 结构化与向量数据联合查询，适配复杂场景Milvus 支持为向量数据附加结构化属性（如文本的分类、图片的标签、数据的时间戳），并提供 SQL 风格的查询语法，实现 “结构化过滤 + 向量相似性检索” 的联合查询。
5. 深度集成 AI 生态，降低开发门槛Milvus 与主流 AI 工具链深度集成：支持 LangChain、LlamaIndex 等大模型应用框架的原生对接，可直接作为向量存储组件使用；兼容 OpenAI、Hugging Face 等主流 Embedding 模型的向量输出格式；还提供 Python/Java/Go 等多语言 SDK，封装了向量存储、索引构建、检索的核心逻辑，开发者只需几行代码即可完成基础功能开发，大幅降低使用门槛。

3.Milvus 向量数据库使用尝试的典型场景

1. 大模型应用的知识库问答（RAG）开发者尝试将 Milvus 作为 RAG 架构的向量存储：将企业文档（PDF、Word、Markdown）转化为 Embedding 向量后存入 Milvus，用户提问时生成向量并检索相似文档，再将文档内容作为上下文输入大模型生成回答。通过尝试验证 Milvus 在检索速度、准确率上的表现，优化 RAG 系统的响应效率与回答质量。
2. 计算机视觉的相似图片检索电商、安防领域的开发者尝试将商品图片、监控图片的特征向量存入 Milvus，实现 “上传图片→检索相似图片” 的功能：例如电商平台的 “拍立淘”，安防系统的人脸相似性检索，验证 Milvus 在百万级、千万级图片向量中的检索性能与精度。
3. 推荐系统的个性化推荐互联网企业尝试将用户行为特征向量、商品特征向量存入 Milvus，通过向量相似性检索为用户推荐相似商品、内容：例如短视频平台的视频推荐，电商平台的商品推荐，验证 Milvus 在高并发场景下的检索稳定性与扩展性。
4. 自然语言处理的文本相似性匹配开发者尝试将新闻、评论、文档的文本向量存入 Milvus，实现文本查重、相似问答匹配、舆情监控等功能：例如客服系统的相似问题匹配，新闻平台的重复新闻检测，验证 Milvus 在文本向量处理上的适配性。
5. 多模态数据检索开发者尝试将文本、图片、音频的多模态向量存入 Milvus，实现跨模态的相似性检索：例如输入文本 “红色手机”，检索相似的图片与视频，验证 Milvus 对多维度、多类型向量的处理能力。

4.使用尝试的核心目标与意义

开发者与企业开展 Milvus 使用尝试，并非单纯的技术验证，而是为了：

1. 技术选型验证：对比 Milvus 与其他向量数据库（如 Pinecone、Weaviate、FAISS）在性能、成本、易用性上的差异，确定适配自身业务的向量存储方案；
2. 性能基准测试：针对业务场景的向量规模（百万级 / 千万级），测试 Milvus 在不同索引算法、硬件配置下的检索速度、准确率、资源消耗，为生产环境部署提供参数依据；
3. 业务适配性验证：验证 Milvus 能否与现有 AI 应用、业务系统无缝集成，解决实际业务中的向量检索痛点；
4. 成本评估：评估 Milvus 分布式部署的硬件成本、运维成本，对比传统存储方式的投入产出比，确定商业化落地的可行性。

综上，Milvus 向量数据库使用尝试的需求，源于 AI 应用对高维向量数据高效处理的迫切诉求，以及传统数据存储方式在向量场景的全面失效。这一尝试不仅是开发者探索新技术的过程，更是企业将 AI 应用从 “演示级原型” 推向 “生产级落地” 的关键步骤 —— 通过验证 Milvus 的能力，解决非结构化数据检索的性能瓶颈，为 AI 应用的规模化部署提供核心数据存储支撑。

二.具体实现

1.安装包

pip install pymilvus

2.引入依赖

import sys import os import random import numpy as np from pymilvus import ( connections, db, Collection, FieldSchema, CollectionSchema, DataType, utility ) # 解决Windows控制台中文乱码问题 if sys.platform == 'win32': # 设置标准输出编码为UTF-8 if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') if hasattr(sys.stderr, 'reconfigure'): sys.stderr.reconfigure(encoding='utf-8') # 设置环境变量 os.environ['PYTHONIOENCODING'] = 'utf-8'

3.创建连接

host = "xxx" port = "19530" database_name = "test_db_01" collection_name = "test_collection_01" # 连接到Milvus print(f"正在连接到 Milvus {host}:{port}...") connections.connect( alias="default", host=host, port=port ) print("连接成功！\n")

4.列举所有数据库

databases = db.list_database()

5.创建数据库

db.create_database(database_name)

6.创建Collection

db.using_database(database_name) # 定义Collection Schema print(f"正在创建Collection {collection_name}...") # 定义字段 fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False), FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=64), FieldSchema(name="a", dtype=DataType.INT64), FieldSchema(name="b", dtype=DataType.FLOAT) ] # 创建Schema schema = CollectionSchema( fields=fields, description="测试Collection，包含64维向量和标量字段a、b" ) # 检查Collection是否存在，如果存在则删除 if utility.has_collection(collection_name): print(f"Collection {collection_name} 已存在，正在删除...") utility.drop_collection(collection_name) # 创建Collection collection = Collection( name=collection_name, schema=schema )

7.插入数据

# 生成随机数据并插入 print("正在生成随机数据并插入...") data_id = random.randint(1, 1000000) data_vector = np.random.random((1, 64)).tolist()[0] # 64维随机向量 data_a = random.randint(1, 100) data_b = random.uniform(0.0, 100.0) data = [ [data_id], [data_vector], [data_a], [data_b] ] print(f"插入数据:") print(f" id: {data_id}") print(f" vector: {data_vector[:5]}... (显示前5维)") print(f" a: {data_a}") print(f" b: {data_b:.2f}") collection.insert(data) print("数据插入成功！\n")

8.创建索引

print("正在为向量字段创建索引...") index_params = { "metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": 128} } collection.create_index( field_name="vector", index_params=index_params ) print("索引创建成功！\n")

9.检索数据

# 加载Collection到内存 print("正在加载Collection到内存...") collection.load() print("加载成功！\n") # 进行查询 print("正在查询数据...") # 使用向量搜索 search_vector = [data_vector] # 使用插入的向量进行搜索 search_params = { "metric_type": "L2", "params": {"nprobe": 10} } results = collection.search( data=search_vector, anns_field="vector", param=search_params, limit=10, output_fields=["id", "a", "b"] ) print("查询结果:") for hits in results: for hit in hits: print(f" id: {hit.id}, 距离: {hit.distance:.4f}, a: {hit.entity.get('a')}, b: {hit.entity.get('b'):.2f}") # 也可以使用表达式查询 print("\n使用表达式查询:") query_result = collection.query( expr=f"id == {data_id}", output_fields=["id", "a", "b", "vector"] ) for result in query_result: print(f" id: {result['id']}") print(f" a: {result['a']}") print(f" b: {result['b']:.2f}") print(f" vector: {result['vector'][:5]}... (显示前5维)")