Я уже писал серию статей о том, как сделать свой 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-аналитика, который будет анализировать наши данные, строить графики и помогать делать расчёты.
ссылка на оригинал статьи https://habr.com/ru/articles/822249/
Добавить комментарий