Создаём полноценного AI-агента в Telegram: асинхронная архитектура на python-telegram-bot, интеграция с LangGraph, стриминг ответов, обработка голосовых и файлов, многопользовательские сессии с изоляцией контекста.
Архитектура: Telegram API → PTB Application → Agent Dispatcher → LangGraph с сессиями и стримингом
python-telegram-bot v21 — асинхронный фреймворк, идеально сочетающийся с async-методами LangGraph. Создаём Application, регистрируем хендлеры и при каждом сообщении запускаем граф агента. Thread_id изолирует контекст по chat_id пользователя.
import os from telegram.ext import Application, CommandHandler, MessageHandler, filters from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.sqlite import SqliteSaver from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o", temperature=0.3) agent = create_react_agent(llm, [search_tool, calculator_tool, weather_tool]) @st.cache_resource def get_checkpointer(): return SqliteSaver.from_conn_string("checkpoints.db") async def handle_message(update, context): user_id = str(update.effective_chat.id) user_msg = update.message.text # Отправляем индикатор «печатает...» await context.bot.send_chat_action(chat_id=user_id, action="typing") # Запускаем агента с изолированным контекстом config = {"configurable": {"thread_id": user_id}} result = await agent.ainvoke( {"messages": [("user", user_msg)]}, config=config ) response = result["messages"][-1].content await update.message.reply_text(response[:4096]) # Telegram limit async def start(update, context): await update.message.reply_text( "🤖 Привет! Я AI-агент. Могу искать информацию, " "считать, узнавать погоду и анализировать данные.\n" "Просто напиши свой вопрос!" ) app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build() app.add_handler(CommandHandler("start", start)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) app.run_polling()
Пользователи не должны ждать 20 секунд. Реализуем стриминг: отправляем пустое сообщение, затем редактируем его каждые 0.5 секунды по мере генерации токенов. Используем astream_events для перехвата токенов LLM и бот.edit_message_text для обновления.
import asyncio async def stream_response(update, context): user_id = str(update.effective_chat.id) msg = update.message.text # Отправляем заглушку, которую будем редактировать placeholder = await update.message.reply_text("⏳ Думаю...") config = {"configurable": {"thread_id": user_id}} response_text = "" last_update = asyncio.get_event_loop().time() async for event in agent.astream_events( {"messages": [("user", msg)]}, config=config, version="v2" ): if event["event"] == "on_chat_model_stream": token = event["data"]["chunk"].content if token: response_text += token # Обновляем сообщение не чаще чем раз в 0.5 сек now = asyncio.get_event_loop().time() if now - last_update > 0.5 and len(response_text) < 4000: try: await context.bot.edit_message_text( response_text + " ⏳", chat_id=user_id, message_id=placeholder.message_id ) last_update = now except Exception: pass # Игнорируем ошибки редактирования # Финальное обновление сообщения try: await context.bot.edit_message_text( response_text[:4096], chat_id=user_id, message_id=placeholder.message_id ) except Exception: await context.bot.send_message(chat_id=user_id, text=response_text[:4096])
Telegram бот должен понимать голосовые. Добавляем обработчик голосовых сообщений: скачиваем аудиофайл, транскрибируем через OpenAI Whisper, передаём текст агенту и отправляем ответ (можно тоже голосом через TTS).
from openai import AsyncOpenAI import io client = AsyncOpenAI() async def handle_voice(update, context): user_id = str(update.effective_chat.id) # Скачиваем голосовое сообщение voice_file = await update.message.voice.get_file() audio_bytes = await voice_file.download_as_bytearray() # Транскрибируем через Whisper await context.bot.send_chat_action(chat_id=user_id, action="typing") transcript = await client.audio.transcriptions.create( model="whisper-1", file=("voice.ogg", io.BytesIO(audio_bytes), "audio/ogg") ) # Показываем распознанный текст await update.message.reply_text(f"🎤 Распознано: {transcript.text}") # Отправляем агенту config = {"configurable": {"thread_id": user_id}} result = await agent.ainvoke( {"messages": [("user", transcript.text)]}, config=config ) await update.message.reply_text(result["messages"][-1].content[:4096]) app.add_handler(MessageHandler(filters.VOICE, handle_voice))
Агент должен работать с файлами: анализировать PDF-документы, описывать изображения через GPT-4V, читать таблицы. Добавляем обработчики для разных типов файлов и интегрируем с мультимодальной моделью.
from telegram import Update from telegram.ext import ContextTypes async def handle_document(update: Update, context: ContextTypes.DEFAULT_TYPE): user_id = str(update.effective_chat.id) doc = update.message.document file = await context.bot.get_file(doc.file_id) file_bytes = await file.download_as_bytearray() caption = update.message.caption or "Проанализируй этот файл" await context.bot.send_chat_action(chat_id=user_id, action="typing") # Маршрутизация по типу файла if doc.mime_type == "application/pdf": content = extract_pdf_text(file_bytes) elif doc.mime_type in ("image/png", "image/jpeg"): # Мультимодальный анализ через GPT-4V import base64 b64 = base64.b64encode(file_bytes).decode() content = f"[IMAGE: data:image/png;base64,{b64[:100]}...]" else: content = file_bytes.decode("utf-8", errors="ignore")[:3000] config = {"configurable": {"thread_id": user_id}} result = await agent.ainvoke( {"messages": [("user", f"{caption}\n\nСодержимое файла:\n{content}")]}, config=config ) await update.message.reply_text(result["messages"][-1].content[:4096]) app.add_handler(MessageHandler(filters.Document.ALL, handle_document)) app.add_handler(MessageHandler(filters.PHOTO, handle_photo))
Когда ботом пользуются сотни людей, нужно надёжное хранение диалогов. SQLite-чекпоинтер LangGraph хранит состояние каждого треда. Для продакшена переключаемся на Redis + Postgres. Добавляем команды: /clear для сброса контекста, /export для выгрузки истории.
from langgraph.checkpoint.postgres import PostgresSaver from contextlib import asynccontextmanager DB_URL = os.getenv("DATABASE_URL", "postgresql://user:pass@localhost/agent_db") async def get_agent(): async with PostgresSaver.from_conn_string(DB_URL) as checkpointer: await checkpointer.setup() agent = create_react_agent(llm, tools, checkpointer=checkpointer) return agent async def clear_context(update, context): user_id = str(update.effective_chat.id) # Создаём новый thread_id — агент забывает контекст context.user_data["thread_id"] = f"{user_id}-{uuid4()}" await update.message.reply_text("🧹 Контекст очищен. Начинаем с чистого листа!") async def export_history(update, context): user_id = str(update.effective_chat.id) config = {"configurable": {"thread_id": context.user_data.get("thread_id", user_id)}} state = await agent.aget_state(config) if state and state.values.get("messages"): history = "\n".join( f"{'👤' if m.type=='human' else '🤖'}: {m.content[:200]}" for m in state.values["messages"][-20:] ) await update.message.reply_text(f"📋 Последние сообщения:\n{history[:4000]}") app.add_handler(CommandHandler("clear", clear_context)) app.add_handler(CommandHandler("export", export_history))
Мы построили полноценного Telegram AI-агента с асинхронной архитектурой, живым стримингом ответов, голосовым вводом через Whisper, обработкой любых файлов и многопользовательскими сессиями на PostgreSQL. Бот помнит контекст каждого пользователя, умеет очищать историю и экспортировать диалоги. Дальше — вебхуки вместо polling, кластеризация через несколько воркеров и интеграция с Telegram Stars для монетизации.