Gemini API

User says: "Play Shape of You on Spotify"
           ↓
    Whisper transcribes (accurate)
           ↓
    Gemini analyzes intent
           ↓
    Result: Actionable = YES
           ↓
    Forward to ClawdBot → Execute

async def _analyze_intent(self, user_text: str) -> tuple[str, dict | None, bool]:
    """Analyze user text to determine if it's actionable."""
    
    # Quick heuristics first (skip LLM for obvious cases)
    if text_lower in ["hi", "hello", "thanks", "ok"]:
        return "conversation", None, False
    
    if any(kw in text_lower for kw in ["open", "play", "search", "send"]):
        return "action", {"original_request": user_text}, True
    
    # Use Gemini for ambiguous cases
    prompt = f"""
    Is this a request to DO something?
    
    User said: "{user_text}"
    
    Answer with just ONE word: YES or NO
    """
    
    response = self.client.models.generate_content(
        model="gemini-3-flash-preview",
        contents=prompt,
    )
    
    is_actionable = response.text.strip().upper().startswith("YES")
    return ("action" if is_actionable else "conversation", ...)

# Skip LLM for greetings (saves 300-800ms)
non_actionable_phrases = [
    "hi", "hello", "hey", "thanks", "ok",
    "bye", "goodbye", "nevermind",
]

if text_lower in non_actionable_phrases:
    return "conversation", None, False

# Quick actionable keywords (skip LLM, go to ClawdBot)
action_keywords = [
    "open", "play", "search", "send",
    "email", "youtube", "spotify",
]

if any(kw in text_lower for kw in action_keywords):
    return "action", {"original_request": text}, True

class ConversationBrain:
    def __init__(
        self,
        api_key: str,
        model: str = "gemini-3-flash-preview",
        telegram_chat_id: str = "",
        call_id: str = "",
    ):
        """Initialize the conversation brain.
        
        Args:
            api_key: Gemini API key
            model: Model for intent understanding
            telegram_chat_id: Telegram chat ID for ClawdBot
            call_id: Call identifier
        """

    def set_callbacks(
        self,
        on_command: Callable[[str, dict], Awaitable[None]] | None = None,
        on_clawdbot_response: Callable[[str], Awaitable[None]] | None = None,
    ):
        """Set event callbacks."""

    def add_user_transcript(self, text: str):
        """Add user transcript fragment."""

    async def flush_user_turn(self):
        """Flush buffered user transcript and analyze intent."""

    def add_assistant_transcript(self, text: str):
        """Add assistant transcript fragment."""

    async def flush_assistant_turn(self):
        """Flush buffered assistant transcript."""

    def get_memory_summary(self) -> str:
        """Get conversation summary."""

import asyncio
from agenticai.core.conversation_brain import ConversationBrain

async def main():
    brain = ConversationBrain(
        api_key="AIzaxxx",
        model="gemini-3-flash-preview",
        call_id="call_123",
    )
    
    # Set callbacks
    async def handle_command(action: str, params: dict):
        print(f"Execute: {action} with {params}")
    
    async def speak_response(text: str):
        print(f"Speak: {text}")
    
    brain.set_callbacks(
        on_command=handle_command,
        on_clawdbot_response=speak_response,
    )
    
    # Add user transcript
    brain.add_user_transcript("Play")
    brain.add_user_transcript(" Spotify")
    
    # Flush and analyze
    await brain.flush_user_turn()
    # Output: Execute: execute_command with {'command': 'Play Spotify'}

asyncio.run(main())

from agenticai.gemini.realtime_handler import GeminiRealtimeHandler

handler = GeminiRealtimeHandler(
    api_key="AIzaxxx",
    model="models/gemini-2.5-flash-native-audio-latest",
    voice="Zephyr",
)

# Similar API to OpenAI Realtime
handler.set_callbacks(
    on_audio=handle_audio,
    on_user_transcript=handle_transcript,
)

await handler.connect()
await handler.send_audio(audio_bytes)

@dataclass 
class ConversationMemory:
    """Memory for the conversation."""
    call_id: str
    turns: list[ConversationTurn] = field(default_factory=list)
    context: dict = field(default_factory=dict)
    extracted_info: dict = field(default_factory=dict)
    
    def add_turn(self, speaker: str, text: str, intent: str = None, command: dict = None):
        """Add a conversation turn."""
        turn = ConversationTurn(
            speaker=speaker,
            text=text,
            timestamp=datetime.now(),
            intent=intent,
            command=command,
        )
        self.turns.append(turn)
    
    def get_recent_context(self, max_turns: int = 10) -> str:
        """Get recent conversation as string."""
        recent = self.turns[-max_turns:]
        lines = []
        for turn in recent:
            speaker = "User" if turn.speaker == "user" else "Assistant"
            lines.append(f"{speaker}: {turn.text}")
        return "\n".join(lines)

context = self.memory.get_recent_context(5)

prompt = f"""
Recent conversation:
{context}

User said: "{user_text}"

Is this a request to DO something? YES or NO
"""