AI-Powered Notifications: Intelligent Alert Prioritization and Delivery

The Notification Overload Problem

SaaS products generate an enormous volume of notifications: task assignments, status changes, comments, system alerts, billing reminders, and feature announcements. When everything is treated as equally important, users either enable all notifications and get overwhelmed, or disable them and miss critical alerts.

AI-powered notifications solve this by scoring each notification for importance, choosing the right delivery channel, and bundling related alerts into digestible summaries.

Notification Scoring Engine

The scoring engine assigns an importance score to each notification based on the event type, the user's relationship to the event, and historical engagement patterns.

flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus<br/>classify"]
    PLAN["Plan and tool<br/>selection"]
    AGENT["Agent loop<br/>LLM plus tools"]
    GUARD{"Guardrails<br/>and policy"}
    EXEC["Execute and<br/>verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus<br/>next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff

from dataclasses import dataclass
from enum import Enum
from datetime import datetime

class NotificationChannel(str, Enum):
    IN_APP = "in_app"
    EMAIL = "email"
    PUSH = "push"
    SMS = "sms"
    SLACK = "slack"

@dataclass
class Notification:
    id: str
    user_id: str
    tenant_id: str
    event_type: str       # e.g., "task_assigned", "comment_mention", "deal_closed"
    title: str
    body: str
    entity_type: str
    entity_id: str
    actor_id: str | None  # Who triggered the event
    created_at: datetime
    metadata: dict

class NotificationScorer:
    # Base importance scores by event type
    BASE_SCORES = {
        "task_assigned": 0.8,
        "task_due_soon": 0.9,
        "task_overdue": 1.0,
        "comment_mention": 0.85,
        "comment_reply": 0.6,
        "deal_closed": 0.7,
        "deal_stage_changed": 0.5,
        "system_maintenance": 0.4,
        "feature_announcement": 0.2,
        "weekly_digest": 0.3,
    }

    def __init__(self, db):
        self.db = db

    async def score(self, notification: Notification) -> float:
        base = self.BASE_SCORES.get(notification.event_type, 0.5)

        # Boost if the actor is someone the user frequently interacts with
        relationship_boost = await self.get_relationship_boost(
            notification.user_id, notification.actor_id
        )

        # Boost if the entity is something the user recently worked on
        recency_boost = await self.get_recency_boost(
            notification.user_id, notification.entity_type,
            notification.entity_id
        )

        # Penalize if the user typically ignores this event type
        engagement_factor = await self.get_engagement_factor(
            notification.user_id, notification.event_type
        )

        score = (base + relationship_boost + recency_boost) * engagement_factor
        return min(max(score, 0.0), 1.0)  # Clamp to [0, 1]

    async def get_relationship_boost(self, user_id: str,
                                      actor_id: str | None) -> float:
        if not actor_id:
            return 0.0
        interaction_count = await self.db.fetchval("""
            SELECT COUNT(*) FROM user_interactions
            WHERE user_id = $1 AND other_user_id = $2
              AND created_at > NOW() - INTERVAL '30 days';
        """, user_id, actor_id)
        if interaction_count > 20:
            return 0.15
        if interaction_count > 5:
            return 0.08
        return 0.0

    async def get_recency_boost(self, user_id: str, entity_type: str,
                                 entity_id: str) -> float:
        last_access = await self.db.fetchval("""
            SELECT MAX(accessed_at) FROM user_activity
            WHERE user_id = $1 AND entity_type = $2 AND entity_id = $3;
        """, user_id, entity_type, entity_id)
        if not last_access:
            return 0.0
        hours_since = (datetime.utcnow() - last_access).total_seconds() / 3600
        if hours_since < 1:
            return 0.15
        if hours_since < 24:
            return 0.08
        return 0.0

    async def get_engagement_factor(self, user_id: str,
                                     event_type: str) -> float:
        stats = await self.db.fetchrow("""
            SELECT COUNT(*) as total,
                   COUNT(*) FILTER (WHERE read_at IS NOT NULL) as read_count
            FROM notifications
            WHERE user_id = $1 AND event_type = $2
              AND created_at > NOW() - INTERVAL '90 days';
        """, user_id, event_type)
        if not stats or stats["total"] == 0:
            return 1.0  # No history, use default
        read_rate = stats["read_count"] / stats["total"]
        return 0.3 + (0.7 * read_rate)  # Floor at 0.3 to never fully suppress

Channel Selection

The delivery channel depends on the notification score and the user's current availability.

class ChannelSelector:
    def __init__(self, db):
        self.db = db

    async def select_channels(self, notification: Notification,
                               score: float) -> list[NotificationChannel]:
        prefs = await self.get_user_preferences(notification.user_id)
        channels = []

        # Always deliver in-app
        channels.append(NotificationChannel.IN_APP)

        # Critical notifications: push + email
        if score >= 0.9:
            if prefs.get("push_enabled", True):
                channels.append(NotificationChannel.PUSH)
            if prefs.get("email_enabled", True):
                channels.append(NotificationChannel.EMAIL)

        # Important notifications: push or email based on preference
        elif score >= 0.7:
            preferred = prefs.get("preferred_channel", "push")
            if preferred == "push" and prefs.get("push_enabled", True):
                channels.append(NotificationChannel.PUSH)
            elif prefs.get("email_enabled", True):
                channels.append(NotificationChannel.EMAIL)

        # Medium notifications: check if user is active in-app
        elif score >= 0.4:
            is_online = await self.is_user_online(notification.user_id)
            if not is_online and prefs.get("email_enabled", True):
                channels.append(NotificationChannel.EMAIL)

        # Low-importance: in-app only (already added)
        return channels

    async def get_user_preferences(self, user_id: str) -> dict:
        row = await self.db.fetchrow(
            "SELECT preferences FROM notification_settings WHERE user_id = $1",
            user_id
        )
        return row["preferences"] if row else {}

    async def is_user_online(self, user_id: str) -> bool:
        last_seen = await self.db.fetchval(
            "SELECT last_seen_at FROM user_presence WHERE user_id = $1",
            user_id
        )
        if not last_seen:
            return False
        return (datetime.utcnow() - last_seen).total_seconds() < 300

Notification Bundling

Group related notifications into a single digest to reduce volume.

from collections import defaultdict

class NotificationBundler:
    def __init__(self, bundle_window_seconds: int = 300):
        self.window = bundle_window_seconds
        self.pending: dict[str, list[Notification]] = defaultdict(list)

    def add(self, notification: Notification):
        key = f"{notification.user_id}:{notification.entity_type}"
        self.pending[key].append(notification)

    async def flush(self) -> list[dict]:
        bundles = []
        for key, notifications in self.pending.items():
            if len(notifications) == 1:
                bundles.append({
                    "type": "single",
                    "notification": notifications[0],
                })
            else:
                bundles.append({
                    "type": "bundle",
                    "summary": self.create_summary(notifications),
                    "count": len(notifications),
                    "notifications": notifications,
                })
        self.pending.clear()
        return bundles

    def create_summary(self, notifications: list[Notification]) -> str:
        event_types = set(n.event_type for n in notifications)
        entity_type = notifications[0].entity_type

        if len(event_types) == 1:
            return (f"{len(notifications)} new {notifications[0].event_type} "
                    f"events on {entity_type} records")
        return (f"{len(notifications)} updates on {entity_type} records "
                f"({', '.join(event_types)})")

The Complete Notification Pipeline

from fastapi import FastAPI

app = FastAPI()

async def process_notification(notification: Notification,
                                scorer: NotificationScorer,
                                channel_selector: ChannelSelector,
                                bundler: NotificationBundler):
    score = await scorer.score(notification)
    channels = await channel_selector.select_channels(notification, score)

    notification.metadata["score"] = score
    notification.metadata["channels"] = [c.value for c in channels]

    # High-priority: deliver immediately
    if score >= 0.8:
        for channel in channels:
            await deliver(notification, channel)
    else:
        # Lower priority: add to bundler for digest delivery
        bundler.add(notification)

FAQ

How do I let users override the AI prioritization?

Provide a notification settings page where users can pin specific event types as "always high priority" or "always mute." These overrides take precedence over AI scoring. Store overrides as explicit rules that the scorer checks before running its scoring logic.

What if a critical notification gets scored too low?

Define a set of event types that bypass scoring entirely — system outages, security alerts, billing failures, and account lockouts should always be treated as maximum priority. Maintain this list in configuration, not in AI logic, so it cannot be affected by model behavior.

How do I measure whether the AI notification system is working?

Track three key metrics: notification read rate (should increase after implementing AI scoring), time-to-action (how quickly users respond to actionable notifications), and unsubscribe rate (should decrease). Compare these metrics to the pre-AI baseline over a 30-day window.

---

#AINotifications #AlertPrioritization #SaaS #IntelligentDelivery #Python #AgenticAI #LearnAI #AIEngineering