基于火山引擎云搜索的混合搜索实战

实现思路

方案实战

pip install -U elasticsearch7==7.10.1pip install -U pandaspip install -U jupyterpip install -U requestspip install -U s3fspip install -U alive_progresspip install -U pillowpip install -U ipython

# Prepare opensearch infofrom elasticsearch7 import Elasticsearch as CloudSearchfrom ssl import create_default_context# opensearch infoopensearch_domain = '{{ OPENSEARCH_DOMAIN }}'opensearch_port = '9200'opensearch_user = 'admin'opensearch_pwd = '{{ OPENSEARCH_PWD }}'# remote config for model servermodel_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 vectorknn_dimension = 384# load cer and create ssl contextssl_context = create_default_context(cafile='./ca.cer')# create CloudSearch clientcloud_search_cli = CloudSearch([opensearch_domain, opensearch_port],                               ssl_context=ssl_context,                               scheme="https",                               http_auth=(opensearch_user, opensearch_pwd)                               )# index nameindex_name = 'index-test'
# pipeline idpipeline_id = 'remote_text_embedding_test'# search pipeline idsearch_pipeline_id = 'rrf_search_pipeline_test'

# Create ingest pipelinepipeline_body = {    "description": "text embedding pipeline for remote inference",    "processors": [{        "remote_text_embedding": {            "remote_config": model_remote_config,            "field_map": {                "caption": "caption_embedding"            }        }    }]}# create requestresp = cloud_search_cli.ingest.put_pipeline(id=pipeline_id, body=pipeline_body)print(resp)

# Create search pipelineimport requestssearch_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 requestresp = 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 indexresp = cloud_search_cli.indices.create(index=index_name, body=index_body)print(resp)

# Prepare dataset
import pandas as pdimport 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.shapemeta = 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()

# Upload dataset
import json
from alive_progress import alive_bar
cnt = 0batch = 0action = 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))

# Search with search pipeline
from urllib import request
from PIL import Imageimport 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 = 10ret = search('shoes', k)
for item in ret['hits']['hits']:    display.display(Image.open(request.urlopen(item['_source']['image_url'])))    print(item['_source']['caption'])

混合搜索演示