Платформа для самостоятельной сборки AI-агентов: разбор архитектуры на 300K строк кода

Я — Денис Шохирев, Enterprise AI architect из Эрлангена. В DennisCraft AI Studio я внедряю AI-системы для клиентов из DACH (логистика, финтех, индустриальная автоматизация) на стеке Claude, Supabase, n8n, Doppler и self-hosted Postgres. Когда запускаешь 14 production AI-агентов за полгода, вылезают реальные боли: неожиданные race conditions, лимиты по токенам, баги в пайплайнах. Здесь разбираю архитектуру своей 300K LOC платформы, которую собирал с нуля — без маркетинговых обёрток, только что реально работает в европротакшене.

Ядро платформы: минимальный набор для production AI-агентов

Архитектурный паттерн

Основная задача — изолировать каждый AI-агент и дать ему максимально стабильный pipeline: очереди задач, трекинг ошибок, контроль внешних вызовов (Claude Code, OpenAI API), мониторинг. Каждый агент — отдельный процесс, взаимодействие через асинхронные очереди (Supabase Realtime, Redis pub/sub). Почему не microservices? В реальности для AI-агентов проще — отдельный process pool, иначе хаос с синхронизацией state и лимитами API.

Стек: что реально в проде

Поток данных: от запроса до лога

Обработка входящего запроса

Любой входящий запрос (API или UI) сначала валидируется через pydantic-схемы. Затем задача кладётся в Supabase queue. Агент-процесс асинхронно извлекает задачу и выполняет все шаги: препроцессинг, вызов LLM (Claude/Anthropic), постпроцессинг, запись результата в Postgres.

from supabase import create_client
import asyncio

async def process_task(supabase_url, supabase_key):
    supabase = create_client(supabase_url, supabase_key)
    while True:
        task = supabase.table('tasks').select('*').eq('status', 'pending').limit(1).execute()
        if task.data:
            # обработка задачи
            result = run_agent_logic(task.data[0])
            supabase.table('tasks').update({'status': 'done', 'result': result}).eq('id', task.data[0]['id']).execute()
        await asyncio.sleep(1)

Логирование и аудит

Все вызовы LLM логируются в отдельную таблицу Postgres: prompt, результат, latency, идентификаторы пользователя. Для GDPR строю отдельную таблицу audit trail: кто, когда, какой prompt, какой output. После инцидента с ошибочным выводом для финтех-клиента ввёл ручной review для 2% случайных задач с помощью n8n и Notion.

Безопасность: не доверять LLM коду

Проблема LLM-генерируемого кода

На практике больше всего уязвимостей ловлю не на входе, а в сгенерированном LLM коде. В трёх последних релизах ловил SQL-инъекции и опасные shell-вызовы в Python-скриптах, которые генерировал Claude. Для проверки использую semgrep, bandit, иногда gitleaks для анализа секретов.

semgrep --config=python security/ --error
bandit -r ./agents/
gitleaks detect --source=./

Sandboxing: LLM-код не должен портить систему

В проде каждый LLM-генерируемый код исполняется в отдельном sandbox-контейнере с ограничениями по времени, памяти, сетевым вызовам (использую firejail + custom docker). После атаки через поддельный prompt в 2024 году (SQL DELETE в RAG-агенте) ввёл фильтрацию prompt-ов через регулярки и runtime sandbox.

Мониторинг и алерты: что реально работает

Метрики и алерты

Собираю метрики через self-hosted Prometheus + Grafana: latency по каждому агенту, количество ошибок, падения очередей. Для критических алертов — интеграция с Telegram-ботом. Пример алерта: если latency > 5 секунд или процент ошибок выше 2% за 10 минут — instant notification.

groups:
- name: ai-agent-alerts
  rules:
  - alert: HighLatency
    expr: avg_over_time(agent_latency[5m]) > 5
    for: 2m
    annotations:
      summary: "Высокая задержка у AI-агента"

FAQ

Почему не беру готовые no-code AI платформы?

Всё что есть на рынке либо не проходит по GDPR/DSGVO (данные у третьих лиц), либо не тянет сложные пайплайны. Для production в Европе — self-hosted + полный контроль.

Как тестировать пайплайны на надёжность?

Пишу unit-тесты на каждый этап, плюс раз в неделю гоняю end-to-end тесты через n8n. Для LLM-агентов — снапшоты output’ов и сравнение с эталоном.

Как решаете rate limit по API?

Слой очередей и ретраев. Supabase queue + отдельный process pool на каждый LLM endpoint, чтобы не ловить 429 ошибки.

Как хранить секреты и токены?

Doppler для централизованного хранения, плюс ограничения по доступу по ролям. Критичные ключи — только на сервере, нигде не логируются.

Что с масштабируемостью?

Пока хватает горизонтального масштабирования: новые процессы агентов, отдельные очереди, отдельный Postgres реплика. Но при росте до 100+ агентов буду смотреть в сторону Kubernetes.

В какой части пайплайна вы чаще всего ловите production-баги: в обработке очередей, в LLM-логике, или в интеграциях с внешними сервисами? Интересен реальный опыт. Провожу бесплатный 30-мин аудит стека для DACH-команд, которые строят AI в регулируемых отраслях. Пишите в LinkedIn или @ger_dennis_ai.