Делаем AI-официанта с помощью OpenAI Assistants API и Vector Store в Telegram

Я уже писал серию статей о том, как сделать свой ChatGPT бот в Telegram. В этих туториалах для того, чтобы бот запоминал историю переписки, мы использовали простой JSON файл и хранили его в object storage Яндекс Облака. И каждый раз, когда пользователь писал новый вопрос, мы отправляли всю предыдущую историю переписки, чтобы модель понимала контекст и могла поддерживать диалог.

Но что, если в качестве контекста нам необходимо, чтобы модель знала не только историю переписки с конкретным пользователем, но ещё и какую-то общую информацию про бизнес или продукт? Можно, конечно, каждый раз добавлять такой контекст вместе с историей переписки, но что, если такой "глобальный" контекст — это большая документация или целая библиотека "базы знаний"? Во-первых: так легко можно выйти за пределы контекстного окна. Во-вторых: токены будут расходоваться не оптимально. Зачем нам посылать всю нашу базу знаний, если пользователь спрашивать про какую-то определённую часть?

Все эти проблемы призваны решить такие продукты OpenAI, как Assistants API и Vector Store. Assistants API берёт на себя задачу ведения истории переписки, так что не надо больше на своей стороне заниматься хранением истории и добавлять её к каждому запросу. Vector Store — это векторное хранилище, в которое можно загрузить файлы с вашей документацией или базой знаний, они автоматически будут трансформированы в векторный формат, и при каждом запросе из хранилища будет выбираться только информация, актуальная для этого конкретного запроса, тем самым помогая модели точнее отвечать на вопросы и экономить токены.

Я решил разобрать работу Assistants API и Vector Store на примере простого Telegram-бота для ресторана. Идея такова: посетители ресторана могут общаться с ботом, который отвечает на вопросы по меню. Как и в прошлых статьях, я построю всю систему с помощью serverless-технологий Яндекс Облака.

Подготовка

Для реализации проекта мне понадобится:

API ключ для OpenAI API с доступом к Assistants API и Vector Store

Прямой доступ к OpenAI API в России заблокирован, зарегистрироваться без иностранного телефона нельзя, оплатить картами российских банков — тоже. Поэтому воспользуемся проверенным способом — сервисом ProxyAPI, который предоставляет доступ к OpenAI API в России, включая Assistants API и Vector Store. Именно то, что нам и нужно.

Регистрируемся на сайте, переходим в раздел Ключи API и создаём ключ. 

Доступ к Assistants API закрыт для обычных пользователей, так что понадобится подписка Pro. Её оформляем здесь.

Отлично, теперь у нас есть ключ API с доступом к OpenAI API, Assistants API и Vector Store.

Аккаунт в Яндекс.Облако

Если у вас ещё нет аккаунта, создайте его здесь. Убедитесь, что ваш платёжный аккаунт подключён и имеет статус ACTIVE или TRIAL_ACTIVE.

Telegram-бот

Для создания и управления своими ботами в Telegram существует специальный бот под названием BotFather. Он поможет вам создать бота и выдаст токен — сохраните его, он нам ещё понадобится.

Облачные ресурсы

Теперь возвращаемся в Яндекс Облако и создаём все ресурсы, необходимые для работы нашего проекта.

1. Сервисный аккаунт

На домашней странице консоли в верхнем меню есть вкладка "Сервисные аккаунты". Переходим туда и создаём новый аккаунт. Здесь и везде далее я использую одно и то же имя для всех ресурсов assistant-telegram-bot просто, чтобы не запутаться. Аккаунту надо присвоить следующие роли:

serverless.functions.invoker

storage.uploader

После того как аккаунт создан, перейдите в него и создайте статический ключ доступа, сохраните полученные идентификатор и секретный ключ, а также идентификатор самого сервисного аккаунта.

2. Бакет

Теперь переходим в раздел "Object Storage" и создаём новый бакет. Я не менял никакие настройки.

3.Облачная функция

Следующий шаг — создание облачной функции. Именно она будет получать ваши запросы от Telegram, перенаправлять их в ProxyAPI и посылать ответ обратно в Telegram-бот.

Переходим в раздел "Cloud Functions" и жмём "Создать функцию".

После создания сразу откроется редактор, в который мы, собственно, и положим код функции. 

Выбираем среду выполнения Python 3.12 и убираем галочку с "Добавить файлы с примерами кода":

requirements.txt

В редакторе сначала создадим новый файл, назовём его requirements.txt и положим туда следующий код:

openai~=1.33.0
boto3~=1.34.122
pyTelegramBotAPI~=4.19.1

Это список зависимостей для Python, которые облако автоматически подгрузит во время сборки так, чтобы мы могли пользоваться ими в коде самой функции.

config.py

Теперь создадим файл, в котором будет код, который отвечает за конфигурацию проекта. Назовём его config.py и запишем следующий код:

import json
import os

import boto3
import openai

YANDEX_KEY_ID = os.environ.get("YANDEX_KEY_ID")
YANDEX_KEY_SECRET = os.environ.get("YANDEX_KEY_SECRET")
YANDEX_BUCKET = os.environ.get("YANDEX_BUCKET")
PROXY_API_KEY = os.environ.get("PROXY_API_KEY")

ASSISTANT_MODEL = os.environ.get("ASSISTANT_MODEL")

TG_BOT_TOKEN = os.environ.get("TG_BOT_TOKEN")
TG_BOT_ADMIN = os.environ.get("TG_BOT_ADMIN")


def get_s3_client():
    session = boto3.session.Session(
        aws_access_key_id=YANDEX_KEY_ID, aws_secret_access_key=YANDEX_KEY_SECRET
    )
    return session.client(
        service_name="s3", endpoint_url="https://storage.yandexcloud.net"
    )


def get_config() -> dict:
    s3client = get_s3_client()
    try:
        response = s3client.get_object(Bucket=YANDEX_BUCKET, Key="config.json")
        return json.loads(response["Body"].read())
    except:
        return {}


def save_config(new_config: dict):
    s3client = get_s3_client()
    s3client.put_object(
        Bucket=YANDEX_BUCKET, Key="config.json", Body=json.dumps(new_config)
    )


proxy_client = openai.Client(
    api_key=PROXY_API_KEY,
    base_url="https://api.proxyapi.ru/openai/v1",
)

Здесь мы читаем все необходимые параметры из переменных окружения и прописываем функции для получения и сохранения дополнительных (динамических) параметров в бакете хранилища Яндекс Облака. Инициируем также клиент для ProxyAPI согласно документации, то есть переопределяем путь к API.

admin.py

Далее создаём файл admin.py и кладём туда весь код, который отвечает за разные административные задачи проекта.

from config import ASSISTANT_MODEL, get_config, proxy_client, save_config


def get_vector_store_id():
    config = get_config()
    if "vector_store_id" not in config or not config["vector_store_id"]:
        new_store = proxy_client.beta.vector_stores.create()
        config["vector_store_id"] = new_store.id
        save_config(config)

    return config["vector_store_id"]


def create_assistant(name, instructions):
    assistant_id = get_assistant_id()
    if not assistant_id:
        new_assistant = proxy_client.beta.assistants.create(
            model=ASSISTANT_MODEL,
            instructions=instructions,
            name=name,
            tools=[
                {
                    "type": "file_search",
                }
            ],
            tool_resources={
                "file_search": {"vector_store_ids": [get_vector_store_id()]}
            },
        )
        config = get_config()
        config["assistant_id"] = new_assistant.id
        save_config(config)
    else:
        proxy_client.beta.assistants.update(
            assistant_id=assistant_id, instructions=instructions
        )


def get_assistant_id():
    config = get_config()
    return config["assistant_id"] if "assistant_id" in config else None


def add_knowledge(filename, file):
    file_object = proxy_client.files.create(file=(filename, file), purpose="assistants")
    store_id = get_vector_store_id()
    proxy_client.beta.vector_stores.files.create(
        vector_store_id=store_id, file_id=file_object.id
    )


def reset_knowledge():
    store_id = get_vector_store_id()
    files = proxy_client.beta.vector_stores.files.list(vector_store_id=store_id)
    for file in files:
        proxy_client.beta.vector_stores.files.delete(
            vector_store_id=store_id, file_id=file.id
        )
        proxy_client.files.delete(file.id)

get_vectore_store_id

Возвращает идентификатор векторного хранилища. Если его ещё нет, то создаёт. Сохраняем идентификатор в конфигурации.

create_assistant

Создаёт или обновляет настройки для ассистента на основе имени и инструкций. Сразу привязываем ассистента к нашему векторному хранилищу. Сохраняем идентификатор ассистента в конфигурации.

get_assistant_id

Возвращает идентификатор ассистента из конфигурации.

add_knowledge

Добавляет содержимое загруженного файла в нашу “базу знаний” - то есть загружает в векторное хранилище (vector store).

reset_knowledge

В случае если данные устарели или мы случайно загрузили неправильную информацию, этот метод поможет нам очистить хранилище и загрузить данные заново.

chat.py

В файле chat.py (тоже создаём его) будем хранить методы для собственно работы с сообщениями пользователей.

from admin import get_assistant_id
from config import get_config, proxy_client, save_config


def get_thread_id(chat_id: str):
    config = get_config()
    if "threads" not in config:
        config["threads"] = {}

    if chat_id not in config["threads"]:
        thread = proxy_client.beta.threads.create()
        config["threads"][chat_id] = thread.id
        save_config(config)

    return config["threads"][chat_id]


def process_message(chat_id: str, message: str) -> list[str]:
    assistant_id = get_assistant_id()
    thread_id = get_thread_id(chat_id)
    proxy_client.beta.threads.messages.create(
        thread_id=thread_id, content=message, role="user"
    )

    run = proxy_client.beta.threads.runs.create_and_poll(
        thread_id=thread_id,
        assistant_id=assistant_id,
    )

    answer = []

    if run.status == "completed":
        messages = proxy_client.beta.threads.messages.list(
            thread_id=thread_id, run_id=run.id
        )
        for message in messages:
            if message.role == "assistant":
                for block in message.content:
                    if block.type == "text":
                        answer.append(block.text.value)

    return answer

get_thread_id

На стороне Assistants API мы будем открывать отдельный тред для каждого пользователя, который общается с нашим ботом. Так что, используя chat_id, который нам присылает Telegram, будем создавать и идентифицировать тред ассистента.

process_message

Этот метод принимает сообщения пользователя, и отправляет его через ProxyAPI на обработку в Assistants API. Assistants API работает немного по-другому, в сравнении с обычной API. Здесь после добавления сообщения в тред ответ модели мы сразу не получим. Надо дополнительно запустить обработку всего треда с помощью метода run. Что мы и делаем, после чего получаем ответ, который может состоять из нескольких сообщений.

index.py

import json
import logging
import threading
import time

import telebot

from admin import add_knowledge, create_assistant, get_assistant_id, reset_knowledge
from chat import process_message
from config import TG_BOT_ADMIN, TG_BOT_TOKEN

logger = telebot.logger
telebot.logger.setLevel(logging.INFO)

bot = telebot.TeleBot(TG_BOT_TOKEN, threaded=False)


is_typing = False


def start_typing(chat_id):
    global is_typing
    is_typing = True
    typing_thread = threading.Thread(target=typing, args=(chat_id,))
    typing_thread.start()


def typing(chat_id):
    global is_typing
    while is_typing:
        bot.send_chat_action(chat_id, "typing")
        time.sleep(4)


def stop_typing():
    global is_typing
    is_typing = False


def check_setup(message):
    if not get_assistant_id():
        if message.from_user.username != TG_BOT_ADMIN:
            bot.send_message(
                message.chat.id, "Бот еще не настроен. Свяжитесь с администратором."
            )
        else:
            bot.send_message(
                message.chat.id,
                "Бот еще не настроен. Используйте команду /create для создания ассистента.",
            )
        return False
    return True


def check_admin(message):
    if message.from_user.username != TG_BOT_ADMIN:
        bot.send_message(message.chat.id, "Доступ запрещен")
        return False
    return True


@bot.message_handler(commands=["help", "start"])
def send_welcome(message):
    if not check_setup(message):
        return

    bot.send_message(
        message.chat.id,
        (
            f"Привет! Я твой виртуальный официант. Спроси меня любой вопрос про наше меню."
        ),
    )


@bot.message_handler(commands=["create"])
def create_assistant_command(message):
    if not check_admin(message):
        return

    instructions = message.text.split("/create")[1].strip()
    if len(instructions) == 0:
        bot.send_message(
            message.chat.id,
            """
Введите подробные инструкции для работы ассистента после команды /create и пробела.

Например: 
/create Ты - виртуальный официант, который помогает посетителю ресторана выбрать блюда из меню. Меню доступно в файлах, к которым у тебя есть доступ в векторном хранилище.
Не используй в ответах markdown и не указывай источники.

Если ассистент уже был ранее создан, инструкции будут обновлены.
            """,
        )
        return

    name = bot.get_me().full_name
    create_assistant(name, instructions)

    bot.send_message(
        message.chat.id,
        "Ассистент успешно создан. Теперь вы можете добавлять документы в базу знаний с помощью команды /upload.",
    )


@bot.message_handler(commands=["upload"])
def add_knowledge_item_command(message):
    if not check_setup(message):
        return

    if not check_admin(message):
        return

    return bot.send_message(
        message.chat.id, "Для добавления нового документа в базу знаний пришлите файл."
    )


@bot.message_handler(content_types=["document"])
def add_knowledge_item(message):
    if not check_setup(message):
        return

    if not check_admin(message):
        return

    file_info = bot.get_file(message.document.file_id)
    downloaded_file = bot.download_file(file_info.file_path)

    try:
        add_knowledge(message.document.file_name, downloaded_file)
    except Exception as e:
        return bot.send_message(
            message.chat.id, f"Ошибка при добавлении документа: {e}"
        )

    return bot.send_message(message.chat.id, "Новый документ добавлен в базу знаний.")


@bot.message_handler(commands=["reset"])
def reset_knowledge_base(message):
    if not check_setup(message):
        return

    if not check_admin(message):
        return

    reset_knowledge()
    return bot.send_message(message.chat.id, "База знаний очищена.")


@bot.message_handler(content_types=["text"])
def handle_message(message):
    if not check_setup(message):
        return

    start_typing(message.chat.id)

    try:
        answers = process_message(str(message.chat.id), message.text)
    except Exception as e:
        bot.send_message(message.chat.id, f"Ошибка при обработке сообщения: {e}")
        return

    stop_typing()

    for answer in answers:
        bot.send_message(message.chat.id, answer)


def handler(event, context):
    message = json.loads(event["body"])
    update = telebot.types.Update.de_json(message)

    if update.message is not None:
        try:
            bot.process_new_updates([update])
        except Exception as e:
            print(e)

    return {
        "statusCode": 200,
        "body": "ok",
    }

typing

Вспомогательная функция, которая посылает статус “набирает сообщение…”, чтобы наши пользователи не скучали, пока модель готовит ответ

check_setup

Вспомогательная функция, которая выводит ошибку, если бот ещё не настроен администратором

check_admin

Вспомогательная функция, которая проверяет, является ли автор сообщения администратором бота

send_welcome

Посылаем приветственное сообщение после команды /start

create_assistant_command

Команда /create доступна только администратору, она создаёт или обновляет инструкции для ассистента.

add_knowledge_item_command

Команда /upload просто информирует администратора, что для добавления данных в базу знаний надо загрузить файл

add_knowledge_item

При загрузке файла отправляем его в векторное хранилище

reset_knowledge_base

Команда /reset доступна только администратору, с её помощью очищаем векторное хранилище

handle_message

Собственно обработчик входящих сообщений от пользователей. Здесь происходит общение с AI-официантом

handler

Точка входа для всей облачной функции. Сюда будут приходить все команды и сообщений от Telegram.

Для удобства я опубликовал весь исходный код функции на GitLab:
https://gitlab.com/evrovas/assistant-telegram-bot

В будущем, если будут какие-то обновления, то они будут именно в репозитории.

Прописываем точку входа равной index.handler:

В параметрах функции поставим таймаут 60 секунд - ответы от OpenAI приходится обычно ждать какое-то время, 60 секунд должно быть достаточно.

А также надо заполнить все переменные окружения, которые использует наша функция.

TG_BOT_TOKEN

Токен Telegram-бота

TG_BOT_ADMIN

Имя пользователя — администратора бота

PROXY_API_KEY

API-ключ от ProxyAPI

YANDEX_KEY_ID

YANDEX_KEY_SECRET

Идентификатор и секретный ключ статического ключа сервисного аккаунта Яндекс

YANDEX_BUCKET

Имя бакета, который вы создали в Object Storage

ASSISTANT_MODEL

Языковая модель, которая будет использоваться ассистентом для генерации ответа. Я рекомендую последнюю из доступных на момент написания статьи - gpt-4o

На этом наша работа с функцией закончена. Жмём "Сохранить изменения" и смотрим, как Облако собирает нашу функцию. Для следующего шага нам понадобится идентификатор функции. Перейдите во вкладку "Обзор" для нашей функции и скопируйте его оттуда.

Шлюз API

Для того чтобы сообщения, которые мы посылаем в Telegram-бот, приходили на обработку в нашу функцию, у неё должен быть какой-то публичный адрес. Сделать это очень легко с помощью инструмента API-шлюз. Переходим в раздел и создаём новый шлюз. В спецификации указываем:

openapi: 3.0.0
info:
  title: Sample API
  version: 1.0.0
paths:
  /:
    post:
      x-yc-apigateway-integration:
        type: cloud-functions
        function_id: <FUNCTION-ID>
        service_account_id: <SERVICE-ACCOUNT-ID>

В спецификации используйте свой идентификатор функции и сервисного аккаунта для значений <FUNCTION-ID> и <SERVICE-ACCOUNT-ID>. Выглядеть это будет вот так:

После сохранения вы увидите сводную информацию о шлюзе. Сохраните оттуда строку "Служебный домен".

Telegram WebHook

Теперь надо сообщить Telegram-боту, куда пересылать сообщения, которые он от нас получает. Для этого достаточно выполнить POST-запрос к API Telegram такого формата:

curl \
  --request POST \
  --url https://api.telegram.org/bot<токен бота>/setWebhook \
  --header 'content-type: application/json' \
  --data '{"url": "<домен API-шлюза>"}'

<токен бота> заменяем на токен Telegram-бота, который мы получили ещё на третьем шаге этого туториала

<домен API-шлюза> заменяем на Служебный домен нашего API-шлюза, созданный на прошлом шаге.

Я использовал Postman для этой задачи, просто удобнее, когда всё наглядно и с user-friendly интерфейсом:

На этом вся наша работа закончена, осталось только проверить.

Тест

Начнём с того, что настроим нашего ассистента. Для этого я запущу команду /create от имени администратора:

Отлично, ассистент настроен. Для примера скачаю меню одного из Московских ресторанов в формате PDF.

Теперь загружу его в нашу базу знаний:

Теперь, наконец, пообщаюсь с ботом так, как будто я посетитель ресторана:

Ура! Всё работает!

Варианты использования

Такой бот можно настроить на работу с любой другой информацией, я сделал AI-официанта для примера. Это может быть AI-работник службы поддержки или AI-ассистент для заказа в интернет-магазине и тому подобное. Достаточно только правильно прописать инструкции для ассистента и загрузить необходимые данные в базу знаний.

Бот можно также дополнительно усовершенствовать. Например, сделать так, чтобы он мог принимать заказ, а не только рассказывать про позиции в меню. Для этого нужно настроить так, чтобы после того, как пользователь согласился разместить заказ, бот посылал его содержимое в определённый чат в Telegram, например на кухню или администратору ресторана.

Тизер

Надеюсь, моя статья помогла вам разобраться с Assistants API и Vestor Store. В следующей статье я хочу разобрать пример работы с ещё одним инструментом OpenAI под названием Code Interpreter. С его помощью мы сделаем "карманного" data-аналитика, который будет анализировать наши данные, строить графики и помогать делать расчёты.