Как подключить ИИ-агента к Telegram-боту: пошаговая инструкция

Telegram-бот - один из самых быстрых способов дать пользователям доступ к ИИ-агенту. Не нужно делать отдельный интерфейс: человек пишет сообщение в Telegram, бот передает его агенту, агент думает, при необходимости вызывает инструменты или RAG, а затем возвращает ответ в чат.

Короткая версия: создайте бота через BotFather, сохраните токен в env, примите сообщение через polling или webhook, передайте текст в функцию агента и отправьте ответ методом sendMessage.

Что мы собираем

Соберем минимальную схему Telegram-бота для ИИ-агента. В статье будет Python-пример для быстрого старта, но сама логика такая же для Laravel, Node.js, n8n, Flowise или LangGraph.

• Пользователь пишет сообщение боту.
• Telegram отправляет update вашему приложению.
• Приложение достает chat_id и текст сообщения.
• Текст передается в agent_answer().
• Ответ отправляется обратно через Telegram Bot API.
• Для теста используем polling, для production - webhook.

Шаг 1. Создайте бота через BotFather

Откройте Telegram и найдите @BotFather. Создайте нового бота командой /newbot, задайте имя и username. В конце BotFather выдаст token - это секретный ключ, через который ваше приложение управляет ботом.

• Не публикуйте token в коде, GitHub, скриншотах и логах.
• Храните token в .env или секретах хостинга.
• Если token утек, перевыпустите его в BotFather.
• Для тестов лучше создать отдельного dev-бота.

Шаг 2. Подготовьте проект

Для простого Python-прототипа удобно использовать библиотеку python-telegram-bot. Она скрывает часть низкоуровневой работы с getUpdates и sendMessage, но важно понимать, что внутри все равно используется Telegram Bot API.

mkdir telegram-ai-agent
cd telegram-ai-agent
python -m venv .venv
.venv\Scripts\activate
pip install -U python-telegram-bot python-dotenv

Создайте файл .env:

TELEGRAM_BOT_TOKEN=123456:your_bot_token_here

Шаг 3. Сделайте функцию агента

На первом шаге не привязывайте Telegram к сложной агентной системе. Сначала сделайте одну функцию, которая принимает user_id и text, а возвращает строку. Потом внутри нее можно подключить LangGraph, Flowise API, n8n webhook, RAG или собственный backend.

async def agent_answer(user_id: str, text: str) -> str:
    # Здесь позже будет вызов LLM, LangGraph, RAG или вашего API.
    if text.lower().strip() in {"/start", "старт"}:
        return "Привет! Напишите вопрос, а я помогу разобраться."

    return f"Я получил сообщение: {text}"

Такой слой полезен тем, что Telegram остается только транспортом, а агентная логика не размазывается по обработчикам сообщений.

Шаг 4. Запустите polling для локального теста

Polling - это режим, когда приложение само регулярно спрашивает Telegram: “есть новые сообщения?”. Для разработки это проще webhook: не нужен публичный HTTPS-адрес и сервер можно запускать прямо на ноутбуке.

import os
from dotenv import load_dotenv
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, ContextTypes, filters

load_dotenv()


async def agent_answer(user_id: str, text: str) -> str:
    if text.lower().strip() in {"/start", "старт"}:
        return "Привет! Напишите вопрос, а я помогу разобраться."

    return f"Я получил сообщение: {text}"


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text("Привет! Задайте вопрос ИИ-агенту.")


async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if update.message is None or update.message.text is None:
        return

    user_id = str(update.effective_user.id)
    answer = await agent_answer(user_id, update.message.text)
    await update.message.reply_text(answer)


def main() -> None:
    token = os.environ["TELEGRAM_BOT_TOKEN"]
    app = Application.builder().token(token).build()

    app.add_handler(CommandHandler("start", start))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

Запустите:

python bot.py

Откройте своего бота в Telegram, отправьте /start и обычное сообщение. Если бот отвечает, транспорт работает.

Шаг 5. Подключите настоящего агента

Теперь замените заглушку agent_answer() на вызов вашей агентной системы. Это может быть локальный LangGraph-граф, HTTP API, n8n webhook или Flowise prediction endpoint. Главное - не смешивать Telegram-обработчик и внутреннюю логику агента.

async def agent_answer(user_id: str, text: str) -> str:
    result = await my_agent.ainvoke(
        {"messages": [{"role": "user", "content": text}]},
        {"configurable": {"thread_id": f"telegram-{user_id}"}},
    )

    return result["messages"][-1].content

Если у агента есть память, используйте Telegram user_id как часть thread_id или namespace. Если бот работает в группах, добавляйте chat_id, иначе память разных чатов может смешаться.

Шаг 6. Добавьте базовые ограничения

Telegram-бот находится ближе к пользователю, чем админка или backend. Поэтому ограничения нужны сразу, даже в прототипе.

• Ограничьте длину входящего сообщения.
• Не отправляйте в модель файлы и вложения без отдельной обработки.
• Добавьте allowlist для приватного тестового бота.
• Логируйте ошибки, но не пишите token и персональные данные в открытые логи.
• Добавьте сообщение “думаю...” или typing action для длинных ответов.
• Разделите ошибки агента и ошибки Telegram API.

Шаг 7. Обработайте длинные ответы

У Telegram есть ограничения на длину сообщения. Даже если точный лимит меняется в деталях по типам сообщений, лучше заранее резать большие ответы на части и не пытаться отправить огромный текст одним sendMessage.

def split_message(text: str, limit: int = 3500) -> list[str]:
    parts = []
    current = text

    while len(current) > limit:
        cut = current.rfind("\n", 0, limit)
        if cut == -1:
            cut = limit
        parts.append(current[:cut].strip())
        current = current[cut:].strip()

    if current:
        parts.append(current)

    return parts

Для статей, отчетов и больших инструкций лучше отправлять краткий ответ и ссылку на документ, а не засыпать чат длинной простыней.

Шаг 8. Перейдите на webhook для production

Webhook - это режим, когда Telegram сам отправляет update на ваш HTTPS endpoint. Для production это обычно надежнее и экономнее, чем постоянный polling-процесс.

Схема такая:

• У вас есть публичный HTTPS endpoint, например https://example.com/telegram/webhook.
• Вы вызываете setWebhook и передаете этот URL.
• Telegram начинает отправлять POST-запросы с update.
• Ваш backend проверяет секретный заголовок, обрабатывает update и отвечает 200 OK.

Пример установки webhook:

curl -X POST "https://api.telegram.org/bot$TELEGRAM_BOT_TOKEN/setWebhook" \
  -d "url=https://example.com/telegram/webhook" \
  -d "secret_token=your-random-secret"

Проверка состояния:

curl "https://api.telegram.org/bot$TELEGRAM_BOT_TOKEN/getWebhookInfo"

Если webhook установлен, getUpdates для polling работать не будет, пока вы не удалите webhook.

Шаг 9. Проверьте webhook security

Webhook нельзя оставлять как открытый endpoint “кто угодно прислал JSON - мы вызвали агента”. Минимально проверьте secret token, который Telegram отправляет в заголовке X-Telegram-Bot-Api-Secret-Token.

def is_valid_telegram_request(headers: dict) -> bool:
    expected = os.environ["TELEGRAM_WEBHOOK_SECRET"]
    return headers.get("X-Telegram-Bot-Api-Secret-Token") == expected

Дополнительно полезны rate limit, проверка allowed_updates, ограничение типов сообщений и отдельная очередь задач, если агент отвечает долго.

Шаг 10. Настройте память под Telegram

У Telegram есть несколько идентификаторов, и важно выбрать правильный. Для личного чата часто достаточно user_id. Для групп и командных чатов лучше учитывать chat_id, потому что один и тот же пользователь может писать в разных контекстах.

• Личный ассистент: thread_id = telegram-user-{user_id}.
• Групповой бот: thread_id = telegram-chat-{chat_id}.
• Агент поддержки: thread_id = ticket-{chat_id}-{user_id}.
• Долгосрочная память: namespace = (user_id, "memories") или (company_id, "memories").

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

Шаг 11. Добавьте очередь для долгих задач

Если агент вызывает RAG, CRM, несколько tools или внешний API, ответ может занимать больше нескольких секунд. В таком случае лучше принять сообщение быстро, положить задачу в очередь и отправить результат отдельным сообщением.

• Webhook endpoint отвечает быстро и не висит на LLM-запросе.
• Очередь выполняет agent run в фоне.
• Пользователь получает “принял, обрабатываю”.
• После завершения бот отправляет итоговый ответ через sendMessage.
• Ошибки попадают в лог и понятное сообщение пользователю.

Шаг 12. Проверьте запуск

Перед публикацией пройдите короткий чеклист. Telegram-боты часто ломаются не из-за модели, а из-за токена, webhook, HTTPS, прав доступа или смешанной памяти.

• /start отвечает.
• Обычное текстовое сообщение проходит до agent_answer().
• Ошибка агента не роняет процесс бота.
• Длинный ответ делится на части.
• Token лежит в env, а не в коде.
• Для webhook настроен HTTPS и secret_token.
• getWebhookInfo не показывает ошибок.
• Память не смешивает пользователей и чаты.
• В логах нет секретов и персональных данных.

Частые вопросы

Что лучше для старта: polling или webhook?

Для локальной разработки проще polling: запустили скрипт и сразу тестируете. Для production обычно лучше webhook, потому что Telegram сам присылает updates на ваш HTTPS endpoint, а приложение не держит постоянный цикл опроса.

Можно ли подключить Telegram к n8n или Flowise без кода?

Да. В n8n можно собрать Telegram Trigger и дальше отправлять текст в AI Agent node или HTTP-запросом в свою агентную систему. В Flowise чаще делают отдельный endpoint агента, а Telegram-слой вызывает его через middleware или automation-платформу.

Где хранить токен Telegram-бота?

В переменных окружения или секретах сервера. Не храните token в репозитории, админке без защиты, публичных логах и примерах кода. При утечке перевыпустите токен через BotFather.

Нужно ли делать отдельную память для Telegram?

Да, как минимум нужно правильно выбрать thread_id. Личный чат, группа и заявка поддержки - разные контексты. Если использовать только user_id, агент может перенести контекст из одного чата в другой.

Что делать, если webhook не работает?

Проверьте HTTPS, правильность URL, getWebhookInfo, логи сервера, статус 200 OK на endpoint, secret token и то, что вы не пытаетесь одновременно использовать webhook и getUpdates polling.