Как построить надежного AI-агента для бизнеса: от контекста до анти-фейл-систем на примере XelaBot

Каждый второй AI-агент, который мне доводилось ревьюить, умирает в первую же неделю продакшена. Не потому что LLM тупая, а потому что инженеры забывают про фундаментальные вещи: контекст, изоляцию сессий, защиту от ложных срабатываний. В этой статье разберём, как построить агента, который не сломается, на примере XelaBot — открытого фреймворка для бизнес-автоматизации, который я собирал по кирпичикам последние полгода.

Забудьте про демки с Jupyter. Здесь будет кровь, пот и реальные интеграционные грабли. Поехали.

Почему 90% AI-агентов — это просто скрипты с API-обёрткой

Когда я говорю "агент", я имею в виду не очередной LangChain-пайплайн, который с вероятностью 30% возвращает "извините, я не могу это сделать". Настоящий бизнес-агент должен:

• Удерживать контекст на протяжении дней и недель (а не 128k токенов диалога)
• Выполнять действия в реальном мире: отправлять email, создавать тикеты, обновлять CRM
• Не врать (ну, хотя бы не врать критично)
• Восстанавливаться после падений и не терять данные

XelaBot (версия 2.4, май 2026) родился из боли: клиент хотел, чтобы агент сам собирал заявки из Telegram, проверял статус заказа в SAP, обзванивал клиентов через Twilio и писал отчёты. Ни одно коробочное решение не справлялось. Пришлось строить своё.

Столп первый: контекст, который не теряется через час

Самый частый баг, который я вижу в production: агент "забывает", что делал 15 минут назад. LLM-провайдеры режут контекст, сдвигают окно, и агент начинает отвечать так, будто задача только началась.

Решение — гибридная память: краткосрочная (in-memory + Redis) и долгосрочная (векторная БД с суммаризациями). Вот как это выглядит в XelaBot:

import redis
from sentence_transformers import SentenceTransformer
import chromadb

class HybridMemory:
    def __init__(self, redis_url="redis://localhost:6379/0", 
                 collection_name="agent_memory"):
        self.redis = redis.from_url(redis_url)
        self.model = SentenceTransformer("all-MiniLM-L6-v2")
        self.chroma = chromadb.PersistentClient(path="./memory_db")
        self.collection = self.chroma.get_or_create_collection(collection_name)

    def add(self, session_id, role, content):
        # краткосрочная: храним последние N сообщений
        self.redis.rpush(f"{session_id}:short", f"{role}: {content}")
        self.redis.ltrim(f"{session_id}:short", -50, -1)  # keep last 50
        
        # долгосрочная: эмбеддинг и сохранение в ChromaDB
        emb = self.model.encode(content).tolist()
        self.collection.add(
            embeddings=[emb],
            documents=[content],
            metadatas={"session_id": session_id, "role": role},
            ids=[f"{session_id}:{hash(content)}"]
        )

    def get_context(self, session_id, query, top_k=5):
        short = self.redis.lrange(f"{session_id}:short", 0, -1)
        short_context = "\n".join(short)
        
        # долгосрочный поиск по релевантности
        q_emb = self.model.encode(query).tolist()
        results = self.collection.query(
            query_embeddings=[q_emb],
            n_results=top_k
        )
        long_context = "\n".join(results["documents"][0])
        
        return short_context + "\n---\n" + long_context

Обратите внимание: я не полагаюсь на single source of truth. Redis — быстрая, Chroma — точная. Агент получает и то, и другое, а LLM сама решает, что важнее. Кстати, для доступа к моделям через единый API-шлюз рекомендую AITunnel — он стабильно работает с GPT-5 Turbo, Claude 4 и новейшими моделями Gemini, что критично для продакшена.

Столп второй: интеграции, которые не падают

Агент без интеграций — как смартфон без интернета. Но подключать CRM, ERP, мессенджеры напрямую из кода агента — самоубийство. Каждый сторонний сервис может лечь, задержаться с ответом или вернуть невалидные данные.

В XelaBot мы используем паттерн Tool Abstraction Layer:

• Каждый инструмент (SendEmail, CreateTicket, CheckOrderStatus) — отдельный микросервис с gRPC или REST, обёрнутый в ретрай-логику с exponential backoff.
• Таймауты — жёсткие. Если инструмент не ответил за 10 секунд — идём к следующему или возвращаемся с ошибкой.
• Все исходящие вызовы логируем в Prometheus (гистограммы latency).

import asyncio
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential

class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register(self, name, tool_func, timeout=10):
        self.tools[name] = (tool_func, timeout)

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    async def call(self, name, **kwargs):
        func, timeout = self.tools[name]
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session:
            async with session.post(f"http://tools/{name}", json=kwargs) as resp:
                return await resp.json()

Обратите внимание: я не использую готовые обёртки типа LangChain Tools — они слишком много магии. Лучше написать 20 строк своего кода, чем потом отлаживать неявные side effects.

Столп третий: браузерные сессии для действий в вебе

Многие бизнес-процессы требуют взаимодействия с веб-интерфейсами: старые ERP, админки, порталы. Чистый API не всегда есть. Тут спасают управляемые браузерные сессии на базе Playwright.

XelaBot использует пул сессий с изоляцией кук и local storage. Каждый инстанс агента работает в своей sandbox-сессии, чтобы не перетирать данные друг друга.

from playwright.async_api import async_playwright

class BrowserSessionPool:
    def __init__(self, max_sessions=10):
        self.sessions = {}
        self.max = max_sessions

    async def get_session(self, session_id):
        if session_id not in self.sessions:
            p = await async_playwright().start()
            browser = await p.chromium.launch(headless=True)
            context = await browser.new_context(
                storage_state=f"./sessions/{session_id}.json"
            )
            page = await context.new_page()
            self.sessions[session_id] = (p, browser, context, page)
        return self.sessions[session_id][-1]  # page

    async def close_session(self, session_id):
        if session_id in self.sessions:
            p, browser, ctx, page = self.sessions.pop(session_id)
            await ctx.storage_state(path=f"./sessions/{session_id}.json")
            await browser.close()
            await p.stop()

Столп четвёртый: anti-false-success — как не дать агенту соврать

Самое опасное в AI-агентах — ложное чувство безопасности. Агент написал "заказ оформлен", а на самом деле CRM вернула 500, но LLM проигнорировала ошибку. Или агент считал, что письмо отправлено, а SMTP-сервер отклонял соединение.

Решение: верификация действий. Каждое значимое действие (запись в БД, отправка сообщения, создание заказа) должно подтверждаться через отдельный сервис-валидатор, который сверяет ожидаемый результат с фактическим.

class ActionVerifier:
    def __init__(self, agent, validator_llm=None):
        self.agent = agent
        self.validator_llm = validator_llm  # можно использовать более дешёвую модель для проверки

    async def perform_and_verify(self, action: str, **params):
        # 1. Выполняем действие
        result = await self.agent.execute(action, **params)
        
        # 2. Проверяем через независимую утилиту (например, curl к API)
        verification = await self.verify_externally(action, params, result)
        
        if not verification["success"]:
            # 3. Если не сошлось — логируем, откатываем, уведомляем
            await self.rollback(action, params)
            await self.alert(f"False success: {action} - {verification['reason']}")
            return {"status": "failed", "reason": verification["reason"]}
        
        return {"status": "ok", "data": result}

    async def verify_externally(self, action, params, claimed_result):
        # Пример: для отправки почты проверяем SMTP лог
        if action == "send_email":
            smtp_logs = await self.get_smtp_logs(params["to"])
            if "250 OK" not in smtp_logs:
                return {"success": False, "reason": "SMTP did not confirm delivery"}
        
        # Для записи в CRM — делаем GET-запрос к созданной записи
        if action == "create_crm_contact":
            contact_id = claimed_result.get("id")
            if contact_id:
                async with aiohttp.ClientSession() as s:
                    async with s.get(f"{CRM_URL}/contacts/{contact_id}") as resp:
                        if resp.status != 200:
                            return {"success": False, "reason": "Contact not found after creation"}
        
        return {"success": True}

Концепция anti-false-success — это аналог тестов на проникновение, только для AI. Вы должны не доверять своему агенту, пока он не докажет обратное. Подробнее о том, как строить production-ready агентов, я писал в статье "Production-ready AI-агенты: как превратить хайп в работающую систему для бизнеса".

Столп пятый: шедулер и cron — автономная работа 24/7

Агент должен просыпаться по расписанию и выполнять рутинные проверки. В XelaBot мы используем обычный cron, но обёрнутый в Distributed Scheduler на базе Redis и APScheduler, чтобы исключить дублирование инстансов.

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.redis import RedisJobStore

jobstore = RedisJobStore(host="localhost", port=6379, db=1)
scheduler = AsyncIOScheduler(jobstores={"default": jobstore})

@scheduler.scheduled_job("cron", hour="9", minute="0", id="daily_summary")
async def daily_summary():
    agent = XelaBot.get_instance()
    report = await agent.run_task("Сформировать ежедневный отчёт по продажам")
    await email.send("manager@company.com", f"Отчёт за {date.today()}", report)

scheduler.start()

Ключевая деталь: каждая задача должна быть идемпотентной. Если агент упал и перезапустился, он не должен создать дубликат отчёта. Мы делаем это через unique_task_id в Redis с TTL.

Ошибки, которые стоили мне двух недель жизни

1. Доверие контексту LLM. Покажите агенту 50 сообщений — он начнёт "выдумывать" детали. Решение: суммируйте старые диалоги раз в 10 шагов.
2. Одна очередь для всех. Когда агентов много, они блокируют друг друга. Используйте отдельные воркер-пулы под разные типы задач (быстрые/медленные).
3. Отсутствие rate limiting на интеграции. CRM-системы не прощают 1000 запросов в секунду. XelaBot использует token bucket алгоритм прямо в ToolRegistry.
4. Хранение сессий в памяти без персистентности. Рестарт контейнера — и агент забывает всё, что узнал за сутки. Всё в Redis с регулярными снепшотами.

Если вы хотите глубже разобраться в проектировании агентов, рекомендую курс "AI-креатор: создаём контент с помощью нейросетей" от Skillbox — там есть блоки по агентной архитектуре (хотя фокус на контент, но фундамент тот же).

Что пошло не по плану (и как это фиксили)

Бета-тест XelaBot в ритейле выявил дикую проблему: агент путал клиентов с одинаковыми именами. Контекст подтягивал историю другого человека. Пришлось добавить строгую изоляцию сессий по user_id + tenant_id во все memory-вызовы. Каждый запрос к памяти теперь фильтруется по двум полям.

Ещё один граб: агент "зацикливался" на попытках выполнить невыполнимое действие (например, позвонить по номеру, который заблокирован). Решение — max_retries с экспоненциальной задержкой и эскалацией на человека после 3 неудач.

FAQ: что ещё нужно знать

Построить надёжного AI-агента — это не про магию промптов. Это про инженерную дисциплину: контекст, интеграции, верификацию, шедулинг. XelaBot — лишь пример, но паттерны универсальны. Начните с одного действия, добавьте проверки, потом масштабируйте. И никогда не верьте агенту на слово.

Кстати, если тема суб-агентов и композиции вам близка, загляните в "Как правильно использовать суб-агентов в AI-разработке: 3 реальных сценария" — там описано, как разбивать монолитного агента на специализированных помощников.

Подписаться на канал