朋友圈装腔：向量数据库让大白话变古诗词

元旦将近，窗外寒风阵阵，你独倚窗边，思绪万千，于是拿出手机，想发一条朋友圈抒发情怀，顺便展示一下文采。

好不容易按出几个字，“今天的风好大……”，但这展示不出你的文采，于是全部删除。

于是，你灵机一动，如果有一个搜索引擎，能搜索出和“今天的风好大”意思相近的古诗词，岂不妙哉！

使用向量数据库就可以实现，代码还不到100行。这两天我试了试，从零开始安装向量数据库 Milvus，向量化古诗词数据集，然后创建集合、导入数据、创建索引，最后实现语义搜索功能。整个流程跑通之后，确实有点意思——你能用现代大白话，直接搜出意境最接近的唐诗宋词。

下面直接进入正题。

准备工作

首先安装向量数据库 Milvus。Milvus 支持本地、Docker 和 K8s 部署。这里用 Docker 运行 Milvus，所以需要先安装 Docker Desktop。MacOS 系统安装方法：Install Docker Desktop on Mac，Windows 系统安装方法：Install Docker Desktop on Windows。

然后安装 Milvus。下载安装脚本：

curl -sfL https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh -o standalone_embed.sh

运行 Milvus：

standalone_embed.sh start

安装依赖：

pip install pymilvus "pymilvus[model]" torch

下载古诗词数据集[1] TangShi.json，它的格式如下：

[
    {
        "author": "太宗皇帝",
        "paragraphs": [
            "秦川雄帝宅，函谷壮皇居。"
        ],
        "title": "帝京篇十首 一",
        "id": 20000001,
        "type": "唐诗"
    },
    ...
]

准备就绪，正式开始。

01 向量化文本

为了实现语义搜索，第一步就是解决“语义”的表示问题。传统的全文检索靠关键词匹配，机械得很——比如“多大了？”和“年龄是？”关键词重合度很低，但意思几乎一样。反过来，关键词重合但多一个“不”字，意思就大相径庭。这叫什么事儿。

现在流行的方法是嵌入模型（embedding）。通过深度神经网络把文字、图像、声音等非结构化数据提取成一堆数字，这堆数字就叫向量，它能够代表语义。在向量空间中，两个向量之间的距离，就是它们所代表数据的语义相似度——两个诗句语义是否相关，或者两张图片是否很像。这些向量可以存储在 Milvus 向量数据库中，方便高效查询。

理解原理后，先写一个把文本向量化的函数 vectorize_text，方便后面反复调用。

import torch
import json
from pymilvus.model.hybrid import BGEM3EmbeddingFunction

# 1 向量化文本数据
def vectorize_text(text, model_name="BAAI/bge-small-zh-v1.5"):
    # 检查是否有可用的CUDA设备
    device = "cuda:0" if torch.cuda.is_a vailable() else "cpu"
    # 根据设备选择是否使用fp16
    use_fp16 = device.startswith("cuda")
    # 创建嵌入模型实例
    bge_m3_ef = BGEM3EmbeddingFunction(
        model_name=model_name,
        device=device,
        use_fp16=use_fp16
    )
    # 把输入的文本向量化
    vectors = bge_m3_ef.encode_documents(text)
    return vectors

函数里用了嵌入模型 BGEM3EmbeddingFunction，就是它把文本变成向量。准备好后，对整个数据集进行向量化。读取 TangShi.json，将 paragraphs 字段转成向量，然后写入 TangShi_vector.json。如果你是第一次使用 Milvus，运行下面的代码时还会自动安装必要的依赖。

# 读取 json 文件，把paragraphs字段向量化
with open("TangShi.json", 'r', encoding='utf-8') as file:
    data_list = json.load(file)
    # 提取该json文件中的所有paragraphs字段的值
    text = [data['paragraphs'][0] for data in data_list]

# 向量化文本数据
vectors = vectorize_text(text)

# 将向量添加到原始文本中
for data, vector in zip(data_list, vectors['dense']):
    data['vector'] = vector.tolist()

# 将更新后的文本内容写入新的json文件
with open("TangShi_vector.json", 'w', encoding='utf-8') as outfile:
    json.dump(data_list, outfile, ensure_ascii=False, indent=4)

一切顺利的话，你会得到 TangShi_vector.json 文件，里面多了 vector 字段，值是一串浮点数——也就是“向量”。

[
    {
        "author": "太宗皇帝",
        "paragraphs": [
            "秦川雄帝宅，函谷壮皇居。"
        ],
        "title": "帝京篇十首 一",
        "id": 20000001,
        "type": "唐诗",
        "vector": [
            0.005114779807627201,
            0.033538609743118286,
            0.020395483821630478,
            ...
        ]
    },
    {
        "author": "太宗皇帝",
        "paragraphs": [
            "绮殿千寻起，离宫百雉余。"
        ],
        "title": "帝京篇十首 一",
        "id": 20000002,
        "type": "唐诗",
        "vector": [
            -0.06334448605775833,
            0.0017451602034270763,
            -0.0010646708542481065,
            ...
        ]
    },
    ...
]

02 创建集合

接下来，要把向量数据导入 Milvus。首先需要在 Milvus 中创建一个集合来容纳它们——类似于在数据库里创建一张表。

from pymilvus import MilvusClient
# 连接向量数据库，创建client实例
client = MilvusClient(uri="http://localhost:19530")

# 指定集合名称
collection_name = "TangShi"

要注意，避免数据库中存在同名集合造成干扰，所以在创建前先检查并删掉旧的。

# 检查同名集合是否存在，如果存在则删除
if client.has_collection(collection_name):
    print(f"Collection {collection_name} already exists")
    try:
        # 删除同名集合
        client.drop_collection(collection_name)
        print(f"Deleted the collection {collection_name}")
    except Exception as e:
        print(f"Error occurred while dropping collection: {e}")

就像往 Excel 表格里填数据前要先设计好表头，指定各个字段的名称和数据类型一样。向量数据库也要定义这样的结构，它的“表头”就是 schema。

from pymilvus import DataType

# 创建集合模式
schema = MilvusClient.create_schema(
    auto_id=False,
    enable_dynamic_field=True,
    description="TangShi"
)

# 添加字段到schema
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=512)
schema.add_field(field_name="title", datatype=DataType.VARCHAR, max_length=1024)
schema.add_field(field_name="author", datatype=DataType.VARCHAR, max_length=256)
schema.add_field(field_name="paragraphs", datatype=DataType.VARCHAR, max_length=10240)
schema.add_field(field_name="type", datatype=DataType.VARCHAR, max_length=128)

schema 创建好后，就可以创建集合了。

# 创建集合
try:
    client.create_collection(
        collection_name=collection_name,
        schema=schema,
        shards_num=2
    )
    print(f"Created collection {collection_name}")
except Exception as e:
    print(f"Error occurred while creating collection: {e}")

03 入库

现在把刚才处理好的 JSON 数据导入 Milvus。

# 读取和处理文件
with open("TangShi_vector.json", 'r') as file:
    data = json.load(file)
    # paragraphs的值是列表，需要从列表中取出字符串，取代列表，以符合Milvus插入数据的要求
    for item in data:
        item["paragraphs"] = item["paragraphs"][0]

# 将数据插入集合
print(f"正在将数据插入集合：{collection_name}")
res = client.insert(
    collection_name=collection_name,
    data=data
)

导入成功了吗？验证一下。

print(f"插入的实体数量: {res['insert_count']}")

返回结果：

插入的实体数量: 4307

看来是成功了。

04 创建索引

向量已经进入 Milvus，可以搜索了吗？别急，要想搜索快，还得先创建索引。什么是指数？你可以回忆一下——有些大部头图书的最后，通常会有一个索引，列出书中间出现的关键术语以及对应的页码，帮你快速定位。如果没有索引，那就只能从第一页翻起。

Milvus 的索引也是类似的道理。不创建索引虽然也能搜，但速度慢得让人着急——它会逐一比较查询向量与数据库中的每一个向量，计算距离，再找出最近的几个。创建索引之后，搜索效率会大幅提升。

索引有不同的类型，适合不同场景，以后有机会再细聊。这里我们先用一个相对简单的索引类型 IVF_FLAT。另外，距离的计算方式也有多种，这里使用 IP，也就是内积。这些都是索引的参数，先设置好。

# 创建IndexParams对象，用于存储索引的各种参数
index_params = client.prepare_index_params()
# 设置索引名称
vector_index_name = "vector_index"
# 设置索引的各种参数
index_params.add_index(
    # 指定为"vector"字段创建索引
    field_name="vector",
    # 设置索引类型
    index_type="IVF_FLAT",
    # 设置度量类型
    metric_type="IP",
    # 设置索引聚类中心的数量
    params={"nlist": 128},
    # 指定索引名称
    index_name=vector_index_name
)

索引参数准备好后，终于可以正式创建索引了。

print(f"开始创建索引：{vector_index_name}")

# 创建索引
client.create_index(
    # 指定为哪个集合创建索引
    collection_name=collection_name,
    # 使用前面创建的索引参数创建索引
    index_params=index_params
)

验证一下索引是否创建成功。

indexes = client.list_indexes(
    collection_name=collection_name
)
print(f"列出创建的索引：{indexes}")

返回结果：

列出创建的索引：['vector_index']

再来查看索引的详情。

# 查看索引详情
index_details = client.describe_index(
    collection_name=collection_name,
    # 指定索引名称，这里假设使用第一个索引
    index_name="vector_index"
)
print(f"索引vector_index详情：{index_details}")

返回的字典里包含我们之前设置的索引参数，比如 nlist、index_type 和 metric_type。简单来说，nlist 就是把一个数据分片里的向量数据分成若干个聚类，方便后续取出最相关的聚类来加速检索。

索引vector_index详情：{'nlist': '128', 'index_type': 'IVF_FLAT', 'metric_type': 'IP', 'field_name': 'vector', 'index_name': 'vector_index', 'total_rows': 0, 'indexed_rows': 0, 'pending_index_rows': 0, 'state': 'Finished'}

05 加载索引

索引创建成功，但还差一步：需要把集合中的数据和索引从硬盘加载到内存中，因为在内存里搜索更快。

print(f"正在加载集合：{collection_name}")
client.load_collection(collection_name=collection_name)

加载完成后，验证一下。

print(client.get_load_state(collection_name=collection_name))

返回加载状态 Loaded，没问题，加载完成。

{'state': }

06 搜索

前面做了这么多准备，现在终于回到最初的问题了——用现代大白话搜索语义相似的古诗词。

首先，把要搜索的现代白话文提取为向量。

# 获取查询向量
text = "今天的雨好大"
query_vectors = [vectorize_text([text])['dense'][0].tolist()]

然后设置搜索参数，告诉 Milvus 怎么搜。

# 设置搜索参数
search_params = {
    # 设置度量类型
    "metric_type": "IP",
    # 指定在搜索过程中要查询的聚类单元数量，增加nprobe值可以提高搜索精度，但会降低搜索速度
    "params": {"nprobe": 16}
}

最后，还得告诉它要返回哪些字段。

# 指定搜索结果的数量，“limit=3”表示返回最相近的前3个搜索结果
limit = 3
# 指定返回的字段
output_fields = ["author", "title", "paragraphs"]

一切就绪，开始搜索。

res1 = client.search(
    collection_name=collection_name,
    # 指定查询向量
    data=query_vectors,
    # 指定搜索的字段
    anns_field="vector",
    # 设置搜索参数
    search_params=search_params,
    # 指定返回搜索结果的数量
    limit=limit,
    # 指定返回的字段
    output_fields=output_fields
)
print(res1)

结果如下：

data: [
    "[
        {
            'id': 20002740,
            'distance': 0.6542239189147949,
            'entity': {
                'title': '郊庙歌辞 享太庙乐章 大明舞',
                'paragraphs': '旱望春雨，云披大风。',
                'author': '张说'
            }
        },
        {
            'id': 20001658,
            'distance': 0.6228379011154175,
            'entity': {
                'title': '三学山夜看圣灯',
                'paragraphs': '细雨湿不暗，好风吹更明。',
                'author': '蜀太妃徐氏'
            }
        },
        {
            'id': 20003360,
            'distance': 0.6123768091201782,
            'entity': {
                'title': '郊庙歌辞 汉宗庙乐舞辞 积善舞',
                'paragraphs': '云行雨施，天成地平。',
                'author': '张昭'
            }
        }
    ]
]

搜索结果中，id、title 等字段都清楚，distance 是新出现的，指的是搜索结果与查询向量之间的“距离”。我们用的度量类型是 IP 内积，数字越大表示越接近。为了看着方便，写一个输出函数。

# 打印向量搜索结果
def print_vector_results(res):
    # hit是搜索结果中的每一个匹配的实体
    res = [hit["entity"] for hit in res[0]]
    for item in res:
        print(f"title: {item['title']}")
        print(f"author: {item['author']}")
        print(f"paragraphs: {item['paragraphs']}")
        print("-"*50)    
    print(f"数量：{len(res)}")

重新输出结果：

print_vector_results(res1)

这下好读多了。

title: 郊庙歌辞 享太庙乐章 大明舞
author: 张说
paragraphs: 旱望春雨，云披大风。
--------------------------------------------------
title: 三学山夜看圣灯
author: 蜀太妃徐氏
paragraphs: 细雨湿不暗，好风吹更明。
--------------------------------------------------
title: 郊庙歌辞 汉宗庙乐舞辞 积善舞
author: 张昭
paragraphs: 云行雨施，天成地平。
--------------------------------------------------
数量：3

如果你不想限制结果数量，而是返回所有质量符合要求的搜索结果，可以修改搜索参数：

# 修改搜索参数，设置距离的范围
search_params = {
    "metric_type": "IP",
    "params": {
        "nprobe": 16,
        "radius": 0.55,
        "range_filter": 1.0
    }
}

这里增加了 radius 和 range_filter 参数，限制 distance 的范围在 0.55 到 1 之间。然后调整搜索代码，删除 limit 参数。

res2 = client.search(
    collection_name=collection_name,
    # 指定查询向量
    data=query_vectors,
    # 指定搜索的字段
    anns_field="vector",
    # 设置搜索参数
    search_params=search_params,
    # 删除limit参数
    # 指定返回的字段
    output_fields=output_fields
)
print(res2)

可以看到，输出结果的 distance 都大于 0.55。

data: [
    "[
        {
            'id': 20002740,
            'distance': 0.6542239189147949,
            'entity': {
                'author': '张说',
                'title': '郊庙歌辞 享太庙乐章 大明舞',
                'paragraphs': '旱望春雨，云披大风。'
            }
        },
        {
            'id': 20001658,
            'distance': 0.6228379011154175,
            'entity': {
                'author': '蜀太妃徐氏',
                'title': '三学山夜看圣灯',
                'paragraphs': '细雨湿不暗，好风吹更明。'
            }
        },
        {
            'id': 20003360,
            'distance': 0.6123768091201782,
            'entity': {
                'author': '张昭',
                'title': '郊庙歌辞 汉宗庙乐舞辞 积善舞',
                'paragraphs': '云行雨施，天成地平。'
            }
        },
        {
            'id': 20003608,
            'distance': 0.5755923986434937,
            'entity': {
                'author': '李端',
                'title': '鼓吹曲辞 巫山高',
                'paragraphs': '回合云藏日，霏微雨带风。'
            }
        },
        {
            'id': 20000992,
            'distance': 0.5700664520263672,
            'entity': {
                'author': '德宗皇帝',
                'title': '九月十八赐百僚追赏因书所怀',
                'paragraphs': '雨霁霜气肃，天高云日明。'
            }
        },
        {
            'id': 20002246,
            'distance': 0.5583387613296509,
            'entity': {
                'author': '不详',
                'title': '郊庙歌辞 祭方丘乐章 顺和',
                'paragraphs': '雨零感节，云飞应序。'
            }
        }
    ]
]

也许你还想知道最喜欢的李白，有没有和你一样感慨“今天的雨好大”。没问题，加一个 filter 参数来指定只搜索 author 为“李白”的内容。

# 通过表达式过滤字段author，筛选出字段“author”的值为“李白”的结果
filter = f"author == '李白'"

res3 = client.search(
    collection_name=collection_name,
    # 指定查询向量
    data=query_vectors,
    # 指定搜索的字段
    anns_field="vector",
    # 设置搜索参数
    search_params=search_params,
    # 通过表达式实现标量过滤，筛选结果
    filter=filter,
    # 指定返回搜索结果的数量
    limit=limit,
    # 指定返回的字段
    output_fields=output_fields
)
print(res3)

返回的结果是空的。

data: ['[]']

原因很简单——我们之前设的 distance 范围下限是 0.55，看来李白和“雨”的缘分还没到这个阈值得分。把 radius 调整为 0.2，再试试。

data: [
    "[
        {
            'id': 20004246,
            'distance': 0.46472394466400146,
            'entity': {
                'author': '李白',
                'title': '横吹曲辞 关山月',
                'paragraphs': '明月出天山，苍茫云海间。'
            }
        },
        {
            'id': 20003707,
            'distance': 0.4347272515296936,
            'entity': {
                'author': '李白',
                'title': '鼓吹曲辞 有所思',
                'paragraphs': '海寒多天风，白波连山倒蓬壶。'
            }
        },
        {
            'id': 20003556,
            'distance': 0.40778297185897827,
            'entity': {
                'author': '李白',
                'title': '鼓吹曲辞 战城南',
                'paragraphs': '去年战桑干源，今年战葱河道。'
            }
        }
    ]
]

观察一下——distance 在 0.4 左右，远小于之前的 0.55，所以被排除了。而且这几句诗词确实跟“雨”关系不大。

如果你非要在搜索结果中看到带“雨”字的诗句，可以用 query 方法做标量搜索。

# paragraphs字段包含“雨”字
filter = f"paragraphs like '%雨%'"

res4 = client.query(
    collection_name=collection_name,
    filter=filter,
    output_fields=output_fields,
    limit=limit
)
print(res4)

标量查询的代码简洁得多——因为省去了向量搜索相关的参数：查询向量 data、指定搜索字段的 anns_field 和搜索参数 search_params，只剩下 filter。观察结果可以发现，标量查询的数据结构少了一层 []，提取字段时需要注意。

data: [
    "{
        "author": "太宗皇帝",
        "title": "咏雨",
        "paragraphs": "罩云飘远岫，喷雨泛长河。",
        "id": 20000305
    },
    {
        "author": "太宗皇帝",
        "title": "咏雨",
        "paragraphs": "和气吹绿野，梅雨洒芳田。",
        "id": 20000402
    },
    {
        "author": "太宗皇帝",
        "title": "赋得花庭雾",
        "paragraphs": "还当杂行雨，髣髴隐遥空。",
        "id": 20000421
    }
]

filter 表达式还能更灵活，比如同时查询两个字段：指定 author 为“杜甫”，同时 paragraphs 仍包含“雨”字。

filter = f"author == '杜甫' && paragraphs like '%雨%'"

res5 = client.query(
    collection_name=collection_name,
    filter=filter,
    output_fields=output_fields,
    limit=limit
)
print(res5)

返回了杜甫含有“雨”字的诗句：

data: [
    "{
        'title': '横吹曲辞 前出塞九首 七', 
        'paragraphs': '驱马天雨雪，军行入高山。', 
        'id': 20004039, 
        'author': '杜甫'
    }
]

更多标量搜索的表达式可以参考Get & Scalar Query。

写在最后

恭喜，到这里你已经完成了一个“大白话搜古诗词”的 demo！完整的数据集和代码可以从文末的下载链接获取，推荐直接去 Colab 里运行体验。

可能前面的搜索结果并没有让你完全满意，这背后有几个原因：首先，数据集太小了，只有 4000 多个句子，语义更接近的句子可能根本不在其中。其次，嵌入模型虽然支持中文，但古诗词并不是它的专长。这就好比找个翻译帮你和老外交流，翻译虽然懂普通话，但你满嘴四川方言，翻译也只能连蒙带猜，质量可想而知。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：朋友圈装腔：向量数据库让大白话变古诗词要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题