PydanticとDDD戦術パターンで設計するMLパイプライン実践ガイド

MLパイプラインの開発では「学習ジョブの状態管理が複雑」「ハイパーパラメータの変更が意図せず伝播する」「モデルレジストリとの結合が密すぎてテストできない」といった問題に直面します。DDD（ドメイン駆動設計）の戦術パターン——集約・値オブジェクト・リポジトリ——は、こうしたMLドメイン固有の複雑さを構造的に解決する手法です。

本記事では、Pydantic v2の型安全な機能を活用しながら、MLパイプラインにDDD戦術パターンを適用する方法を解説します。DDD戦術パターンの基本概念についてはTypeScript・Pythonで実装するDDD戦術的パターン実践ガイドもあわせてご参照ください。

この記事でわかること

• DDD戦術パターン（集約・値オブジェクト・リポジトリ）をMLドメインに適用する考え方
• Pydantic v2のfrozenモデル・RootModel・model_validatorを使った値オブジェクトの実装
• 学習ジョブを集約としてモデリングし、状態遷移の整合性を保証する設計
• リポジトリパターンでMLflow等のML基盤を抽象化し、テスト容易性を確保する手法
• 集約設計でよくある失敗パターンと、MLドメインでの回避策

対象読者

• 想定読者: MLパイプラインの設計・運用に携わるMLエンジニア
• 必要な前提知識:
  • Pythonの基礎文法（クラス、データクラス、型ヒント）
  • Pydanticの基本的な使い方（BaseModel、バリデーション）
  • ML学習パイプラインの基本概念（学習ジョブ、ハイパーパラメータ、モデルレジストリ）

DDD自体が初めての方向けに、各パターンの基本概念をMLの類推で解説します。@dataclass(frozen=True) や BaseModel に馴染みがあれば、値オブジェクトの概念はすぐに理解できるでしょう。

結論・成果

DDD戦術パターンをMLパイプラインに適用すると、以下の効果が期待できます。

• 値オブジェクトによりハイパーパラメータのバリデーションロジックを一箇所に集約し、パイプライン全体でのバリデーション重複を排除できる
• 集約で学習ジョブの状態遷移を保護し、「完了済みジョブが再実行される」「学習中にハイパーパラメータが変更される」といった不正操作をコンパイル時に近い形で防止できる
• リポジトリパターンでMLflow・W&B等のML基盤を抽象化し、インメモリ実装でのユニットテストが可能になる（外部依存なしでテスト実行時間を短縮）
• Vaughn Vernonの調査によると、集約の約70%はルートエンティティと値オブジェクトのみで構成でき、MLドメインでもこの傾向は一致する（Effective Aggregate Design Part I）

MLドメインをDDD戦術パターンでモデリングする

DDD戦術パターンは、ビジネスロジックをコードに正確に反映するための設計手法です。MLエンジニアにとっては「学習パイプラインのドメインルールを、型システムとオブジェクト設計で表現する」と捉えると理解しやすいでしょう。まず、3つのパターンがMLドメインでどう対応するかを見ていきます。

DDD戦術パターンの3要素とMLでの対応

DDD戦術パターン	概念	MLでの対応例
値オブジェクト	不変で、値が等しければ同一とみなすオブジェクト	Hyperparameters、Metrics、LearningRate、ModelVersion
エンティティ	一意な識別子を持ち、ライフサイクルを通じて追跡されるオブジェクト	TrainingRun、Experiment
集約	エンティティと値オブジェクトのまとまり。整合性の境界	TrainingJob（ルート）+ Hyperparameters + TrainingConfig
リポジトリ	集約の永続化・復元を担うインターフェース	ExperimentRepository、ModelRepository

MLエンジニアにとって馴染みのある表現で説明すると、値オブジェクトはPythonの@dataclass(frozen=True)のような不変データ構造です。ハイパーパラメータを変えたい場合は既存のオブジェクトを変更するのではなく、新しいオブジェクトを作ります。これはNumPyの配列操作で.copy()してから変更する感覚に似ています。

集約は「トランザクション境界」と考えてください。MLflowで学習を開始するとき、ハイパーパラメータの設定・ジョブの状態変更・メトリクスの初期化は「すべて成功するか、すべて失敗するか」であるべきです。この一貫性を保証する範囲が集約です。

MLドメインの集約マッピング

MLパイプラインを集約としてモデリングする際の具体的なマッピングを示します。

なぜこのように分割するか:

• TrainingJobとMLModelを別の集約にする理由は、学習完了とモデル登録は異なるタイミングで発生し、それぞれ独立した整合性要件を持つためです
• Hyperparametersを値オブジェクトにする理由は、学習率0.001とバッチサイズ32の組み合わせが「同じ設定」であれば、それらは区別する必要がないためです（識別子ではなく値で比較する）

注意: 集約の境界は「データベースのテーブル設計」ではなく「ビジネスルール（不変条件）の整合性境界」で決めます。「学習中にハイパーパラメータは変更できない」というルールがあるなら、HyperparametersはTrainingJob集約の内部に置くべきです。

Pydantic v2で値オブジェクトを実装する

値オブジェクトはDDD戦術パターンの基盤です。MLドメインでは、ハイパーパラメータ・メトリクス・モデルバージョン・リソース仕様など、多くの概念が値オブジェクトとしてモデリングできます。Pydantic v2のConfigDict(frozen=True)は、不変性・バリデーション・構造的等価性を一度に実現でき、値オブジェクトの実装に適しています。

単一フィールドの値オブジェクト: RootModel

学習率やエポック数のように、単一の値にドメインルールを付与したい場合はRootModelを使います。

from pydantic import RootModel, field_validator, ConfigDict


class LearningRate(RootModel[float]):
    """学習率を表す値オブジェクト。0より大きく1未満の値のみ許可する。"""
    model_config = ConfigDict(frozen=True)

    @field_validator("root")
    @classmethod
    def validate_range(cls, v: float) -> float:
        if not (0 < v < 1):
            raise ValueError(f"学習率は0より大きく1未満である必要があります: {v}")
        return v


class BatchSize(RootModel[int]):
    """バッチサイズを表す値オブジェクト。2の冪乗のみ許可する。"""
    model_config = ConfigDict(frozen=True)

    @field_validator("root")
    @classmethod
    def validate_power_of_two(cls, v: int) -> int:
        if v <= 0 or (v & (v - 1)) != 0:
            raise ValueError(f"バッチサイズは2の冪乗である必要があります: {v}")
        return v


class Epoch(RootModel[int]):
    """エポック数を表す値オブジェクト。1以上の正の整数のみ許可する。"""
    model_config = ConfigDict(frozen=True)

    @field_validator("root")
    @classmethod
    def validate_positive(cls, v: int) -> int:
        if v < 1:
            raise ValueError(f"エポック数は1以上である必要があります: {v}")
        return v


lr = LearningRate(0.001)
print(lr.root)       # 0.001
print(LearningRate(0.001) == LearningRate(0.001))  # True（構造的等価性）

# lr.root = 0.01  # ValidationError: frozen（不変性を保証）

RootModelを使うと、LearningRate(0.001)のように直接値を渡すだけで、バリデーションと不変性が自動的に適用されます。これにより「学習率に負の値を渡してしまった」「バッチサイズに奇数を指定してGPUメモリが非効率になった」といったバグを型レベルで防止できます。

複合値オブジェクト: Hyperparametersの実装

複数のフィールドを持つ値オブジェクトでは、BaseModelにfrozen=Trueを設定し、model_validatorでフィールド間の制約を表現します。

from pydantic import BaseModel, ConfigDict, model_validator
from typing import Literal


class Hyperparameters(BaseModel):
    """学習ハイパーパラメータを表す値オブジェクト。"""
    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)

    learning_rate: LearningRate
    batch_size: BatchSize
    epochs: Epoch
    optimizer: Literal["adam", "sgd", "adamw"]
    weight_decay: float = 0.0

    @model_validator(mode="after")
    def validate_optimizer_constraints(self) -> "Hyperparameters":
        """optimizer種別に応じた制約を検証する。"""
        if self.optimizer == "sgd" and self.learning_rate.root > 0.1:
            raise ValueError(
                f"SGDでは学習率0.1以下を推奨します（現在: {self.learning_rate.root}）"
            )
        if self.optimizer != "adamw" and self.weight_decay > 0:
            raise ValueError(
                f"weight_decayはAdamW以外では0にしてください（現在: {self.weight_decay}）"
            )
        return self


params = Hyperparameters(
    learning_rate=LearningRate(0.001),
    batch_size=BatchSize(32),
    epochs=Epoch(10),
    optimizer="adamw",
    weight_decay=0.01,
)

# 値を変更したい場合は新しいインスタンスを生成する（不変性の維持）
new_params = params.model_copy(update={"learning_rate": LearningRate(0.0005)})
print(new_params.learning_rate.root)  # 0.0005
print(params.learning_rate.root)      # 0.001（元は変わらない）

なぜmodel_copyを使うか:

model_copy(update={...})は、frozenモデルの値を「変更」する唯一の方法です。新しいインスタンスが返され、元のインスタンスは変更されません。MLの文脈では「ハイパーパラメータチューニングで学習率だけ変えた新しい設定を作る」操作に対応します。Optunaの試行ごとに新しいHyperparametersが生成されるイメージです。

設定情報の値オブジェクト: TrainingConfig

学習ジョブの設定情報も値オブジェクトとして定義します。後述の集約でこの型を使用します。

class TrainingConfig(BaseModel):
    """学習設定を表す値オブジェクト。"""
    model_config = ConfigDict(frozen=True, extra="forbid")

    framework: Literal["pytorch", "tensorflow", "jax"]
    device: Literal["cpu", "cuda", "tpu"]
    mixed_precision: bool = False
    gradient_accumulation_steps: int = 1

    @model_validator(mode="after")
    def validate_device_constraints(self) -> "TrainingConfig":
        if self.mixed_precision and self.device == "cpu":
            raise ValueError("混合精度学習はCPUでは利用できません")
        return self

メトリクスの値オブジェクト

学習中に記録されるメトリクスも値オブジェクトとしてモデリングします。

from pydantic import BaseModel, ConfigDict, computed_field
from datetime import datetime


class TrainingMetrics(BaseModel):
    """学習メトリクスを表す値オブジェクト。"""
    model_config = ConfigDict(frozen=True, extra="forbid")

    train_loss: float
    val_loss: float
    val_accuracy: float
    step: int
    recorded_at: datetime

    @computed_field
    @property
    def is_overfitting(self) -> bool:
        """train_lossとval_lossの乖離から過学習の兆候を判定する。"""
        if self.train_loss == 0:
            return False
        return (self.val_loss / self.train_loss) > 2.0

    @computed_field
    @property
    def generalization_gap(self) -> float:
        """汎化ギャップを計算する。"""
        return self.val_loss - self.train_loss

@computed_fieldは派生的な属性を定義します。メトリクスの生データから「過学習の兆候」や「汎化ギャップ」を導出する処理を、値オブジェクト内に閉じ込めることで、パイプラインの各所で同じ計算を重複させずに済みます。

注意点:

Pydantic v2のfrozen=Trueには一つ注意があります。内部にミュータブルなオブジェクト（例: listやdict）を持つ場合、フィールドへの再代入は防止されますが、内部のオブジェクト自体の変更は防げません。値オブジェクト内でリストを使う場合はtupleに変換するか、@model_validatorで防御的にコピーすることを検討してください（GitHub Issue #12361）。

集約で学習ジョブの整合性を保証する

集約はDDD戦術パターンの中核です。「複数のオブジェクトをまとめて、ビジネスルール（不変条件）の整合性を単一のトランザクション内で保証する」境界を定義します。MLドメインでは、学習ジョブが集約の代表的な例です。

TrainingJob集約の設計

学習ジョブには以下の不変条件があります。

1. 学習開始後、ハイパーパラメータは変更できない
2. 状態遷移は決められたパスのみ許可される（例: COMPLETED → RUNNINGへの逆行は不可）
3. メトリクスは学習中（RUNNING状態）のみ記録できる

これらの不変条件を集約の内部で保護します。

from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field
from enum import Enum
from datetime import datetime
from uuid import UUID, uuid4


class JobStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"


VALID_TRANSITIONS: dict[JobStatus, set[JobStatus]] = {
    JobStatus.PENDING: {JobStatus.RUNNING, JobStatus.CANCELLED},
    JobStatus.RUNNING: {JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED},
    JobStatus.COMPLETED: set(),
    JobStatus.FAILED: set(),
    JobStatus.CANCELLED: set(),
}


class DomainEvent(BaseModel):
    """ドメインイベントの基底クラス。"""
    model_config = ConfigDict(frozen=True)
    event_id: UUID = Field(default_factory=uuid4)
    occurred_at: datetime = Field(default_factory=datetime.now)


class TrainingStarted(DomainEvent):
    job_id: UUID
    hyperparameters: Hyperparameters


class TrainingCompleted(DomainEvent):
    job_id: UUID
    final_metrics: TrainingMetrics


class TrainingFailed(DomainEvent):
    job_id: UUID
    error_message: str


class TrainingJob:
    """学習ジョブ集約。状態遷移と不変条件を保護する集約ルート。"""

    def __init__(
        self,
        job_id: UUID,
        experiment_id: UUID,
        hyperparameters: Hyperparameters,
        config: TrainingConfig,
    ) -> None:
        self._job_id = job_id
        self._experiment_id = experiment_id
        self._hyperparameters = hyperparameters
        self._config = config
        self._status = JobStatus.PENDING
        self._metrics_history: list[TrainingMetrics] = []
        self._events: list[DomainEvent] = []
        self._created_at = datetime.now()
        self._started_at: datetime | None = None
        self._completed_at: datetime | None = None

    @property
    def job_id(self) -> UUID:
        return self._job_id

    @property
    def status(self) -> JobStatus:
        return self._status

    @property
    def hyperparameters(self) -> Hyperparameters:
        return self._hyperparameters

    @property
    def metrics_history(self) -> tuple[TrainingMetrics, ...]:
        return tuple(self._metrics_history)

    @property
    def events(self) -> list[DomainEvent]:
        return list(self._events)

    def clear_events(self) -> None:
        self._events.clear()

    def start(self) -> None:
        """学習を開始する。PENDING状態からのみ遷移可能。"""
        self._transition_to(JobStatus.RUNNING)
        self._started_at = datetime.now()
        self._events.append(
            TrainingStarted(
                job_id=self._job_id,
                hyperparameters=self._hyperparameters,
            )
        )

    def record_metrics(self, metrics: TrainingMetrics) -> None:
        """メトリクスを記録する。RUNNING状態でのみ実行可能。"""
        if self._status != JobStatus.RUNNING:
            raise InvalidOperationError(
                f"メトリクスはRUNNING状態でのみ記録可能です（現在: {self._status.value}）"
            )
        self._metrics_history.append(metrics)

    def complete(self, final_metrics: TrainingMetrics) -> None:
        """学習を正常完了する。"""
        self._transition_to(JobStatus.COMPLETED)
        self._metrics_history.append(final_metrics)
        self._completed_at = datetime.now()
        self._events.append(
            TrainingCompleted(
                job_id=self._job_id,
                final_metrics=final_metrics,
            )
        )

    def fail(self, error_message: str) -> None:
        """学習を失敗として記録する。"""
        self._transition_to(JobStatus.FAILED)
        self._completed_at = datetime.now()
        self._events.append(
            TrainingFailed(
                job_id=self._job_id,
                error_message=error_message,
            )
        )

    def cancel(self) -> None:
        """学習をキャンセルする。PENDINGまたはRUNNING状態から遷移可能。"""
        self._transition_to(JobStatus.CANCELLED)
        self._completed_at = datetime.now()

    def _transition_to(self, new_status: JobStatus) -> None:
        """状態遷移を検証して実行する。"""
        valid = VALID_TRANSITIONS.get(self._status, set())
        if new_status not in valid:
            raise InvalidOperationError(
                f"状態遷移 {self._status.value} → {new_status.value} は許可されていません。"
                f"許可される遷移先: {[s.value for s in valid]}"
            )
        self._status = new_status


class InvalidOperationError(Exception):
    """ドメインルール違反時に送出される例外。"""
    pass

集約設計の判断基準

集約の境界を決める際に重要なのは、何を同じトランザクション内で一貫して保つ必要があるかです。

# NG: TrainingJobの中にMLModelを含めてしまう
class TrainingJob:
    def __init__(self, ...):
        self._model = MLModel(...)  # 集約が肥大化する

# OK: IDで参照し、結果整合性で連携する
class TrainingJob:
    def __init__(self, ...):
        self._model_id: UUID = model_id  # IDのみ保持

なぜIDで参照するか:

Vaughn Vernonの集約設計ルールでは「他の集約はIDでのみ参照する」とされています（Effective Aggregate Design Part I）。学習完了後にモデルを登録する処理は、ドメインイベントTrainingCompletedを通じて別のトランザクションで実行します。これにより、学習ジョブの保存とモデル登録が独立し、一方の失敗が他方に波及しません。

TheCodeForgeの報告では、20以上のエンティティを持つ集約で楽観的ロック失敗が発生し、トランザクションの12%が失敗した事例があります。集約のサイズ目標は3〜7の内部エンティティとされています。MLドメインでは、TrainingJob集約の内部をHyperparameters（値オブジェクト）+ TrainingConfig（値オブジェクト）+ ResourceSpec（値オブジェクト）+ メトリクス履歴に限定し、ExperimentやMLModelは別の集約として分離するのが適切です。

よくある失敗: 貧血ドメインモデル

MLパイプラインでありがちな設計ミスが「貧血ドメインモデル」です。

# NG: 貧血ドメインモデル（ロジックがサービス層に漏れ出す）
class TrainingJobData:
    job_id: UUID
    status: str
    hyperparameters: dict  # 生のdictで管理

class TrainingService:
    def start_job(self, job: TrainingJobData) -> None:
        if job.status != "pending":  # バリデーションがサービスに分散
            raise ValueError("...")
        job.status = "running"       # 状態を外部から直接変更

この設計では、状態遷移のルールがTrainingServiceに分散し、別のサービスが同じTrainingJobDataを操作する場合に同じバリデーションを重複実装する必要があります。集約パターンでは、状態遷移のロジックをTrainingJob内部に閉じ込めることで、どこから呼び出しても一貫したルールが適用されます。

リポジトリパターンでML基盤を抽象化する

リポジトリパターンは、集約の永続化と復元を担うインターフェースです。ドメイン層は「集約をどこに保存するか」を知らず、抽象的なインターフェースを通じて操作します。MLドメインでは、MLflow・Weights & Biases・SageMakerなどのML基盤への依存をリポジトリの背後に隠すことで、テスト容易性と基盤の差し替え可能性を確保できます。

リポジトリインターフェースの定義

Cosmic Pythonで解説されているパターンに従い、ドメイン層にインターフェース（抽象基底クラス）を定義し、インフラ層に実装を配置します。

from abc import ABC, abstractmethod
from uuid import UUID


class ExperimentRepository(ABC):
    """Experiment集約のリポジトリインターフェース。
    ドメイン層に定義し、インフラ層で実装する。
    """

    @abstractmethod
    def save(self, job: TrainingJob) -> None:
        """TrainingJob集約を永続化する。"""
        ...

    @abstractmethod
    def find_by_id(self, job_id: UUID) -> TrainingJob | None:
        """IDでTrainingJob集約を復元する。"""
        ...

    @abstractmethod
    def find_by_experiment(self, experiment_id: UUID) -> list[TrainingJob]:
        """Experiment IDに紐づくTrainingJob一覧を取得する。"""
        ...


class ModelRepository(ABC):
    """MLModel集約のリポジトリインターフェース。"""

    @abstractmethod
    def save(self, model: "MLModel") -> None: ...

    @abstractmethod
    def find_by_id(self, model_id: UUID) -> "MLModel | None": ...

    @abstractmethod
    def find_latest_version(self, model_name: str) -> "MLModel | None": ...

インメモリ実装（テスト用）

テスト時はML基盤への接続なしでドメインロジックを検証します。

class InMemoryExperimentRepository(ExperimentRepository):
    """テスト用のインメモリリポジトリ実装。"""

    def __init__(self) -> None:
        self._store: dict[UUID, TrainingJob] = {}

    def save(self, job: TrainingJob) -> None:
        self._store[job.job_id] = job

    def find_by_id(self, job_id: UUID) -> TrainingJob | None:
        return self._store.get(job_id)

    def find_by_experiment(self, experiment_id: UUID) -> list[TrainingJob]:
        return [
            job for job in self._store.values()
            if job._experiment_id == experiment_id
        ]

MLflow実装（本番用）

本番環境ではMLflow等のML基盤に接続するリポジトリを実装します。

import mlflow
from mlflow.tracking import MlflowClient


class MLflowExperimentRepository(ExperimentRepository):
    """MLflowをバックエンドとするリポジトリ実装。"""

    def __init__(self, tracking_uri: str) -> None:
        mlflow.set_tracking_uri(tracking_uri)
        self._client = MlflowClient()

    def save(self, job: TrainingJob) -> None:
        with mlflow.start_run(run_name=str(job.job_id)) as run:
            mlflow.log_params(
                job.hyperparameters.model_dump(mode="python")
            )
            for metrics in job.metrics_history:
                mlflow.log_metrics(
                    {
                        "train_loss": metrics.train_loss,
                        "val_loss": metrics.val_loss,
                        "val_accuracy": metrics.val_accuracy,
                    },
                    step=metrics.step,
                )

    def find_by_id(self, job_id: UUID) -> TrainingJob | None:
        runs = self._client.search_runs(
            experiment_ids=["0"],
            filter_string=f"tags.job_id = '{job_id}'",
            max_results=1,
        )
        if not runs:
            return None
        return self._reconstruct_job(runs[0])

    def find_by_experiment(self, experiment_id: UUID) -> list[TrainingJob]:
        runs = self._client.search_runs(
            experiment_ids=[str(experiment_id)],
        )
        return [self._reconstruct_job(run) for run in runs]

    def _reconstruct_job(self, run: mlflow.entities.Run) -> TrainingJob:
        """MLflowのRunからTrainingJob集約を復元する。"""
        params = run.data.params
        job = TrainingJob(
            job_id=UUID(run.info.run_name),
            experiment_id=UUID(run.info.experiment_id),
            hyperparameters=Hyperparameters(
                learning_rate=LearningRate(float(params["learning_rate"])),
                batch_size=BatchSize(int(params["batch_size"])),
                epochs=Epoch(int(params["epochs"])),
                optimizer=params["optimizer"],
                weight_decay=float(params.get("weight_decay", 0.0)),
            ),
            config=TrainingConfig.from_mlflow_tags(run.data.tags),
        )
        return job

テストでの活用

リポジトリパターンの主要な利点は、テスト容易性です。インメモリ実装を使うことで、MLflowサーバーなしで集約のビジネスロジックをテストできます。

import pytest
from uuid import uuid4
from datetime import datetime


class TestTrainingJob:
    def setup_method(self) -> None:
        self.repo = InMemoryExperimentRepository()
        self.params = Hyperparameters(
            learning_rate=LearningRate(0.001),
            batch_size=BatchSize(32),
            epochs=Epoch(10),
            optimizer="adamw",
            weight_decay=0.01,
        )

    def test_start_transitions_to_running(self) -> None:
        job = TrainingJob(
            job_id=uuid4(),
            experiment_id=uuid4(),
            hyperparameters=self.params,
            config=TrainingConfig(framework="pytorch", device="cuda"),
        )
        job.start()
        assert job.status == JobStatus.RUNNING

    def test_cannot_record_metrics_when_pending(self) -> None:
        job = TrainingJob(
            job_id=uuid4(),
            experiment_id=uuid4(),
            hyperparameters=self.params,
            config=TrainingConfig(framework="pytorch", device="cuda"),
        )
        metrics = TrainingMetrics(
            train_loss=0.5, val_loss=0.6, val_accuracy=0.8,
            step=1, recorded_at=datetime.now(),
        )
        with pytest.raises(InvalidOperationError):
            job.record_metrics(metrics)

    def test_completed_job_cannot_restart(self) -> None:
        job = TrainingJob(
            job_id=uuid4(),
            experiment_id=uuid4(),
            hyperparameters=self.params,
            config=TrainingConfig(framework="pytorch", device="cuda"),
        )
        job.start()
        final = TrainingMetrics(
            train_loss=0.1, val_loss=0.15, val_accuracy=0.95,
            step=100, recorded_at=datetime.now(),
        )
        job.complete(final)
        with pytest.raises(InvalidOperationError):
            job.start()

    def test_repository_save_and_retrieve(self) -> None:
        job = TrainingJob(
            job_id=uuid4(),
            experiment_id=uuid4(),
            hyperparameters=self.params,
            config=TrainingConfig(framework="pytorch", device="cuda"),
        )
        self.repo.save(job)
        retrieved = self.repo.find_by_id(job.job_id)
        assert retrieved is not None
        assert retrieved.job_id == job.job_id

よくある問題と解決方法

DDD戦術パターンをMLドメインに適用する際に遭遇しやすい問題をまとめます。

問題	原因	解決方法
集約が巨大化してメモリを圧迫する	TrainingJobにメトリクス全履歴を保持している	メトリクス履歴は別の集約（MetricsLog）に分離し、サマリのみ保持する
ハイパーパラメータのバリデーションが散在する	値オブジェクトを使わずdictで管理している	Hyperparameters値オブジェクトにバリデーションを集約する
MLflowへの依存でテストが遅い	リポジトリパターンを使わず直接MLflow APIを呼んでいる	InMemoryExperimentRepositoryでドメインロジックのテストを分離する
学習ジョブの状態が不整合になる	外部から直接statusを書き換えている	集約ルートのメソッド（start(), complete()）経由のみで状態変更を許可する
集約間の更新が部分的に失敗する	TrainingJob完了とMLModel登録を同一トランザクションで処理している	ドメインイベント（TrainingCompleted）で結果整合性を実現する
frozenモデル内のリストが外部から変更される	Pydanticのfrozen=Trueはフィールド再代入のみ防止する	リスト型フィールドはtupleに変換して返すか、@model_validatorで防御する

トレードオフの認識:

DDD戦術パターンの適用にはコストが伴います。値オブジェクトのクラス定義、集約のメソッド設計、リポジトリのインターフェース分離は、単純なCRUD操作に比べてコード量が増加します。2026年のDDDガイドでも指摘されているように、戦術パターンはドメインロジックが複雑な「コアドメイン」に限定して適用し、単純なサポートドメイン（例: ユーザー設定管理）にはCRUDで十分です。MLパイプラインでは、学習ジョブの状態管理やモデルバージョニングがコアドメインに該当し、ログ出力やメール通知はサポートドメインです。

まとめと次のステップ

まとめ:

• 値オブジェクト（Pydantic frozenモデル）で、ハイパーパラメータやメトリクスのバリデーションを型レベルで保証し、不正な値の混入を構造的に防止できる
• 集約（TrainingJob）で学習ジョブの状態遷移を保護し、不変条件をドメインモデル内に閉じ込めることで、どこからでも一貫したルールが適用される
• リポジトリパターンでMLflow等のML基盤への依存を抽象化し、インメモリ実装によるテストの高速化と基盤の差し替え可能性を確保できる
• 集約の境界は「データ構造」ではなく「ビジネスルール（不変条件）の整合性範囲」で決定する。MLドメインでは、学習ジョブ・実験・モデルをそれぞれ独立した集約として設計するのが適切
• DDD戦術パターンはドメインが複雑な箇所に限定して適用し、単純なCRUDには使わない

次にやるべきこと:

• 自分のMLパイプラインで「状態遷移が複雑」「バリデーションが散在している」箇所を特定し、集約と値オブジェクトの導入を検討する
• Cosmic Pythonのリポジトリパターンの章を読み、Unit of Workパターンとの組み合わせを学ぶ
• DDDの戦略的パターン（境界づけられたコンテキスト、ユビキタス言語）を学び、MLパイプライン全体の設計に適用する。戦術パターン単体では、コンテキスト境界が曖昧なまま「DDDっぽいクラスがあるが結局絡み合う」状態になりがちである点に注意

参考

• Effective Aggregate Design Part I (Vaughn Vernon) - 集約設計の基本4ルールを定義した論文
• Use Tactical DDD to Design Microservices (Microsoft Learn, 2026) - マイクロサービスにおけるDDD戦術パターンの適用ガイド
• Repository Pattern (Cosmic Python) - Pythonでのリポジトリパターン実装の解説
• DDD Value Objects: Mastering Data Validation in Python - Pydanticによる値オブジェクト実装ガイド
• DDD Aggregate Sizing and Production Failures (TheCodeForge) - 集約サイズの目標値と本番障害事例
• Domain-Driven Design: A Complete Guide (2026) - 2026年版の包括的DDDガイド
• AI/ML & Domain Driven Design (AKF Partners) - DDDとML/AIの共通点
• The Similarities between Machine Learning and DDD (Philippe Bourgau) - DDDとMLの類似性分析
• Pydantic V2 Deep Dive: Immutable Data Architectures - Pydantic v2のfrozenモデル詳細解説
• TypeScript・Pythonで実装するDDD戦術的パターン実践ガイド - DDD戦術パターンの基本概念と実装（本シリーズ前編）

---

注意: この記事はAI（Claude Code）により自動生成されました。内容の正確性については複数の情報源で検証していますが、実際の利用時は公式ドキュメントもご確認ください。