予測ツールの実装

ツール設計の原則

AIエージェントが使うツールは、以下の原則に従って実装する。

Tool 1: 需要予測ツール

日次の売上データを受け取り、指定期間の予測値を返すツールを実装する。

from langchain_core.tools import tool
from pydantic import BaseModel, Field
import pandas as pd
import numpy as np
from prophet import Prophet
import lightgbm as lgb

class ForecastInput(BaseModel):
    store_nbr: int = Field(description="店舗番号")
    family: str = Field(description="商品カテゴリ")
    horizon_days: int = Field(default=15, description="予測日数")
    method: str = Field(default="ensemble", description="予測手法: prophet, lgbm, ensemble")

class ForecastResult(BaseModel):
    store_nbr: int
    family: str
    method: str
    predictions: list[dict]  # [{"date": "2017-08-01", "predicted": 1234.5}, ...]
    metrics: dict  # {"mae": 100.0, "mape": 5.2}

@tool
def forecast_demand(input: ForecastInput) -> ForecastResult:
    """指定した店舗・カテゴリの需要を予測する。
    Prophetモデル、LightGBMモデル、またはアンサンブルモデルから選択可能。
    """
    # データ読み込み
    df = load_store_data(input.store_nbr, input.family)
    train_df = df[df['date'] <= '2017-07-31']

    if input.method == "prophet":
        predictions = _predict_prophet(train_df, input.horizon_days)
    elif input.method == "lgbm":
        predictions = _predict_lgbm(train_df, input.horizon_days)
    else:
        pred_prophet = _predict_prophet(train_df, input.horizon_days)
        pred_lgbm = _predict_lgbm(train_df, input.horizon_days)
        predictions = _ensemble(pred_prophet, pred_lgbm, weights=[0.4, 0.6])

    return ForecastResult(
        store_nbr=input.store_nbr,
        family=input.family,
        method=input.method,
        predictions=predictions,
        metrics=_calculate_metrics(predictions, df)
    )

Prophet予測の内部実装

def _predict_prophet(train_df: pd.DataFrame, horizon: int) -> list[dict]:
    """Prophet による需要予測"""
    prophet_df = train_df[['date', 'sales']].rename(
        columns={'date': 'ds', 'sales': 'y'}
    )

    model = Prophet(
        yearly_seasonality=True,
        weekly_seasonality=True,
        changepoint_prior_scale=0.05
    )
    model.fit(prophet_df)

    future = model.make_future_dataframe(periods=horizon)
    forecast = model.predict(future)

    result = forecast.tail(horizon)[['ds', 'yhat']].copy()
    result['yhat'] = result['yhat'].clip(lower=0)

    return [
        {"date": row['ds'].strftime('%Y-%m-%d'), "predicted": round(row['yhat'], 2)}
        for _, row in result.iterrows()
    ]

LightGBM予測の内部実装

def _predict_lgbm(train_df: pd.DataFrame, horizon: int) -> list[dict]:
    """LightGBM による需要予測"""
    df = _create_features(train_df)

    feature_cols = [
        'dayofweek', 'month', 'day', 'year',
        'lag_7', 'lag_14', 'lag_28',
        'rolling_mean_7', 'rolling_std_7', 'rolling_mean_28',
        'is_payday', 'is_holiday'
    ]

    X_train = df.dropna()[feature_cols]
    y_train = df.dropna()['sales']

    model = lgb.LGBMRegressor(
        n_estimators=500,
        learning_rate=0.05,
        num_leaves=31,
        min_child_samples=20
    )
    model.fit(X_train, y_train)

    # 再帰的に1日ずつ予測
    predictions = []
    current_data = train_df.copy()

    for i in range(horizon):
        next_date = current_data['date'].max() + pd.Timedelta(days=1)
        next_features = _create_single_features(current_data, next_date)
        pred = max(0, model.predict(next_features[feature_cols])[0])
        predictions.append({
            "date": next_date.strftime('%Y-%m-%d'),
            "predicted": round(pred, 2)
        })
        # 予測値を次のステップのラグ計算に使用
        new_row = pd.DataFrame({'date': [next_date], 'sales': [pred]})
        current_data = pd.concat([current_data, new_row], ignore_index=True)

    return predictions

Tool 2: 発注量計算ツール

予測値と現在の在庫量から、最適な発注量を算出するツールを実装する。

class OrderInput(BaseModel):
    store_nbr: int = Field(description="店舗番号")
    family: str = Field(description="商品カテゴリ")
    current_stock: float = Field(description="現在在庫量")
    lead_time_days: int = Field(default=3, description="リードタイム（日）")
    service_level: float = Field(default=0.95, description="目標サービスレベル")
    predictions: list[dict] = Field(description="予測結果リスト")

class OrderResult(BaseModel):
    recommended_quantity: float
    safety_stock: float
    reorder_point: float
    expected_stockout_risk: float
    reasoning: str

@tool
def calculate_order(input: OrderInput) -> OrderResult:
    """予測値に基づき最適な発注量を計算する。
    安全在庫、発注点、推奨発注量を算出し、欠品リスクも評価する。
    """
    from scipy import stats

    pred_values = [p['predicted'] for p in input.predictions]

    # リードタイム期間の需要予測
    lt_demand = sum(pred_values[:input.lead_time_days])

    # 予測の標準偏差（過去データから推定）
    demand_std = np.std(pred_values) * np.sqrt(input.lead_time_days)

    # 安全在庫 = z値 × 標準偏差
    z_score = stats.norm.ppf(input.service_level)
    safety_stock = z_score * demand_std

    # 発注点 = リードタイム期間の予測需要 + 安全在庫
    reorder_point = lt_demand + safety_stock

    # 推奨発注量
    if input.current_stock <= reorder_point:
        # 予測期間全体の需要 + 安全在庫 - 現在在庫
        total_demand = sum(pred_values)
        recommended = max(0, total_demand + safety_stock - input.current_stock)
    else:
        recommended = 0.0

    # 欠品リスク
    stockout_risk = 1 - stats.norm.cdf(
        (input.current_stock - lt_demand) / max(demand_std, 1e-6)
    )

    reasoning = (
        f"リードタイム{input.lead_time_days}日間の予測需要: {lt_demand:.0f}、"
        f"安全在庫: {safety_stock:.0f}、"
        f"現在在庫: {input.current_stock:.0f}、"
        f"欠品リスク: {stockout_risk:.1%}"
    )

    return OrderResult(
        recommended_quantity=round(recommended, 0),
        safety_stock=round(safety_stock, 0),
        reorder_point=round(reorder_point, 0),
        expected_stockout_risk=round(stockout_risk, 4),
        reasoning=reasoning
    )

Tool 3: 異常検知ツール

予測値と実績値を比較し、異常な乖離を検出するツールを実装する。

class AnomalyInput(BaseModel):
    actual_values: list[dict] = Field(description="実績値リスト")
    predicted_values: list[dict] = Field(description="予測値リスト")
    threshold_sigma: float = Field(default=2.0, description="異常判定の閾値（σ）")

class AnomalyResult(BaseModel):
    anomalies: list[dict]  # [{"date": ..., "actual": ..., "predicted": ..., "deviation": ...}]
    anomaly_count: int
    alert_level: str  # "NORMAL", "WARNING", "CRITICAL"
    summary: str

@tool
def detect_anomalies(input: AnomalyInput) -> AnomalyResult:
    """予測値と実績値の乖離から異常を検出する。
    指定したσ閾値を超える乖離があった場合にアラートを生成する。
    """
    deviations = []
    for actual, predicted in zip(input.actual_values, input.predicted_values):
        deviation = actual['value'] - predicted['predicted']
        deviations.append({
            'date': actual['date'],
            'actual': actual['value'],
            'predicted': predicted['predicted'],
            'deviation': deviation
        })

    dev_values = [d['deviation'] for d in deviations]
    mean_dev = np.mean(dev_values)
    std_dev = np.std(dev_values) if len(dev_values) > 1 else 1.0

    anomalies = []
    for d in deviations:
        z = abs(d['deviation'] - mean_dev) / max(std_dev, 1e-6)
        if z > input.threshold_sigma:
            d['z_score'] = round(z, 2)
            anomalies.append(d)

    # アラートレベル判定
    if len(anomalies) == 0:
        alert_level = "NORMAL"
    elif len(anomalies) <= 2:
        alert_level = "WARNING"
    else:
        alert_level = "CRITICAL"

    summary = (
        f"検査期間: {len(deviations)}日間、"
        f"異常検出: {len(anomalies)}件、"
        f"平均乖離: {mean_dev:.1f}、"
        f"乖離標準偏差: {std_dev:.1f}"
    )

    return AnomalyResult(
        anomalies=anomalies,
        anomaly_count=len(anomalies),
        alert_level=alert_level,
        summary=summary
    )

ツールのテスト方法

各ツールは単体でテスト可能に設計されている。

# 予測ツールのテスト
result = forecast_demand.invoke({
    "store_nbr": 1,
    "family": "GROCERY I",
    "horizon_days": 15,
    "method": "ensemble"
})
print(f"予測件数: {len(result.predictions)}")
print(f"精度: MAE={result.metrics['mae']:.2f}")

# 発注量計算ツールのテスト
order = calculate_order.invoke({
    "store_nbr": 1,
    "family": "GROCERY I",
    "current_stock": 5000,
    "lead_time_days": 3,
    "service_level": 0.95,
    "predictions": result.predictions
})
print(f"推奨発注量: {order.recommended_quantity}")
print(f"安全在庫: {order.safety_stock}")
print(f"欠品リスク: {order.expected_stockout_risk:.1%}")

まとめ

チェックリスト

• 需要予測ツールのProphet/LightGBM実装を理解した
• 発注量計算の安全在庫・発注点の算出ロジックを理解した
• 異常検知ツールのσ閾値による判定ロジックを理解した
• 各ツールを単体でテストする方法を把握した

次のステップへ

個別のツールが完成したところで、次はこれらをLangGraphワークフローとして連携させ、日次の自動予測パイプラインを構築しよう。

推定読了時間: 30分