ajarbot/example_bot_with_pulse_brain.py

"""
Example: Bot with Pulse & Brain monitoring.

The most cost-effective approach:
- Pulse checks run constantly (pure Python, $0 cost)
- Brain only invokes Agent when needed
- 92% cost savings vs always-on agent

Usage:
    python example_bot_with_pulse_brain.py
"""

import asyncio

from adapters.base import AdapterConfig
from adapters.runtime import AdapterRuntime
from adapters.slack.adapter import SlackAdapter
from adapters.telegram.adapter import TelegramAdapter
from agent import Agent
from pulse_brain import PulseBrain

# Cost estimation constants
_AVERAGE_TOKENS_PER_CALL = 1000
_COST_PER_TOKEN = 0.000003


async def main() -> None:
    print("=" * 60)
    print("Ajarbot with Pulse & Brain")
    print("=" * 60)
    print("\nPulse: Pure Python checks (zero cost)")
    print("Brain: Agent/SDK (only when needed)\n")

    # 1. Create agent
    agent = Agent(
        provider="claude",
        workspace_dir="./memory_workspace",
        enable_heartbeat=False,
    )

    # 2. Create runtime with adapters
    runtime = AdapterRuntime(agent)

    slack_adapter = SlackAdapter(AdapterConfig(
        platform="slack",
        enabled=True,
        credentials={
            "bot_token": "xoxb-YOUR-TOKEN",
            "app_token": "xapp-YOUR-TOKEN",
        },
    ))
    runtime.add_adapter(slack_adapter)

    telegram_adapter = TelegramAdapter(AdapterConfig(
        platform="telegram",
        enabled=True,
        credentials={"bot_token": "YOUR-TELEGRAM-TOKEN"},
    ))
    runtime.add_adapter(telegram_adapter)

    # 3. Create Pulse & Brain system
    pb = PulseBrain(agent, pulse_interval=60)

    pb.add_adapter("slack", slack_adapter)
    pb.add_adapter("telegram", telegram_adapter)

    # Optional: Apply custom configuration
    try:
        from config.pulse_brain_config import apply_custom_config
        apply_custom_config(pb)
        print("[Setup] Custom pulse/brain config loaded")
    except ImportError:
        print("[Setup] Using default pulse/brain config")

    # 4. Show configuration
    print("\n" + "=" * 60)
    print("Configuration")
    print("=" * 60)

    print(f"\nPulse checks ({len(pb.pulse_checks)}):")
    for check in pb.pulse_checks:
        print(
            f"  [Pulse] {check.name} "
            f"(every {check.interval_seconds}s)"
        )

    print(f"\nBrain tasks ({len(pb.brain_tasks)}):")
    for task in pb.brain_tasks:
        if task.check_type.value == "scheduled":
            print(
                f"  [Brain] {task.name} "
                f"(scheduled {task.schedule_time})"
            )
        else:
            print(f"  [Brain] {task.name} (conditional)")

    # 5. Start everything
    print("\n" + "=" * 60)
    print("Starting system...")
    print("=" * 60 + "\n")

    await runtime.start()
    pb.start()

    print("\n" + "=" * 60)
    print("System is running!")
    print("=" * 60)
    print("\nWhat's happening:")
    print("  - Users can chat with bot on Slack/Telegram")
    print("  - Pulse checks run every 60s (zero cost)")
    print("  - Brain only invokes when:")
    print("    - Error detected")
    print("    - Threshold exceeded")
    print("    - Scheduled time (8am, 6pm)")
    print("\nPress Ctrl+C to stop.\n")

    try:
        iteration = 0
        while True:
            await asyncio.sleep(30)
            iteration += 1

            if iteration % 2 == 0:
                status = pb.get_status()
                invocations = status["brain_invocations"]
                print(
                    f"[Status] Brain invoked "
                    f"{invocations} times"
                )

    except KeyboardInterrupt:
        print("\n\n[Shutdown] Stopping system...")
    finally:
        pb.stop()
        await runtime.stop()

        final_status = pb.get_status()
        invocations = final_status["brain_invocations"]
        total_tokens = invocations * _AVERAGE_TOKENS_PER_CALL
        estimated_cost = total_tokens * _COST_PER_TOKEN

        print("\n" + "=" * 60)
        print("Final Statistics")
        print("=" * 60)
        print(f"Brain invocations: {invocations}")
        print(f"Estimated tokens: {total_tokens:,}")
        print(f"Estimated cost: ${estimated_cost:.4f}")
        print("\nCompare to always-on agent:")
        print("  Pulse checks: FREE")
        print(
            f"  Brain calls: {invocations} "
            f"(only when needed)"
        )
        print(
            "  Savings: ~92% vs running agent every minute"
        )


if __name__ == "__main__":
    asyncio.run(main())