Пример кода на Python для векторного поиска

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94from opensearchpy import OpenSearch

# Create a client.
def get_client(hosts: list, user: str = None, password: str = None):
    if user and password:
        return OpenSearch(hosts, http_auth=(user, password), verify_certs=False, ssl_show_warn=False)
    else:
        return OpenSearch(hosts)

# Создать таблицу индекса.
def create(client: OpenSearch, index: str):
    # Информация о сопоставлении индекса
    index_mapping = {
        "settings": {
            "index": {
                "vector": "true",  # Включить векторную функцию.
                "number_of_shards": 1,  # Установите количество шардов индекса по мере необходимости.
                "number_of_replicas": 0  # Установите количество реплик индекса по мере необходимости.
            }
        },
        "mappings": {
            "свойства": {
                "my_vector": {
                    "тип": "вектор",
                    "размерность": 2,
                    "индексация": True,
                    "алгоритм": "GRAPH",
                    "метрика": "евклидовый"
                }
                # Другие поля могут быть добавлены при необходимости.
            }
        }
    }
    рез = клиент.индексы.создать(индекс=индекс, тело=отображение_индекса)
    печать("результат создания индекса: ", res)

# Записать данные.
def write(client: OpenSearch, index: str, vecs: list, bulk_size=500):
    for i in range(0, len(vecs), bulk_size):
        actions = ""
        for vec in vecs[i: i + bulk_size]:
            actions += '{"index": {"_index": "%s"}}\n' % индекс
            действия += '{"my_vector": %s}\n' % строка(вектор)
        клиент.массовый(тело=действия, request_timeout=3600)
    клиент.индексы.обновить(индекс=индекс, request_timeout=3600)
    печать("запись индекса успешна!")

# Запрос векторного индекса.
def поиск(клиент: OpenSearch, индекс: str, запрос: список, size: int):
    # Запрос. Выберите подходящий метод запроса.
    query_body = {
        "size": size,
        "запрос": {
            "вектор": {
                "my_vector": {
                    "вектор": запрос,
                    "topk": размер
                }
            }
        }
    }
    результат = клиент.поиск(индекс=индекс, тело=тело_запроса)
    print("результат поиска индекса: ", res)

# Удалить индекс.
def delete(client: OpenSearch, индекс: str):
    рез = клиент.индексы.удалить(индекс=индекс)
    print("удалить индекс результат: ", рез)

если __name__ == '__main__':
    os_client = get_client(hosts=['http://xx.xx.xx.xx:9200'])

    # Для кластера в режиме безопасности, использующего HTTPS, выполните следующее:
    # os_client = get_client(hosts=['https://xx.xx.xx.xx:9200', 'https://xx.xx.xx.xx:9200'], user='xxxxx', password='xxxxx')

    # Для кластера в режиме безопасности с отключённым HTTPS выполните следующее:
    # os_client = get_client(hosts=['http://xx.xx.xx.xx:9200', 'http://xx.xx.xx.xx:9200'], user='xxxxx', password='xxxxx')

    # Проверьте имя индекса.
    index_name = "my_index"

    # Создайте индекс.
    create(os_client, index=index_name)

    # Записать данные.
    data = [[1.0, 1.0], [2.0, 2.0], [3.0, 3.0]]
    write(os_client, index=index_name, vecs=data)

    # Запрос к индексу.
    query_vector = [1.0, 1.0]
    search(os_client, index=index_name, запрос=query_vector, размер=3)

    # Удалить индекс.
    удалить(os_client, индекс=index_name)