Ai Generating Notifications

Build an AI Notification Generator

Guide the user through building AI that turns structured events into useful, channel-appropriate notification messages. Uses DSPy to produce consistent, personalized notification copy with urgency calibration and digest aggregation.

Step 1: Understand the notification task

Ask the user: 1. What events trigger notifications? (system alerts, user activity, scheduled digests, thresholds crossed?) 2. What channels do you target? (push/iOS/Android, email, Slack, SMS?) 3. Do you need personalization? (user name, role, preferences, history?) 4. Real-time or digest? (one notification per event, or aggregate multiple events into one message?)

Step 2: Build a single-event notifier

Basic signature

import dspy
from typing import Literal

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

class GenerateNotification(dspy.Signature):
    """Write a clear, concise notification message for the target channel and recipient."""
    event: str = dspy.InputField(desc="Structured event data or description")
    recipient_profile: str = dspy.InputField(desc="Who receives the notification - role, name, preferences")
    channel: Literal["push", "email", "slack", "sms"] = dspy.InputField(desc="Delivery channel")
    notification_text: str = dspy.OutputField(desc="The notification message, respecting channel length limits")
    urgency_level: Literal["low", "medium", "high", "critical"] = dspy.OutputField(
        desc="Urgency level - low=informational, medium=needs attention, high=act soon, critical=act now"
    )

notifier = dspy.ChainOfThought(GenerateNotification)

result = notifier(
    event="User account login from new device - IP 203.0.113.42, Berlin, Germany",
    recipient_profile="Account owner, security-conscious, email preferred",
    channel="email",
)
print(result.notification_text)
print(result.urgency_level)

Step 3: Channel-specific constraints

Each channel has hard limits. Define them explicitly and enforce with a reward function.

CHANNEL_LIMITS = {
    "push": 150,    # title + body combined
    "email": 500,   # subject + preview text
    "slack": 500,
    "sms": 160,
}

def channel_length_reward(args, pred):
    """Hard penalty for exceeding channel length limits."""
    limit = CHANNEL_LIMITS.get(args["channel"], 300)
    text_len = len(pred.notification_text)
    if text_len <= limit:
        return 1.0
    # Hard fail above 2x limit, graduated penalty between limit and 2x
    if text_len > limit * 2:
        return 0.0
    return max(0.0, 1.0 - (text_len - limit) / limit)

notifier_enforced = dspy.Refine(
    module=dspy.ChainOfThought(GenerateNotification),
    N=3,
    reward_fn=channel_length_reward,
    threshold=0.9,
)

Step 4: Digest aggregation

Group multiple events into a single summary notification — reduces alert fatigue.

from pydantic import BaseModel, Field

class DigestOutput(BaseModel):
    subject: str = Field(description="Email subject line, max 60 chars")
    headline: str = Field(description="One-sentence summary of the most important event")
    event_groups: list[str] = Field(description="Events grouped by type, e.g. '3 new comments, 2 deployments'")
    call_to_action: str = Field(description="What the user should do next, if anything")

class GenerateDigest(dspy.Signature):
    """Aggregate multiple events into a single digest notification. Group similar events, highlight the most important, and keep it scannable."""
    events: list[str] = dspy.InputField(desc="List of events to include in the digest")
    recipient_profile: str = dspy.InputField(desc="Who receives the digest")
    time_period: str = dspy.InputField(desc="Time window covered - e.g. 'last 24 hours', 'this week'")
    digest: DigestOutput = dspy.OutputField()

class DigestNotifier(dspy.Module):
    def __init__(self):
        self.group = dspy.ChainOfThought("events -> grouped_events: list[str]")
        self.write = dspy.ChainOfThought(GenerateDigest)

    def forward(self, events, recipient_profile, time_period):
        # Group similar events first, then write the digest
        grouped = self.group(events=events).grouped_events
        return self.write(
            events=grouped,
            recipient_profile=recipient_profile,
            time_period=time_period,
        )

Step 5: Urgency calibration

Prevent over-alerting by calibrating urgency against event severity and recipient fatigue.

class CalibrateUrgency(dspy.Signature):
    """Assess the urgency of this event for this recipient. Consider event severity, recipient role, and whether action is required."""
    event: str = dspy.InputField(desc="Event description")
    recipient_profile: str = dspy.InputField(desc="Recipient role and preferences")
    recent_notification_count: int = dspy.InputField(
        desc="Number of notifications sent to this recipient in the last hour"
    )
    urgency_level: Literal["low", "medium", "high", "critical"] = dspy.OutputField()
    should_send: bool = dspy.OutputField(
        desc="False if recipient is already overloaded with high-urgency alerts"
    )
    rationale: str = dspy.OutputField(desc="One sentence explaining the urgency decision")

def urgency_reward(args, pred):
    """Penalize assigning high/critical urgency to clearly informational events."""
    score = 1.0
    informational_keywords = ["viewed", "logged in", "updated preferences", "exported"]
    event_lower = args["event"].lower()
    if any(kw in event_lower for kw in informational_keywords):
        if pred.urgency_level in ("high", "critical"):
            score -= 0.5  # soft: informational events should not be urgent
    return score

urgency_calibrator = dspy.Refine(
    module=dspy.ChainOfThought(CalibrateUrgency),
    N=3,
    reward_fn=urgency_reward,
    threshold=0.8,
)

Step 6: Personalization

Recipient context should influence tone, detail level, and channel preference.

class PersonalizedNotification(dspy.Signature):
    """Write a notification tailored to the recipient. Match tone to their role, include relevant context, and use their preferred channel style."""
    event: str = dspy.InputField(desc="Structured event data")
    recipient_name: str = dspy.InputField(desc="Recipient's name")
    recipient_role: str = dspy.InputField(desc="e.g. 'developer', 'executive', 'end user'")
    recipient_preferences: str = dspy.InputField(
        desc="e.g. 'brief and technical', 'plain language', 'include numbers'"
    )
    channel: Literal["push", "email", "slack", "sms"] = dspy.InputField()
    notification_text: str = dspy.OutputField()
    urgency_level: Literal["low", "medium", "high", "critical"] = dspy.OutputField()

Tone by role example:

ROLE_HINTS = {
    "developer": "technical details, stack traces welcome, use markdown in Slack",
    "executive": "business impact only, no jargon, one sentence if possible",
    "end_user": "plain language, friendly tone, tell them exactly what to do",
    "on_call": "all relevant details, include timestamp, severity, and system affected",
}

Step 7: Evaluate and optimize

Notification quality metric

class JudgeNotification(dspy.Signature):
    """Judge the quality of a notification message on clarity, actionability, and channel fit."""
    event: str = dspy.InputField(desc="Original event that triggered the notification")
    channel: str = dspy.InputField()
    notification_text: str = dspy.InputField()
    urgency_level: str = dspy.InputField()
    clarity: float = dspy.OutputField(desc="0.0-1.0 - is the message immediately understandable?")
    actionability: float = dspy.OutputField(desc="0.0-1.0 - does the recipient know what to do?")
    channel_fit: float = dspy.OutputField(desc="0.0-1.0 - is length and format right for the channel?")

def notification_metric(example, prediction, trace=None):
    judge = dspy.Predict(JudgeNotification)
    result = judge(
        event=example.event,
        channel=example.channel,
        notification_text=prediction.notification_text,
        urgency_level=prediction.urgency_level,
    )
    return (result.clarity + result.actionability + result.channel_fit) / 3

optimizer = dspy.BootstrapFewShot(metric=notification_metric, max_bootstrapped_demos=4)
optimized = optimizer.compile(notifier, trainset=trainset)

When NOT to use AI notifications

• Transactional messages (order confirmations, password resets, receipt emails) — use templates. The text must be exact and predictable; AI adds variability without value.
• Regulatory or compliance messages (GDPR notices, financial disclosures, legal alerts) — wording is fixed by requirement; AI-generated copy introduces compliance risk.
• Simple threshold alerts ("CPU > 90%", "balance below $10") — a format string is faster, cheaper, and more reliable than an LM call.

Key patterns

Gotchas

• Claude generates text that exceeds channel limits. Passing max_chars=160 in a field description is not enough — the model treats it as a suggestion. Always wrap with dspy.Refine and a programmatic length check reward function that reads len(pred.notification_text).
• Claude treats all events as equally urgent. Without explicit calibration, routine events ("user viewed a file") get marked high urgency. Add a CalibrateUrgency step and a reward function that penalizes over-classification of low-severity events.
• Claude uses `dspy.Assert`/`dspy.Suggest` for constraints. Use dspy.Refine with a reward function instead — it handles retries with feedback and is the current DSPy pattern for enforcing output constraints.
• Claude generates generic notifications that ignore recipient context. Without recipient_profile in the signature, every user gets the same message. Always pass name, role, and preferences as inputs to get personalized copy.
• Claude creates digests by listing events sequentially instead of grouping. "3 events happened: X, Y, Z" is not a digest — it is a log. Build a separate GroupEvents step before the notification writer to cluster similar events and count them before writing copy.

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Aggregate events intelligently before notifying — see /ai-summarizing
• Parse structured event payloads (JSON, logs) before feeding to notifier — see /ai-parsing-data
• Score notification quality automatically — see /ai-scoring
• Enforce output constraints with retry loops — see /dspy-refine
• Sample multiple notification variants and pick the best — see /dspy-best-of-n
• Write DSPy signatures for input/output contracts — see /dspy-modules
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Additional resources

• For worked examples (push notifications, weekly digest, incident Slack alerts), see examples.md

AI Generating Notifications — Worked Examples

Example 1: Push notification generator

Turn app events into concise iOS/Android push notification copy.

Setup

import dspy
from typing import Literal

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

Signatures and module

from pydantic import BaseModel, Field

class PushNotification(BaseModel):
    title: str = Field(description="Push notification title, max 50 chars")
    body: str = Field(description="Push notification body, max 100 chars")
    urgency_level: Literal["low", "medium", "high", "critical"]

class GeneratePush(dspy.Signature):
    """Write a push notification for a mobile app. Title must be under 50 chars. Body must be under 100 chars. Be direct — the user sees this on their lock screen."""
    event: str = dspy.InputField(desc="App event that triggered the notification")
    recipient_name: str = dspy.InputField(desc="User's first name for personalization")
    notification: PushNotification = dspy.OutputField()


def push_length_reward(args, pred):
    """Hard fail if title or body exceeds character limits."""
    title_ok = len(pred.notification.title) <= 50
    body_ok = len(pred.notification.body) <= 100
    if not title_ok and not body_ok:
        return 0.0
    if not title_ok or not body_ok:
        return 0.5  # partial: one field out of spec
    return 1.0

push_notifier = dspy.Refine(
    module=dspy.ChainOfThought(GeneratePush),
    N=4,
    reward_fn=push_length_reward,
    threshold=1.0,
)

Usage

events = [
    ("New comment on your post 'Launching v2.0'", "Alex"),
    ("Your export is ready to download", "Maria"),
    ("Payment failed - subscription renewal", "Sam"),
    ("Someone mentioned you in #general", "Jordan"),
]

for event, name in events:
    result = push_notifier(event=event, recipient_name=name)
    n = result.notification
    print(f"[{n.urgency_level.upper()}] {n.title}")
    print(f"  {n.body}")
    print()

# [LOW] Alex, new comment waiting
#   "Great post! When does v2.0 ship?" — view now
#
# [LOW] Maria, your export is ready
#   Download your file before it expires in 24 hours
#
# [CRITICAL] Payment failed, Sam
#   Your subscription renewal failed. Update your card now.
#
# [MEDIUM] Jordan, you were mentioned
#   Someone tagged you in #general — tap to read

Metric and optimization

def push_metric(example, prediction, trace=None):
    n = prediction.notification
    title_within = len(n.title) <= 50
    body_within = len(n.body) <= 100
    # Both fields must be within limits
    if not (title_within and body_within):
        return 0.0
    # Urgency must match expected level if provided
    if hasattr(example, "expected_urgency") and n.urgency_level != example.expected_urgency:
        return 0.5
    return 1.0

optimizer = dspy.BootstrapFewShot(metric=push_metric, max_bootstrapped_demos=4)
optimized = optimizer.compile(dspy.ChainOfThought(GeneratePush), trainset=trainset)

---

Example 2: Weekly digest email

Aggregate many events from the past week into a single scannable email digest.

Signatures and module

from pydantic import BaseModel, Field

class DigestSection(BaseModel):
    group_label: str = Field(description="Category label, e.g. 'Comments (4)', 'Deployments (2)'")
    summary: str = Field(description="One sentence summarizing this group")

class WeeklyDigest(BaseModel):
    subject_line: str = Field(description="Email subject, max 60 chars, no clickbait")
    headline: str = Field(description="Most important thing that happened this week, one sentence")
    sections: list[DigestSection] = Field(description="Event groups, 2-5 sections")
    call_to_action: str = Field(description="What the user should do, or empty string if none")

class GroupEvents(dspy.Signature):
    """Group a list of app events by type. Return a list of group labels with counts, e.g. '5 new comments', '2 deployments'."""
    events: list[str] = dspy.InputField(desc="Raw event list")
    grouped: list[str] = dspy.OutputField(desc="Grouped summaries, one per event type")

class WriteDigest(dspy.Signature):
    """Write a weekly digest email from grouped event summaries. Keep it scannable - the user is skimming."""
    grouped_events: list[str] = dspy.InputField(desc="Event groups from GroupEvents step")
    recipient_name: str = dspy.InputField()
    total_event_count: int = dspy.InputField(desc="Total number of raw events this week")
    digest: WeeklyDigest = dspy.OutputField()

class WeeklyDigestNotifier(dspy.Module):
    def __init__(self):
        self.group = dspy.ChainOfThought(GroupEvents)
        self.write = dspy.ChainOfThought(WriteDigest)

    def forward(self, events, recipient_name):
        grouped = self.group(events=events).grouped
        return self.write(
            grouped_events=grouped,
            recipient_name=recipient_name,
            total_event_count=len(events),
        )


def digest_reward(args, pred):
    """Encourage concise subject line and at least 2 event sections."""
    score = 1.0
    if len(pred.digest.subject_line) > 60:
        score -= 0.3  # soft: subject is too long for email clients
    if len(pred.digest.sections) < 2:
        score -= 0.4  # soft: digest needs multiple sections to be useful
    return score

digest_notifier = dspy.Refine(
    module=WeeklyDigestNotifier(),
    N=3,
    reward_fn=digest_reward,
    threshold=0.8,
)

Usage

weekly_events = [
    "User commented on post #42",
    "User commented on post #45",
    "User commented on post #48",
    "Deployment to production succeeded",
    "New follower: @devrel_team",
    "API key rotated",
    "User liked post #42",
    "User liked post #50",
    "Report export completed",
]

result = digest_notifier(events=weekly_events, recipient_name="Taylor")
d = result.digest

print(d.subject_line)
# "Your week in review — 3 comments, 2 likes, 1 deployment"

print(d.headline)
# "Three new comments on your posts this week, including post #42 which got the most engagement."

for section in d.sections:
    print(f"  {section.group_label} - {section.summary}")
# Comments (3) - Readers engaged with posts #42, #45, and #48
# Likes (2) - Posts #42 and #50 received likes
# Deployments (1) - Production deployment succeeded
# Other (2) - API key rotated, report export ready

print(d.call_to_action)
# "Check your comments and reply to keep the conversation going."

---

Example 3: Incident alert generator

Turn system events and log snippets into structured Slack alerts with severity and context.

Signatures and module

from pydantic import BaseModel, Field
from typing import Literal

class IncidentAlert(BaseModel):
    severity: Literal["info", "warning", "error", "critical"]
    title: str = Field(description="Short incident title, max 60 chars")
    summary: str = Field(description="What happened and what is affected, 1-2 sentences")
    impact: str = Field(description="Who or what is impacted, e.g. 'US-East users', 'checkout API'")
    suggested_action: str = Field(description="Immediate next step for on-call engineer")
    runbook_hint: str = Field(description="Which runbook or playbook to check, or empty string")

class GenerateIncidentAlert(dspy.Signature):
    """Generate a Slack incident alert for an on-call engineer. Include severity, what happened, impact, and the immediate next action. Be precise - no filler text."""
    system_event: str = dspy.InputField(desc="Raw system event, error, or log snippet")
    service_name: str = dspy.InputField(desc="Name of the affected service")
    environment: Literal["production", "staging", "dev"] = dspy.InputField()
    alert: IncidentAlert = dspy.OutputField()

class IncidentNotifier(dspy.Module):
    def __init__(self):
        self.generate = dspy.ChainOfThought(GenerateIncidentAlert)

    def forward(self, system_event, service_name, environment):
        return self.generate(
            system_event=system_event,
            service_name=service_name,
            environment=environment,
        )

    def format_slack_block(self, alert: IncidentAlert) -> str:
        """Format the alert as a Slack markdown block."""
        severity_emoji = {
            "info": ":information_source:",
            "warning": ":warning:",
            "error": ":x:",
            "critical": ":rotating_light:",
        }
        icon = severity_emoji.get(alert.severity, ":bell:")
        lines = [
            f"{icon} *[{alert.severity.upper()}] {alert.title}*",
            f"> {alert.summary}",
            f"*Impact* - {alert.impact}",
            f"*Action* - {alert.suggested_action}",
        ]
        if alert.runbook_hint:
            lines.append(f"*Runbook* - {alert.runbook_hint}")
        return "\n".join(lines)


def incident_reward(args, pred):
    """Hard fail on empty summary or action. Soft penalty for overly long blocks."""
    if not pred.alert.summary or not pred.alert.suggested_action:
        return 0.0  # hard: both fields are required
    slack_text = pred.alert.summary + pred.alert.suggested_action
    if len(slack_text) > 400:
        return 0.6  # soft: Slack blocks should stay readable
    return 1.0

incident_notifier_enforced = dspy.Refine(
    module=IncidentNotifier(),
    N=3,
    reward_fn=incident_reward,
    threshold=0.9,
)

Usage

notifier = IncidentNotifier()

result = notifier(
    system_event="""
    ERROR [2026-05-04T14:32:11Z] checkout-service: Connection pool exhausted
    java.sql.SQLException: Timeout waiting for connection from pool (30000ms)
    at com.example.checkout.OrderRepository.findById(OrderRepository.java:142)
    Active connections: 100/100, Pending requests: 847
    """,
    service_name="checkout-service",
    environment="production",
)

slack_message = notifier.format_slack_block(result.alert)
print(slack_message)
# :rotating_light: *[CRITICAL] checkout-service DB pool exhausted*
# > Connection pool fully exhausted with 847 pending requests; all checkout transactions are failing.
# *Impact* - All users attempting to complete a purchase in production
# *Action* - Scale up DB connection pool or restart checkout-service pods immediately
# *Runbook* - db-connection-pool-runbook.md

Metric

class JudgeIncidentAlert(dspy.Signature):
    """Judge whether an incident alert gives an on-call engineer everything they need to act."""
    system_event: str = dspy.InputField()
    alert_text: str = dspy.InputField()
    has_clear_action: bool = dspy.OutputField(desc="Does the alert say exactly what to do?")
    severity_appropriate: bool = dspy.OutputField(desc="Is the severity correct for the event?")
    no_filler: bool = dspy.OutputField(desc="Is the alert free of vague or generic text?")

def incident_metric(example, prediction, trace=None):
    judge = dspy.Predict(JudgeIncidentAlert)
    notifier = IncidentNotifier()
    slack_text = notifier.format_slack_block(prediction.alert)
    result = judge(system_event=example.system_event, alert_text=slack_text)
    return (result.has_clear_action + result.severity_appropriate + result.no_filler) / 3