От Telegram-бота к AI-агенту: как собрать своего «исполнителя задач» на Python в 2025-м

В 2023–2024 почти каждый второй pet-проект с LLM выглядел как чатик: ты спрашиваешь — модель отвечает, иногда с RAG, иногда без. В 2025-м тренд сместился: на рынке всё чаще говорят про AI-агентов — системы, которые не просто болтают, а сами инициируют действия, ходят в API, планируют шаги и живут в продакшене как часть инфраструктуры.

В прошлых проектах я уже собирал Telegram-ботов: от простого «ресепшена» для малого бизнеса на aiogram 3.x до RAG-консультанта по железу «Кремний» на бесплатном стеке Groq + sentence-transformers. Логичный следующий шаг — научить бота не только отвечать в диалоге, но и самостоятельно выполнять задачи в фоне: следить за ценами на железо, мониторить статусы заказов или пинговать при аномалиях.

В этой статье разберём на практике минимальный AI-агент вокруг Telegram-бота: архитектуру, стек и рабочий код на Python. Получится небольшой, но честный «исполнитель задач», которого можно дорастить до чего-то полезного в проде.

Что такое AI-агент «в полях»

Если отбросить маркетинговые определения, AI-агент в инженерном смысле — это:

• Цель. Чётко сформулированная задача: «следить за X и сигналить, когда Y».

• Инструменты. Набор функций/обёрток над внешними системами (HTTP-API, БД, файловая система, очереди), которые агент умеет вызывать.

• Память. Хранилище состояния между запусками (что уже видел, что отправил пользователю, какие ошибки поймал).

• Контур принятия решений. Обычно это LLM, которая на основе целей, контекста и результатов инструментов решает, что делать дальше.

Главное отличие от «просто бота» в том, что агент может запускаться без участия пользователя: по расписанию, по событию из очереди, по веб-хуку от другого сервиса и т.д. Он сам инициирует действия и сам приходит к человеку с результатом.

В продакшене такие агенты всё чаще используют не как «умные игрушки», а как прослойку между LLM и реальными бизнес-процессами: обработка тикетов, автоматизация части саппорта, мониторинг логов, подготовка отчётов и др.

Архитектура мини-проекта: бот + агент

Возьмём приземлённый кейс: агент раз в N минут проверяет цены на видеокарты по HTTP-API магазина (или вашему внутреннему сервису), сравнивает с прошлым состоянием и, если есть интересные изменения, отправляет сводку в Telegram.

Высокоуровневая схема будет такой:

• Telegram-бот
  Обёртка вокруг Bot API (aiogram или python-telegram-bot), принимает команды от тебя как владельца и отсылает уведомления, которые генерирует агент.

• Агент-воркер
  Отдельный процесс (или сервис), который:

  • по расписанию (cron/systemd timer/встроенный планировщик) просыпается;

  • вызывает инструмент fetch_prices() (HTTP-клиент);

  • сравнивает результат с предыдущим снимком в БД;

  • если есть интересные изменения — формирует человекочитаемое резюме через LLM и шлёт его в Telegram.

• Хранилище
  Для демо хватит SQLite/файла JSON, в проде — Postgres/Redis/что-то ещё. Здесь лежат:

  • последний известный список цен;

  • метаданные уведомлений (когда и по какому товару уже слали нотификацию).

• LLM-провайдер
  Любой OpenAI-совместимый API: это может быть тот же Groq с Llama 3.1 8B, который уже неплохо показал себя в RAG-боте, или любая другая модель.

Telegram-бот и агент могут жить в одном репозитории, но как разные entrypoint’ы: bot_main.py и agent_worker.py. Так проще деплоить, перезапускать и масштабировать.

Стек и подготовка окружения

Стек возьмём максимально доступный:

• Python 3.11+

• aiogram 3.x — для Telegram-бота

• httpx — асинхронный HTTP-клиент для вызова API

• SQLite через sqlite3 — как простое хранилище состояния

• Любой OpenAI-совместимый клиент (openai/groq/любой аналог)

• python-dotenv — для конфигурации через .env

requirements.txt может выглядеть так:

aiogram>=3.4.0
httpx>=0.27.0
python-dotenv>=1.0.0
openai>=1.50.0  # или groq / другой OpenAI-совместимый клиент

Пример .env для локального запуска:

BOT_TOKEN=123456:ABC...
ADMIN_CHAT_ID=123456789
LLM_API_KEY=sk-...
LLM_MODEL=gpt-4.1-mini  # или llama-3.1-8b-instant у Groq
PRICE_API_URL=https://api.example.com/gpu-prices
PRICE_CHECK_INTERVAL_MIN=15

Конфиг на Python:

# config.py

from dataclasses import dataclass
import os
from dotenv import load_dotenv

load_dotenv()

@dataclass
class Settings:
bot_token: str
admin_chat_id: int
llm_api_key: str
llm_model: str
price_api_url: str
price_check_interval_min: int

def get_settings() -> Settings:
return Settings(
bot_token=os.environ["BOT_TOKEN"],
admin_chat_id=int(os.environ["ADMIN_CHAT_ID"]),
llm_api_key=os.environ["LLM_API_KEY"],
llm_model=os.environ.get("LLM_MODEL", "gpt-4.1-mini"),
price_api_url=os.environ["PRICE_API_URL"],
price_check_interval_min=int(os.environ.get("PRICE_CHECK_INTERVAL_MIN", "15")),
)

settings = get_settings()

Реализация агента: от инструмента до цикла

Начнём с простого инструмента fetch_prices, который ходит в внешнее API и возвращает словарь {"sku": {"name": ..., "price": ...}}.

# tools.py

from typing import Dict, Any
import httpx
from config import settings

async def fetch_prices() -> Dict[str, Dict[str, Any]]:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(settings.price_api_url)
resp.raise_for_status()
data = resp.json()

    # Ожидаемый формат data зависит от вашего API.
    # Для примера приведём нормализацию к словарю вида:
    # {"sku_4070": {"name": "...", "price": 65000}, ...}
    normalized: Dict[str, Dict[str, Any]] = {}
    for item in data["items"]:
        sku = item["sku"]
        normalized[sku] = {
            "name": item["name"],
            "price": float(item["price"]),
        }
    return normalized

Теперь — слой хранилища на SQLite: один файл, одна таблица с последними ценами.

# storage.py

import sqlite3
from contextlib import contextmanager
from typing import Dict, Any

DB_PATH = "agent_state.sqlite3"

@contextmanager
def get_conn():
conn = sqlite3.connect(DB_PATH)
try:
yield conn
finally:
conn.close()

def init_db() -> None:
with get_conn() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS prices (
sku TEXT PRIMARY KEY,
name TEXT NOT NULL,
price REAL NOT NULL
)
"""
)
conn.commit()

def load_last_prices() -> Dict[str, Dict[str, Any]]:
with get_conn() as conn:
cur = conn.execute("SELECT sku, name, price FROM prices")
rows = cur.fetchall()
return {sku: {"name": name, "price": price} for sku, name, price in rows}

def save_prices(prices: Dict[str, Dict[str, Any]]) -> None:
with get_conn() as conn:
conn.execute("DELETE FROM prices")
conn.executemany(
"INSERT INTO prices (sku, name, price) VALUES (?, ?, ?)",
[(sku, v["name"], v["price"]) for sku, v in prices.items()],
)
conn.commit()

Функция для вычисления дельт между старым и новым состоянием:

# diff.py

from typing import Dict, Any, List, TypedDict

class PriceChange(TypedDict):
sku: str
name: str
old_price: float
new_price: float
diff_abs: float
diff_rel: float

def compute_changes(
old: Dict[str, Dict[str, Any]],
new: Dict[str, Dict[str, Any]],
min_rel_change: float = 0.05,  # 5%
) -> List[PriceChange]:
changes: List[PriceChange] = []
for sku, item in new.items():
if sku not in old:
# Новая позиция — можно отдельно обрабатывать
continue
old_price = float(old[sku]["price"])
new_price = float(item["price"])
if old_price <= 0:
continue
diff_abs = new_price - old_price
diff_rel = diff_abs / old_price
if abs(diff_rel) < min_rel_change:
continue
changes.append(
PriceChange(
sku=sku,
name=item["name"],
old_price=old_price,
new_price=new_price,
diff_abs=diff_abs,
diff_rel=diff_rel,
)
)
return changes

Теперь — обёртка над LLM для генерации человекочитаемого отчёта. Для определённости используем openai, но интерфейс у Groq/Ollama-проксей похожий.

# llm_client.py

from typing import List
from openai import OpenAI
from config import settings
from diff import PriceChange

client = OpenAI(api_key=settings.llm_api_key)

SYSTEM_PROMPT = (
"Ты помощник по железу. "
"На вход получаешь список изменений цен на комплектующие "
"и должен коротко и понятно описать их для тех, кто собирает ПК."
)

def format_changes_for_prompt(changes: List[PriceChange]) -> str:
lines = []
for ch in changes:
direction = "подешевел" if ch["diff_abs"] < 0 else "подорожал"
rel = round(ch["diff_rel"] * 100, 1)
lines.append(
f"- {ch['name']} ({ch['sku']}) {direction} с {ch['old_price']:.0f} "
f"до {ch['new_price']:.0f} руб. ({rel:+.1f}%)."
)
return "n".join(lines)

def build_user_prompt(changes: List[PriceChange]) -> str:
base = "Ниже список изменений цен:nn"
base += format_changes_for_prompt(changes)
base += (
"nnСделай краткий отчёт для чата в Telegram: "
"что изменилось и на что стоит обратить внимание энтузиасту сборок ПК."
)
return base

def generate_report(changes: List[PriceChange]) -> str:
user_prompt = build_user_prompt(changes)
completion = client.chat.completions.create(
model=settings.llm_model,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
temperature=0.4,
max_tokens=800,
)
return completion.choices.message.content.strip()

И, наконец, сам агент-воркер: связать всё вместе и отправить результат в Telegram.

# agent_worker.py

import asyncio
import logging
from contextlib import suppress

from aiogram import Bot
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode

from config import settings
from storage import init_db, load_last_prices, save_prices
from tools import fetch_prices
from diff import compute_changes
from llm_client import generate_report

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def run_once(bot: Bot) -> None:
logger.info("Запуск проверки цен")
old = load_last_prices()
new = await fetch_prices()

    changes = compute_changes(old, new, min_rel_change=0.05)
    if not changes:
        logger.info("Существенных изменений нет")
        save_prices(new)
        return
    
    logger.info("Найдены изменения: %s позиций", len(changes))
    report = generate_report(changes)
    
    with suppress(Exception):
        await bot.send_message(
            chat_id=settings.admin_chat_id,
            text=report,
            parse_mode=ParseMode.HTML,
        )
    
    save_prices(new)
    logger.info("Проверка завершена")
    async def main() -> None:
init_db()
bot = Bot(
token=settings.bot_token,
default=DefaultBotProperties(parse_mode=ParseMode.HTML),
)
interval = settings.price_check_interval_min * 60

    while True:
        try:
            await run_once(bot)
        except Exception as e:  # в проде лучше ловить аккуратнее
            logger.exception("Ошибка при выполнении агента: %s", e)
        await asyncio.sleep(interval)
    if __name__ == "__main__":
asyncio.run(main())

Такой воркер можно запускать как отдельный сервис (systemd unit, Docker-контейнер) параллельно с основным Telegram-ботом или вообще без него, если уведомления уходят только тебе.

Telegram-слой: команда для ручного запуска и проверка

Чтобы агент не был полностью «чёрным ящиком», полезно добавить в бота команду, которая руками триггерит один прогон и показывает последние изменения.

Минимальный bot_main.py:

# bot_main.py

import asyncio
import logging
import sys

from aiogram import Bot, Dispatcher, F
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import CommandStart, Command
from aiogram.types import Message

from config import settings
from agent_worker import run_once  # реиспользуем логику

logging.basicConfig(level=logging.INFO, stream=sys.stdout)
dp = Dispatcher()

@dp.message(CommandStart())
async def cmd_start(message: Message) -> None:
await message.answer(
"Привет! Это менеджер цен на железо.n"
"Агент сам проверяет цены по расписанию, "
"а командой /check можно запустить проверку вручную."
)

@dp.message(Command("check"))
async def cmd_check(message: Message) -> None:
await message.answer("Запускаю проверку цен, подожди пару секунд...")
# В идеале здесь делегировать задачу в очередь, а не блокировать хэндлер,
# но для демо можно сделать так:
bot = message.bot
await run_once(bot)
await message.answer("Готово. Если были изменения, отчёт уже в этом чате.")

@dp.message(F.text)
async def fallback(message: Message) -> None:
await message.answer(
"Пока я умею только /start и /check. "
"Остальное доучим по мере развития проекта."
)

async def main() -> None:
bot = Bot(
token=settings.bot_token,
default=DefaultBotProperties(parse_mode=ParseMode.HTML),
)
await dp.start_polling(bot)

if __name__ == "__main__":
asyncio.run(main())

Теперь у тебя есть связка:

• фонового агента, который сам по себе живёт и проверяет цены;

• Telegram-бота, через которого можно руками инициировать проверку и получать уведомления.

Типичные грабли и как их обойти

AI-агент — это не только про LLM, но и про «грязную» операционку вокруг неё.

На что стоит обратить внимание уже в первой версии:

• Идемпотентность.
  Если воркер перезапустился на середине run_once, есть риск продублировать уведомления. Решение — хранить в БД не только последние цены, но и, например, хэш последнего отчёта или timestamp последней нотификации по каждому SKU.

• Rate limiting и ошибки API.
  Внешний сервис цен и LLM-провайдер могут резать по лимитам. Добавь экспоненциальный backoff, ограничение числа попыток и метрики (даже простые счётчики в логах), чтобы видеть, когда ты упёрся в лимит.

• Стоимость токенов.
  Если список изменений длинный, счёт за LLM может неприятно удивить. Помогает предфильтрация на Python: сгруппировать изменения по категориям, отсечь совсем мелкие дельты и отдавать в LLM только агрегированный объём.

• Наблюдаемость.
  Даже для pet-проекта полезно писать осмысленные логи и, по возможности, прокидывать их хотя бы в journald или отдельный файл. Следующий шаг — Prometheus/Grafana, но это уже отдельная история.

Куда развивать агента дальше

Даже такой минимальный агент — уже не просто «бот с кнопками», а самостоятельный исполнитель задач, который живёт рядом с твоей инфраструктурой.

Несколько направлений, куда его можно прокачать:

• Мультитаскинг
  Завести несколько типов задач: мониторинг цен, проверка наличия, отслеживание статусов заказов, автоответы на типовые тикеты. Хранить их в очереди (RabbitMQ, Redis, Kafka) и давать LLM роль диспетчера.

• Богатая память
  Вместо SQLite — нормальную БД с историей изменений, чтобы агент мог отвечать на вопросы «как менялась цена этой карты за месяц» прямо в чате.

• Мультиагентная схема
  Разнести роли: один агент только собирает сырые данные, второй агрегирует, третий общается с пользователем в Telegram. В 2025-м как раз много фреймворков, которые упрощают такие сценарии (workflow-движки, LangGraph-подобные решения и т.п.).

• Интеграция в бизнес-процессы
  Вместо «игрушечного» мониторинга — реальные задачи: подсветка аномалий в логах, подготовка ежедневных отчётов, авто-драфты писем клиентам по шаблонам.

Если ты уже писал ботов и RAG-сервисы, шаг к AI-агентам — это не про магию, а про ещё один слой инженерии: планирование, инструменты, состояние и ответственность за действия кода. И такой маленький агент на Python — нормальная точка входа, чтобы почувствовать, как эта архитектура живёт в реальном окружении.