Cuculi AI · Notification Engineering

Cuculi AI Notification System – Python Function Library

Progressively uncover the 32 core functions, configuration blueprints, and execution choreography that power Cuculi AI's psychology-driven notification engine. Each section is designed for progressive disclosure—skim the summaries, expand the layers, and capture every implementation detail.

32

Core Python Functions

Organized across 7 interlocking categories.

6

Execution Phases

Critical-to-low priority pipeline orchestration.

4

Configuration Files

Environment, templates, delivery, and tests.

Progressive Disclosure Roadmap

Follow the curated sequence below. Each card reveals a deeper layer—from architecture context to exact YAML configurations.

01

System Overview & Data Flow

Understand how configuration, context, AI, and delivery connect.
02

Source Directory Blueprint

Map every folder that houses the functions and configs.
03

Function Categories

Reveal tables for configuration, database, psychology, AI, delivery, and monitoring.
04

Execution Sequence

Step-by-step flows grouped by priority phase.
05

Priority Matrix

Clarify what must launch first versus continuous optimizations.
06

Configuration Management

Inspect every YAML file powering environments, nudges, and delivery.
07

Testing Framework

Review mandatory validation, performance, and integration coverage.

System Overview

The Cuculi AI notification system orchestrates 32 core functions across seven categories to craft personalized nudges rooted in behavioral psychology. Each layer—from configuration to analytics—feeds the next, enabling a disciplined, testable pipeline.

Data Flow Spine

Every notification follows a strict progression so no dependency is skipped.

  1. Configuration Load
  2. Database Initialization
  3. User Action Intake
  4. Context Assembly
  5. Psychology Selection
  6. AI Content Generation
  7. Delivery Execution
  8. Analytics & Monitoring

What Progressive Disclosure Means Here

Each section below unlocks more fidelity:

  • Start broad: Understand categories at a glance.
  • Expand tables: Reveal exact function signatures, outputs, and tunable parameters.
  • Drill into YAML: See the literal configuration values driving runtime behavior.
  • Validate: Close with the testing obligations for every function.

Directory Structure

Trace where each component of the notification engine lives. Expand the tree mentally as you map functions, configs, scripts, and test fixtures.

cuculi_ai_notifications/
├── config/
│   ├── settings.yaml              # Main system configuration
│   ├── nudge_templates.yaml       # Psychology-driven templates
│   ├── delivery_channels.yaml     # Channel configurations
│   └── test_settings.yaml         # Testing configurations
├── src/
│   ├── __init__.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config/
│   │   │   ├── __init__.py
│   │   │   └── config_manager.py
│   │   ├── database/
│   │   │   ├── __init__.py
│   │   │   ├── mongodb_client.py
│   │   │   ├── vector_operations.py
│   │   │   └── data_retrieval.py
│   │   ├── ai/
│   │   │   ├── __init__.py
│   │   │   ├── openai_client.py
│   │   │   ├── prompt_engine.py
│   │   │   └── content_generator.py
│   │   ├── psychology/
│   │   │   ├── __init__.py
│   │   │   ├── behavioral_analyzer.py
│   │   │   ├── nudge_selector.py
│   │   │   └── user_segmenter.py
│   │   ├── delivery/
│   │   │   ├── __init__.py
│   │   │   ├── channel_manager.py
│   │   │   ├── timing_optimizer.py
│   │   │   └── delivery_handler.py
│   │   ├── monitoring/
│   │   │   ├── __init__.py
│   │   │   ├── logger.py
│   │   │   ├── performance_tracker.py
│   │   │   └── quality_controller.py
│   │   └── utils/
│   │       ├── __init__.py
│   │       ├── validators.py
│   │       ├── formatters.py
│   │       └── helpers.py
├── tests/
│   ├── __init__.py
│   ├── unit/
│   │   ├── test_config/
│   │   ├── test_database/
│   │   ├── test_ai/
│   │   ├── test_psychology/
│   │   ├── test_delivery/
│   │   └── test_monitoring/
│   ├── integration/
│   │   ├── test_end_to_end.py
│   │   └── test_api_flows.py
│   ├── fixtures/
│   │   ├── sample_users.json
│   │   ├── sample_events.json
│   │   └── mock_responses.json
│   └── conftest.py
├── scripts/
│   ├── setup_database.py
│   ├── create_indexes.py
│   ├── migrate_data.py
│   └── run_benchmarks.py
├── docs/
│   ├── api_reference.md
│   ├── configuration_guide.md
│   └── deployment_guide.md
├── requirements/
│   ├── base.txt
│   ├── dev.txt
│   └── test.txt
└── README.md

Function Categories

Expand each accordion to reveal every function, file location, signature, return object, and configurable parameter. This is the heart of the library—32 functions divided into seven mission-specific categories.

Category 1

Configuration & Initialization

Foundation functions for system setup and configuration management.

⟶
Function Name File Location Input Parameters Output Type Configurable Parameters
load_yaml_configuration config/config_manager.py config_path: str, env: str ConfigDict environment, validation_schema
validate_configuration_schema config/config_manager.py config: dict, schema: dict ValidationResult strict_mode, error_handling
initialize_environment_variables config/config_manager.py config: ConfigDict EnvVars override_existing, prefix
setup_logging_configuration config/config_manager.py log_config: dict Logger log_level, file_rotation, format

Category 2

Database Connectivity & Operations

MongoDB connection, vector search, and data operations.

⟶
Function Name File Location Input Parameters Output Type Configurable Parameters
create_mongodb_connection database/mongodb_client.py connection_string: str, config: dict MongoClient timeout, pool_size, retry_writes
initialize_database_collections database/mongodb_client.py client: MongoClient, db_name: str Database collection_names, indexes
setup_vector_search_indexes database/vector_operations.py db: Database, index_config: dict IndexResults vector_dimensions, similarity_algorithm
create_aggregation_pipeline database/data_retrieval.py pipeline_type: str, params: dict Pipeline optimization_level, cache_results
execute_vector_similarity_search database/vector_operations.py collection: Collection, query_vector: List[float], params: dict SimilarityResults num_candidates, similarity_threshold

Category 3

Data Retrieval & Context Assembly

Functions for gathering user, event, and contextual data.

⟶
Function Name File Location Input Parameters Output Type Configurable Parameters
fetch_user_profile_complete database/data_retrieval.py user_id: ObjectId, include_fields: List[str] UserProfile cache_ttl, include_behavioral, include_social
fetch_event_details_with_venue database/data_retrieval.py event_id: ObjectId, include_participants: bool EventContext participant_limit, venue_details_depth
fetch_social_proof_indicators database/data_retrieval.py user_id: ObjectId, event_id: ObjectId SocialProofData friend_network_depth, mutual_connections
calculate_user_event_compatibility database/vector_operations.py user_vector: List[float], event_vector: List[float] float similarity_metric, normalization
assemble_message_context database/data_retrieval.py user_id: ObjectId, event_id: ObjectId, context_type: str MessageContext context_depth, historical_data

Category 4

Psychology & Behavioral Analysis

User psychology profiling and nudge selection.

⟶
Function Name File Location Input Parameters Output Type Configurable Parameters
analyze_user_behavioral_patterns psychology/behavioral_analyzer.py user_data: UserProfile, timeframe: int BehaviorProfile pattern_window, significance_threshold
calculate_engagement_score psychology/behavioral_analyzer.py behavioral_data: dict, weights: dict float recency_weight, frequency_weight, quality_weight
determine_psychology_profile psychology/behavioral_analyzer.py user_behavior: BehaviorProfile PsychologyProfile profile_confidence, fallback_default
segment_user_by_engagement psychology/user_segmenter.py user_data: UserProfile, segmentation_rules: dict UserSegment segment_boundaries, custom_rules
select_optimal_nudge_type psychology/nudge_selector.py psychology_profile: PsychologyProfile, context: MessageContext NudgeStrategy effectiveness_weights, historical_performance
check_notification_frequency_limits psychology/nudge_selector.py user_id: ObjectId, proposed_nudge: str FrequencyCheckResult daily_limit, weekly_limit, cooldown_period

Category 5

AI Content Generation

OpenAI integration and personalized message creation.

⟶
Function Name File Location Input Parameters Output Type Configurable Parameters
initialize_openai_client ai/openai_client.py api_key: str, model_config: dict OpenAIClient model_name, temperature, max_tokens, timeout
load_prompt_templates ai/prompt_engine.py template_config: dict, nudge_type: str PromptTemplate template_version, personalization_level
generate_psychology_driven_prompt ai/prompt_engine.py template: PromptTemplate, context: MessageContext GeneratedPrompt dynamic_variables, context_injection
create_personalized_message ai/content_generator.py prompt: GeneratedPrompt, generation_params: dict GeneratedMessage creativity_level, length_constraint, tone
validate_message_quality ai/content_generator.py message: str, quality_criteria: dict QualityScore grammar_check, relevance_score, appropriateness
optimize_message_for_channel ai/content_generator.py message: str, channel: str, constraints: dict OptimizedMessage character_limits, formatting_rules

Category 6

Delivery & Channel Management

Message delivery, timing, and channel optimization.

⟶
Function Name File Location Input Parameters Output Type Configurable Parameters
determine_optimal_delivery_time delivery/timing_optimizer.py user_timezone: str, user_activity: dict, event_timing: datetime DeliverySchedule timezone_handling, activity_windows
select_best_delivery_channel delivery/channel_manager.py user_preferences: dict, message_urgency: str ChannelSelection channel_priority, fallback_options
format_message_for_channel delivery/channel_manager.py message: str, channel: str FormattedMessage channel_specifications, emoji_support
queue_notification_for_delivery delivery/delivery_handler.py notification: Notification, priority: str QueueResult batch_size, priority_levels, delay_scheduling
send_notification_via_channel delivery/delivery_handler.py message: FormattedMessage, channel: str, recipient: str DeliveryResult retry_attempts, timeout, delivery_confirmation
handle_delivery_failures delivery/delivery_handler.py failed_delivery: DeliveryResult RetryResult backoff_strategy, max_retries, fallback_channels

Category 7

Monitoring & Analytics

Logging, performance tracking, and quality control.

⟶
Function Name File Location Input Parameters Output Type Configurable Parameters
log_function_execution monitoring/logger.py function_name: str, inputs: dict, outputs: dict, metadata: dict LogEntry log_level, retention_period, sensitive_data_masking
track_message_generation_metrics monitoring/performance_tracker.py generation_data: dict, performance_metrics: dict GenerationMetrics metric_types, aggregation_period
monitor_system_health monitoring/performance_tracker.py system_components: List[str] HealthStatus check_interval, alert_thresholds, escalation_rules
calculate_nudge_effectiveness monitoring/performance_tracker.py nudge_history: List[dict], success_criteria: dict EffectivenessReport time_window, conversion_definitions
generate_analytics_dashboard_data monitoring/performance_tracker.py date_range: tuple, metrics: List[str] AnalyticsData aggregation_level, real_time_updates
perform_quality_control_review monitoring/quality_controller.py generated_content: dict, review_criteria: dict QualityReview automated_checks, human_review_threshold

Execution Sequence & Dependencies

Progressively step through each priority band. Expand a phase to reveal the ordered function calls that must complete before moving forward.

Phase 1 · Critical

System Bootstrap

Must execute first to make the system operational.

⟶
  1. load_yaml_configuration
  2. validate_configuration_schema
  3. initialize_environment_variables
  4. setup_logging_configuration
  5. create_mongodb_connection
  6. initialize_database_collections
  7. setup_vector_search_indexes
  8. initialize_openai_client
  9. load_prompt_templates

Phase 2 · High

User Action Trigger

Real-time context assembly triggered by user activity.

⟶
  1. fetch_user_profile_complete
  2. fetch_event_details_with_venue
  3. fetch_social_proof_indicators
  4. calculate_user_event_compatibility
  5. assemble_message_context

Phase 3 · High

Psychology Analysis & Nudge Selection

Core logic that chooses the right persuasion technique.

⟶
  1. analyze_user_behavioral_patterns
  2. calculate_engagement_score
  3. determine_psychology_profile
  4. segment_user_by_engagement
  5. select_optimal_nudge_type
  6. check_notification_frequency_limits

Phase 4 · High

AI Content Generation

Prompt crafting, generation, validation, and optimization.

⟶
  1. generate_psychology_driven_prompt
  2. create_personalized_message
  3. validate_message_quality
  4. optimize_message_for_channel

Phase 5 · Medium

Delivery Orchestration

Timing, channel selection, and resilient delivery operations.

⟶
  1. determine_optimal_delivery_time
  2. select_best_delivery_channel
  3. format_message_for_channel
  4. queue_notification_for_delivery
  5. send_notification_via_channel
  6. handle_delivery_failures

Phase 6 · Low

Monitoring & Analytics

Runs continuously in the background to measure and improve.

⟶
  1. log_function_execution (runs parallel to all functions)
  2. track_message_generation_metrics
  3. monitor_system_health
  4. calculate_nudge_effectiveness
  5. generate_analytics_dashboard_data
  6. perform_quality_control_review

Priority Matrix

Align stakeholders on what must be shipped before go-live and what can iterate after stabilization.

🔴 Critical Priority

  • Configuration loading and validation
  • Database connection initialization
  • OpenAI client setup
  • Basic logging setup

🟠 High Priority

  • User data retrieval
  • Psychology analysis
  • Message generation
  • Nudge type selection
  • Basic delivery functionality

🟡 Medium Priority

  • Advanced delivery optimization
  • Channel-specific formatting
  • Retry mechanisms
  • Performance monitoring

🟢 Low Priority

  • Advanced analytics
  • Quality control automation
  • Dashboard data generation
  • A/B testing support

Configuration Management

Expand each YAML file to see the exact configuration powering environments, psychology heuristics, delivery channels, and testing guarantees.

Main Configuration File config/settings.yaml

Environment defaults, database tuning, vector search, psychology, delivery, and monitoring.

⟶ 
# Environment Configuration
environment:
  name: "${ENV:-development}"
  debug_mode: true
  log_level: "INFO"
  timezone: "UTC"

# Database Configuration
database:
  connection_string: "${MONGO_CONNECTION_STRING}"
  database_name: "cuculi_ai_notifications"
  timeout_seconds: 30
  retry_attempts: 3
  connection_pool:
    min_size: 5
    max_size: 50
  collections:
    users: "users"
    events: "events"
    notifications: "notifications"
    venues: "venues"

# Vector Search Configuration
vector_search:
  enabled: true
  indexes:
    user_vector:
      name: "user_vector_index"
      path: "behavioral_data.user_vector"
      dimensions: 3072
      similarity: "cosine"
    event_vector:
      name: "event_vector_index" 
      path: "event_vector"
      dimensions: 3072
      similarity: "cosine"
  search_params:
    num_candidates: 100
    similarity_threshold: 0.7

# OpenAI Configuration
openai:
  api_key: "${OPENAI_API_KEY}"
  model: "gpt-4o"
  temperature: 0.7
  max_tokens: 200
  timeout_seconds: 30
  retry_attempts: 3
  rate_limits:
    requests_per_minute: 500
    tokens_per_minute: 80000

# Psychology Configuration
psychology:
  nudge_types:
    - name: "friend_interest"
      psychology_principle: "social_proof"
      effectiveness_weight: 0.25
      min_friends_threshold: 1
    - name: "scarcity_alert"
      psychology_principle: "urgency"
      effectiveness_weight: 0.20
      min_spots_threshold: 5
    - name: "similar_taste"
      psychology_principle: "personalization"
      effectiveness_weight: 0.18
      similarity_threshold: 0.8
    - name: "host_reputation"
      psychology_principle: "authority"
      effectiveness_weight: 0.15
      min_rating_threshold: 4.5
    - name: "new_match"
      psychology_principle: "discovery"
      effectiveness_weight: 0.12
      compatibility_threshold: 0.75

  user_segmentation:
    engagement_levels:
      high: 
        min_events_attended: 3
        min_recent_activity_days: 7
      medium:
        min_events_attended: 1
        min_recent_activity_days: 14
      low:
        min_events_attended: 0
        min_recent_activity_days: 30
      dormant:
        max_recent_activity_days: 30

# Notification Frequency Limits
frequency_limits:
  max_notifications:
    daily: 1
    weekly: 3
    monthly: 8
  cooldown_periods:
    same_nudge_type: 24  # hours
    same_event: 12       # hours
    any_notification: 6  # hours
  priority_overrides:
    urgent_events: true
    friend_invitations: true

# Content Generation
content_generation:
  templates:
    max_length: 160
    personalization_depth: "high"
    creativity_level: 0.7
    tone: "enthusiastic"
  quality_control:
    grammar_check: true
    appropriateness_filter: true
    relevance_threshold: 0.8
    minimum_quality_score: 0.75

# Delivery Configuration
delivery:
  channels:
    push:
      provider: "onesignal"
      priority: 1
      max_retries: 3
      timeout_seconds: 10
    sms:
      provider: "twilio"
      priority: 2
      max_retries: 2
      timeout_seconds: 15
    email:
      provider: "mailchimp"
      priority: 3
      max_retries: 2
      timeout_seconds: 20
  timing_optimization:
    user_timezone_aware: true
    respect_do_not_disturb: true
    optimal_windows:
      morning: "09:00-11:00"
      lunch: "12:00-13:30"
      evening: "17:00-20:00"
  retry_strategy:
    backoff_multiplier: 2
    max_backoff_seconds: 300
    jitter_enabled: true

# Monitoring & Analytics
monitoring:
  logging:
    level: "INFO"
    format: "json"
    retention_days: 90
    sensitive_data_masking: true
  performance_tracking:
    enabled: true
    metrics_retention_days: 365
    real_time_alerts: true
  health_checks:
    interval_seconds: 300
    timeout_seconds: 10
    alert_thresholds:
      error_rate: 0.05
      response_time_ms: 2000
      queue_depth: 1000
  analytics:
    dashboard_refresh_seconds: 60
    historical_data_months: 12
    export_formats: ["json", "csv"]

# Testing Configuration
testing:
  mock_external_apis: true
  test_data_cleanup: true
  performance_benchmarks:
    max_response_time_ms: 1000
    max_memory_mb: 512
  quality_gates:
    min_test_coverage: 0.80
    max_complexity_score: 10

Nudge Templates config/nudge_templates.yaml

Social proof, urgency, and personalization patterns with variables and constraints.

⟶ 
templates:
  friend_interest:
    psychology_principle: "social_proof"
    base_template: "{friend_count} {friend_names} {interest_verb} {event_title}"
    variations:
      single_friend: "{friend_name} is interested in {event_title}!"
      multiple_friends: "{friend_count} friends are interested in {event_title}"
      close_friends: "Your friends {friend_names} are going to {event_title}"
    variables:
      friend_count: "integer"
      friend_names: "string_list"
      friend_name: "string"
      interest_verb: "enum[interested in, going to, excited about]"
      event_title: "string"
    constraints:
      max_length: 150
      min_friends: 1
      max_friends_shown: 3

  scarcity_alert:
    psychology_principle: "urgency"
    base_template: "Only {spots_remaining} spots left for {event_title}!"
    variations:
      low_availability: "Only {spots_remaining} spots left for {event_title}"
      very_low: "Last {spots_remaining} spots for {event_title} - grab yours!"
      filling_fast: "{event_title} is filling up fast - {spots_remaining} spots left"
    variables:
      spots_remaining: "integer"
      event_title: "string"
      urgency_modifier: "enum[quickly, fast, soon]"
    constraints:
      max_length: 140
      max_spots_threshold: 5
      min_spots_threshold: 1

  similar_taste:
    psychology_principle: "personalization"
    base_template: "Perfect {event_type} for your love of {cuisine_preference}"
    variations:
      cuisine_match: "Love {cuisine}? You'll enjoy this {event_title}"
      venue_return: "Another great event at {venue_name} - you loved it last time!"
      style_match: "Based on your preferences: {event_title}"
    variables:
      event_type: "string"
      cuisine_preference: "string"
      cuisine: "string"
      event_title: "string"
      venue_name: "string"
    constraints:
      max_length: 155
      min_similarity_score: 0.8

Delivery Channels config/delivery_channels.yaml

Provider credentials, formatting limits, and delivery options for push, SMS, and email.

⟶ 
channels:
  push:
    provider: "onesignal"
    configuration:
      app_id: "${ONESIGNAL_APP_ID}"
      api_key: "${ONESIGNAL_API_KEY}"
      timeout_seconds: 10
      retry_attempts: 3
    formatting:
      max_title_length: 50
      max_body_length: 160
      supports_emoji: true
      supports_rich_media: true
    delivery_options:
      immediate: true
      scheduled: true
      timezone_aware: true
      badge_updates: true

  sms:
    provider: "twilio"
    configuration:
      account_sid: "${TWILIO_ACCOUNT_SID}"
      auth_token: "${TWILIO_AUTH_TOKEN}"
      from_number: "${TWILIO_FROM_NUMBER}"
      timeout_seconds: 15
      retry_attempts: 2
    formatting:
      max_length: 160
      supports_emoji: false
      supports_links: true
      link_shortening: true
    delivery_options:
      immediate: true
      scheduled: true
      delivery_receipts: true

  email:
    provider: "mailchimp"
    configuration:
      api_key: "${MAILCHIMP_API_KEY}"
      list_id: "${MAILCHIMP_LIST_ID}"
      timeout_seconds: 20
      retry_attempts: 2
    formatting:
      max_subject_length: 78
      max_body_length: 5000
      supports_html: true
      supports_attachments: false
    delivery_options:
      immediate: true
      scheduled: true
      open_tracking: true
      click_tracking: true

Testing Configuration config/test_settings.yaml

Test database handling, mocking requirements, performance benchmarks, and quality gates.

⟶ 
testing:
  database:
    test_db_name: "cuculi_ai_test"
    cleanup_after_tests: true
    use_transactions: true
  
  mocking:
    mock_openai_api: true
    mock_delivery_channels: true
    mock_vector_search: false
    
  performance_benchmarks:
    max_response_time_ms: 1000
    max_memory_usage_mb: 512
    concurrent_request_limit: 100
    
  quality_gates:
    min_code_coverage: 0.85
    max_cyclomatic_complexity: 10
    min_test_success_rate: 0.98

Testing Framework

Quality gates ensure every function—configuration through analytics—behaves reliably under ideal and adverse conditions.

Unit Testing Requirements

Input Validation Tests ⟶
  • Valid input parameters with expected data types
  • Invalid/missing required parameters
  • Edge cases and boundary conditions
  • Malformed data structures
Output Verification Tests ⟶
  • Correct return data types and structures
  • Required fields presence in output objects
  • Data consistency and integrity checks
  • Performance benchmarks (response time, memory usage)
Error Handling Tests ⟶
  • Network connectivity failures
  • API rate limit handling
  • Database connection timeouts
  • Invalid configuration scenarios
  • Graceful degradation behaviors

Integration Testing

  • Database connection and query execution: Validate MongoDB setup, index creation, and aggregation pipelines end-to-end.
  • External API integrations: Ensure OpenAI and delivery channel connectors behave with mocks and live credentials.
  • End-to-end workflow validation: Simulate the entire notification journey from trigger through analytics logging.
  • Cross-function dependency testing: Confirm downstream functions respond correctly to upstream outputs and failure states.

Test Configuration Snapshot

Pulled directly from config/test_settings.yaml to enforce non-negotiable quality thresholds.

testing:
  database:
    test_db_name: "cuculi_ai_test"
    cleanup_after_tests: true
    use_transactions: true
  
  mocking:
    mock_openai_api: true
    mock_delivery_channels: true
    mock_vector_search: false
    
  performance_benchmarks:
    max_response_time_ms: 1000
    max_memory_usage_mb: 512
    concurrent_request_limit: 100
    
  quality_gates:
    min_code_coverage: 0.85
    max_cyclomatic_complexity: 10
    min_test_success_rate: 0.98

Putting It All Together

This Python library is more than a collection of helper functions—it is a disciplined operating system for Cuculi AI's psychology-driven notifications. Progressive disclosure gives stakeholders a way to scan quickly, dig deep, and trace every dependency before code ever hits production.

Return to Blog Index Discuss Implementation