Automation Suggestions & Blueprints

Consent-first automation suggestions and reusable blueprint templates — surfaced through chat, CLI, or Python.

Automation Suggestions turn recurring requests into scheduled jobs — proposed explicitly, or offered automatically when the assistant notices the same intent recurring — but only ever scheduled when you tap Accept.

Two paths feed one consent gate — a manual propose() call, or an automatic observe() that fires once a request recurs.

Quick Start

1. Let the assistant offer it (feed each turn)

   Feed each completed turn’s inferred blueprint + slots. After the same intent recurs a few times, the engine offers it — you still tap Accept.

   from praisonai.scheduler.suggestion_engine import SuggestionEngine
   
   engine = SuggestionEngine()
   
   # Called from your bot/gateway/CLI after each completed turn:
   sug_id = engine.observe(
       "morning-brief",
       slots={"hour": 8, "weekdays": "mon-fri"},
       principal="telegram:12345",  # optional, isolates per user
   )
   
   if sug_id:
       print(f"Offered automation as {sug_id} — waiting for user to accept.")

   Nothing runs until the user taps ✓ Accept. Below the threshold (default 3), observe() just counts and returns None.

2. See pending suggestions

   Type /automations in chat. The bot lists each pending suggestion with its own Accept/Dismiss buttons.

   /automations

3. Accept from chat

   Tap the ✓ Accept button on a suggestion. Exactly one scheduled job is created — nothing runs until you accept.

4. Or create from a template

   Skip suggestions and build a job straight from a blueprint:

   /blueprint morning-brief hour=8 weekdays=mon-fri

How It Works

Two paths lead to the same result — a scheduled job — but only ever on an explicit accept.

Auto-generating suggestions with observe()

observe() is the recurrence generator — a caller feeds every turn’s inferred intent, and the engine offers an automation once it recurs.

Both entry points end in the same store and both require Accept — pick the one that matches who notices the recurrence.

Use observe() when the assistant should notice recurrence itself (per-turn feed from a bot/gateway). Use propose() when your code already knows an automation is worth offering (manual CLI, dashboard, explicit rule).

observe() is deterministic, not ML — a per-intent trailing count keyed on (principal, blueprint_name, frozen(slots)). Two module-level constants set its defaults:

Parameters

Example

from praisonai.scheduler.suggestion_engine import SuggestionEngine

engine = SuggestionEngine()

# Feed one observation per completed turn.
# 1st and 2nd calls: returns None (below threshold=3).
engine.observe("morning-brief", slots={"hour": 8}, principal="telegram:alice")
engine.observe("morning-brief", slots={"hour": 8}, principal="telegram:alice")

# 3rd call: threshold met — engine.propose() runs, store returns a suggestion id.
sug_id = engine.observe(
    "morning-brief",
    slots={"hour": 8},
    principal="telegram:alice",
    reason="Alice asks for a morning brief around 8am every day.",
)

# Different slots for the same user = distinct intent. This is turn 1 of its own count.
engine.observe("morning-brief", slots={"hour": 9}, principal="telegram:alice")

# Different principal = fully separate count. Bob's first observation is turn 1.
engine.observe("morning-brief", slots={"hour": 8}, principal="telegram:bob")

Tuning

Adjust threshold and window_seconds to fit the workflow:

# Fire immediately on the first observation — handy for testing.
engine.observe("morning-brief", slots={"hour": 8}, threshold=1)

# Wait for stronger evidence over a longer window (30 days).
engine.observe(
    "morning-brief",
    slots={"hour": 8},
    threshold=5,
    window_seconds=3600 * 24 * 30,
)

# Nested slots are supported — the recurrence key is frozen recursively.
engine.observe(
    "morning-brief",
    slots={"weekdays": ["mon", "tue"], "opts": {"tz": "UTC"}},
)

Immediately after a fire, another observation with the same key is below-threshold again and the store dedups the identical intent, so there is no double-offer. self._observations is process-local best-effort memory — a restart merely resets counters; the durable state stays in the SuggestionStore.

Built-in Blueprints

Three blueprints ship ready to use, straight from blueprint_catalogue.py.

Custom Blueprints (YAML)

Author your own blueprints under ~/.praisonai/blueprints/<name>/blueprint.yaml; a custom blueprint overrides a built-in of the same name.

name: standup-digest
version: "1.0.0"
description: Team standup digest
category: daily
default_deliver: telegram
prompt_template: "Summarise today's standup for {team}."
schedule_template: "cron:{minute} {hour} * * {weekdays_expression}"
slots:
  - name: hour
    type: int
    default: 9
  - name: minute
    type: int
    default: 30
  - name: weekdays
    type: choice
    default: mon-fri
    choices: [mon-fri, daily, weekends]
  - name: team
    type: str
    default: engineering

Discovery scans each <name>/blueprint.yaml subdirectory — see BlueprintCatalogue._load_from_directory for the exact rule.

Chat Commands

CLI Commands

Drive the same engine from the shell — full flags in Schedule CLI.

observe() is a Python-level generator called from your bot/gateway; there is no CLI entry point today. Use praisonai schedule suggestion-propose <blueprint> for a one-shot manual proposal.

Python Usage

Propose and accept a suggestion programmatically.

from praisonai.scheduler.suggestion_engine import SuggestionEngine
from praisonai.scheduler.blueprint_catalogue import BlueprintCatalogue

engine = SuggestionEngine()

sug_id = engine.propose(
    "morning-brief",
    slots={"hour": 8, "weekdays": "mon-fri"},
    deliver="telegram",
    reason="Detected daily morning request pattern",
)

if sug_id:
    catalogue = BlueprintCatalogue()
    bp = catalogue.get_blueprint("morning-brief")
    sug = engine.get_suggestion(sug_id)
    resolved = catalogue.resolve_slots(bp, sug.slots)
    prompt = catalogue.materialize_prompt(bp, resolved)
    schedule = catalogue.materialize_schedule(bp, resolved)
    engine.accept(sug_id)
    print(prompt, schedule)

propose() returns None when the pending cap (20) is reached or the same blueprint + slots was suggested within the 24-hour dedup window.

On a bot gateway the engine reads SessionContext.unified_user_id for you. Pass principal= yourself only from a custom script that has no session installed.

from praisonai.scheduler.suggestion_engine import SuggestionEngine

engine = SuggestionEngine()

# Primary generator: feed each completed turn's inferred intent.
# Inside an agent turn on a bot gateway — no principal= needed.
# The engine defaults it from SessionContext.unified_user_id.
sug_id = engine.observe(
    "morning-brief",
    slots={"hour": 8, "weekdays": "mon-fri"},
    deliver="telegram",
)

# Fallback: propose directly when your code already knows to offer.
engine.propose(
    "morning-brief",
    slots={"hour": 8, "weekdays": "mon-fri"},
    deliver="telegram",
    reason="Detected daily morning request pattern",
)

engine.pending()   # → only this user's suggestions

from praisonai.scheduler.suggestion_engine import SuggestionEngine

engine = SuggestionEngine()

session_user_id = "telegram:12345"

sug_id = engine.propose(
    "morning-brief",
    slots={"hour": 8, "weekdays": "mon-fri"},
    deliver="telegram",
    reason="Detected daily morning request pattern",
    principal=session_user_id,
)

engine.pending(principal=session_user_id)   # → only this user's suggestions

User Interaction Flows

Flow A — Accept an existing suggestion in chat:

User → /automations
Bot  → "💡 You have 1 pending automation suggestion:"
Bot  → "💡 Detected daily morning request pattern
        morning-brief · hour=8, weekdays=mon-fri"
        [✓ Accept]  [✕ Dismiss]
User → taps ✓ Accept
Bot  → "✅ Automation scheduled. Added job 'morning-brief' (schedule cron:0 8 * * mon,tue,wed,thu,fri)"

Flow B — Create directly from a blueprint:

User → /blueprint
Bot  → "📋 Create an automation from a template:
        • morning-brief — Daily morning briefing with news and priorities
        • important-mail — Check for important emails at a regular interval
        • weekly-review — End-of-week summary and review
        Usage: /blueprint <name> [slot=value ...]
        Example: /blueprint morning-brief hour=8 weekdays=mon-fri"
User → /blueprint morning-brief hour=8 weekdays=mon-fri focus=tech
Bot  → "✅ Automation scheduled from 'morning-brief'. Added job 'morning-brief' (schedule cron:0 8 * * mon,tue,wed,thu,fri)"

Flow D — Auto-offered after a recurring request:

User → "Give me the morning brief"           (turn 1 → observe → below threshold)
User → next day: "morning brief please"      (turn 2 → observe → below threshold)
User → next day: "brief me on the morning"   (turn 3 → observe → threshold met → propose)
Bot  → "💡 Noticed a recurring 'morning-brief' request (3x); offering to automate it."
       morning-brief · hour=8
       [✓ Accept]  [✕ Dismiss]
User → taps ✓ Accept
Bot  → "✅ Automation scheduled. Added job 'morning-brief' ..."

Mapping natural-language turns → blueprint_name + slots is the caller’s job (bot/gateway), not something observe() infers by itself.

Flow C — Two users, one bot, no cross-talk (multi-tenant):

On a bot gateway you don’t call this by hand — the gateway installs a SessionContext per turn, and SuggestionEngine.propose(...) reads it automatically. The pseudo-code below shows what the store sees end-to-end.

Alice → DMs the bot: "remind me to file expenses every Friday at 5pm"
Bot   → add(Suggestion(principal="tg:alice", name="weekly-expenses", ...))
Bob   → DMs the same bot: "remind me to file expenses every Friday at 5pm"   (identical text)
Bot   → add(Suggestion(principal="tg:bob", ...))  # admitted — dedup is per-principal
Alice → /automations   → sees only her own pending suggestion
Alice → accepts        → job created with principal="tg:alice"
Bob   → accepts        → his own job created; no name collision, both coexist

Before per-principal scoping, Bob’s identical proposal was silently blocked by Alice’s dedup entry, and either user could accept or dismiss the other’s suggestion.

Multi-tenant isolation

Each end-user sees only their own suggestions once the gateway passes their resolved identity as principal=. Alice and Bob can share one gateway yet never see or mutate each other’s pending queue; a single-user deployment leaves principal=None and keeps the global pool.

principal is any stable, opaque string that names the caller — a user id, an email hash, a tenant id. The store never parses or validates it; it only compares it. Leave it unset (None) for the global, single-tenant behaviour used by the CLI and single-user deployments.

A gateway handler that already knows who’s calling threads that identity through as principal:

from praisonaiagents.scheduler import Suggestion, SuggestionStore

store = SuggestionStore()  # ~/.praisonai/suggestions.json

# In a gateway handler that already knows who's calling:
def handle_user_message(user_id: str) -> None:
    store.add(Suggestion(
        id=f"sug_{user_id}_standup",
        blueprint_name="morning-brief",
        slots={"hour": 9, "weekdays": "mon-fri"},
        reason="Detected daily standup request",
        principal=user_id,          # isolate to this caller
    ))                              # pending cap + dedup window scoped per user_id

# Later, when the same user pulls their pending list:
def list_for(user_id: str) -> list[Suggestion]:
    return store.list_pending(principal=user_id)

Cross-owner access is rejected, not silently merged:

• list_pending(principal="alice") never returns Bob’s suggestions.
• accept(sug_id, principal="alice") returns False if the suggestion is Bob’s.
• dismiss(sug_id, principal="alice") returns False if the suggestion is Bob’s.
• The pending cap (MAX_PENDING_CAP, 20) and 24-hour dedup window are measured per principal, so one tenant can neither exhaust another’s slots nor block their identical proposal.

There is no admin bypass — a call passing principal="bob" cannot see Alice’s items even when the process runs as root. Scoping is enforced only by the principal the caller passes.

Safe by Default

Since PR #3507, Suggestion carries an optional principal owner and the store isolates list_pending / accept / dismiss per owner. See Per-User Scheduler Isolation for the full multi-tenant model. Access is still gated by CommandAccessPolicy (the automations permission is re-checked on every accept/dismiss tap).

Nothing is ever auto-created — the engine only materialises a job on an explicit accept. MAX_PENDING_CAP (20), the 24-hour dedup window, and the 3-day TTL are enforced within each principal, so one tenant cannot exhaust the cap or shadow another’s suggestion. Leaving principal=None measures them against the global pool.

Platform Coverage

The commands also degrade gracefully: if the scheduler extra isn’t installed, they reply ❌ Automations are not available (scheduler not installed).

Best Practices

from praisonai.scheduler.suggestion_engine import SuggestionEngine

engine = SuggestionEngine()

# After each completed turn:
engine.observe(
    "morning-brief",
    slots={"hour": 8, "weekdays": "mon-fri"},
    principal="telegram:12345",
)

from praisonaiagents.scheduler.suggestion_store import SuggestionStore

store = SuggestionStore()

# Only this user's pending suggestions
mine = store.list_pending(principal="telegram:12345")

# Cross-owner accept / dismiss are refused (returns False)
store.accept("sug_abc123", principal="telegram:12345")
store.dismiss("sug_abc123", principal="telegram:12345")

Related