Я уже писал серию статей о том, как сделать свой ChatGPT бот в Telegram. В этих туториалах для того, чтобы бот запоминал историю переписки, мы использовали простой JSON файл и хранили его в object storage Яндекс Облака. И каждый раз, когда пользователь писал новый вопрос, мы отправляли всю предыдущую историю переписки, чтобы модель понимала контекст и могла поддерживать диалог.
Но что, если в качестве контекста нам необходимо, чтобы модель знала не только историю переписки с конкретным пользователем, но ещё и какую-то общую информацию про бизнес или продукт? Можно, конечно, каждый раз добавлять такой контекст вместе с историей переписки, но что, если такой "глобальный" контекст — это большая документация или целая библиотека "базы знаний"? Во-первых: так легко можно выйти за пределы контекстного окна. Во-вторых: токены будут расходоваться не оптимально. Зачем нам посылать всю нашу базу знаний, если пользователь спрашивать про какую-то определённую часть?
Все эти проблемы призваны решить такие продукты OpenAI, как Assistants API и Vector Store. Assistants API берёт на себя задачу ведения истории переписки, так что не надо больше на своей стороне заниматься хранением истории и добавлять её к каждому запросу. Vector Store — это векторное хранилище, в которое можно загрузить файлы с вашей документацией или базой знаний, они автоматически будут трансформированы в векторный формат, и при каждом запросе из хранилища будет выбираться только информация, актуальная для этого конкретного запроса, тем самым помогая модели точнее отвечать на вопросы и экономить токены.
Я решил разобрать работу Assistants API и Vector Store на примере простого Telegram-бота для ресторана. Идея такова: посетители ресторана могут общаться с ботом, который отвечает на вопросы по меню. Как и в прошлых статьях, я построю всю систему с помощью serverless-технологий Яндекс Облака.
Для реализации проекта мне понадобится:
Прямой доступ к OpenAI API в России заблокирован, зарегистрироваться без иностранного телефона нельзя, оплатить картами российских банков — тоже. Поэтому воспользуемся проверенным способом — сервисом ProxyAPI, который предоставляет доступ к OpenAI API в России, включая Assistants API и Vector Store. Именно то, что нам и нужно.
Регистрируемся на сайте, переходим в раздел Ключи API и создаём ключ.
Доступ к Assistants API закрыт для обычных пользователей, так что понадобится подписка Pro. Её оформляем здесь.
Отлично, теперь у нас есть ключ API с доступом к OpenAI API, Assistants API и Vector Store.
Если у вас ещё нет аккаунта, создайте его здесь. Убедитесь, что ваш платёжный аккаунт подключён и имеет статус ACTIVE
или TRIAL_ACTIVE
.
Для создания и управления своими ботами в Telegram существует специальный бот под названием BotFather. Он поможет вам создать бота и выдаст токен — сохраните его, он нам ещё понадобится.
Теперь возвращаемся в Яндекс Облако и создаём все ресурсы, необходимые для работы нашего проекта.
На домашней странице консоли в верхнем меню есть вкладка "Сервисные аккаунты". Переходим туда и создаём новый аккаунт. Здесь и везде далее я использую одно и то же имя для всех ресурсов assistant-telegram-bot
просто, чтобы не запутаться. Аккаунту надо присвоить следующие роли:
serverless.functions.invoker
storage.uploader
После того как аккаунт создан, перейдите в него и создайте статический ключ доступа, сохраните полученные идентификатор и секретный ключ, а также идентификатор самого сервисного аккаунта.
Теперь переходим в раздел "Object Storage" и создаём новый бакет. Я не менял никакие настройки.
Следующий шаг — создание облачной функции. Именно она будет получать ваши запросы от Telegram, перенаправлять их в ProxyAPI и посылать ответ обратно в Telegram-бот.
Переходим в раздел "Cloud Functions" и жмём "Создать функцию".
После создания сразу откроется редактор, в который мы, собственно, и положим код функции.
Выбираем среду выполнения Python 3.12 и убираем галочку с "Добавить файлы с примерами кода":
В редакторе сначала создадим новый файл, назовём его requirements.txt и положим туда следующий код:
openai~=1.33.0
boto3~=1.34.122
pyTelegramBotAPI~=4.19.1
Это список зависимостей для Python, которые облако автоматически подгрузит во время сборки так, чтобы мы могли пользоваться ими в коде самой функции.
Теперь создадим файл, в котором будет код, который отвечает за конфигурацию проекта. Назовём его 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
и кладём туда весь код, который отвечает за разные административные задачи проекта.
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
(тоже создаём его) будем хранить методы для собственно работы с сообщениями пользователей.
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. Что мы и делаем, после чего получаем ответ, который может состоять из нескольких сообщений.
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
На этом наша работа с функцией закончена. Жмём "Сохранить изменения" и смотрим, как Облако собирает нашу функцию. Для следующего шага нам понадобится идентификатор функции. Перейдите во вкладку "Обзор" для нашей функции и скопируйте его оттуда.
Для того чтобы сообщения, которые мы посылаем в 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-боту, куда пересылать сообщения, которые он от нас получает. Для этого достаточно выполнить 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-аналитика, который будет анализировать наши данные, строить графики и помогать делать расчёты.