MCP Zero Shot Agentic Forecaster 是一个生产级、基于微服务的时间序列预测引擎，通过 Model Context Protocol (MCP) 对外暴露。它由最先进的基础模型（Google TimesFM 2.5 XReg 和 Amazon Chronos-2）驱动，允许自主 AI 智能体（LangGraph 状态机、CrewAI 集群、由 OPA/Rego 治理的 NeSy 栈，以及标准 ReAct 循环）按需查询概率性需求预测——无需离线模型训练、超参数调优或按 SKU 准备数据集。
消除冷启动延迟：为新产品发布、促销活动和短历史 SKU 提供即时零样本概率预测，无需训练流水线。
分位数约束的风险控制：输出经过校准的 $p_{10}$、$p_{50}$ 和 $p_{90}$ 需求分位数，使自主采购智能体能够在安全库存缓冲与资金持有成本之间取得平衡。
降低总拥有成本（TCO）：用统一的三层回退引擎取代复杂的微调流水线，大幅降低 GPU 计算需求和基础设施漂移。
智能体弹性：当输入无效时，返回带有修复提示的结构化 AgentFriendlyError 负载，使调用智能体能够在执行循环中自我纠正，而不会静默失败或抛出未处理的异常。
智能体调用：调用智能体通过 MCP 工具接口，基于 stdio/HTTP 调用 forecast_demand 或 forecast_batch。
契约强制：Pydantic v2 模式执行严格的有限数值和时间边界检查（[Type-Safe Input Contract]）。
非阻塞推理：FastMCP 通过 asyncio.to_thread 将繁重的张量操作卸载到线程池，以保持网关的响应性。
统一三层流水线：始终通过 TimesFM 2.5（第 1 层）路由，回退到 Chronos-2（第 2 层，若存在协变量则保留）和 ARIMA111（第 3 层，丢弃协变量）。在 CUDA OOM 或失败时，引擎在降级到下一层之前触发内存回收（gc.collect() + torch.cuda.empty_cache()）。
数学净化：应用等渗排序来保证输出分位数的单调性（$p_{10} \le p_{50} \le p_{90}$），并在返回结构化 JSON 负载之前对认知置信度分数进行归一化。
Related MCP server: forecast-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| AgentFastMCP 网关（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。线程安全的双重检查锁定可防止并发冷启动下的重复实例化。
# 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无论负载内容如何，该引擎都维护一条统一且确定性的降级链。
层级 | 后端 | 模式 |
|
|
1 | TimesFM 2.5 | XReg / 单变量 |
|
|
2 | Chronos-2 | 多变量 / 单变量 |
|
|
3 | ARIMA111 | 基线 |
|
|
当 TimesFM 失败时（包括 torch.cuda.OutOfMemoryError）：
gc.collect() + torch.cuda.empty_cache()
通过 _build_chronos_covariates()（过去/未来协变量）为 Chronos-2 保留外生信号
仅当回退降级到第 3 层（ARIMA111）时，exogenous_dropped = true
执行路由到 Chronos-2
如果 Chronos 失败 → ARIMA111 基线
Pydantic v2 校验器在边界处拒绝无效的遥测数据：
校验器 | 规则 |
上下文长度边界 |
|
预测范围边界 |
|
有限数值 |
|
外生变量对齐 |
|
二进制标志 |
|
所有后端（TimesFM、Chronos-2、AutoARIMA）在极端 OOD 输入下偶尔会输出交叉的原始分位数（$p_{10} > p_{50}$ 或 $p_{50} > p_{90}$）。引擎应用一个轻量级后处理步骤 _enforce_quantile_monotonicity()，该步骤对每个时间步执行等渗排序——堆叠 $(p_{10}, p_{50}, p_{90})$，沿分位数轴排序，并返回有序三元组。这保证了每个预测范围步长在数学上有效的 $p_{10} \le p_{50} \le p_{90}$，而不会扭曲分布形态。
forecast_batch MCP 工具使用 asyncio.gather 并发执行多 SKU 推理，并通过 asyncio.Semaphore(4) 限制并发数。这提供了并行吞吐量，同时保护 GPU/CPU 内存免受无界并发张量分配的影响。每个条目获取信号量、验证其负载、通过 asyncio.to_thread 将 engine.predict 卸载到线程池，并返回带有逐条目回退元数据的结构化 ForecastResponse。摘要块报告总条目数、错误计数和按后端统计的使用情况（model_usage）。
每个基础模型后端都维护一个独立的熔断器（CircuitBreakerState），以防止在模型中心不可达或持续出错时发生级联故障。在连续 5 次失败后，熔断器打开，并立即将流量路由到下一个回退层，持续 60 秒，之后才允许一次测试调用。
后端 | 失败阈值 | 冷却时间 | 断开行为 |
TimesFM 2.5 | 5 次失败 | 60 秒 | 抛出 |
Chronos-2 | 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 策略治理的神经符号（Neuro-Symbolic）栈、LangGraph 状态机、CrewAI 集群或标准 ReAct 循环。
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
# 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_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 模式。
{
"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]
}
}uv run pytest tests/ -q
# or
python -m pytest tests/ -q预期输出：
.............................................................. [100%]
62 passed in ~3sAC | 标准 | 测试函数 |
AC-1 | 非阻塞事件循环 |
|
AC-2 | 懒加载与 VRAM 节省 |
|
AC-3 | 优雅回退与信号剥离 |
|
AC-4 | CUDA OOM 恢复 |
|
AC-5 | 智能体自纠错载荷 |
|
全部 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 -qzero-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 fileMIT 许可证。详见 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
This server cannot be deployed
Probabilistic time-series forecasts from zero-shot foundation models: routed, single or ensembled.
Forecast product demand using historical sales and market signals.
Market forecasts with their own track record, inside Claude, ChatGPT and other MCP apps.
PredictOracle - 12 forecasting tools: time-series, scenario analysis, risk projections.
