Milvus实战：基于向量检索的问答系统搭建完整指南

此前我们介绍了Milvus向量数据库的安装部署，现在直接进入一个实战案例——如何利用Milvus与Huggingface的大模型构建一个简洁高效的问答系统。首先，Milvus的相似性搜索能力强大，应用场景极为广泛，具体包括：

• 图像相似性搜索：从海量图片库中快速定位最相似的图像
• 视频相似度搜索：将关键帧转化为向量，实现数十亿视频的近乎实时检索
• 音频相似度搜索：快速匹配语音、音乐与音效片段
• 推荐系统：基于用户行为数据精准推荐内容
• 问答系统：构建交互式数字问答机器人，自动回答用户提问
• DNA序列分类：毫秒级比对DNA序列，完成基因分类识别
• 文本搜索引擎：通过关键词匹配帮助用户高效查找信息

本文聚焦于问答系统，详细演示Milvus与Huggingface大模型如何协同工作，实现智能问答。

Huggingface

Hugging Face是自然语言处理领域广泛使用的开源平台，提供大量预训练模型与数据集。本次使用的模型与数据集如下：

Model：bert-base-uncased（来自Google的BERT模型）

Dataset：squad（斯坦福问答数据集）

0. 环境准备

请先安装所需依赖：

pip install transformers datasets pymilvus torch

1. 创建Collection

确认本地Milvus服务已启动，然后在Milvus中创建Collection并建立索引：

from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility

DATASET = 'squad'  # Huggingface Dataset to use
MODEL = 'bert-base-uncased'  # Transformer to use for embeddings
TOKENIZATION_BATCH_SIZE = 1000  # Batch size for tokenizing operation
INFERENCE_BATCH_SIZE = 64  # batch size for transformer
INSERT_RATIO = .001  # How many titles to embed and insert
COLLECTION_NAME = 'huggingface_db'  # Collection name
DIMENSION = 768  # Embeddings size
LIMIT = 10  # How many results to search for

URI = "http://192.168.153.100:19530"
TOKEN = "root:Milvus"

connections.connect(uri=URI, token=TOKEN)

if utility.has_collection(COLLECTION_NAME):
    utility.drop_collection(COLLECTION_NAME)

fields = [
    FieldSchema(name='id', dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name='original_question', dtype=DataType.VARCHAR, max_length=1000),
    FieldSchema(name='answer', dtype=DataType.VARCHAR, max_length=1000),
    FieldSchema(name='original_question_embedding', dtype=DataType.FLOAT_VECTOR, dim=DIMENSION)
]
schema = CollectionSchema(fields=fields)
collection = Collection(name=COLLECTION_NAME, schema=schema)

index_params = {
    'metric_type':'L2',
    'index_type':"IVF_FLAT",
    'params':{"nlist":1536}
}
collection.create_index(field_name="original_question_embedding", index_params=index_params)
print("Create index done.")

2. 插入数据

Collection创建完成后，即可向其中插入数据。整体流程分为三步：

1. 对数据集中的问题进行分词处理
2. 将分词结果转换为向量表示
3. 将问题、向量以及对应的答案一并插入Milvus

from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility
from datasets import load_dataset_builder, load_dataset, Dataset
from transformers import AutoTokenizer, AutoModel
from torch import clamp, sum


DATASET = 'squad'
MODEL = 'bert-base-uncased'
TOKENIZATION_BATCH_SIZE = 1000
INFERENCE_BATCH_SIZE = 64
INSERT_RATIO = .001
COLLECTION_NAME = 'huggingface_db'
DIMENSION = 768
LIMIT = 10

URI = "http://192.168.153.100:19530"
TOKEN = "root:Milvus"

connections.connect(uri=URI, token=TOKEN)

data_dataset = load_dataset(DATASET, split='all')
data_dataset = data_dataset.train_test_split(test_size=INSERT_RATIO, seed=42)['test']
data_dataset = data_dataset.map(lambda val: {'answer': val['answers']['text'][0]}, remove_columns=['answers'])

tokenizer = AutoTokenizer.from_pretrained(MODEL)

def tokenize_question(batch):
    results = tokenizer(batch['question'], add_special_tokens = True, truncation = True, padding = "max_length", return_attention_mask = True, return_tensors = "pt")
    batch['input_ids'] = results['input_ids']
    batch['token_type_ids'] = results['token_type_ids']
    batch['attention_mask'] = results['attention_mask']
    return batch

data_dataset = data_dataset.map(tokenize_question, batch_size=TOKENIZATION_BATCH_SIZE, batched=True)
data_dataset.set_format('torch', columns=['input_ids', 'token_type_ids', 'attention_mask'], output_all_columns=True)

model = AutoModel.from_pretrained(MODEL)
def embed(batch):
    sentence_embs = model(
                input_ids=batch['input_ids'],
                token_type_ids=batch['token_type_ids'],
                attention_mask=batch['attention_mask']
                )[0]
    input_mask_expanded = batch['attention_mask'].unsqueeze(-1).expand(sentence_embs.size()).float()
    batch['question_embedding'] = sum(sentence_embs * input_mask_expanded, 1) / clamp(input_mask_expanded.sum(1), min=1e-9)
    return batch

data_dataset = data_dataset.map(embed, remove_columns=['input_ids', 'token_type_ids', 'attention_mask'], batched = True, batch_size=INFERENCE_BATCH_SIZE)


fields = [
    FieldSchema(name='id', dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name='original_question', dtype=DataType.VARCHAR, max_length=1000),
    FieldSchema(name='answer', dtype=DataType.VARCHAR, max_length=1000),
    FieldSchema(name='original_question_embedding', dtype=DataType.FLOAT_VECTOR, dim=DIMENSION)
]
schema = CollectionSchema(fields=fields)
collection = Collection(name=COLLECTION_NAME, schema=schema)
collection.load()
def insert_function(batch):
    insertable = [
        batch['question'],
        [x[:995] + '...' if len(x) > 999 else x for x in batch['answer']],
        batch['question_embedding'].tolist()
    ]
    collection.insert(insertable)

data_dataset.map(insert_function, batched=True, batch_size=64)
collection.flush()
print("Insert data done.")

3. 提问

数据入库完成后，即可向Milvus发起查询。系统会返回与提问最相似的若干条原始记录及其对应答案。

from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility
from datasets import load_dataset_builder, load_dataset, Dataset
from transformers import AutoTokenizer, AutoModel
from torch import clamp, sum


DATASET = 'squad'
MODEL = 'bert-base-uncased'
TOKENIZATION_BATCH_SIZE = 1000
INFERENCE_BATCH_SIZE = 64
INSERT_RATIO = .001
COLLECTION_NAME = 'huggingface_db'
DIMENSION = 768
LIMIT = 10

URI = "http://192.168.153.100:19530"
TOKEN = "root:Milvus"

connections.connect(uri=URI, token=TOKEN)
fields = [
    FieldSchema(name='id', dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name='original_question', dtype=DataType.VARCHAR, max_length=1000),
    FieldSchema(name='answer', dtype=DataType.VARCHAR, max_length=1000),
    FieldSchema(name='original_question_embedding', dtype=DataType.FLOAT_VECTOR, dim=DIMENSION)
]
schema = CollectionSchema(fields=fields)
collection = Collection(name=COLLECTION_NAME, schema=schema)
collection.load()

tokenizer = AutoTokenizer.from_pretrained(MODEL)
def tokenize_question(batch):
    results = tokenizer(batch['question'], add_special_tokens = True, truncation = True, padding = "max_length", return_attention_mask = True, return_tensors = "pt")
    batch['input_ids'] = results['input_ids']
    batch['token_type_ids'] = results['token_type_ids']
    batch['attention_mask'] = results['attention_mask']
    return batch

model = AutoModel.from_pretrained(MODEL)
def embed(batch):
    sentence_embs = model(
                input_ids=batch['input_ids'],
                token_type_ids=batch['token_type_ids'],
                attention_mask=batch['attention_mask']
                )[0]
    input_mask_expanded = batch['attention_mask'].unsqueeze(-1).expand(sentence_embs.size()).float()
    batch['question_embedding'] = sum(sentence_embs * input_mask_expanded, 1) / clamp(input_mask_expanded.sum(1), min=1e-9)
    return batch

questions = {'question':['When was chemistry invented?', 'When was Eisenhower born?']}
question_dataset = Dataset.from_dict(questions)

question_dataset = question_dataset.map(tokenize_question, batched = True, batch_size=TOKENIZATION_BATCH_SIZE)
question_dataset.set_format('torch', columns=['input_ids', 'token_type_ids', 'attention_mask'], output_all_columns=True)
question_dataset = question_dataset.map(embed, remove_columns=['input_ids', 'token_type_ids', 'attention_mask'], batched = True, batch_size=INFERENCE_BATCH_SIZE)

def search(batch):
    res = collection.search(batch['question_embedding'].tolist(), anns_field='original_question_embedding', param = {}, output_fields=['answer', 'original_question'], limit = LIMIT)
    overall_id = []
    overall_distance = []
    overall_answer = []
    overall_original_question = []
    for hits in res:
        ids = []
        distance = []
        answer = []
        original_question = []
        for hit in hits:
            ids.append(hit.id)
            distance.append(hit.distance)
            answer.append(hit.entity.get('answer'))
            original_question.append(hit.entity.get('original_question'))
        overall_id.append(ids)
        overall_distance.append(distance)
        overall_answer.append(answer)
        overall_original_question.append(original_question)
    return {
        'id': overall_id,
        'distance': overall_distance,
        'answer': overall_answer,
        'original_question': overall_original_question
    }
question_dataset = question_dataset.map(search, batched=True, batch_size = 1)
for x in question_dataset:
    print()
    print('Question:')
    print(x['question'])
    print('Answer, Distance, Original Question')
    for x in zip(x['answer'], x['distance'], x['original_question']):
        print(x)

输出结果示例如下（以“When was chemistry invented?”为例）：

Question:
When was chemistry invented?
Answer, Distance, Original Question
('until 1870', tensor(34.1343), 'When did the Papal States exist?')
('1787', tensor(35.6709), 'When was the Tower constructed?')
('October 1992', tensor(38.9599), 'When were free elections held?')
('6,000 years', tensor(44.8685), 'How old did biblical scholars think the Earth was?')
('Poland, Bulgaria, the Czech Republic, Slovakia, Hungary, Albania, former East Germany and Cuba', tensor(45.7986), 'Where was Russian schooling mandatory in the 20th century?')
('1992', tensor(47.0607), 'In what year was the Premier League created?')
('1981', tensor(48.3999), "When was ZE's Mutant Disco released?")
('Medieval Latin', tensor(50.9613), "What was the Latin of Charlemagne's era later known as?")
('taxation', tensor(51.0803), 'How did Hobson argue to rid the world of imperialism?')
('military education', tensor(52.5620), 'What Prussian system was superior to the French example?')

从输出结果可以看出，系统返回了与提问最相似的10个原始问题及其答案。尽管部分答案（如“until 1870”）并不准确，但作为一次基于相似性搜索的实践，整个流程已成功跑通。在实际应用中，可以通过微调模型、优化嵌入方式或增加数据量来进一步提升准确率。

除了与Huggingface模型集成外，你还可以替换为OpenAI或其他预训练模型，进而实现图像搜索、音频搜索等更多场景。核心思路始终一致：利用向量数据库进行相似性匹配，借助预训练模型完成特征提取。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：Milvus实战：基于向量检索的问答系统搭建完整指南要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题