来源|字节跳动云搜索团队
Keyword Search 和 Semantic Search 都存在明显的优劣势,那么是否可以通过组合它们的优点来整体提高搜索的相关性?答案是简单的算术组合并不能收到预期的效果,主要原因有两个:
首先是不同类型查询的评分并不在同一个可比较的维度,因此不能直接进行简单的算术计算。
其次是在分布式检索系统中,评分通常在分片级别,需要对所有分片的评分进行全局归一化处理。
综上,我们需要寻找一种理想的查询类型来解决这些问题,它能单独执行每个查询子句,同时收集分片级别的查询结果,最后对所有查询的评分进行归一化合并后返回最终的结果,这就是混合搜索(Hybrid Search) 方案。
通常一次混合搜索查询可以分为以下几步:
查询阶段:使用混合查询子句进行 Keyword Search 和 Semantic Search。
评分归一化和合并阶段,该阶段在查询阶段之后。
由于每种查询类型都会提供不同范围的评分,该阶段对每一个查询子句的评分结果执行归一化操作,支持的归一化方法有 min_max、l2、rrf。
对归一化后的评分进行组合,组合方法有 arithmetic_mean、geometric_mean、harmonic_mean。
根据组合后的评分对文档重新排序并返回给用户。
实现思路
从前面原理介绍,我们可以看到要实现一个混合检索应用,至少需要用到这些基础技术设施
全文检索引擎
向量检索引擎
用于向量 Embedding 的机器学习模型
将文本、音频、视频等数据转化成向量的数据管道
融合排序
火山引擎云搜索构建在开源的 Elasticsearch 和 OpenSearch 项目上,从第一天上线就支持了完善成熟的文本检索和向量检索能力,同时针对混合搜索场景也进行了一系列的功能迭代和演进,提供了开箱即用的混合搜索解决方案。本文将以图像搜索应用为例,介绍如何借助火山引擎云搜索服务的解决方案快速开发一个混合搜索应用。
其端到端流程概括如下:
配置创建相关对象
Ingestion Pipeline:支持自动调用模型把图片转换向量并存到索引中
Search Pipeline:支持把文本查询语句自动转换成向量以便进行相似度计算
k-NN索引:存放向量的索引
将图像数据集数据写入 OpenSearch 实例,OpenSearch 会自动调用机器学习模型将文本转为 Embedding 向量。
Client 端发起混合搜索查询时,OpenSearch 调用机器学习模型将传入的查询转为 Embedding 向量。
OpenSearch 执行混合搜索请求处理,组合 Keyword Seach 和 Semantic Seach 的评分,返回搜索结果。
方案实战
环境准备
待实例创建完毕,启用 AI 节点。
在模型选择上,可以创建自己的模型,也可以选择公共模型。这里我们选择公共模型,完成配置后,点击立即启动。
至此,准备好了 OpenSearch 实例和混合搜索所依赖的机器学习服务。
数据集准备
使用 Amazon Berkeley Objects Dataset(https://registry.opendata.aws/amazon-berkeley-objects/)作为数据集,数据集无需本地下载,直接通过代码逻辑上传到 OpenSearch,详见下面代码内容。
操作步骤
pip install -U elasticsearch7==7.10.1
pip install -U pandas
pip install -U jupyter
pip install -U requests
pip install -U s3fs
pip install -U alive_progress
pip install -U pillow
pip install -U ipython
# Prepare opensearch info
from elasticsearch7 import Elasticsearch as CloudSearch
from ssl import create_default_context
# opensearch info
opensearch_domain = '{{ OPENSEARCH_DOMAIN }}'
opensearch_port = '9200'
opensearch_user = 'admin'
opensearch_pwd = '{{ OPENSEARCH_PWD }}'
# remote config for model server
model_remote_config = {
"method": "POST",
"url": "{{ REMOTE_MODEL_URL }}",
"params": {},
"headers": {
"Content-Type": "application/json"
},
"advance_request_body": {
"model": "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
}
}
# dimension for knn vector
knn_dimension = 384
# load cer and create ssl context
ssl_context = create_default_context(cafile='./ca.cer')
# create CloudSearch client
cloud_search_cli = CloudSearch([opensearch_domain, opensearch_port],
ssl_context=ssl_context,
scheme="https",
http_auth=(opensearch_user, opensearch_pwd)
)
# index name
index_name = 'index-test'
# pipeline id
pipeline_id = 'remote_text_embedding_test'
# search pipeline id
search_pipeline_id = 'rrf_search_pipeline_test'
# Create ingest pipeline
pipeline_body = {
"description": "text embedding pipeline for remote inference",
"processors": [{
"remote_text_embedding": {
"remote_config": model_remote_config,
"field_map": {
"caption": "caption_embedding"
}
}
}]
}
# create request
resp = cloud_search_cli.ingest.put_pipeline(id=pipeline_id, body=pipeline_body)
print(resp)
创建查询需要使用的 Pipeline,配置好远程模型。
支持的归一化方法和加权求和方法:
归一化方法:min_max
,l2
,rrf
加权求和方法:arithmetic_mean
,geometric_mean
,harmonic_mean
这里选择 rrf 归一化方法。
# Create search pipeline
import requests
search_pipeline_body = {
"description": "post processor for hybrid search",
"request_processors": [{
"remote_embedding": {
"remote_config": model_remote_config
}
}],
"phase_results_processors": [ # normalization and combination
{
"normalization-processor": {
"normalization": {
"technique": "rrf", # the normalization technique in the processor is set to rrf
"parameters": {
"rank_constant": 60 # param
}
},
"combination": {
"technique": "arithmetic_mean", # the combination technique is set to arithmetic mean
"parameters": {
"weights": [
0.4,
0.6
]
}
}
}
}
]
}
headers = {
'Content-Type': 'application/json',
}
# create request
resp = requests.put(
url="https://" + opensearch_domain + ':' + opensearch_port + '/_search/pipeline/' + search_pipeline_id,
auth=(opensearch_user, opensearch_pwd),
json=search_pipeline_body,
headers=headers,
verify='./ca.cer')
print(resp.text)
# Create k-NN index
# create index and set settings, mappings, and properties as needed.
index_body = {
"settings": {
"index.knn": True,
"number_of_shards": 1,
"number_of_replicas": 0,
"default_pipeline": pipeline_id # ingest pipeline
},
"mappings": {
"properties": {
"image_url": {
"type": "text"
},
"caption_embedding": {
"type": "knn_vector",
"dimension": knn_dimension,
"method": {
"engine": "faiss",
"space_type": "l2",
"name": "hnsw",
"parameters": {}
}
},
"caption": {
"type": "text"
}
}
}
}
# create index
resp = cloud_search_cli.indices.create(index=index_name, body=index_body)
print(resp)
读取数据集到内存中,并过滤出部分需要使用的数据。
# Prepare dataset
import pandas as pd
import string
appended_data = []
for character in string.digits[0:] + string.ascii_lowercase:
if character == '1':
break
try:
meta = pd.read_json("s3://amazon-berkeley-objects/listings/metadata/listings_" + character + ".json.gz",
lines=True)
except FileNotFoundError:
continue
appended_data.append(meta)
appended_data_frame = pd.concat(appended_data)
appended_data_frame.shape
meta = appended_data_frame
def func_(x):
us_texts = [item["value"] for item in x if item["language_tag"] == "en_US"]
return us_texts[0] if us_texts else None
meta = meta.assign(item_name_in_en_us=meta.item_name.apply(func_))
meta = meta[~meta.item_name_in_en_us.isna()][["item_id", "item_name_in_en_us", "main_image_id"]]
print(f"#products with US English title: {len(meta)}")
meta.head()
image_meta = pd.read_csv("s3://amazon-berkeley-objects/images/metadata/images.csv.gz")
dataset = meta.merge(image_meta, left_on="main_image_id", right_on="image_id")
dataset.head()
上传数据集到 Opensearch,针对每条数据,传入 image_url 和 caption。无需传入 caption_embedding,将通过远程机器学习模型自动生成。
# Upload dataset
import json
from alive_progress import alive_bar
cnt = 0
batch = 0
action = json.dumps({"index": {"_index": index_name}})
body_ = ''
with alive_bar(len(dataset), force_tty=True) as bar:
for index, row in (dataset.iterrows()):
if row['path'] == '87/874f86c4.jpg':
continue
payload = {}
payload['image_url'] = "https://amazon-berkeley-objects.s3.amazonaws.com/images/small/" + row['path']
payload['caption'] = row['item_name_in_en_us']
body_ = body_ + action + "\n" + json.dumps(payload) + "\n"
cnt = cnt + 1
if cnt == 100:
resp = cloud_search_cli.bulk(
request_timeout=1000,
index=index_name,
body=body_)
cnt = 0
batch = batch + 1
body_ = ''
bar()
print("Total Bulk batches completed: " + str(batch))
以查询 shoes 为例,查询中包含两个查询子句,一个是 match
查询,一个是 remote_neural
查询。查询时将事先创建好的 Search Pipeline 指定为查询参数,Search Pipeline 会将传入的文本转为向量,存储到 caption_embedding 字段,用于后续查询。
# Search with search pipeline
from urllib import request
from PIL import Image
import IPython.display as display
def search(text, size):
resp = cloud_search_cli.search(
index=index_name,
body={
"_source": ["image_url", "caption"],
"query": {
"hybrid": {
"queries": [
{
"match": {
"caption": {
"query": text
}
}
},
{
"remote_neural": {
"caption_embedding": {
"query_text": text,
"k": size
}
}
}
]
}
}
},
params={"search_pipeline": search_pipeline_id},
)
return resp
k = 10
ret = search('shoes', k)
for item in ret['hits']['hits']:
display.display(Image.open(request.urlopen(item['_source']['image_url'])))
print(item['_source']['caption'])
混合搜索演示
以上就是以图像搜索应用为例,介绍如何借助火山引擎云搜索服务的解决方案快速开发一个混合搜索应用的实战过程,欢迎大家登陆火山引擎控制台操作!