Создание Telegram-бота с ИИ на Python
Локальная модель уже работает на srv-ai01 и отвечает через API — мы разбирали установку Ollama и веб-интерфейс Open WebUI в предыдущих статьях серии. Логичный следующий шаг — вынести ИИ туда, где реально общаются люди в компании: в Telegram. Бот с локальной LLM внутри удобен тем, что не тянет данные во внешние облака, не требует от коллег заходить куда-то отдельно и работает там, где и так открыт мессенджер весь день.
В этой статье напишем рабочего Telegram-бота на Python с библиотекой aiogram, подключим его к Ollama на srv-ai01 и задеплоим как systemd-сервис, который переживает перезагрузки и падения.
Подготовка окружения
Заведём отдельного пользователя и директорию под бота на сервере, чтобы не мешать процессы друг другу:
sudo mkdir -p /opt/ai-bot
sudo chown $USER:$USER /opt/ai-bot
cd /opt/ai-bot
python3 -m venv venv
source venv/bin/activate
pip install aiogram httpx python-dotenvaiogram — асинхронный фреймворк для Telegram Bot API, httpx понадобится для обращения к Ollama асинхронно, python-dotenv — чтобы не хардкодить токены в коде.
Получение токена бота
Открой Telegram, найди @BotFather, отправь команду /newbot и следуй подсказкам — укажи имя и username бота (username должен заканчиваться на bot). BotFather выдаст токен вида 123456789:AAExampleTokenHere. Сохрани его в файл окружения, не в код:
echo "BOT_TOKEN=твой_токен_здесь" > /opt/ai-bot/.env
echo "OLLAMA_URL=http://127.0.0.1:11434" >> /opt/ai-bot/.env
echo "MODEL_NAME=llama3.1:8b" >> /opt/ai-bot/.envФайл .env не должен попадать в систему контроля версий — добавь его в .gitignore, если планируешь хранить код бота в git-репозитории.
Базовый код бота
Создадим файл bot.py с минимальной логикой: приём сообщения, обращение к Ollama, отправка ответа. Начнём с простого варианта без памяти диалога:
import asyncio
import os
import logging
import httpx
from aiogram import Bot, Dispatcher, types
from aiogram.filters import CommandStart
from dotenv import load_dotenv
load_dotenv()
BOT_TOKEN = os.getenv("BOT_TOKEN")
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://127.0.0.1:11434")
MODEL_NAME = os.getenv("MODEL_NAME", "llama3.1:8b")
logging.basicConfig(level=logging.INFO)
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()
async def ask_model(prompt: str) -> str:
async with httpx.AsyncClient(timeout=120) as client:
response = await client.post(
f"{OLLAMA_URL}/api/generate",
json={"model": MODEL_NAME, "prompt": prompt, "stream": False},
)
response.raise_for_status()
return response.json()["response"]
@dp.message(CommandStart())
async def handle_start(message: types.Message) -> None:
await message.answer(
"Привет! Я ИИ-помощник IT-Pages на локальной модели. Просто напиши вопрос."
)
@dp.message()
async def handle_message(message: types.Message) -> None:
await bot.send_chat_action(message.chat.id, "typing")
try:
answer = await ask_model(message.text)
except Exception as exc:
logging.exception("Ошибка при обращении к модели")
answer = "Не получилось получить ответ от модели, попробуй ещё раз чуть позже."
await message.answer(answer)
async def main() -> None:
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())Запусти бота вручную для проверки:
python bot.pyНапиши боту в Telegram что угодно — через несколько секунд должен прийти ответ модели. Если ничего не приходит, проверь логи в консоли: чаще всего проблема либо в неверном токене, либо в недоступности Ollama по указанному адресу.
Добавляем память диалога
Без контекста бот забывает предыдущие сообщения и отвечает на каждое как на первое. Добавим простое хранение истории в памяти процесса — для продакшена с несколькими сотнями пользователей стоит перейти на Redis, но для команды из десятка человек хватит и словаря:
from collections import defaultdict
history: dict[int, list[dict]] = defaultdict(list)
MAX_HISTORY = 10
async def ask_model_with_history(chat_id: int, prompt: str) -> str:
history[chat_id].append({"role": "user", "content": prompt})
history[chat_id] = history[chat_id][-MAX_HISTORY:]
async with httpx.AsyncClient(timeout=120) as client:
response = await client.post(
f"{OLLAMA_URL}/api/chat",
json={
"model": MODEL_NAME,
"messages": history[chat_id],
"stream": False,
},
)
response.raise_for_status()
answer = response.json()["message"]["content"]
history[chat_id].append({"role": "assistant", "content": answer})
return answerОбрати внимание — мы перешли с эндпоинта /api/generate на /api/chat, который в Ollama специально предназначен для диалогов с историей в формате role/content, привычном по OpenAI API. Замени вызов ask_model на ask_model_with_history(message.chat.id, message.text) в обработчике handle_message.
Добавь команду сброса контекста — она пригодится, когда пользователь захочет начать новую тему с чистого листа:
from aiogram.filters import Command
@dp.message(Command("reset"))
async def handle_reset(message: types.Message) -> None:
history.pop(message.chat.id, None)
await message.answer("Контекст диалога очищен.")Ограничение доступа
Если бот работает с внутренней моделью компании, вряд ли ты хочешь пускать в него весь интернет. Добавь простую проверку по списку разрешённых Telegram ID:
ALLOWED_USERS = {111111111, 222222222}
@dp.message()
async def handle_message(message: types.Message) -> None:
if message.from_user.id not in ALLOWED_USERS:
await message.answer("Доступ к боту ограничен.")
return
# остальная логикаСвой Telegram ID можно узнать у бота @userinfobot — просто напиши ему что угодно, и он вернёт числовой идентификатор.
Список разрешённых пользователей, зашитый в код, неудобно менять каждый раз через редактирование файла и перезапуск сервиса. Для команды побольше вынеси список в отдельный JSON-файл на диске и перечитывай его при каждом сообщении — так добавить нового коллегу сможет любой администратор без доступа к самому коду бота:
import json
def load_allowed_users() -> set[int]:
with open("/opt/ai-bot/allowed_users.json") as f:
return set(json.load(f))Потоковая генерация ответа
Ждать полный ответ модели молча — не лучший UX, особенно если модель большая и генерирует ответ 10-15 секунд. В Telegram нет настоящего стриминга текста, как в веб-чатах, но можно сымитировать эффект печати, редактируя одно и то же сообщение по мере поступления токенов от Ollama в потоковом режиме:
async def ask_model_streaming(chat_id: int, prompt: str, message: types.Message) -> None:
history[chat_id].append({"role": "user", "content": prompt})
sent = await message.answer("...")
buffer = ""
async with httpx.AsyncClient(timeout=120) as client:
async with client.stream(
"POST",
f"{OLLAMA_URL}/api/chat",
json={"model": MODEL_NAME, "messages": history[chat_id], "stream": True},
) as response:
async for line in response.aiter_lines():
if not line:
continue
chunk = httpx.Response(200, content=line).json()
buffer += chunk["message"]["content"]
if len(buffer) % 40 < 5:
await sent.edit_text(buffer)
await sent.edit_text(buffer)
history[chat_id].append({"role": "assistant", "content": buffer})Редактировать сообщение на каждый токен нельзя — Telegram API ограничивает частоту правок одного сообщения, поэтому обновляем текст не чаще, чем раз в несколько десятков символов, и отправляем финальную версию отдельно после завершения потока.
Ограничение частоты запросов
Без rate limiting один активный пользователь может завалить бота (и модель на GPU) очередью запросов быстрее, чем модель успевает отвечать. Простейшая защита — блокировка повторного запроса, пока предыдущий не завершился:
busy_users: set[int] = set()
@dp.message()
async def handle_message(message: types.Message) -> None:
user_id = message.from_user.id
if user_id in busy_users:
await message.answer("Дождись ответа на предыдущий вопрос.")
return
busy_users.add(user_id)
try:
answer = await ask_model_with_history(message.chat.id, message.text)
await message.answer(answer)
finally:
busy_users.discard(user_id)Для более серьёзной нагрузки имеет смысл добавить очередь задач (например, через asyncio.Queue или отдельный воркер) вместо блокировки — тогда сообщения не будут теряться, а просто встанут в очередь на обработку.
Webhook вместо long polling
Режим start_polling из наших примеров прост, но не оптимален для продакшена — бот постоянно опрашивает серверы Telegram. Если сервер srv-ai01 доступен из интернета по HTTPS (через тот же Caddy, что мы настраивали для Open WebUI), выгоднее переключиться на webhook — Telegram сам будет присылать обновления на твой эндпоинт:
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
from aiohttp import web
WEBHOOK_PATH = "/webhook"
WEBHOOK_URL = "https://bot.itpages.lab.local" + WEBHOOK_PATH
async def on_startup(bot: Bot) -> None:
await bot.set_webhook(WEBHOOK_URL)
app = web.Application()
SimpleRequestHandler(dispatcher=dp, bot=bot).register(app, path=WEBHOOK_PATH)
setup_application(app, dp, bot=bot)
dp.startup.register(on_startup)
if __name__ == "__main__":
web.run_app(app, host="127.0.0.1", port=8081)За портом 8081 ставится реверс-прокси с валидным TLS-сертификатом — Telegram не примет webhook на самоподписанный сертификат без явного указания собственного корневого сертификата, поэтому для боевого бота проще всего использовать публичный домен с сертификатом от Let’s Encrypt.
Деплой как systemd-сервис
Чтобы бот работал постоянно и переживал перезагрузку сервера, оформим его как systemd-сервис. Создай файл /etc/systemd/system/ai-bot.service:
[Unit]
Description=Telegram AI Bot
After=network.target
[Service]
Type=simple
User=aibot
WorkingDirectory=/opt/ai-bot
ExecStart=/opt/ai-bot/venv/bin/python /opt/ai-bot/bot.py
Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.targetСоздай отдельного системного пользователя без прав на shell, от имени которого будет работать бот, — это ограничит ущерб в случае уязвимости в коде:
sudo useradd -r -s /usr/sbin/nologin aibot
sudo chown -R aibot:aibot /opt/ai-botВключи и запусти сервис:
sudo systemctl daemon-reload
sudo systemctl enable --now ai-bot
sudo systemctl status ai-botЛоги бота смотри стандартно через journalctl — это первое место для диагностики, если бот перестал отвечать:
journalctl -u ai-bot -fЛогирование и мониторинг
Стандартного вывода в journalctl достаточно для отладки, но для продакшен-бота полезно писать логи ещё и в файл с ротацией, чтобы можно было анализировать историю ошибок за недели, не гоняя journalctl с огромным диапазоном дат. Добавь в начало bot.py настройку логирования через RotatingFileHandler:
from logging.handlers import RotatingFileHandler
file_handler = RotatingFileHandler(
"/var/log/ai-bot/bot.log", maxBytes=10_000_000, backupCount=5
)
file_handler.setFormatter(
logging.Formatter("%(asctime)s %(levelname)s %(message)s")
)
logging.getLogger().addHandler(file_handler)Не забудь создать директорию под логи и выдать права пользователю aibot:
sudo mkdir -p /var/log/ai-bot
sudo chown aibot:aibot /var/log/ai-botПолезно также логировать не только ошибки, но и базовую статистику — количество запросов в час, среднее время ответа модели. Даже простой счётчик в памяти процесса, который раз в сутки пишет сводку в лог, сильно облегчает разговор с руководством о том, стоит ли докупать GPU или менять модель на более быструю.
Частые ошибки и как их избежать
- Бот молчит на длинные запросы и падает по таймауту. Модель на CPU или под нагрузкой может отвечать дольше стандартного таймаута httpx. Увеличь timeout в клиенте и обязательно отправляй chat_action «typing», чтобы пользователь видел, что бот работает, а не завис.
- Неограниченный рост истории диалога. Если не обрезать список history, контекст будет расти бесконечно, каждый запрос станет медленнее и в итоге упрётся в лимит контекстного окна модели. Всегда ограничивай историю разумным числом последних сообщений.
- Хранение токена бота в коде. Токен, закоммиченный в открытый репозиторий, будет найден и использован ботами-сканерами за считаные минуты. Держи его только в .env и переменных окружения systemd-сервиса.
Что дальше
Тема ИИ-инструментов быстро развивается — вот соседние практические темы серии: