В предыдущей статье мы говорили о текстовом поиске, а в сегодняшней я расскажу о векторном (семантическом) поиске.
Итак, если мы используем OpenSearch, в Yandex Cloud представляется логичным использовать модели вложений этого же облака.
Рассмотрим пример yc-os-explicit-embedding-cloud-function.py.
Этот код можно запустить как Python Cloud Function. Написан он исходя из того, что в каталоге сервисного аккаунта, под которым запускается функция, доступна модель вложений (embedding). Детали подключения к кластеру описаны в документации.
Рассмотрим один крайний случай: если мы подключаемся, указывая FQDN DATA-узлов, у которых не включен публичный доступ, то функция должна запускаться в сети кластера OpenSearch, иначе они будут недоступны. Альтернативные варианты: подключаться через «Особый FQDN» или узел DASHBOARD с публичным доступом.
Код создаёт тестовый индекс с текстовым и векторным полем, явно вызывает embedding model через REST API, создавая векторы вложений для документов и запроса, и выполняет векторный поиск, демонстрируя способ интеграции. Обратите внимание на способ выбора разных моделей для документов и запросов.
Получение вложений через OpenSearch Ingest Pipelines
Другой вариант интеграции с моделью вложений – OS ingest pipelines. В этом случае клиент отправляет только текст документа, а вызов модели векторизации выполняет OpenSearch.
В общем виде процесс подключения внешней модели описан тут. Последовательность шагов для подключения модели вложений из YC AI Studio описана в заявке на добавление в документацию OS Connector Blueprints. Пришлось столкнуться со следующими трудностями:
Подключение Managed OpenSearch к моделям AI Studio
DATA-хосты подключаются к моделям YC через интернет (до тех пор, пока не реализовано что-то подобное). Соответственно, DATA-узел без публичного доступа не может вызвать модель AI Studio. Есть два варианта решения:
Включаем публичный доступ для DATA-хостов. Это немного контринтуитивно. Кажется, что «публичность» определяет входящий трафик, но после некоторых размышлений становится очевидно, что она же требуется и для исходящего подключения к модели.
Создание NAT-шлюза с таблицей маршрутизации (Next hop) на этот шлюз и привязка этой таблицы к сети кластера OpenSearch. Тарифицируется NAT отдельно (дополнительно), но это более безопасно, чем позволять всему интернету подбирать пароль к DATA-хостам.
Решение проблемы физического подключения не отменяет необходимости разрешить OpenSearch отправлять запросы в AI Stutio, как указано в последовательности шагов.
Функции процессинга в коннекторе
Другая трудность – функции пре/пост-процессинга. К счастью, подошли функции от Amazon Bedrock. Эти функции нужны для адаптации интерфейса вызова коннектора в ml_commons и Embedding REST API. Грубо говоря, из списка строк выбирается первый элемент, который принимает в виде одной строки text REST API. У меня возникли некоторые сомнения насчёт предсказуемости всего конвейера ml_commons: OpenSearch унаследовал от Elasticsearch свободное смешивание списков и отдельных значений, поэтому мне показалось, что не исключена ситуация, когда вместо списка из одного элемента на вход придут несколько строк и будут отброшены. Попробую это уточнить в дальнейшем. Базовые примеры работают ожидаемо.
В случае необходимости, функции пре/пост-процессинга можно написать самому, но советую быть в стороне от этого «увлекательного» занятия: пишутся они на Painless (диалекте JavaScript), скрипт этот формирует JSON (Java Script Object Notation, you know), потом этот скрипт эскейпится и передаётся строкой в JSON. В результате имеем интуитивно понятные нет-серии обратных слэшей и кавычек \"\\\", разбираться c которыми – то ещё удовольствие. Диагностика проблем затруднительна – немногословные ответы с ошибками без стэк-трейсов и логов. Мне пришлось запускать OpenSearch локально под отладчиком, но даже так некоторые исходники не скачались из-за каких-то конфликтов версий.
Другой возможный подход – использование Open AI совместимого Embedding API. Возможно, это упростит конфигурацию коннектора, но исходя из того, что несколько строк в один запрос этого API не отправить, преимущества такого подхода неочевидны.
Токены доступа
В конфигурации коннектора используется API Key. Другой возможный способ – Bearer ${IAM_TOKEN}. Но «IAM-токен действует не больше 12 часов». В принципе, Cloud Function может периодически обновлять Bearer, получая его из контекста, как в примерах, рассмотренных выше. Если быть честным, мне не удалось обновить токен, обновляя конфигурацию коннектора. Требует дальнейшего исследования.
Конвейер индексации
После проверки модели /_predict можно симулировать процессор без создания пайплайна, отправив в него пару одностроковых документов.
POST /_ingest/pipeline/_simulate { "pipeline" : { "processors": [ { "text_embedding": { "model_id": "6cbj05oB7Vkgz3is7UJH", "field_map": { "text": "text_embedding" } } } ] }, "docs": [ { "_index": "my_hybrid_index", "_id": "1", "_source": { "text": "Poet birthday" } }, { "_index": "my_hybrid_index", "_id": "2", "_source": { "text": "Blossing plant" } } ] } // ответ { "docs": [ { "doc": { "_index": "my_hybrid_index", "_id": "1", "_source": { "text_embedding": [ 0.068237305, -0.03338623, 0.023925781, ... -0.076538086, 0.021759033, 0.065979004 ], "text": "Poet birthday" }, "_ingest": { "timestamp": "2025-11-30T08:41:11.016017656Z" } } }, { "doc": { "_index": "my_hybrid_index", "_id": "2", "_source": { "text_embedding": [ -0.038879395, -0.037139893, ... -0.023345947, -0.001789093 ], "text": "Blossing plant" }, "_ingest": { "timestamp": "2025-11-30T08:41:11.016022038Z" } } } ] }
Видно, как тесты были дополнены векторами. Ранее я писал о том, что ml_commons, как мне кажется, не полностью обрабатывает все возможные виды документов.
Вот пример:
POST /_ingest/pipeline/_simulate { "pipeline" : { "processors": [ { "text_embedding": { "model_id": "6cbj05oB7Vkgz3is7UJH", "field_map": { "text_field": "vector_field", "obj.text_field": "vector_field" } } } ] }, "docs": [ { "_index": "second-index", "_id": "1", "_source": { "text_field": ["array","isn't handled properly"], "obj.text_field": "another way ", "obj":{ "text_field": "to pass array " } } } ] }
В общем, если вам действительно нужно что-то такое обрабатывать, то придётся приложить некоторые усилия.
Создаём пайплайн для индексации.
PUT /_ingest/pipeline/_yc_embeddings_pipeline { "description": "Pipeleine with YC embeddings", "processors": [ { "text_embedding": { "model_id": "6cbj05oB7Vkgz3is7UJH", "field_map": { "text": "text_embedding" } } } ] }
Код клиента индексации и поиска
Модифицируем код демо-функции: yc-os-pipeline-neural-cloud-function.py при индексировании указываем пайплайн для конверсии текстов в векторы. При поиске используем немного другой запрос, передавая в него идентификатор созданной модели, которая конвертирует текст запроса в вектор. Вы же помните, что для запроса нужно указывать другую модель вложений? При запуске кода видим, что векторы были созданы при индексации и запросе.
Результат поиска:
[ { "_index": "my_hybrid_index", "_id": "8saY1JoB7Vkgz3ish0Ky", "_score": 0.5327221, "_source": { "text_embedding": [ -0.027313232, -0.077697754, 0.043762207, ... -0.0029201508, -0.065979004 ], "text": "Alexander Sergeyevich Pushkin ....." } }, { "_index": "my_hybrid_index", "_id": "88aY1JoB7Vkgz3ish0LY", "_score": 0.47184184, "_source": { "text_embedding": [ 0.03225708, ..., 0.0029239655, 0.1027832 ], "text": "Matricaria is a genus of annual flowering plants ...." } } ] }
В отличие от предыдущей версии кода, при использовании пайплайнов и ссылки на модель функции не нужен сервисный аккаунт, а только пароль OpenSearch, т.к. сама функция не вызывает модель из AI Studio.
Использование конвейера запроса
Следующая попытка: использовать процессор в конвейере запроса, который векторизует текст запроса, вызывая модель вложений YC. На мой взгляд, очень монструозно. Продемонстрирую его, добавляя конфигурацию процессора запроса сразу в запрос. В рабочем режиме предполагается, что пайплайн конфигурируется и применяется к запросам по имени или неявно.
POST my_hybrid_index/_search?verbose_pipeline=true { "query": { "match": { "text": {"query":"blossing plant"} } }, "search_pipeline": { "request_processors": [ { "ml_inference": { "model_id": "6cbj05oB7Vkgz3is7UJH", "input_map": [ { "inputText": "query.match.text.query" } ], "output_map": [ { "vector_out": "$.inference_results.*.output.*.data" } ], "query_template": """{ "size": 2, "query": { "knn": { "text_embedding": { "vector": ${vector_out}, "k": 5 } } } }""" } } ]} }
Здесь, кстати, используется трюк с тремя ка��ычками, немного облегчающий ад эскейпинга, на который я жаловался выше. Но работает он только в DevTools. Вот так это значение выглядит при Copy as cURL.
"query_template": "{\n \"size\": 2,\n \"query\": {\n \"knn\": {\n \"text_embedding\": {\n \"vector\": ${vector_out},\n \"k\": 5\n }\n }\n }\n }
Получаем аналогичный результат поиска по вектору. Кроме того, verbose_pipeline=true показывает результат работы процессора – вектор knn запроса.
{ "took": 29, .. "hits": { ... }, "processor_results": [ { "processor_name": "ml_inference", "duration_millis": 27756171, "status": "success", "input_data": { ....}, "output_data": { "size": 2, "query": { "knn": { "text_embedding": { "vector": [ 0.060516357, -0.04724121, ... 0.06149292, -0.009025574 ], "boost": 1, "k": 5 } } }
На этом пока все. В следующих постах хотелось бы обсудить использование генеративной и ранжирующей моделей.
Благодарю за внимание!
PS. PR на добавление Yandex Cloud Model Blueprints в документацию Open Search.
