一、引言：为什么AI平台需要DDD
领域驱动设计（Domain-Driven Design，DDD）作为一种成熟的软件设计方法论，在处理复杂业务领域时展现出独特的优势。AI算法平台恰恰是一个业务复杂度极高的领域：模型管理、训练调度、推理服务、数据处理、权限控制等核心领域相互交织，传统的三层架构（Controller-Service-DAO）往往难以清晰地表达业务意图，导致代码随着业务增长而逐渐腐化。
我所在团队在从零构建AI算法平台的过程中，选择了DDD作为核心架构方法。经过一年多的实践，我们深刻体会到DDD在处理复杂业务边界、统一语言、保证系统可演化性方面的价值。本文将详细记录我们在AI算法平台上应用DDD的完整实践，包括限界上下文划分、领域模型设计、领域事件机制、CQRS应用等核心内容。
二、DDD基础概念回顾
2.1 战略设计与限界上下文
DDD分为战略设计和战术设计两个层面。战略设计关注"做什么"，战术设计关注"怎么做"。
战略设计的核心是限界上下文（Bounded Context）。限界上下文是一个业务领域的边界，在这个边界内有一套统一的术语、业务规则和数据模型。限界上下文之间的交互通过上下文映射（Context Mapping）来定义。
┌─────────────────────────────────────────────────────────────────────────┐
│ AI算法平台限界上下文划分 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────────┐ │
│ │ 用户管理上下文 │ User Context │
│ │ ┌────────────┐ │ │
│ │ │ 用户 │ │ │
│ │ │ 角色 │ │ │
│ │ │ 权限 │ │ │
│ │ └────────────┘ │ │
│ └────────┬─────────┘ │
│ │ │ │
│ │ │ ACL (防腐层) │
│ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 核心业务领域 │ │
│ │ │ │
│ │ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ │
│ │ │ 模型管理上下文 │ │ 训练管理上下文 │ │ 推理服务上下文 │ │ │
│ │ │ ┌────────────┐ │ │ ┌────────────┐ │ │ ┌────────────┐ │ │ │
│ │ │ │ 模型注册 │ │ │ │ 训练任务 │ │ │ │ 推理服务 │ │ │ │
│ │ │ │ 模型版本 │ │ │ │ 数据集 │ │ │ │ 部署配置 │ │ │ │
│ │ │ │ 模型元数据│ │ │ │ 训练配置 │ │ │ │ 调用记录 │ │ │ │
│ │ │ │ 模型评测 │ │ │ │ 资源调度 │ │ │ │ 性能监控 │ │ │ │
│ │ │ └────────────┘ │ │ └────────────┘ │ │ └────────────┘ │ │ │
│ │ └──────────────────┘ └────────┬─────────┘ └──────────────────┘ │ │
│ │ │ │ │
│ │ ┌──────────────────────────────┴──────────────────────────────┐ │ │
│ │ │ 领域事件总线 (Domain Event Bus) │ │ │
│ │ │ ModelRegistered | TrainingCompleted | DeployPromoted │ │ │
│ │ └───────────────────────────────────────────────────────────────┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘2.2 战术设计：聚合根、实体、值对象
战术设计关注领域模型的具体实现，包括：
聚合根（Aggregate Root）：聚合的入口点，负责维护聚合内的一致性边界。一个聚合只有一个聚合根。
实体（Entity）：具有唯一标识的对象，其生命周期内标识不变。
值对象（Value Object）：没有唯一标识，仅通过属性值定义的对象，不可变。
┌─────────────────────────────────────────────────────────────────────────┐
│ 聚合设计示例：训练任务 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ TrainingTask (聚合根) │ │
│ │ ┌───────────────────────────────────────────────────────────┐ │ │
│ │ │ id: UUID # 聚合根标识 │ │ │
│ │ │ name: str # 任务名称 │ │ │
│ │ │ status: TaskStatus # 任务状态（实体） │ │ │
│ │ │ config: TrainingConfig # 训练配置（值对象） │ │ │
│ │ │ resources: ResourceSpec # 资源规格（值对象） │ │ │
│ │ │ metrics: TrainingMetrics # 训练指标（实体） │ │ │
│ │ └───────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ┌───────────────────┼───────────────────┐ │ │
│ │ ▼ ▼ ▼ │ │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │
│ │ │ Checkpoint │ │ TrainingLog │ │ Hyperparam │ │ │
│ │ │ (实体) │ │ (实体) │ │ (值对象) │ │ │
│ │ │ - id │ │ - id │ │ - lr │ │ │
│ │ │ - path │ │ - lines │ │ - batch_size │ │ │
│ │ │ - step │ │ - timestamp │ │ - epochs │ │ │
│ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ 业务规则： │
│ - 只有处于PENDING状态的训练任务才能启动 │
│ - Checkpoint只能由TrainingTask创建和管理 │
│ - 状态变更必须通过聚合根的方法完成 │
│ │
└─────────────────────────────────────────────────────────────────────────┘三、限界上下文划分详解
3.1 模型管理上下文（Model Management Context）
模型管理上下文负责模型的注册、版本管理、元数据存储和评测。这是AI平台的核心上下文之一，与训练和推理上下文都有密切交互。
统一语言（Ubiquitous Language）：
模型（Model）：一个可部署的AI模型
模型版本（Model Version）：模型的特定版本
基座模型（Base Model）：用于微调的基础模型
评测任务（Evaluation Task）：评估模型性能的测试任务
评测指标（Metrics）：衡量模型性能的具体指标
3.2 训练管理上下文（Training Context）
训练管理上下文处理模型训练的全生命周期，包括数据准备、训练配置、任务调度、 checkpoint管理和训练监控。
统一语言：
训练任务（Training Task）：一次完整的训练作业
数据集（Dataset）：训练用的数据集
超参数（Hyperparameters）：训练超参数配置
Checkpoint：训练过程中的模型快照
训练指标（Training Metrics）：Loss、Accuracy等训练过程指标
3.3 推理服务上下文（Inference Context）
推理服务上下文管理模型的部署、推理API和性能监控。
统一语言：
推理服务（Inference Service）：对外提供推理能力的Service
部署配置（Deployment Config）：部署规格配置
推理请求（Inference Request）：一次推理调用
推理响应（Inference Response）：推理返回结果
服务指标（Service Metrics）：QPS、延迟等监控指标
3.4 数据管理上下文（Data Context）
数据管理上下文处理数据集的上传、存储、版本化和数据质量控制。
统一语言：
数据集（Dataset）：数据集合
数据版本（Data Version）：数据集的特定版本
数据质量（Data Quality）：数据质量评估结果
数据血缘（Data Lineage）：数据来源和加工关系
四、领域模型设计
4.1 模型管理领域模型
以下是模型管理上下文中核心领域模型的Python实现：
# domain/models/model_management.py
# 模型管理领域模型 - 展示DDD战术设计实践
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import List, Optional, Dict, Any
class ModelStatus(Enum):
"""模型状态枚举"""
DRAFT = "draft" # 草稿
VALIDATING = "validating" # 验证中
READY = "ready" # 就绪
DEPLOYED = "deployed" # 已部署
DEPRECATED = "deprecated" # 已废弃
class QuantizationType(Enum):
"""量化类型枚举"""
FP16 = "fp16"
FP32 = "fp32"
INT8 = "int8"
INT4 = "int4"
AWQ = "awq"
GPTQ = "gptq"
@dataclass(frozen=True)
class ModelId:
"""模型唯一标识（值对象）"""
value: str
def __post_init__(self):
if not self.value:
raise ValueError("模型标识不能为空")
def __str__(self) -> str:
return self.value
@dataclass(frozen=True)
class ModelVersionId:
"""模型版本唯一标识（值对象）"""
value: str
def __post_init__(self):
if not self.value:
raise ValueError("版本标识不能为空")
@dataclass(frozen=True)
class QuantizationConfig:
"""量化配置（值对象）"""
quantization_type: QuantizationType
calibration_data_size: int = 1000
bits: int = 4
def __post_init__(self):
if self.quantization_type in (QuantizationType.INT4, QuantizationType.INT8):
if self.bits not in (4, 8):
raise ValueError(f"量化位数必须是4或8，实际为{self.bits}")
@dataclass
class ModelMetrics:
"""模型评测指标（实体）"""
bleu_score: Optional[float] = None
rouge_score: Optional[float] = None
accuracy: Optional[float] = None
loss: Optional[float] = None
latency_p50_ms: Optional[float] = None
latency_p99_ms: Optional[float] = None
custom_metrics: Dict[str, float] = field(default_factory=dict)
def add_metric(self, name: str, value: float) -> None:
"""添加自定义指标"""
self.custom_metrics[name] = value
def get_metric(self, name: str) -> Optional[float]:
"""获取指标值"""
return self.custom_metrics.get(name)
@dataclass
class Model:
"""
模型聚合根
聚合 invariants（不变式）：
1. 模型名称在租户内唯一
2. 至少包含一个版本才能标记为 READY
3. 只有 READY 状态的模型才能部署
"""
# 基础属性
id: ModelId
tenant_id: str
name: str
description: str
# 版本相关
_versions: List[ModelVersion] = field(default_factory=list)
# 元数据
base_model_type: Optional[str] = None
framework: str = "pytorch"
tags: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
# 状态
status: ModelStatus = ModelStatus.DRAFT
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
created_by: str = ""
# 聚合版本控制（用于乐观锁）
_version: int = 0
def register_version(self, version: ModelVersion) -> None:
"""
注册新版本
业务规则：
1. 版本号必须唯一
2. 如果是第一个版本，模型状态变为 READY
"""
# 检查版本号唯一性
if any(v.version == version.version for v in self._versions):
raise ValueError(f"版本号 {version.version} 已存在")
self._versions.append(version)
self.updated_at = datetime.now()
# 如果是第一个版本，更新模型状态
if len(self._versions) == 1:
self.status = ModelStatus.READY
def get_latest_version(self) -> Optional[ModelVersion]:
"""获取最新版本"""
if not self._versions:
return None
return max(self._versions, key=lambda v: v.created_at)
def deprecate(self) -> None:
"""
废弃模型
业务规则：
1. 已部署的模型不能直接废弃，需先下线
"""
if self.status == ModelStatus.DEPLOYED:
raise ValueError("已部署的模型需先下线才能废弃")
self.status = ModelStatus.DEPRECATED
self.updated_at = datetime.now()
def can_deploy(self) -> bool:
"""检查是否可以部署"""
return (
self.status == ModelStatus.READY and
len(self._versions) > 0
)
@dataclass
class ModelVersion:
"""
模型版本（实体）
作为 Model 聚合内的实体，由聚合根管理其生命周期
"""
id: ModelVersionId
model_id: ModelId
version: str
model_path: str
# 模型规格
parameter_count: int
context_length: int = 4096
# 量化配置
quantization: Optional[QuantizationConfig] = None
# 存储信息
size_bytes: int
checkpoint_path: Optional[str] = None
# 配置
config: Dict[str, Any] = field(default_factory=dict)
requirements: List[str] = field(default_factory=list)
# 评测指标
metrics: ModelMetrics = field(default_factory=ModelMetrics)
# 状态
status: ModelStatus = ModelStatus.VALIDATING
created_at: datetime = field(default_factory=datetime.now)
created_by: str = ""
def validate(self) -> None:
"""验证模型文件"""
# TODO: 实现模型文件验证逻辑
self.status = ModelStatus.READY
def record_metrics(self, metrics: ModelMetrics) -> None:
"""记录评测指标"""
self.metrics = metrics
@dataclass
class EvaluationTask:
"""
评测任务（聚合根）
管理模型评测的完整生命周期
"""
class Status(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
class Type(Enum):
AUTOMATIC = "automatic" # 自动评测
HUMAN = "human" # 人工评测
LLM_JUDGE = "llm_judge" # LLM作为裁判
id: str = field(default_factory=lambda: str(uuid.uuid4()))
tenant_id: str = ""
model_version_id: Optional[ModelVersionId] = None
# 评测配置
task_type: Type = Type.AUTOMATIC
benchmark_id: str = ""
prompt_template: str = ""
evaluation_config: Dict[str, Any] = field(default_factory=dict)
# 状态
status: Status = Status.PENDING
progress: float = 0.0
# 结果
results: Optional[ModelMetrics] = None
error_message: Optional[str] = None
# 时间戳
created_at: datetime = field(default_factory=datetime.now)
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None4.2 训练管理领域模型
# domain/models/training.py
# 训练管理领域模型
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import List, Optional, Dict, Any
import uuid
class TrainingStatus(Enum):
"""训练任务状态"""
PENDING = "pending" # 等待调度
QUEUED = "queued" # 已进入队列
RUNNING = "running" # 运行中
PAUSED = "paused" # 暂停
SUCCEEDED = "succeeded" # 成功完成
FAILED = "failed" # 失败
CANCELLED = "cancelled" # 已取消
@dataclass(frozen=True)
class GPUResourceSpec:
"""GPU资源规格（值对象）"""
gpu_count: int
gpu_type: str # 如 "A100", "V100", "A10"
memory_per_gpu_gb: int
def total_memory_gb(self) -> int:
return self.gpu_count * self.memory_per_gpu_gb
def __post_init__(self):
if self.gpu_count <= 0:
raise ValueError("GPU数量必须大于0")
if self.gpu_type not in ("A100", "V100", "A10", "H100", "L40"):
raise ValueError(f"不支持的GPU类型: {self.gpu_type}")
@dataclass(frozen=True)
class TrainingHyperparameters:
"""训练超参数（值对象）"""
learning_rate: float
batch_size: int
epochs: int
warmup_steps: int = 0
weight_decay: float = 0.01
gradient_accumulation_steps: int = 1
# LLM特定参数
max_seq_length: int = 2048
use_fp16: bool = True
use_gradient_checkpointing: bool = False
def __post_init__(self):
if self.learning_rate <= 0:
raise ValueError("学习率必须大于0")
if self.batch_size <= 0:
raise ValueError("批量大小必须大于0")
if self.epochs <= 0:
raise ValueError("训练轮数必须大于0")
@dataclass
class Checkpoint:
"""
训练Checkpoint（实体）
由 TrainingTask 聚合根管理
"""
id: str = field(default_factory=lambda: str(uuid.uuid4()))
training_task_id: str = ""
# Checkpoint信息
step: int = 0
epoch: int = 0
path: str = ""
size_bytes: int = 0
# 指标快照
global_step: int = 0
loss: Optional[float] = None
learning_rate: Optional[float] = None
metrics: Dict[str, float] = field(default_factory=dict)
created_at: datetime = field(default_factory=datetime.now)
def is_complete(self) -> bool:
"""检查Checkpoint是否完整"""
return self.path and self.size_bytes > 0
@dataclass
class TrainingTask:
"""
训练任务（聚合根）
管理训练任务的完整生命周期和内部实体
聚合 invariants：
1. 只有 PENDING 状态才能启动
2. Checkpoint只能通过 save_checkpoint() 方法创建
3. 状态变更必须遵循合法转换路径
"""
# 基本信息
id: str = field(default_factory=lambda: str(uuid.uuid4()))
tenant_id: str = ""
name: str = ""
description: str = ""
# 关联资源
dataset_id: str = ""
base_model_id: str = ""
output_model_name: str = ""
# 配置
hyperparameters: Optional[TrainingHyperparameters] = None
resource_spec: Optional[GPUResourceSpec] = None
# 训练配置
entry_point: str = "train.py"
python_packages: List[str] = field(default_factory=list)
environment_vars: Dict[str, str] = field(default_factory=dict)
# 状态
status: TrainingStatus = TrainingStatus.PENDING
progress: float = 0.0
# 内部实体
_checkpoints: List[Checkpoint] = field(default_factory=list)
_current_step: int = 0
# 结果
final_metrics: Optional[Dict[str, float]] = None
error_message: Optional[str] = None
# Kubernetes相关信息
kubernetes_namespace: Optional[str] = None
kubernetes_job_name: Optional[str] = None
# 时间戳
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
created_by: str = ""
# 乐观锁版本
_version: int = 0
# 允许的状态转换映射
_allowed_transitions: Dict[TrainingStatus, List[TrainingStatus]] = {
TrainingStatus.PENDING: [TrainingStatus.QUEUED, TrainingStatus.CANCELLED],
TrainingStatus.QUEUED: [TrainingStatus.RUNNING, TrainingStatus.CANCELLED],
TrainingStatus.RUNNING: [
TrainingStatus.PAUSED,
TrainingStatus.SUCCEEDED,
TrainingStatus.FAILED,
TrainingStatus.CANCELLED
],
TrainingStatus.PAUSED: [TrainingStatus.RUNNING, TrainingStatus.CANCELLED],
TrainingStatus.FAILED: [], # 终态
TrainingStatus.SUCCEEDED: [], # 终态
TrainingStatus.CANCELLED: [], # 终态
}
def start(self) -> None:
"""启动训练任务"""
self._validate_transition(TrainingStatus.RUNNING)
self.status = TrainingStatus.RUNNING
self.started_at = datetime.now()
self.updated_at = datetime.now()
def pause(self) -> None:
"""暂停训练任务"""
self._validate_transition(TrainingStatus.PAUSED)
self.status = TrainingStatus.PAUSED
self.updated_at = datetime.now()
def resume(self) -> None:
"""恢复训练任务"""
self._validate_transition(TrainingStatus.RUNNING)
self.status = TrainingStatus.RUNNING
self.updated_at = datetime.now()
def complete(self, final_metrics: Dict[str, float]) -> None:
"""
完成任务
Args:
final_metrics: 最终训练指标
"""
self._validate_transition(TrainingStatus.SUCCEEDED)
self.status = TrainingStatus.SUCCEEDED
self.final_metrics = final_metrics
self.progress = 1.0
self.completed_at = datetime.now()
self.updated_at = datetime.now()
def fail(self, error_message: str) -> None:
"""标记任务失败"""
self._validate_transition(TrainingStatus.FAILED)
self.status = TrainingStatus.FAILED
self.error_message = error_message
self.completed_at = datetime.now()
self.updated_at = datetime.now()
def cancel(self) -> None:
"""取消任务"""
self._validate_transition(TrainingStatus.CANCELLED)
self.status = TrainingStatus.CANCELLED
self.completed_at = datetime.now()
self.updated_at = datetime.now()
def update_progress(self, step: int, metrics: Dict[str, float]) -> None:
"""更新训练进度"""
if self.status != TrainingStatus.RUNNING:
raise ValueError(f"当前状态 {self.status} 不能更新进度")
self._current_step = step
self._checkpoints.append(Checkpoint(
training_task_id=self.id,
step=step,
path=f"/checkpoints/{self.id}/step_{step}",
global_step=step,
loss=metrics.get("loss"),
learning_rate=metrics.get("learning_rate"),
metrics=metrics
))
self.updated_at = datetime.now()
def save_checkpoint(self, step: int, metrics: Dict[str, float]) -> Checkpoint:
"""
保存Checkpoint
这是创建Checkpoint的唯一入口
"""
if self.status != TrainingStatus.RUNNING:
raise ValueError("只有运行中的任务才能保存Checkpoint")
checkpoint = Checkpoint(
training_task_id=self.id,
step=step,
path=f"/checkpoints/{self.id}/step_{step}",
global_step=step,
loss=metrics.get("loss"),
learning_rate=metrics.get("learning_rate"),
metrics=metrics
)
self._checkpoints.append(checkpoint)
return checkpoint
def get_latest_checkpoint(self) -> Optional[Checkpoint]:
"""获取最新Checkpoint"""
if not self._checkpoints:
return None
return max(self._checkpoints, key=lambda c: c.step)
def _validate_transition(self, new_status: TrainingStatus) -> None:
"""验证状态转换是否合法"""
allowed = self._allowed_transitions.get(self.status, [])
if new_status not in allowed:
raise ValueError(
f"不允许的状态转换: {self.status} -> {new_status}"
)五、领域事件设计
5.1 领域事件概述
领域事件（Domain Event）是DDD中非常重要的概念，它描述了领域中发生的、对业务有意义的事件。通过领域事件，不同限界上下文可以解耦地通信，实现最终一致性。
┌─────────────────────────────────────────────────────────────────────────┐
│ 领域事件流转架构 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────────┐ │
│ │ 模型管理上下文 │ │
│ │ │ │
│ │ DomainEvent │ │
│ │ ├─ ModelRegisted │ │
│ │ └─ ModelValidated│ │
│ └────────┬─────────┘ │
│ │ publish │
│ ▼ │
│ ┌──────────────────────────────────────────────────────────────────┐ │
│ │ 领域事件总线 (Event Bus) │ │
│ │ ┌─────────────────────────────────────────────────────────────┐ │ │
│ │ │ Event Store (持久化) │ │ │
│ │ │ - 所有发布的事件都会被记录 │ │ │
│ │ │ - 支持事件回放和重试 │ │ │
│ │ └─────────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────────┘ │
│ │ dispatch │
│ ▼ │
│ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │
│ │ 训练管理上下文 │ │ 推理服务上下文 │ │ 通知上下文 │ │
│ │ │ │ │ │ │ │
│ │ EventHandler │ │ EventHandler │ │ EventHandler │ │
│ │ └─ 启动训练任务 │ │ └─ 更新模型列表 │ │ └─ 发送通知 │ │
│ └──────────────────┘ └──────────────────┘ └──────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘5.2 领域事件实现
# domain/events.py
# 领域事件定义和发布机制
from __future__ import annotations
import json
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, Optional, Callable, List
from abc import ABC, abstractmethod
import logging
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class DomainEvent:
"""
领域事件基类
所有领域事件都应该：
1. 继承此基类
2. 是不可变的（frozen=True）
3. 包含事件发生的时间戳
4. 包含事件的唯一标识
"""
event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
occurred_on: datetime = field(default_factory=datetime.now)
tenant_id: str = ""
@property
@abstractmethod
def event_type(self) -> str:
"""事件类型，必须在子类中定义"""
pass
def to_dict(self) -> Dict[str, Any]:
"""序列化为字典"""
return {
'event_id': self.event_id,
'event_type': self.event_type,
'occurred_on': self.occurred_on.isoformat(),
'tenant_id': self.tenant_id,
'data': self._extract_data()
}
@abstractmethod
def _extract_data(self) -> Dict[str, Any]:
"""提取事件数据"""
pass
# ==================== 模型管理领域事件 ====================
@dataclass(frozen=True)
class ModelRegistered(DomainEvent):
"""
模型注册事件
当新模型被注册到平台时触发
"""
model_id: str = ""
model_name: str = ""
base_model: str = ""
created_by: str = ""
@property
def event_type(self) -> str:
return "model.registered"
def _extract_data(self) -> Dict[str, Any]:
return {
'model_id': self.model_id,
'model_name': self.model_name,
'base_model': self.base_model,
'created_by': self.created_by
}
@dataclass(frozen=True)
class ModelVersionRegistered(DomainEvent):
"""
模型版本注册事件
当新版本被添加到模型时触发
"""
model_id: str = ""
version_id: str = ""
version: str = ""
parameter_count: int = 0
size_bytes: int = 0
@property
def event_type(self) -> str:
return "model.version.registered"
def _extract_data(self) -> Dict[str, Any]:
return {
'model_id': self.model_id,
'version_id': self.version_id,
'version': self.version,
'parameter_count': self.parameter_count,
'size_bytes': self.size_bytes
}
@dataclass(frozen=True)
class ModelValidated(DomainEvent):
"""模型验证完成事件"""
model_id: str = ""
version_id: str = ""
validation_passed: bool = False
validation_errors: List[str] = field(default_factory=list)
@property
def event_type(self) -> str:
return "model.validated"
def _extract_data(self) -> Dict[str, Any]:
return {
'model_id': self.model_id,
'version_id': self.version_id,
'validation_passed': self.validation_passed,
'validation_errors': self.validation_errors
}
@dataclass(frozen=True)
class EvaluationCompleted(DomainEvent):
"""评测完成事件"""
evaluation_id: str = ""
model_id: str = ""
model_version_id: str = ""
benchmark_id: str = ""
metrics: Dict[str, float] = field(default_factory=dict)
passed: bool = True
@property
def event_type(self) -> str:
return "evaluation.completed"
def _extract_data(self) -> Dict[str, Any]:
return {
'evaluation_id': self.evaluation_id,
'model_id': self.model_id,
'model_version_id': self.model_version_id,
'benchmark_id': self.benchmark_id,
'metrics': self.metrics,
'passed': self.passed
}
# ==================== 训练管理领域事件 ====================
@dataclass(frozen=True)
class TrainingStarted(DomainEvent):
"""训练任务启动事件"""
training_task_id: str = ""
model_id: str = ""
dataset_id: str = ""
resource_spec: Dict[str, Any] = field(default_factory=dict)
@property
def event_type(self) -> str:
return "training.started"
def _extract_data(self) -> Dict[str, Any]:
return {
'training_task_id': self.training_task_id,
'model_id': self.model_id,
'dataset_id': self.dataset_id,
'resource_spec': self.resource_spec
}
@dataclass(frozen=True)
class TrainingCompleted(DomainEvent):
"""
训练完成事件
这是最重要的领域事件之一，会触发多个下游操作：
1. 模型注册
2. 触发评测
3. 发送通知
"""
training_task_id: str = ""
model_id: str = ""
output_model_version: str = ""
final_metrics: Dict[str, float] = field(default_factory=dict)
checkpoint_path: str = ""
duration_seconds: int = 0
@property
def event_type(self) -> str:
return "training.completed"
def _extract_data(self) -> Dict[str, Any]:
return {
'training_task_id': self.training_task_id,
'model_id': self.model_id,
'output_model_version': self.output_model_version,
'final_metrics': self.final_metrics,
'checkpoint_path': self.checkpoint_path,
'duration_seconds': self.duration_seconds
}
@dataclass(frozen=True)
class TrainingFailed(DomainEvent):
"""训练失败事件"""
training_task_id: str = ""
model_id: str = ""
error_message: str = ""
failed_at_step: int = 0
@property
def event_type(self) -> str:
return "training.failed"
def _extract_data(self) -> Dict[str, Any]:
return {
'training_task_id': self.training_task_id,
'model_id': self.model_id,
'error_message': self.error_message,
'failed_at_step': self.failed_at_step
}
@dataclass(frozen=True)
class CheckpointCreated(DomainEvent):
"""Checkpoint创建事件"""
training_task_id: str = ""
checkpoint_id: str = ""
step: int = 0
path: str = ""
size_bytes: int = 0
@property
def event_type(self) -> str:
return "training.checkpoint.created"
def _extract_data(self) -> Dict[str, Any]:
return {
'training_task_id': self.training_task_id,
'checkpoint_id': self.checkpoint_id,
'step': self.step,
'path': self.path,
'size_bytes': self.size_bytes
}
# ==================== 推理服务领域事件 ====================
@dataclass(frozen=True)
class DeployPromoted(DomainEvent):
"""
模型部署事件
当模型被部署到生产环境时触发
"""
model_id: str = ""
version_id: str = ""
deployment_id: str = ""
endpoint: str = ""
replica_count: int = 1
@property
def event_type(self) -> str:
return "inference.deployment.promoted"
def _extract_data(self) -> Dict[str, Any]:
return {
'model_id': self.model_id,
'version_id': self.version_id,
'deployment_id': self.deployment_id,
'endpoint': self.endpoint,
'replica_count': self.replica_count
}
@dataclass(frozen=True)
class InferenceAnomalyDetected(DomainEvent):
"""推理异常检测事件"""
deployment_id: str = ""
anomaly_type: str = "" # high_latency, high_error_rate, etc.
metric_value: float = 0.0
threshold: float = 0.0
@property
def event_type(self) -> str:
return "inference.anomaly.detected"
def _extract_data(self) -> Dict[str, Any]:
return {
'deployment_id': self.deployment_id,
'anomaly_type': self.anomaly_type,
'metric_value': self.metric_value,
'threshold': self.threshold
}
# ==================== 事件发布器 ====================
class EventPublisher:
"""
领域事件发布器
负责：
1. 发布事件到事件总线
2. 持久化事件到事件存储
3. 处理发布失败的情况
"""
def __init__(self):
self._handlers: Dict[str, List[Callable]] = {}
self._event_store: List[DomainEvent] = []
def subscribe(self, event_type: str, handler: Callable) -> None:
"""订阅事件"""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
logger.info(f"订阅事件: {event_type}")
def unsubscribe(self, event_type: str, handler: Callable) -> None:
"""取消订阅"""
if event_type in self._handlers:
self._handlers[event_type].remove(handler)
def publish(self, event: DomainEvent) -> None:
"""
发布事件
1. 持久化到事件存储
2. 同步调用所有处理器
3. 处理失败情况
"""
event_type = event.event_type
# 1. 持久化事件
self._event_store.append(event)
logger.info(f"发布领域事件: {event_type} [id={event.event_id}]")
# 2. 调用处理器
if event_type in self._handlers:
for handler in self._handlers[event_type]:
try:
handler(event)
except Exception as e:
logger.error(
f"事件处理器执行失败: {event_type}, "
f"handler={handler.__name__}, error={e}"
)
# TODO: 实现重试机制
def get_events(
self,
event_type: Optional[str] = None,
since: Optional[datetime] = None,
limit: int = 100
) -> List[DomainEvent]:
"""查询历史事件"""
events = self._event_store
if event_type:
events = [e for e in events if e.event_type == event_type]
if since:
events = [e for e in events if e.occurred_on >= since]
return events[-limit:]
# 全局事件发布器实例
event_publisher = EventPublisher()5.3 事件处理器实现
# domain/event_handlers.py
# 领域事件处理器实现
from domain.events import (
event_publisher,
TrainingCompleted,
ModelRegistered,
EvaluationCompleted,
DeployPromoted,
)
import logging
logger = logging.getLogger(__name__)
# ==================== 训练完成事件处理器 ====================
def handle_training_completed_training(event: TrainingCompleted) -> None:
"""
训练完成 -> 启动模型注册流程
当训练完成时：
1. 创建新的模型版本记录
2. 触发模型验证
"""
logger.info(
f"处理训练完成事件: task_id={event.training_task_id}, "
f"output={event.output_model_version}"
)
# TODO: 实现模型注册逻辑
# model_version = ModelVersion(
# model_id=event.model_id,
# version=event.output_model_version,
# checkpoint_path=event.checkpoint_path,
# metrics=event.final_metrics
# )
# model_repository.save(model_version)
# event_publisher.publish(ModelVersionRegistered(...))
logger.info(f"模型注册流程已启动: {event.output_model_version}")
def handle_training_completed_notification(event: TrainingCompleted) -> None:
"""
训练完成 -> 发送通知
当训练完成时通知相关人员
"""
logger.info(f"发送训练完成通知: task_id={event.training_task_id}")
# TODO: 实现通知逻辑
# notification_service.send(
# recipient=training_task.created_by,
# title="训练任务完成",
# content=f"任务 {training_task.name} 已完成，最终指标: {event.final_metrics}"
# )
def handle_training_completed_evaluation(event: TrainingCompleted) -> None:
"""
训练完成 -> 触发自动评测
当训练完成时，自动触发模型评测
"""
logger.info(f"触发自动评测: model={event.output_model_version}")
# TODO: 实现评测触发逻辑
# evaluation_service.create_task(
# model_version_id=model_version.id,
# benchmark_id="default",
# task_type=EvaluationTask.Type.AUTOMATIC
# )
# ==================== 模型注册事件处理器 ====================
def handle_model_registered_inference(event: ModelRegistered) -> None:
"""
模型注册 -> 更新推理服务模型列表
当新模型注册时，更新推理服务的可用模型列表
"""
logger.info(f"更新推理服务模型列表: model={event.model_name}")
# TODO: 实现推理服务更新逻辑
# inference_service.refresh_available_models()
# ==================== 部署事件处理器 ====================
def handle_deploy_promoted_monitoring(event: DeployPromoted) -> None:
"""
模型部署 -> 启动监控
当模型部署时，设置监控告警
"""
logger.info(f"启动模型监控: deployment_id={event.deployment_id}")
# TODO: 实现监控配置逻辑
# monitoring_service.setup_deployment_monitoring(
# deployment_id=event.deployment_id,
# endpoint=event.endpoint
# )
def handle_deploy_promoted_notification(event: DeployPromoted) -> None:
"""
模型部署 -> 发送通知
"""
logger.info(f"发送部署完成通知: deployment_id={event.deployment_id}")
# TODO: 实现通知逻辑
# ==================== 事件订阅初始化 ====================
def initialize_event_handlers() -> None:
"""
初始化事件处理器订阅
在应用启动时调用此函数注册所有事件处理器
"""
# 训练完成事件 -> 多个处理器
event_publisher.subscribe(
"training.completed",
handle_training_completed_training
)
event_publisher.subscribe(
"training.completed",
handle_training_completed_notification
)
event_publisher.subscribe(
"training.completed",
handle_training_completed_evaluation
)
# 模型注册事件
event_publisher.subscribe(
"model.registered",
handle_model_registered_inference
)
# 部署完成事件
event_publisher.subscribe(
"inference.deployment.promoted",
handle_deploy_promoted_monitoring
)
event_publisher.subscribe(
"inference.deployment.promoted",
handle_deploy_promoted_notification
)
logger.info("领域事件处理器初始化完成")六、应用服务层设计
6.1 应用服务架构
应用服务层是DDD中连接领域模型和外部世界的桥梁，它负责协调多个领域对象完成业务操作，但不包含业务逻辑。
┌─────────────────────────────────────────────────────────────────────────┐
│ 应用服务层架构 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 应用服务层 (Application Services) │ │
│ │ │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ ModelAppService │ │TrainingAppService│ │InferenceAppSvc │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ │ - register() │ │ - create_task() │ │ - deploy() │ │ │
│ │ │ - evaluate() │ │ - start() │ │ - predict() │ │ │
│ │ │ - deprecate() │ │ - cancel() │ │ - rollback() │ │ │
│ │ └────────┬────────┘ └────────┬─────────┘ └────────┬────────┘ │ │
│ │ │ │ │ │ │
│ │ ▼ ▼ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ 领域服务层 (Domain Services) │ │ │
│ │ │ ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │ │ │
│ │ │ │ModelValidator │ │TrainingScheduler│ │InferenceRouter│ │ │ │
│ │ │ └───────────────┘ └───────────────┘ └───────────────┘ │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ 基础设施层 (Infrastructure) │ │ │
│ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │
│ │ │ │Repository │ │EventPublisher│ │K8sClient │ │ │ │
│ │ │ │(MySQL/Redis)│ │ │ │ │ │ │ │
│ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘6.2 应用服务实现
# application/training_service.py
# 训练管理应用服务
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
import logging
from domain.models.training import (
TrainingTask,
TrainingStatus,
TrainingHyperparameters,
GPUResourceSpec,
Checkpoint,
)
from domain.events import (
event_publisher,
TrainingStarted,
TrainingCompleted,
TrainingFailed,
CheckpointCreated,
)
from infrastructure.repositories import TrainingTaskRepository
from infrastructure.k8s import KubernetesClient
logger = logging.getLogger(__name__)
@dataclass
class CreateTrainingTaskCommand:
"""创建训练任务的命令"""
tenant_id: str
name: str
description: str
dataset_id: str
base_model_id: str
output_model_name: str
hyperparameters: Dict[str, Any]
gpu_count: int
gpu_type: str
created_by: str
@dataclass
class TrainingTaskDTO:
"""训练任务数据传输对象"""
id: str
tenant_id: str
name: str
status: str
progress: float
created_at: str
started_at: Optional[str]
completed_at: Optional[str]
final_metrics: Optional[Dict[str, float]]
class TrainingApplicationService:
"""
训练管理应用服务
负责协调训练任务的创建、启动、监控等操作
"""
def __init__(
self,
task_repository: TrainingTaskRepository,
k8s_client: KubernetesClient,
):
self.task_repository = task_repository
self.k8s_client = k8s_client
def create_task(self, command: CreateTrainingTaskCommand) -> TrainingTaskDTO:
"""
创建训练任务
流程：
1. 验证输入参数
2. 创建领域模型
3. 持久化到数据库
"""
# 1. 验证数据集和模型是否存在
# dataset = self.dataset_repository.find_by_id(command.dataset_id)
# if not dataset:
# raise ValueError(f"数据集不存在: {command.dataset_id}")
# 2. 创建领域模型
hyperparameters = TrainingHyperparameters(
learning_rate=command.hyperparameters.get('learning_rate', 1e-5),
batch_size=command.hyperparameters.get('batch_size', 8),
epochs=command.hyperparameters.get('epochs', 3),
max_seq_length=command.hyperparameters.get('max_seq_length', 2048),
warmup_steps=command.hyperparameters.get('warmup_steps', 100),
weight_decay=command.hyperparameters.get('weight_decay', 0.01),
)
resource_spec = GPUResourceSpec(
gpu_count=command.gpu_count,
gpu_type=command.gpu_type,
memory_per_gpu_gb=80 if command.gpu_type == 'A100' else 32
)
task = TrainingTask(
tenant_id=command.tenant_id,
name=command.name,
description=command.description,
dataset_id=command.dataset_id,
base_model_id=command.base_model_id,
output_model_name=command.output_model_name,
hyperparameters=hyperparameters,
resource_spec=resource_spec,
created_by=command.created_by,
)
# 3. 持久化
self.task_repository.save(task)
logger.info(f"训练任务创建成功: {task.id}")
return self._to_dto(task)
def start_task(self, task_id: str) -> TrainingTaskDTO:
"""
启动训练任务
流程：
1. 获取任务
2. 验证状态
3. 提交到K8s
4. 更新状态
5. 发布事件
"""
task = self.task_repository.find_by_id(task_id)
if not task:
raise ValueError(f"任务不存在: {task_id}")
# 验证状态
if task.status != TrainingStatus.PENDING:
raise ValueError(f"任务状态不允许启动: {task.status}")
# 提交到Kubernetes
job_name = self._submit_to_kubernetes(task)
# 更新任务状态
task.kubernetes_job_name = job_name
task.start()
self.task_repository.save(task)
# 发布领域事件
event_publisher.publish(TrainingStarted(
tenant_id=task.tenant_id,
training_task_id=task.id,
model_id=task.base_model_id,
dataset_id=task.dataset_id,
resource_spec={
'gpu_count': task.resource_spec.gpu_count,
'gpu_type': task.resource_spec.gpu_type,
} if task.resource_spec else {}
))
logger.info(f"训练任务已启动: {task_id}, job={job_name}")
return self._to_dto(task)
def handle_task_completed(
self,
task_id: str,
final_metrics: Dict[str, float]
) -> None:
"""
处理任务完成回调
由Kubernetes Operator或定时任务调用
"""
task = self.task_repository.find_by_id(task_id)
if not task:
logger.error(f"任务不存在: {task_id}")
return
# 更新任务状态
task.complete(final_metrics)
self.task_repository.save(task)
# 发布完成事件
event_publisher.publish(TrainingCompleted(
tenant_id=task.tenant_id,
training_task_id=task.id,
model_id=task.base_model_id,
output_model_version=task.output_model_name,
final_metrics=final_metrics,
checkpoint_path=task.get_latest_checkpoint().path if task.get_latest_checkpoint() else "",
duration_seconds=int(
(task.completed_at - task.started_at).total_seconds()
) if task.started_at and task.completed_at else 0
))
logger.info(f"训练任务完成处理: {task_id}")
def cancel_task(self, task_id: str) -> TrainingTaskDTO:
"""取消训练任务"""
task = self.task_repository.find_by_id(task_id)
if not task:
raise ValueError(f"任务不存在: {task_id}")
# 取消K8s Job
if task.kubernetes_job_name:
self.k8s_client.delete_job(
namespace=task.kubernetes_namespace or task.tenant_id,
job_name=task.kubernetes_job_name
)
# 更新状态
task.cancel()
self.task_repository.save(task)
logger.info(f"训练任务已取消: {task_id}")
return self._to_dto(task)
def get_task(self, task_id: str) -> Optional[TrainingTaskDTO]:
"""获取任务详情"""
task = self.task_repository.find_by_id(task_id)
return self._to_dto(task) if task else None
def list_tasks(
self,
tenant_id: str,
status: Optional[str] = None,
limit: int = 20,
offset: int = 0
) -> List[TrainingTaskDTO]:
"""查询任务列表"""
tasks = self.task_repository.find_by_tenant(
tenant_id=tenant_id,
status=status,
limit=limit,
offset=offset
)
return [self._to_dto(task) for task in tasks]
def _submit_to_kubernetes(self, task: TrainingTask) -> str:
"""提交训练任务到Kubernetes"""
job_manifest = {
'apiVersion': 'batch/v1',
'kind': 'Job',
'metadata': {
'name': f'training-{task.id[:8]}',
'namespace': task.tenant_id
},
'spec': {
'backoffLimit': 3,
'template': {
'spec': {
'restartPolicy': 'OnFailure',
'containers': [{
'name': 'training',
'image': 'your-registry.com/training:latest',
'env': [
{'name': 'TASK_ID', 'value': task.id},
{'name': 'DATASET_ID', 'value': task.dataset_id},
{'name': 'BASE_MODEL_ID', 'value': task.base_model_id},
],
'resources': {
'limits': {
'nvidia.com/gpu': str(
task.resource_spec.gpu_count
) if task.resource_spec else '1'
}
}
}]
}
}
}
}
# 使用K8s Client提交
# result = self.k8s_client.create_job(job_manifest)
# return result['metadata']['name']
# 模拟返回
return f'training-{task.id[:8]}'
def _to_dto(self, task: TrainingTask) -> TrainingTaskDTO:
"""领域模型转DTO"""
return TrainingTaskDTO(
id=task.id,
tenant_id=task.tenant_id,
name=task.name,
status=task.status.value,
progress=task.progress,
created_at=task.created_at.isoformat(),
started_at=task.started_at.isoformat() if task.started_at else None,
completed_at=task.completed_at.isoformat() if task.completed_at else None,
final_metrics=task.final_metrics
)七、CQRS在查询场景的应用
7.1 为什么AI平台需要CQRS
在AI平台中，命令操作（写）和查询操作（读）有明显的特点差异：
命令操作：模型注册、训练任务创建、部署等，写入频率相对较低，但一致性要求高
查询操作：模型列表、训练历史、监控指标等，读取频率很高，需要快速响应
CQRS（Command Query Responsibility Segregation）将命令和查询分离，使用不同的数据模型和数据存储来优化各自的场景。
┌─────────────────────────────────────────────────────────────────────────┐
│ CQRS 架构 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 命令端 (Command) │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Model │ │ Training │ │ Inference │ │ │
│ │ │ Command │ │ Command │ │ Command │ │ │
│ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │
│ │ │ │ │ │ │
│ │ ▼ ▼ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ 领域模型 (写入模型) │ │ │
│ │ │ - 聚合根管理业务规则 │ │ │
│ │ │ - 丰富的领域逻辑 │ │ │
│ │ │ - 强一致性 │ │ │
│ │ └─────────────────────────────┬───────────────────────────┘ │ │
│ │ │ │ │
│ │ write │ │ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ MySQL (写库) │ │ │
│ │ │ - 范式化设计 │ │ │
│ │ │ - 支持事务 │ │ │
│ │ │ - 复杂查询支持 │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ │ event │ │
│ │ │ 同步/异步 │
│ │ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 查询端 (Query) │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Model │ │ Training │ │ Dashboard │ │ │
│ │ │ Query │ │ Query │ │ Query │ │ │
│ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │
│ │ │ │ │ │ │
│ │ ▼ ▼ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Read Model (读取模型) │ │ │
│ │ │ - 反范式化设计 │ │ │
│ │ │ - 针对查询优化 │ │ │
│ │ │ - 可用性优先 │ │ │
│ │ └─────────────────────────────┬───────────────────────────┘ │ │
│ │ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Elasticsearch/Redis (读库) │ │ │
│ │ │ - 全文搜索支持 │ │ │
│ │ │ - 高速缓存 │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘7.2 读写分离的查询服务实现
# application/query_service.py
# CQRS 查询端实现
from typing import List, Optional, Dict, Any
from dataclasses import dataclass
from datetime import datetime
import logging
from infrastructure.query_db import QueryDatabase
logger = logging.getLogger(__name__)
@dataclass
class ModelListItem:
"""模型列表项（优化过的读取模型）"""
model_id: str
tenant_id: str
name: str
description: str
latest_version: str
status: str
base_model: str
tags: List[str]
created_at: datetime
updated_at: datetime
# 预聚合的统计信息
version_count: int
total_size_bytes: int
evaluation_count: int
avg_score: Optional[float]
# 用于列表展示的简短信息
short_description: str
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'ModelListItem':
"""从字典创建"""
return cls(
model_id=data['model_id'],
tenant_id=data['tenant_id'],
name=data['name'],
description=data['description'],
latest_version=data['latest_version'],
status=data['status'],
base_model=data['base_model'],
tags=data.get('tags', []),
created_at=data['created_at'],
updated_at=data['updated_at'],
version_count=data.get('version_count', 0),
total_size_bytes=data.get('total_size_bytes', 0),
evaluation_count=data.get('evaluation_count', 0),
avg_score=data.get('avg_score'),
short_description=data['description'][:100] + '...' if len(data['description']) > 100 else data['description']
)
@dataclass
class TrainingTaskSummary:
"""训练任务汇总（读取模型）"""
task_id: str
tenant_id: str
name: str
status: str
progress: float
created_at: datetime
# 关联信息
dataset_name: str
base_model_name: str
# 时间信息
duration_seconds: Optional[int]
started_at: Optional[datetime]
completed_at: Optional[datetime]
# 关键指标
current_step: int
loss: Optional[float]
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'TrainingTaskSummary':
started = data.get('started_at')
completed = data.get('completed_at')
duration = None
if started and completed:
duration = int((completed - started).total_seconds())
return cls(
task_id=data['task_id'],
tenant_id=data['tenant_id'],
name=data['name'],
status=data['status'],
progress=data.get('progress', 0.0),
created_at=data['created_at'],
dataset_name=data.get('dataset_name', ''),
base_model_name=data.get('base_model_name', ''),
duration_seconds=duration,
started_at=started,
completed_at=completed,
current_step=data.get('current_step', 0),
loss=data.get('loss')
)
class ModelQueryService:
"""
模型查询服务（Read Side）
负责处理所有模型相关的查询请求
使用专门的读取数据库优化查询性能
"""
def __init__(self, query_db: QueryDatabase):
self.query_db = query_db
def search_models(
self,
tenant_id: str,
keyword: Optional[str] = None,
tags: Optional[List[str]] = None,
status: Optional[str] = None,
sort_by: str = 'updated_at',
sort_order: str = 'desc',
limit: int = 20,
offset: int = 0
) -> List[ModelListItem]:
"""
搜索模型列表
支持：
- 关键词全文搜索
- 标签过滤
- 状态过滤
- 排序
- 分页
"""
query = f"""
SELECT
m.model_id,
m.tenant_id,
m.name,
m.description,
m.latest_version,
m.status,
m.base_model,
m.tags,
m.created_at,
m.updated_at,
COUNT(DISTINCT mv.version_id) as version_count,
SUM(mv.size_bytes) as total_size_bytes,
COUNT(DISTINCT e.evaluation_id) as evaluation_count,
AVG(e.score) as avg_score
FROM models m
LEFT JOIN model_versions mv ON m.model_id = mv.model_id
LEFT JOIN evaluations e ON m.model_id = e.model_id
WHERE m.tenant_id = %s
"""
params = [tenant_id]
# 关键词搜索
if keyword:
query += " AND (m.name LIKE %s OR m.description LIKE %s)"
params.extend([f'%{keyword}%', f'%{keyword}%'])
# 标签过滤
if tags:
# 使用JSON数组包含检查
for tag in tags:
query += " AND m.tags @> %s"
params.append(f'["{tag}"]')
# 状态过滤
if status:
query += " AND m.status = %s"
params.append(status)
# 分组和排序
query += f"""
GROUP BY m.model_id
ORDER BY m.{sort_by} {sort_order}
LIMIT %s OFFSET %s
"""
params.extend([limit, offset])
results = self.query_db.execute(query, params)
return [ModelListItem.from_dict(row) for row in results]
def get_model_detail(self, model_id: str) -> Optional[Dict[str, Any]]:
"""
获取模型详情
返回包含完整信息的读取模型
"""
# 预编译的查询，一次性获取所有需要的信息
query = """
SELECT
m.*,
-- 版本列表（JSON聚合）
COALESCE(JSON_AGG(
JSON_BUILD_OBJECT(
'version_id', mv.version_id,
'version', mv.version,
'status', mv.status,
'parameter_count', mv.parameter_count,
'size_bytes', mv.size_bytes,
'created_at', mv.created_at
)
) FILTER (WHERE mv.version_id IS NOT NULL), '[]') as versions,
-- 最新版本的指标
(
SELECT JSON_BUILD_OBJECT(
'bleu_score', mv2.bleu_score,
'rouge_score', mv2.rouge_score,
'accuracy', mv2.accuracy
)
FROM model_versions mv2
WHERE mv2.model_id = m.model_id
ORDER BY mv2.created_at DESC
LIMIT 1
) as latest_metrics,
-- 训练历史（最近10条）
(
SELECT COALESCE(JSON_AGG(t.*), '[]')
FROM (
SELECT
tt.task_id,
tt.name,
tt.status,
tt.progress,
tt.created_at,
tt.started_at,
tt.completed_at
FROM training_tasks tt
WHERE tt.base_model_id = m.model_id
ORDER BY tt.created_at DESC
LIMIT 10
) t
) as training_history
FROM models m
WHERE m.model_id = %s
GROUP BY m.model_id
"""
results = self.query_db.execute(query, [model_id])
if not results:
return None
return results[0]
def get_model_versions(self, model_id: str) -> List[Dict[str, Any]]:
"""获取模型版本列表"""
query = """
SELECT
mv.*,
e.metrics,
e.status as eval_status
FROM model_versions mv
LEFT JOIN evaluations e ON mv.version_id = e.model_version_id
WHERE mv.model_id = %s
ORDER BY mv.created_at DESC
"""
return self.query_db.execute(query, [model_id])
class DashboardQueryService:
"""
仪表盘查询服务
提供聚合后的统计数据，用于仪表盘展示
"""
def __init__(self, query_db: QueryDatabase):
self.query_db = query_db
def get_tenant_overview(self, tenant_id: str) -> Dict[str, Any]:
"""
获取租户概览统计
返回聚合后的统计数据
"""
query = """
SELECT
-- 模型统计
COUNT(DISTINCT m.model_id) as total_models,
COUNT(DISTINCT CASE WHEN m.status = 'deployed' THEN m.model_id END) as deployed_models,
-- 训练统计
COUNT(DISTINCT tt.task_id) as total_training_tasks,
COUNT(DISTINCT CASE WHEN tt.status = 'running' THEN tt.task_id END) as running_tasks,
SUM(CASE WHEN tt.status = 'succeeded' THEN 1 ELSE 0 END) as completed_tasks,
-- 资源使用
SUM(CASE WHEN tt.status = 'running' THEN tt.requested_gpu_hours ELSE 0 END) as current_gpu_usage,
-- 最近7天活跃度
COUNT(DISTINCT CASE WHEN tt.created_at >= NOW() - INTERVAL '7 days'
THEN tt.task_id END) as tasks_last_7_days
FROM models m
CROSS JOIN training_tasks tt
WHERE m.tenant_id = %s AND tt.tenant_id = %s
"""
results = self.query_db.execute(query, [tenant_id, tenant_id])
return results[0] if results else {}
def get_training_trend(
self,
tenant_id: str,
days: int = 30
) -> List[Dict[str, Any]]:
"""
获取训练趋势数据
按天聚合训练任务数量和成功率
"""
query = """
SELECT
DATE_TRUNC('day', created_at) as date,
COUNT(*) as total_tasks,
COUNT(CASE WHEN status = 'succeeded' THEN 1 END) as succeeded,
COUNT(CASE WHEN status = 'failed' THEN 1 END) as failed,
AVG(progress) as avg_progress
FROM training_tasks
WHERE tenant_id = %s
AND created_at >= NOW() - INTERVAL '%s days'
GROUP BY DATE_TRUNC('day', created_at)
ORDER BY date
"""
return self.query_db.execute(query, [tenant_id, days])八、项目结构
8.1 DDD项目结构设计
以下是完整的Django DDD项目结构：
ai_platform/
├── domain/ # 领域层 - 核心业务逻辑
│ ├── __init__.py
│ ├── models/ # 领域模型
│ │ ├── __init__.py
│ │ ├── model_management.py # 模型管理领域模型
│ │ ├── training.py # 训练管理领域模型
│ │ ├── inference.py # 推理服务领域模型
│ │ └── data.py # 数据管理领域模型
│ ├── events.py # 领域事件定义
│ ├── repositories.py # 仓储接口（抽象）
│ └── services.py # 领域服务
│
├── application/ # 应用层 - 用例编排
│ ├── __init__.py
│ ├── commands/ # 命令（写操作）
│ │ ├── __init__.py
│ │ ├── training_commands.py
│ │ └── model_commands.py
│ ├── queries/ # 查询（读操作）
│ │ ├── __init__.py
│ │ ├── training_queries.py
│ │ └── model_queries.py
│ └── services/ # 应用服务
│ ├── __init__.py
│ ├── training_service.py
│ └── model_service.py
│
├── infrastructure/ # 基础设施层
│ ├── __init__.py
│ ├── repositories/ # 仓储实现
│ │ ├── __init__.py
│ │ ├── mysql_repository.py
│ │ └── elasticsearch_repository.py
│ ├── k8s/ # Kubernetes集成
│ │ ├── __init__.py
│ │ └── client.py
│ ├── messaging/ # 消息队列
│ │ ├── __init__.py
│ │ └── event_bus.py
│ └── query_db.py # 查询数据库
│
├── api/ # 接口层
│ ├── __init__.py
│ ├── views/ # Django视图
│ │ ├── __init__.py
│ │ ├── training_views.py
│ │ └── model_views.py
│ ├── serializers.py # 序列化器
│ └── urls.py # 路由
│
├── core/ # 核心模块
│ ├── __init__.py
│ ├── aggregates.py # 聚合工厂
│ ├── value_objects.py # 值对象定义
│ └── exceptions.py # 领域异常
│
└── tests/ # 测试
├── __init__.py
├── domain/
├── application/
└── integration/九、总结
9.1 DDD实践要点
通过在AI算法平台后端实践DDD，我们总结了以下关键要点：
限界上下文划分是第一步：清晰的限界上下文划分是DDD成功的关键。我们根据业务能力边界将平台划分为模型管理、训练管理、推理服务、数据管理、用户管理等多个上下文，每个上下文有清晰的职责和统一的语言。
聚合设计要保持精简：聚合是DDD中最核心的战术设计元素。我们坚持"聚合要小"的原则，每个聚合只包含最核心的实体和值对象，通过聚合根控制对外暴露的接口。
领域事件实现解耦：通过领域事件机制，不同限界上下文可以松耦合地通信。当训练完成时，会自动触发模型注册、评测、通知等多个下游操作，整个流程清晰且易于维护。
CQRS优化查询性能：在AI平台中，查询场景远多于写入场景。通过CQRS分离读写模型，我们可以针对不同场景进行专门优化，既保证了写入的一致性，又提升了查询的性能。
9.2 踩坑经验
教训一：不要过早细化。刚开始实施DDD时，我们试图把每个领域的每个概念都建模得非常精细，结果导致领域模型过于复杂，难以维护。后来我们采用了"先跑通核心流程，再逐步演进"的策略，效果好得多。
教训二：防腐层必不可少。在对接外部系统（如Kubernetes、对象存储）时，一定要建立防腐层，将外部系统的概念转换为领域概念，避免外部变化影响核心领域。
教训三：团队DDD能力建设很重要。DDD对团队的设计能力有较高要求。我们通过内部培训、代码审查、结对编程等方式，逐步提升了团队的DDD实践能力。
DDD不是银弹，但它确实是处理复杂业务领域的有力工具。希望本文的实践分享能为正在进行类似架构转型的团队提供一些参考。
评论区