MCP Zero Shot Agentic Forecaster

Обзор для руководства и бизнес-ценность

Что представляет собой этот репозиторий?

MCP Zero Shot Agentic Forecaster — это производственный микросервисный движок прогнозирования временных рядов, доступный через Model Context Protocol (MCP). Работая на основе передовых фундаментальных моделей (Google TimesFM 2.5 XReg и Amazon Chronos-2), он позволяет автономным ИИ-агентам (конечным автоматам LangGraph, роям CrewAI, нейросимволическим стекам под управлением OPA/Rego и стандартным циклам ReAct) запрашивать вероятностные прогнозы спроса по требованию — без офлайн-обучения моделей, подбора гиперпараметров или подготовки наборов данных для каждого SKU.

Бизнес-ценность и ROI

• Устраняет задержку холодного старта: обеспечивает мгновенные zero-shot вероятностные прогнозы для запуска новых продуктов, акций и SKU с короткой историей без конвейеров обучения.

• Контроль рисков с ограничением по квантилям: формирует калиброванные квантили спроса $p_{10}$, $p_{50}$ и $p_{90}$, позволяя автономным агентам закупок балансировать страховые запасы и затраты на хранение капитала.

• Снижение совокупной стоимости владения (TCO): заменяет сложные конвейеры дообучения единым механизмом резервирования из 3 уровней, значительно снижая требования к GPU-вычислениям и дрейф инфраструктуры.

• Устойчивость для агентов: возвращает структурированные полезные нагрузки AgentFriendlyError с подсказками по исправлению при недопустимых входных данных, позволяя вызывающим агентам самостоятельно исправляться в циклах выполнения без тихого сбоя или необработанных исключений.

Как это работает

1. Вызов агента: вызывающие агенты обращаются к forecast_demand или forecast_batch через stdio/HTTP посредством интерфейса MCP-инструментов.

2. Проверка контракта: схемы Pydantic v2 выполняют строгие проверки конечности чисел и временных границ ([Type-Safe Input Contract]).

3. Неблокирующий инференс: FastMCP выгружает тяжелые тензорные операции в пулы потоков через asyncio.to_thread, сохраняя отзывчивость шлюза.

4. Единый конвейер из 3 уровней: всегда направляет запросы через TimesFM 2.5 (уровень 1), переключаясь на Chronos-2 (уровень 2, сохраняет ковариаты при их наличии) и ARIMA111 (уровень 3, отбрасывает ковариаты). При CUDA OOM или сбое движок запускает восстановление памяти (gc.collect() + torch.cuda.empty_cache()) перед переходом на следующий уровень.

5. Математическая санация: применяет изотоническую сортировку для гарантии монотонности выходных квантилей ($p_{10} \le p_{50} \le p_{90}$) и нормализует эпистемические оценки уверенности перед возвратом структурированных JSON-полезных нагрузок.

Related MCP server: Geneva Forecasting MCP

Системная архитектура

Микросервис придерживается строгого разделения ответственности: уровень MCP-инструментов управляет неблокирующим транспортом, безопасностью памяти и санацией выходных данных на уровне моделей, оставляя доменные бизнес-политики нижестоящим оркестраторам агентов.

graph TD
    subgraph External Agent Orchestrator
        Agent[LLM Agent / Swarm / State Machine<br/>LangGraph / OPA Sidecar / ReAct Loop]
    end

    subgraph MCP Microservice Boundary
        Gateway[FastMCP Async Gateway Server<br/>mcp_server.py]
        Sanitizer[Pydantic v2 Input Contract<br/>TimeSeriesInputPayload]
        ErrorFormatter[AgentFriendlyError Formatter]
        
        subgraph Engine Memory & Concurrency Boundary
            ExecThread[Thread Executor<br/>asyncio.to_thread]
            Engine[ZeroShotForecastingEngine<br/>src/models/forecaster.py<br/>Lazy-Load Lock Protected]
            
            subgraph 3-Tier Fallback Model Chain
                T1[Tier 1: TimesFM 2.5<br/>XReg / Univariate]
                T2[Tier 2: Chronos-2<br/>Multivariate / Univariate]
                T3[Tier 3: ARIMA111<br/>CPU Baseline Fallback]
            end
            
            IsoSanitizer[Isotonic Quantile Sanitizer<br/>Enforces p10 ≤ p50 ≤ p90]
        end
    end

    Agent -->|FastMCP Tool Call<br/>forecast_demand / forecast_batch| Gateway
    Gateway -->|1. Validate Schema| Sanitizer
    Sanitizer -->|Validation Error| ErrorFormatter
    ErrorFormatter -.->|Structured Error + Remediation| Agent
    
    Sanitizer -->|2. Valid Payload| ExecThread
    ExecThread -->|3. Route Request| Engine
    Engine --> T1
    T1 -.->|CUDA OOM / Fail| T2
    T2 -.->|Fail| T3
    
    T1 -->|Raw Quantiles| IsoSanitizer
    T2 -->|Raw Quantiles| IsoSanitizer
    T3 -->|Raw Quantiles| IsoSanitizer
    
    IsoSanitizer -->|4. Validated ForecastResponse| Gateway
    Gateway -->|5. Return JSON Payload| Agent

Разбивка компонентов архитектуры

Шлюз FastMCP (mcp_server.py): обеспечивает асинхронный JSON-RPC транспорт и ограничивает пакетную конкурентность (asyncio.Semaphore(4)).

Типобезопасная граница контракта (src/schemas/payloads.py): обеспечивает временное выравнивание, гарантии конечности чисел и ограничения контекста/горизонта.

Потокобезопасное ядро прогнозирования (src/models/forecaster.py): использует блокировку с двойной проверкой (threading.Lock()) для ленивой загрузки моделей и обрабатывает автоматическое восстановление после CUDA OOM (gc.collect() + torch.cuda.empty_cache()).

Изотонический санитайзер вывода: постобрабатывает сырые квантили фундаментальных моделей с помощью монотонной сортировки для устранения статистических аномалий ($p_{10} > p_{50}$) перед возвратом прогнозов агентам.

Поток выполнения системы (диаграмма последовательности)

sequenceDiagram
    autonumber
    actor Agent as LLM Agent / Orchestrator
    participant Gateway as FastMCP Gateway (Async)
    participant Sanitizer as Pydantic Input Contract
    participant Executor as Thread Executor (asyncio.to_thread)
    participant Pipeline as 3-Tier Fallback Pipeline
    participant Std as Isotonic Quantile Sanitizer

    Agent->>Gateway: forecast_demand / forecast_batch (JSON)
    Gateway->>Sanitizer: Validate TimeSeriesInputPayload
    alt Validation Failure
        Sanitizer-->>Gateway: AgentFriendlyError {error_code, expected, received, remediation}
        Gateway-->>Agent: Structured Error Response
    else Validation Success
        Sanitizer->>Executor: Offload sync inference
        Executor->>Pipeline: Execute prediction
        Note right of Pipeline: Tier 1: TimesFM 2.5 → Tier 2: Chronos-2 → Tier 3: ARIMA111
        alt CUDA OOM / Transient Failure
            Pipeline->>Pipeline: gc.collect() + torch.cuda.empty_cache()
            Pipeline->>Pipeline: Degrade to next tier (retain covariates where possible)
        end
        Pipeline->>Std: Apply _enforce_quantile_monotonicity()
        Std-->>Executor: ForecastResponse {model_used, exogenous_dropped, warnings}
        Executor-->>Gateway: Return validated response
        Gateway-->>Agent: 200 OK with ForecastResponse
    end

Основные архитектурные принципы

Неблокирующий асинхронный транспорт

Все прямые тензорные проходы выполняются в пуле потоков через asyncio.to_thread, сохраняя цикл событий FastMCP отзывчивым к параллельным проверкам работоспособности и вызовам инструментов под нагрузкой.

# mcp_server.py
result = await asyncio.to_thread(engine.predict, validated_payload)

Ленивая загрузка в VRAM

Веса моделей материализуются только при первом использовании через методы доступа — VRAM не расходуется при запуске. Потокобезопасная блокировка с двойной проверкой предотвращает дублирующую инстанциацию при параллельных холодных стартах.

# src/models/forecaster.py
def _get_timesfm(self):
    if self._timesfm is None:
        with self._timesfm_lock:
            if self._timesfm is None:
                import timesfm
                logger.info(f"Lazily loading TimesFM-2.5 ({self.timesfm_repo_id}) onto {self._device}")
                self._timesfm = timesfm.TimesFM_2p5_200M_torch.from_pretrained(self.timesfm_repo_id)
    return self._timesfm

def _get_chronos(self):
    if self._chronos is None:
        with self._chronos_lock:
            if self._chronos is None:
                from chronos import BaseChronosPipeline
                logger.info(f"Lazily loading Chronos-2 ({self.chronos_repo_id}) onto {self._device}")
                self._chronos = BaseChronosPipeline.from_pretrained(
                    self.chronos_repo_id, device_map=self._device, dtype=torch.float32
                )
    return self._chronos

Механизм резервирования из 3 уровней

Движок поддерживает единую детерминированную цепочку деградации независимо от содержимого полезной нагрузки.

При сбое TimesFM (включая torch.cuda.OutOfMemoryError):

1. gc.collect() + torch.cuda.empty_cache()

2. Экзогенные сигналы сохраняются для Chronos-2 через _build_chronos_covariates() (прошлые/будущие ковариаты)

3. exogenous_dropped = true только если резервный уровень деградирует до уровня 3 (ARIMA111)

4. Выполнение направляется в Chronos-2

5. Если Chronos завершается сбоем → базовый ARIMA111

Обеспечение временной регулярности

Валидаторы Pydantic v2 отклоняют недопустимую телеметрию на границе:

Изотоническая санация квантилей

Все бэкенды (TimesFM, Chronos-2, AutoARIMA) выдают сырые квантили, которые иногда могут пересекаться ($p_{10} > p_{50}$ или $p_{50} > p_{90}$) при экстремальных OOD-входных данных. Движок применяет легкий этап постобработки _enforce_quantile_monotonicity(), который выполняет изотоническую сортировку для каждого временного шага — объединяя $(p_{10}, p_{50}, p_{90})$, сортируя по оси квантилей и возвращая упорядоченные тройки. Это гарантирует математически корректные $p_{10} \le p_{50} \le p_{90}$ для каждого шага горизонта прогнозирования без искажения формы распределения.

Ограниченная пакетная конкурентность

MCP-инструмент forecast_batch выполняет многоклассовый (multi-SKU) инференс параллельно с помощью asyncio.gather, ограниченного asyncio.Semaphore(4). Это обеспечивает параллельную пропускную способность, защищая память GPU/CPU от неограниченных параллельных выделений тензоров. Каждый элемент получает семафор, проверяет свою полезную нагрузку, выгружает engine.predict в пул потоков через asyncio.to_thread и возвращает структурированный ForecastResponse с метаданными резервирования для каждого элемента. Сводный блок сообщает общее количество элементов, количество ошибок и использование по бэкендам (model_usage).

Автоматические выключатели для каждого бэкенда

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

Это гарантирует, что временный сбой HuggingFace Hub или поврежденная загрузка весов не заблокируют агента на неопределенный срок.

Нормализованная метрика уверенности

Оценка уверенности использует ограниченное относительное отношение неопределенности вместо линейного минимума, который сжимает широкую дисперсию до 0.0:

$$\text{Confidence} = \frac{1}{1 + \frac{p_{90} - p_{10}}{\vert p_{50}\vert + \epsilon}}$$

где $\epsilon = 10^{-5}$. Свойства:

• Диапазон вывода $(0, 1]$ — никогда не отрицательный, никогда не сжимается до 0

• При стремлении разброса $(p_{90} - p_{10}) \to 0$ уверенность $\to 1$ (узкие границы)

• При стремлении разброса $\to \infty$ уверенность асимптотически $\to 0$ (экстремальная неопределенность)

• Масштабно-инвариантна благодаря делению на медианную величину $|p_{50}|$

Агностичность к архитектуре агента

Будучи микросервисом MCP-инструментов без состояния и со строгой схемой, этот движок бесшовно интегрируется с любым оркестратором агентов — включая нейросимволические стеки под управлением политик OPA/Rego, конечные автоматы LangGraph, рои CrewAI или стандартные циклы ReAct.

Спецификации MCP-инструментов и API-контракты

forecast_demand

Прогноз одного временного ряда.

Запрос (TimeSeriesInputPayload)

{
  "target_series": [120.5, 115.0, 130.2, 125.8, 140.1],
  "forecast_horizon": 30,
  "price_index": [19.99, 19.99, 24.99, 24.99, 24.99, 24.99, ...],
  "promo_flag": [0, 0, 1, 0, 1, 0, ...]
}

Ответ (ForecastResponse)

{
  "model_used": "Chronos-2-Fallback",
  "mean_prediction": [142.3, 145.1, 140.8, 148.2, 150.0],
  "p10_quantile": [120.1, 122.4, 118.7, 125.3, 127.9],
  "p50_quantile": [142.3, 145.1, 140.8, 148.2, 150.0],
  "p90_quantile": [165.2, 168.5, 162.1, 170.4, 172.8],
  "confidence_score": 0.87,
  "horizon_length": 30,
  "exogenous_dropped": false,
  "warnings": ["CUDA unavailable; running on CPU. Expect degraded inference performance."]
}

forecast_batch

Пакетный прогноз для нескольких SKU на основе массива со сводкой резервирования по каждому элементу.

Запрос

{
  "payloads": [
    {"target_series": [10.0]*20, "forecast_horizon": 5},
    {"target_series": [11.0]*30, "forecast_horizon": 3, "price_index": [20.0]*33}
  ]
}

Ответ

{
  "results": [
    {"model_used": "Chronos-2-Fallback", "mean_prediction": [...], ...},
    {"model_used": "TimesFM-2.5", "mean_prediction": [...], ...}
  ],
  "summary": {
    "total": 2,
    "errors": 0,
    "model_usage": {"Chronos-2-Fallback": 1, "TimesFM-2.5": 1}
  }
}

AgentFriendlyError

Схема ошибки для самокоррекции, возвращаемая при сбое проверки или выполнения.

{
  "error_code": "VALIDATION_ERROR",
  "message": "Input validation failed at 'price_index': Price array misalignment. Expected 25 elements (Context: 20 + Horizon: 5), got 2.",
  "expected": "Payload matching TimeSeriesInputPayload schema (context 16-16000 finite values, aligned exogenous signals).",
  "received": "{\"location\": \"price_index\", \"message\": \"Price array misalignment. Expected 25 elements (Context: 20 + Horizon: 5), got 2.\", \"context\": {\"expected\": \"25\", \"got\": \"2\"}}",
  "remediation_suggestion": "Correct field 'price_index' (Price array misalignment. Expected 25 elements (Context: 20 + Horizon: 5), got 2.) and resubmit. Ensure context length is between 16 and 16000, values are finite (no NaN/Inf), and exogenous arrays align to len(target_series) + forecast_horizon."
}

Коды ошибок: VALIDATION_ERROR, MODEL_UNAVAILABLE, TRANSIENT_FAILURE, MODEL_FAILURE, INTERNAL_ERROR

Быстрый старт и настройка MCP

Установка

# Requires Python 3.11+
uv sync --extra gpu   # or: pip install -r requirements.txt

Переменные окружения

# Optional: force CPU if GPU memory constrained
export MODEL_CONFIG_PATH=configs/model_config.yaml
export DATA_STORAGE_ROOT=data/

Движок автоматически определяет CUDA. Если она недоступна, переключается на CPU и выдает предупреждение в поле warnings.

Конфигурация MCP-клиента (Claude Desktop / OpenCode / LangGraph / CrewAI)

Добавьте в конфигурацию вашего MCP-клиента (claude_desktop_config.json, opencode.json или аналогичный файл):

{
  "mcpServers": {
    "zero-shot-forecaster": {
      "command": "python",
      "args": ["mcp_server.py"],
      "cwd": "/absolute/path/to/zero-shot-demand-foundation",
      "env": {
        "MODEL_CONFIG_PATH": "configs/model_config.yaml"
      }
    }
  }
}

Перезапустите ваш MCP-клиент. Инструменты forecast_demand и forecast_batch автоматически зарегистрируются со своими полными JSON-схемами.

Пример вызова (Claude / LLM-агент)

{
  "tool": "forecast_demand",
  "arguments": {
    "target_series": [120, 115, 130, 125, 140, 135, 150, 145, 155, 160, 155, 165, 170, 168, 172, 175, 180, 178, 185, 190],
    "forecast_horizon": 7,
    "price_index": [19.99, 19.99, 19.99, 19.99, 19.99, 19.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99, 24.99],
    "promo_flag": [0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0]
  }
}

Проверка и тестирование

Запуск полного набора тестов (62 теста, AC-1–AC-5)

uv run pytest tests/ -q
# or
python -m pytest tests/ -q

Ожидаемый вывод:

.............................................................. [100%]
62 passed in ~3s

Матрица покрытия тестами

Легаси-тесты (обратная совместимость)

Все 48 исходных тестов по-прежнему проходят:

pytest tests/test_forecasting_engine.py tests/test_forecaster_router.py tests/test_mcp_server.py tests/test_schemas.py tests/test_metrics.py -q

Структура проекта (после рефакторинга)

zero-shot-demand-foundation/
├── configs/
│   └── model_config.yaml            # Model IDs, device_map, num_samples
├── data/                            # Git-ignored (CSV, ZIP)
├── scripts/
│   ├── download_m5.py               # M5 dataset fetcher
│   └── download_favorita.py         # Favorita dataset fetcher
├── src/
│   ├── models/
│   │   └── forecaster.py            # ZeroShotForecastingEngine (refactored)
│   ├── schemas/
│   │   └── payloads.py              # TimeSeriesInputPayload, ForecastResponse, AgentFriendlyError
│   └── utils/
│       ├── data_loader.py           # DemandDataEngine, FavoritaDataLoader
│       └── metrics.py               # WAPE, RMSSE, Pinball Loss, CRPS
├── tests/
│   ├── test_forecasting_engine.py   # Updated for lazy loading
│   ├── test_forecaster_router.py    # Updated fixtures
│   ├── test_mcp_server.py           # Async + AgentFriendlyError
│   ├── test_metrics.py              # Unchanged
│   ├── test_schemas.py              # Unchanged
│   └── test_refactored_forecaster.py # NEW: AC-1..5 coverage
├── main.py                          # CLI evaluation entry point
├── mcp_server.py                    # FastMCP server (async, batch, errors)
├── requirements.txt
├── .gitignore                       # Ignores *.md, data/, __pycache__/
└── README.md                        # This file

Лицензия

Лицензия MIT. Подробности см. в LICENSE.

Ссылки

• Chronos-2: Ansari et al., Chronos: Learning the Language of Time Series, arXiv:2403.07815

• TimesFM: Das et al., TimesFM: A Decoder-Only Foundation Model for Time-Series Forecasting, arXiv:2402.02592

• M5 Competition: Makridakis et al., M5 Accuracy Competition, IJF 2022

• Corporación Favorita: Kaggle Favorita Grocery Sales Forecasting

• Model Context Protocol: Anthropic MCP Specification