Как построить мультиагентный AI-прогнозатор событий: архитектура Seldon Vault на Python

Почему ваши прогнозы событий всегда ошибаются?

Вы берете последние новости, скармливаете их GPT-5.5 или DeepSeek-V3, просите "предсказать, что будет дальше" и получаете туманную генерацию текста. Через неделю понимаете, что реальность пошла по другому сценарию. Знакомо? Проблема не в модели, а в архитектуре.

Одиночный LLM, даже самый продвинутый, не справляется с прогнозированием. Ему не хватает специализации, долгосрочной памяти и главное - способности оспаривать собственные выводы. Тот самый когнитивный bias, который заставляет человека делать глупости, у нейросети тоже есть. Решение - распределить интеллект между несколькими агентами, каждый со своей ролью и критическим мышлением.

Seldon Vault: архитектура, которая не даст вам облажаться

Название - отсылка к Hari Seldon из "Академии" Азимова. Это не просто пафос. Seldon Vault строится на трех китах:

1. Разделение обязанностей: вместо одного всезнайки - три узких специалиста, которые спорят между собой.
2. Память с обратной связью: система запоминает свои прогнозы, а потом сравнивает их с реальностью, чтобы учиться на ошибках. Без этого любая аналитика превращается в гадание на кофейной гуще.
3. Координация через арбитра: отдельный агент-менеджер, который не генерирует контент, а только управляет потоком данных и принимает финальное решение на основе "голосов" экспертов.

Из чего состоит наш прогнозатор

Представьте аналитический отдел в миниатюре:

• Data Collector Agent: паук, который ползает по RSS, Reddit, новостным API и вытаскивает сырые данные. Его задача - не анализ, а сбор и первичная очистка.
• Analyst Agent: скептик. Берет данные от сборщика, ищет противоречия, проверяет источники, выделяет ключевые тренды. Не верит никому, требует доказательств.
• Forecaster Agent: футуролог. На основе анализа строит вероятностные сценарии. "С вероятностью 70% компания X объявит о слиянии в течение 14 дней".
• Coordinator: тихий менеджер. Получает выводы от аналитика и прогнозиста, сверяется с исторической памятью (помните про MemV?), выдает финальный вердикт.
• Memory Vault: хранилище, где лежат все прошлые прогнозы и реальные исходы. Без этого компонента система никогда не научится. Можно реализовать через Temple Vault для простоты или векторную базу с временными метками.

Собираем систему по шагам

Забудьте про Jupyter Notebook с кучей ячеек. Мы строим продакшен-готовую систему, которая будет работать в cron или как демон. Код пишем модульно, с логированием и обработкой ошибок. Используем Python 3.11+ (на 2026 это все еще актуально, хотя 3.12 уже стабилен).

1 Подготовка и зависимости

Создаем виртуальное окружение и ставим пакеты. Ключевое на 2026:

python -m venv .venv
source .venv/bin/activate  # или .venv\Scripts\activate на Windows

# Основные зависимости
pip install httpx aiohttp beautifulsoup4 feedparser asyncpraw  # для сбора данных
pip install openai==1.12.0 deepseek-api  # DeepSeek-V3 API клиент
pip install pydantic>=2.5.0  # валидация данных
pip install redis>=5.0.0  # для кэша и памяти
pip install structlog  # логирование, которое не стыдно показать в продакшене

2 Агент-сборщик данных

Пишем не просто парсер, а устойчивого к ошибкам робота. Он должен падать, перезапускаться и продолжать с того же места.

import asyncio
import httpx
from datetime import datetime, timedelta
from typing import List, Optional
from pydantic import BaseModel
import structlog

logger = structlog.get_logger()

class NewsItem(BaseModel):
    title: str
    source: str  # 'rss', 'reddit', 'twitter'
    content: str
    published_at: datetime
    url: str
    confidence: float = 1.0  # достоверность источника

class DataCollectorAgent:
    def __init__(self, api_keys: dict):
        self.reddit_client = asyncpraw.Reddit(
            client_id=api_keys['reddit_id'],
            client_secret=api_keys['reddit_secret'],
            user_agent="SeldonVault/1.0"
        )
        self.timeout = httpx.Timeout(10.0, connect=15.0)
        
    async def fetch_rss(self, feeds: List[str]) -> List[NewsItem]:
        """Тянем RSS с обработкой 429 и 503 ошибок"""
        items = []
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            for feed_url in feeds:
                try:
                    resp = await client.get(feed_url)
                    resp.raise_for_status()
                    # ... парсинг XML ...
                    items.append(NewsItem(...))
                except httpx.HTTPStatusError as e:
                    logger.warning("RSS fetch failed", url=feed_url, error=str(e))
                    await asyncio.sleep(5)  # exponential backoff тут
        return items
    
    # Аналогичные методы для Reddit, Twitter API v3, NewsAPI

Совет: не смешивайте источники разной достоверности без весов. Запись с Reddit r/wallstreetbets и новость с Bloomberg должны иметь разный confidence. Иначе аналитик запутается.

3 Агент-аналитик: профессиональный скептик

Это самый сложный агент. Его промпт - это не "проанализируй", а конкретная инструкция по поиску противоречий. Используем DeepSeek-V3, потому что у него лучшая работа с длинными контекстами на март 2026.

from deepseek_api import DeepSeekClient
import json

class AnalystAgent:
    def __init__(self, api_key: str):
        self.client = DeepSeekClient(api_key=api_key, model="deepseek-chat-v3-128k")
        self.system_prompt = """Ты - старший аналитик разведки. Твоя задача:
1. Найди противоречия между разными источниками.
2. Оцени надежность каждого источника (1-10).
3. Выдели факты, отдели их от мнений и спекуляций.
4. Верни JSON строго по схеме: {"conflicts": [...], "reliability_scores": {...}, "key_facts": [...]}

Никаких рассуждений, только структурированный вывод."""
    
    async def analyze_batch(self, news_items: List[NewsItem]) -> dict:
        context = self._prepare_context(news_items)
        response = await self.client.chat.completions.create(
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": context}
            ],
            response_format={"type": "json_object"},
            temperature=0.1  # минимум креативности
        )
        return json.loads(response.choices[0].message.content)

4 Агент-прогнозист и координатор

Прогнозист получает очищенные данные от аналитика и строит сценарии. Координатор - это просто роутер с логикой голосования. Он обращается к памяти, чтобы найти похожие исторические ситуации.

Вот как выглядит вызов памяти:

class Coordinator:
    def __init__(self, memory_client):
        self.memory = memory_client  # Redis или Temple Vault
        
    async def make_decision(self, analyst_report, forecast_scenarios):
        # Ищем в памяти похожие ситуации
        similar_cases = await self.memory.search_similar(
            query=analyst_report['key_facts'],
            limit=5
        )
        
        # Если в прошлом мы уже ошибались в похожем случае, снижаем уверенность
        confidence_penalty = 0
        for case in similar_cases:
            if case['actual_outcome'] != case['predicted_outcome']:
                confidence_penalty += 0.15  # эвристика
                
        # Принимаем решение на основе голосования агентов
        final_confidence = forecast_scenarios['avg_confidence'] - confidence_penalty
        
        return {
            "prediction": forecast_scenarios['most_likely'],
            "confidence": max(0.1, min(0.95, final_confidence)),
            "historical_similarity": len(similar_cases) > 0
        }

Координатор не использует LLM. Это детерминированная логика, потому что нам нужна воспроизводимость. Если арбитр тоже нейросеть, отлаживать систему становится адом.

5 Интеграция памяти с обратной связью

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

async def update_memory_with_actual_outcome(
    prediction_id: str, 
    actual_outcome: str, 
    metadata: dict
):
    """Этот джоб запускается по расписанию раз в день"""
    prediction = await memory.get_prediction(prediction_id)
    
    if prediction['outcome'] == actual_outcome:
        score = 1.0
    else:
        score = -0.5  # штраф за ошибку
    
    await memory.store_feedback(
        prediction_id=prediction_id,
        actual=actual_outcome,
        score=score,
        context=metadata
    )
    
    # Пересчитываем веса агентов, если они часто ошибаются
    await adjust_agent_weights(prediction['agents_involved'], score)

Именно эта петля обратной связи превращает набор промптов в самообучающуюся систему. Подробнее о проектировании такой памяти читайте в "Agent-memory-state".

Где вы наверняка накосячите

Что делать дальше, когда основа работает

Базовая Seldon Vault предсказывает отдельные события. Но настоящая сила раскрывается, когда вы добавляете:

1. Временные цепочки: прогнозист учится видеть не просто "событие A", а "если A, то через 3-7 дней вероятно B". Для этого нужна память с временными метками и графовая база.
2. Доверительные веса агентов: если прогнозист три раза подряд ошибся по технологическим темам, но точен в политике, координатор должен это учитывать. Реализуется через простую Bayesian оценку в памяти.
3. Human-in-the-loop: выводите прогнозы с низкой уверенностью (confidence < 0.4) на проверку человеку. Ответ человека становится золотым стандартом для обучения.

Если хотите пойти дальше, изучите архитектуру универсального AI-агента - она отлично сочетается с Seldon Vault.

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