System Architecture: Phoenix ML Platform

1. Overview

Phoenix ML is an end-to-end MLOps system built with Domain-Driven Design (DDD) and Clean Architecture. It provides:

• Real-time ML inference (REST + gRPC)
• Automatic data drift and anomaly detection
• Self-healing: alert → rollback → retrain
• Multi-model serving with A/B testing and canary deployment
• Full observability stack (metrics, tracing, logging)

2. Architecture Diagram

2.1 Overall System Architecture

graph TD
    Client["External Client"] -->|REST/gRPC| API["Phoenix API Gateway<br/>FastAPI + gRPC"]

    subgraph "Inference Core"
        API --> Handler["PredictHandler<br/>CQRS Command Handler"]
        Handler --> Pipeline["Request Pipeline<br/>Chain of Responsibility"]
        Pipeline --> Router["Routing Strategy<br/>A/B · Canary · Shadow"]
        Router --> CB["Circuit Breaker<br/>Closed/Open/Half-Open"]
        CB --> Engines

        subgraph Engines["Model Executors"]
            ONNX["ONNX Runtime<br/>(Default - CPU/GPU)"]
            TRT["TensorRT Executor<br/>(FP16 GPU)"]
            Triton["Triton Client<br/>(HTTP v2 API)"]
        end
    end

    subgraph "Data Layer"
        Handler -->|"HMGET/HSET"| Redis["Redis<br/>(Online Feature Store)"]
        Handler -->|Read| Parquet["Parquet Files<br/>(Offline Features)"]
        API -->|"SQLAlchemy async"| Postgres["PostgreSQL<br/>(Models, Logs, Reports)"]
        API -->|"boto3"| MinIO["MinIO / S3<br/>(Model Artifacts)"]
    end

    subgraph "Event Streaming"
        Handler -.->|"JSON publish"| Kafka{"Apache Kafka<br/>(KRaft mode)"}
        Kafka -->|"Consumer group"| Logger["Prediction Logger"]
        Kafka -->|"Consumer group"| Analytics["Stream Analytics"]
    end

    subgraph "Self-Healing Subsystem"
        Monitor["MonitoringService<br/>(Background loop)"] --> Drift["DriftCalculator<br/>(KS · PSI · Chi2)"]
        Monitor --> Anomaly["AnomalyDetector<br/>(Z-score · IQR)"]
        Monitor --> Alert["AlertManager<br/>(Rules + Cooldown)"]
        Alert --> Notifier["AlertNotifier<br/>(Slack / Discord webhook)"]
        Alert --> Rollback["RollbackManager<br/>(Auto-revert model)"]
    end

    subgraph "Observability"
        API -->|"/metrics"| Prometheus["Prometheus<br/>(Scrape metrics)"]
        Prometheus --> Grafana["Grafana<br/>(Dashboards)"]
        API -->|"OTLP export"| Jaeger["Jaeger<br/>(Distributed Tracing)"]
    end

    subgraph "Frontend"
        Dashboard["React Dashboard<br/>(TypeScript + Recharts)"]
        Dashboard -->|"fetch API"| API
        Dashboard -->|"iframe embed"| Grafana
    end

    subgraph "MLOps Pipeline"
        Airflow["Apache Airflow<br/>(Scheduler)"] -->|"DAG tasks"| Train["Train Models<br/>(sklearn → ONNX)"]
        Train -->|"MLflow log"| MLflow["MLflow<br/>(Experiment Tracking)"]
    end

2.2 Request Flow (Single Prediction)

sequenceDiagram
    participant C as Client
    participant API as FastAPI Router
    participant PH as PredictHandler
    participant FS as FeatureStore
    participant IS as InferenceService
    participant RS as RoutingStrategy
    participant CB as CircuitBreaker
    participant IE as ONNX Engine
    participant EB as EventBus
    participant KF as Kafka
    participant PG as PostgreSQL

    C->>API: POST /predict {model_id, features}
    API->>PH: handle(PredictCommand)

    opt entity_id provided (no raw features)
        PH->>FS: get_online_features(entity_id)
        FS-->>PH: [0.5, 1.2, 0.8, ...]
    end

    PH->>IS: predict(model, features)
    IS->>RS: select_model(champion, challenger)
    RS-->>IS: selected_model (based on strategy)
    IS->>CB: check_state()

    alt Circuit CLOSED
        IS->>IE: predict(model, feature_vector)
        IE-->>IS: Prediction(result, confidence, latency)
    else Circuit OPEN
        IS-->>PH: raise CircuitOpenError
    end

    IS-->>PH: Prediction

    par Async Task 1
        PH->>EB: publish(PredictionCompleted)
    and Async Task 2
        EB->>KF: publish("inference-events", event)
    and Async Task 3
        PH->>PG: log_prediction(command, prediction)
    end

    PH-->>API: PredictionResponse
    API-->>C: 200 OK {result, confidence, latency_ms}

3. Layer Architecture

Phoenix ML uses a layered architecture — dependency rule: outer layers depend on inner layers, never the reverse.

graph TD
    subgraph Infrastructure
        I["FastAPI · gRPC · PostgreSQL · Redis · Kafka<br/>ONNX Runtime · Prometheus · S3 · MLflow"]
    end
    subgraph Application
        A["PredictHandler · BatchPredictHandler<br/>MonitoringService · Commands · Queries · DTOs"]
    end
    subgraph Domain
        D["Model · Prediction · InferenceService<br/>DriftCalculator · AlertManager · RoutingStrategy<br/>CircuitBreaker · FeatureStore (ABC) · EventBus"]
    end
    subgraph Shared
        S["Exceptions · Interfaces · Utilities"]
    end
    Infrastructure --> Application --> Domain --> Shared

3.1 Dependency Rule

graph LR
    Infrastructure --> Application --> Domain --> Shared

• Domain Layer does NOT import FastAPI, SQLAlchemy, Redis, Kafka, or ONNX Runtime
• Application Layer only imports Domain entities/services
• Infrastructure Layer implements Domain ABCs (interfaces)

3.2 Source Code Structure

phoenix_ml/
├── config/                         # 5 files — Settings (Pydantic, reads .env)
│   ├── app.py                      #   APP_VERSION, DEFAULT_MODEL_ID
│   ├── inference.py                #   BATCH_MAX_SIZE, INFERENCE_ENGINE
│   ├── infrastructure.py           #   DATABASE_URL, KAFKA_URL, REDIS_URL
│   └── monitoring.py               #   MONITORING_INTERVAL_SECONDS
│
├── domain/                         # 40+ files — Pure business logic
│   ├── inference/                  #   ML prediction bounded context
│   │   ├── entities/               #     Model, Prediction
│   │   ├── value_objects/          #     ConfidenceScore, FeatureVector, LatencyBudget
│   │   ├── services/               #     InferenceEngine, InferenceService, BatchManager
│   │   │                           #     RoutingStrategy, CircuitBreaker, RequestPipeline
│   │   └── events/                 #     ModelLoaded, PredictionMade
│   ├── monitoring/                 #   Drift & alerting bounded context
│   │   ├── entities/               #     DriftReport
│   │   ├── services/               #     DriftCalculator, AlertManager, AnomalyDetector
│   │   │                           #     MetricsPublisher, ModelEvaluator, RollbackManager
│   │   └── repositories/          #     DriftReportRepository, PredictionLogRepository
│   ├── training/                   #   Training bounded context
│   │   ├── entities/               #     TrainingConfig, TrainingJob
│   │   ├── services/               #     TrainingService, IDataLoader, ITrainer
│   │   └── events/                 #     TrainingCompleted
│   ├── feature_store/              #   Feature management bounded context
│   │   ├── entities/               #     FeatureRegistry
│   │   └── repositories/          #     FeatureStore, OfflineFeatureStore
│   ├── model_registry/            #   Model versioning bounded context
│   │   └── repositories/          #     ModelRepository, ArtifactStorage
│   └── shared/                     #   Shared kernel
│       ├── domain_events.py        #     PredictionCompleted, DriftDetected, ModelRetrained
│       ├── event_bus.py            #     DomainEventBus (Observer pattern)
│       └── plugin_registry.py     #     PluginRegistry (pre/post processors)
│
├── application/                    # 14 files — Use cases (CQRS)
│   ├── commands/                   #   PredictCommand, BatchPredictCommand, LoadModelCommand
│   ├── handlers/                   #   PredictHandler, BatchPredictHandler, RetrainHandler
│   │                               #   QueryHandlers (GetModel, GetDrift, GetPerformance)
│   ├── services/                   #   MonitoringService
│   ├── dto/                        #   PredictionRequest, PredictionResponse
│   └── decorators.py              #   @timing, @retry decorators
│
├── infrastructure/                 # 37 files — Framework adapters
│   ├── bootstrap/                  #   container.py (DI), lifespan.py (startup/shutdown)
│   ├── http/                       #   FastAPI routes, dependencies
│   ├── grpc/                       #   gRPC server + proto definitions
│   ├── ml_engines/                 #   ONNX, TensorRT, Triton, Mock
│   ├── messaging/                  #   Kafka producer + consumer
│   ├── feature_store/              #   Redis, InMemory, Parquet
│   ├── persistence/                #   PostgreSQL repos, MLflow, InMemory
│   ├── monitoring/                 #   Prometheus, Jaeger, AlertNotifier
│   ├── artifact_storage/          #   Local, S3/MinIO
│   └── data_loaders/             #   Tabular (CSV), Image (NPZ/dir)
│
└── shared/                         # 9 files — Cross-cutting concerns
    ├── exceptions/                 #   ModelNotFoundError, InferenceError, etc.
    ├── ingestion/                  #   API/Redis data ingestors, DataCollector
    └── utils/                      #   model_generator (ONNX for CI)

4. Design Patterns

4.1 Pattern Catalog

4.2 Strategy Pattern — Routing

classDiagram
    class RoutingStrategy {
        <<abstract>>
        +select(champion, challengers) Model
    }
    class SingleModelStrategy {
        +select() → always champion
    }
    class ABTestStrategy {
        -traffic_ratio: float
        +select() → random split
    }
    class CanaryStrategy {
        -canary_percentage: float
        +select() → % small to challenger
    }
    class ShadowStrategy {
        +select() → mirror to both
    }

    RoutingStrategy <|-- SingleModelStrategy
    RoutingStrategy <|-- ABTestStrategy
    RoutingStrategy <|-- CanaryStrategy
    RoutingStrategy <|-- ShadowStrategy

4.3 Circuit Breaker — State Machine

stateDiagram-v2
    [*] --> CLOSED
    CLOSED --> OPEN: failure_count > threshold
    OPEN --> HALF_OPEN: after timeout period
    HALF_OPEN --> CLOSED: success
    HALF_OPEN --> OPEN: failure
    CLOSED --> CLOSED: success (reset counter)

5. Self-Healing Flow

5.1 Detection → Response Pipeline

sequenceDiagram
    participant MLoop as Monitoring Loop (Background)
    participant DC as DriftCalculator
    participant AD as AnomalyDetector
    participant AM as AlertManager
    participant AN as AlertNotifier (Slack)
    participant RM as RollbackManager
    participant EB as DomainEventBus
    participant AF as Airflow DAG

    MLoop->>DC: calculate_drift(reference, current)
    DC-->>MLoop: DriftReport(score=0.45, is_drifted=true)

    MLoop->>AD: detect_anomaly(recent_latencies)
    AD-->>MLoop: anomaly_indices=[42, 67, 89]

    MLoop->>AM: evaluate(rules, drift_score=0.45)

    alt score > 0.3 (CRITICAL threshold)
        AM->>AN: notify(Alert: "high_drift_score", severity=CRITICAL)
        AN->>AN: POST webhook → Slack
        AM->>RM: evaluate_rollback(model_id)
        RM->>RM: archive challengers, keep champion
    else score > 0.1 (WARNING threshold)
        AM->>AN: notify(Alert: "moderate_drift", severity=WARNING)
    end

    MLoop->>EB: publish(DriftDetected)
    EB->>AF: trigger retrain_pipeline DAG

5.2 Multi-Model Monitoring

The monitoring loop runs for all models configured in model_configs/:

• Each model has its own reference data (models/<id>/v1/reference_data.json)
• Each model has its own drift test type (ks, psi, chi2 — from YAML config)
• Monitoring interval is configured via MONITORING_INTERVAL_SECONDS (default: 30 s in production)

6. Infrastructure Services

6.1 Docker Compose Stack

6.2 Airflow Stack (docker-compose.airflow.yaml)

7. Data Flow

7.1 Prediction Data Flow

graph TD
    Client["Client Request"] --> API["FastAPI Router (routes.py)"]
    API --> Handler["PredictHandler (application/handlers/predict_handler.py)"]

    Handler --> Features{"Feature Retrieval"}
    Features -->|entity_id provided| FS["FeatureStore.get(id)"]
    Features -->|raw features| Inference["InferenceService"]
    FS --> Inference

    Inference --> Router["RoutingStrategy → select model"]
    Router --> CB{"CircuitBreaker"}
    CB -->|Proceed| Engine["InferenceEngine.predict(model, features)"]
    CB -->|OPEN| Error["Return 503 Error"]
    Engine --> Prediction["Prediction Result"]

    Prediction --> BG["Background Tasks (async)"]
    BG -.-> PG["Log to PostgreSQL"]
    BG -.-> Kafka["Publish to Kafka 'inference-events'"]
    BG -.-> Prom["Publish Prometheus metrics"]
    BG -.-> EB["Update EventBus subscribers"]

    Prediction --> Return["Return PredictionResponse to Client"]

7.2 Training Data Flow

graph TD
    DAG["Airflow DAG (dags/retrain_pipeline.py)"] --> Gen["generate_datasets.py"]
    Gen -->|"data/{model}/dataset.csv or .npz"| Train["examples/{model}/train.py"]

    Train --> Loader["DataLoader"]
    Loader --> Fit["sklearn/xgboost model → ONNX export"]
    Fit --> Save["models/{model}/v1/model.onnx + metrics.json"]

    Save --> Flow["MLflow: log params, metrics, artifacts"]
    Save --> PG["PostgreSQL: register model version"]
    PG --> Promote["Promote to champion stage"]

    subgraph "Self-Healing Retrain"
        Drift["Drift detected"] --> Export["POST /data/export-training"]
        Export --> Query["Query prediction_logs WHERE ground_truth = true"]
        Query --> Merge["Merge: baseline + fresh dataset"]
        Merge --> ReTrain["examples/{model}/train.py --data combined.csv"]
        ReTrain --> Register["POST /models/register (challenger)"]
    end

8. Test Architecture

8.1 Test Pyramid

8.2 Coverage

8.3 Quality Gates (CI)

uv run ruff check .           # Lint: 0 errors
uv run ruff format --check .  # Format: 243 files consistent
uv run mypy phoenix_ml/               # Type check: 0 issues
uv run pytest tests/           # Tests: all pass, 87% coverage
npx tsc --noEmit               # Frontend type check: 0 errors
npx eslint phoenix_ml/                # Frontend lint: 0 errors
npx vitest run                 # Frontend tests: 104/104 pass

Document Status: v4.0 — Updated March 2026