Robert Helewka
ecf37658ce
Complete project scaffolding and core implementation of an AI-powered telephony system that calls companies, navigates IVR menus, waits on hold, and transfers to the user when a human answers. Key components: - FastAPI server with REST API, WebSocket, and MCP (SSE) interfaces - SIP/VoIP call management via PJSUA2 with RTP audio streaming - LLM-powered IVR navigation using OpenAI/Anthropic with tool calling - Hold detection service combining audio analysis and silence detection - Real-time STT (Whisper/Deepgram) and TTS (OpenAI/Piper) pipelines - Call recording with per-channel and mixed audio capture - Event bus (asyncio pub/sub) for real-time client updates - Web dashboard with live call monitoring - SQLite persistence via SQLAlchemy with call history and analytics - Notification support (email, SMS, webhook, desktop) - Docker Compose deployment with Opal VoIP and Opal Media containers - Comprehensive test suite with unit, integration, and E2E tests - Simplified .gitignore and full project documentation in README
109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
"""
|
|
Call Flow models — IVR navigation trees.
|
|
|
|
Store known IVR structures for phone numbers you call regularly.
|
|
The Hold Slayer follows the map instead of exploring blind.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from typing import Optional
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class ActionType(str, Enum):
|
|
"""Actions the Hold Slayer can take at each IVR step."""
|
|
|
|
DTMF = "dtmf" # Press a button
|
|
SPEAK = "speak" # Say something (for speech-recognition IVRs)
|
|
WAIT = "wait" # Wait for prompt
|
|
LISTEN = "listen" # Listen and let LLM decide
|
|
HOLD = "hold" # On hold — activate hold detection
|
|
TRANSFER = "transfer" # Transfer to user's device
|
|
|
|
|
|
class CallFlowStep(BaseModel):
|
|
"""A single step in an IVR navigation tree."""
|
|
|
|
id: str
|
|
description: str # Human-readable: "Main menu"
|
|
expect: Optional[str] = None # What we expect to hear (regex or keywords)
|
|
action: ActionType
|
|
action_value: Optional[str] = None # DTMF digit(s), speech text, device target
|
|
timeout: int = 30 # Seconds to wait before retry/fallback
|
|
next_step: Optional[str] = None # Next step ID on success
|
|
fallback_step: Optional[str] = None # Step ID if unexpected response
|
|
notes: Optional[str] = None # "They changed this menu in Jan 2025"
|
|
|
|
|
|
class CallFlow(BaseModel):
|
|
"""A complete IVR navigation tree for a phone number."""
|
|
|
|
id: str
|
|
name: str # "Chase Bank - Main Line"
|
|
phone_number: str # "+18005551234"
|
|
description: str = ""
|
|
last_verified: Optional[datetime] = None
|
|
steps: list[CallFlowStep]
|
|
tags: list[str] = Field(default_factory=list)
|
|
notes: Optional[str] = None
|
|
|
|
# Stats from previous runs
|
|
avg_hold_time: Optional[int] = None # seconds
|
|
success_rate: Optional[float] = None # 0.0 - 1.0
|
|
last_used: Optional[datetime] = None
|
|
times_used: int = 0
|
|
|
|
def get_step(self, step_id: str) -> Optional[CallFlowStep]:
|
|
"""Look up a step by ID."""
|
|
for step in self.steps:
|
|
if step.id == step_id:
|
|
return step
|
|
return None
|
|
|
|
def first_step(self) -> Optional[CallFlowStep]:
|
|
"""Get the first step in the flow."""
|
|
return self.steps[0] if self.steps else None
|
|
|
|
def steps_by_id(self) -> dict[str, CallFlowStep]:
|
|
"""Return a dict mapping step ID -> step for fast lookups."""
|
|
return {s.id: s for s in self.steps}
|
|
|
|
|
|
class CallFlowCreate(BaseModel):
|
|
"""Request model for creating a new call flow."""
|
|
|
|
name: str
|
|
phone_number: str
|
|
description: str = ""
|
|
steps: list[CallFlowStep]
|
|
tags: list[str] = Field(default_factory=list)
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class CallFlowUpdate(BaseModel):
|
|
"""Request model for updating an existing call flow."""
|
|
|
|
name: Optional[str] = None
|
|
description: Optional[str] = None
|
|
steps: Optional[list[CallFlowStep]] = None
|
|
tags: Optional[list[str]] = None
|
|
notes: Optional[str] = None
|
|
last_verified: Optional[datetime] = None
|
|
|
|
|
|
class CallFlowSummary(BaseModel):
|
|
"""Lightweight summary for list views."""
|
|
|
|
id: str
|
|
name: str
|
|
phone_number: str
|
|
description: str = ""
|
|
step_count: int
|
|
avg_hold_time: Optional[int] = None
|
|
success_rate: Optional[float] = None
|
|
last_used: Optional[datetime] = None
|
|
times_used: int = 0
|
|
tags: list[str] = Field(default_factory=list)
|