CQRS for AI Systems

Apply Command Query Responsibility Segregation to AI systems - separate write models (commands) from read models (queries) for scalable, auditable ML architectures.

What is CQRS?

CQRS separates the write side (commands that change state) from the read side (queries that return data). For AI systems, this means separating the data ingestion/training pipeline from the prediction/serving pipeline - a natural fit since these have fundamentally different performance requirements.

CQRS Applied to AI

CQRS + Event Sourcing for ML

# Command side: process events and update state
class MLCommandHandler:
    def __init__(self):
        self.event_store = EventStore()
        self.feature_computer = FeatureComputer()

    async def handle_user_interaction(self, command):
        # Validate command
        event = UserInteractionEvent(
            user_id=command.user_id,
            action=command.action,
            item_id=command.item_id,
            timestamp=datetime.utcnow(),
        )

        # Store event (source of truth)
        await self.event_store.append(event)

        # Update materialized features
        await self.feature_computer.update(event)

        # Publish for downstream consumers
        await self.event_bus.publish("user-interactions", event)


# Query side: serve predictions from optimized read models
class MLQueryHandler:
    def __init__(self):
        self.feature_store = OnlineFeatureStore()  # Redis-backed
        self.model = load_model("recommendation-v3")

    async def get_recommendations(self, query):
        # Read from optimized read model
        features = await self.feature_store.get(query.user_id)
        predictions = self.model.predict(features)
        return RankingResponse(items=predictions.top_k(10))

Projections for ML

Projections build read-optimized views from event streams. For ML, projections create materialized feature tables:

# Projection: build user feature table from events
class UserFeatureProjection:
    """Reads user events and maintains a materialized feature table"""

    async def process(self, event):
        if event.type == "UserInteractionEvent":
            user_features = await self.read_store.get(event.user_id)

            # Update features
            user_features.total_interactions += 1
            user_features.last_active = event.timestamp
            user_features.category_counts[event.category] += 1

            # Compute derived features
            user_features.engagement_score = compute_engagement(
                user_features.total_interactions,
                user_features.session_count,
                user_features.last_active,
            )

            # Write to read-optimized store
            await self.read_store.set(event.user_id, user_features)

Benefits of CQRS for AI