Proactive Agents: AI That Initiates Conversations and Suggests Next Actions

Beyond Reactive Conversations

Most conversational agents are purely reactive — they wait for the user to say something and respond. Proactive agents flip this dynamic by identifying opportunities to initiate helpful interactions. A shipping agent that notifies you about a delay before you ask, or an onboarding assistant that suggests the next step when you have been idle — these create significantly better user experiences.

The challenge is doing this without being annoying. Proactive agents must balance helpfulness against interruption cost, respect user preferences, and time their outreach for maximum relevance.

Trigger System Design

Every proactive interaction starts with a trigger — an event or condition that warrants reaching out to the user.

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 datetime import datetime, timedelta
from enum import Enum
from typing import Callable, Optional

class TriggerType(Enum):
    EVENT = "event"       # Something happened
    TIME = "time"         # Scheduled or deadline-based
    INACTIVITY = "inactivity"  # User has been idle
    THRESHOLD = "threshold"    # A metric crossed a limit

@dataclass
class Trigger:
    name: str
    trigger_type: TriggerType
    condition: Callable[..., bool]
    message_template: str
    relevance_score: float  # 0.0-1.0
    cooldown_minutes: int = 60  # Minimum gap between firings
    last_fired: Optional[datetime] = None

    def can_fire(self) -> bool:
        if self.last_fired is None:
            return True
        elapsed = datetime.now() - self.last_fired
        return elapsed > timedelta(minutes=self.cooldown_minutes)

    def fire(self, context: dict) -> Optional[str]:
        if not self.can_fire():
            return None
        if not self.condition(context):
            return None
        self.last_fired = datetime.now()
        return self.message_template.format(**context)

The cooldown mechanism is essential. Without it, a trigger that remains true (like "user has not completed onboarding") would fire repeatedly.

Relevance Scoring and Priority

When multiple triggers fire simultaneously, the agent needs to pick the most relevant one. Sending three proactive messages at once overwhelms users.

class ProactiveEngine:
    def __init__(self, max_messages_per_hour: int = 2):
        self.triggers: list[Trigger] = []
        self.max_per_hour = max_messages_per_hour
        self.sent_this_hour: int = 0
        self.hour_start: datetime = datetime.now()
        self.user_preferences = {
            "proactive_enabled": True,
            "quiet_hours_start": 22,  # 10 PM
            "quiet_hours_end": 8,     # 8 AM
        }

    def add_trigger(self, trigger: Trigger):
        self.triggers.append(trigger)

    def check_quiet_hours(self) -> bool:
        hour = datetime.now().hour
        start = self.user_preferences["quiet_hours_start"]
        end = self.user_preferences["quiet_hours_end"]
        if start > end:  # Spans midnight
            return hour >= start or hour < end
        return start <= hour < end

    def evaluate(self, context: dict) -> Optional[str]:
        if not self.user_preferences["proactive_enabled"]:
            return None

        if self.check_quiet_hours():
            return None

        # Reset hourly counter
        if datetime.now() - self.hour_start > timedelta(hours=1):
            self.sent_this_hour = 0
            self.hour_start = datetime.now()

        if self.sent_this_hour >= self.max_per_hour:
            return None

        # Collect and rank eligible triggers
        candidates = []
        for trigger in self.triggers:
            message = trigger.fire(context)
            if message:
                candidates.append((trigger.relevance_score, message))

        if not candidates:
            return None

        candidates.sort(key=lambda x: x[0], reverse=True)
        self.sent_this_hour += 1
        return candidates[0][1]

Defining Practical Triggers

engine = ProactiveEngine(max_messages_per_hour=2)

engine.add_trigger(Trigger(
    name="onboarding_incomplete",
    trigger_type=TriggerType.INACTIVITY,
    condition=lambda ctx: (
        not ctx.get("onboarding_complete")
        and ctx.get("idle_minutes", 0) > 10
    ),
    message_template=(
        "I noticed you haven't finished setting up your profile. "
        "Would you like help completing step {next_step}?"
    ),
    relevance_score=0.7,
    cooldown_minutes=120,
))

engine.add_trigger(Trigger(
    name="shipping_delay",
    trigger_type=TriggerType.EVENT,
    condition=lambda ctx: ctx.get("shipping_delayed", False),
    message_template=(
        "Heads up: your order {order_id} has a shipping delay. "
        "New estimated delivery is {new_eta}. "
        "Would you like to see options?"
    ),
    relevance_score=0.95,
    cooldown_minutes=30,
))

# Evaluate with current context
context = {
    "onboarding_complete": False,
    "idle_minutes": 15,
    "next_step": 3,
    "shipping_delayed": True,
    "order_id": "ORD-4821",
    "new_eta": "March 20",
}

message = engine.evaluate(context)
print(message)  # Shipping delay wins (higher relevance)

Respecting User Preferences

Proactive agents must provide opt-out controls. Store user preferences for notification types, frequency limits, and quiet hours. Always honor "do not disturb" signals immediately.

def update_preferences(engine: ProactiveEngine, user_input: str):
    lower = user_input.lower()
    if "stop" in lower or "no more" in lower:
        engine.user_preferences["proactive_enabled"] = False
        return "Proactive notifications disabled. You can re-enable anytime."
    if "quiet" in lower:
        engine.user_preferences["quiet_hours_start"] = 20
        engine.user_preferences["quiet_hours_end"] = 9
        return "Quiet hours set from 8 PM to 9 AM."
    return None

FAQ

How do you prevent proactive agents from feeling spammy?

Three mechanisms work together: cooldown periods between trigger firings, hourly message caps, and relevance thresholds that filter out low-value notifications. Additionally, track user engagement with proactive messages — if a user dismisses three in a row, automatically reduce frequency or pause until they initiate a conversation.

What triggers justify unsolicited outreach?

High-urgency, time-sensitive events are the best candidates: security alerts, delivery issues, approaching deadlines, or service disruptions. Low-urgency suggestions like "did you know about this feature?" should be rate-limited aggressively and tied to specific user activity patterns that suggest genuine need.

How do you measure the success of proactive interactions?

Track the engagement rate (what percentage of proactive messages get a user response), the resolution rate (did the proactive message lead to a completed action), and the opt-out rate. A healthy proactive system has engagement above 30 percent and opt-out below 5 percent per month.

---

#ProactiveAI #AgentDesign #TriggerSystems #ConversationalAI #Python #AgenticAI #LearnAI #AIEngineering