#!/usr/bin/env python3 """ Rivet Voice Assistant - Real-time phone AI using Pipecat Simpler implementation using available Pipecat APIs """ import asyncio import base64 import json import os import sys from typing import Optional from dotenv import load_dotenv from loguru import logger import websockets # Pipecat imports - using non-deprecated paths from pipecat.services.anthropic.llm import AnthropicLLMService from pipecat.services.elevenlabs.tts import ElevenLabsTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.frames.frames import ( TextFrame, TranscriptionFrame, EndFrame, AudioRawFrame, ) # Load environment load_dotenv() # Configure logger logger.remove() logger.add(sys.stderr, level="INFO") logger.add("logs/{time:YYYY-MM-DD}.log", rotation="1 day", retention="7 days", level="DEBUG") # System prompt SYSTEM_PROMPT = """You are Rivet, the COO system for RateRight — a construction hiring marketplace in Australia. You're talking on the phone with Michael, the founder. Be direct, calm, and supportive. Keep responses SHORT (1-2 sentences). You're his right hand — you know the business, you push back when needed, you're honest. RateRight: $50 flat fee per hire. Workers pay nothing. Connects contractors with construction workers. PROACTIVE: Morning and evening calls have different purposes. **MORNING CALL (6:30 AM): Site Brief + Today's Priorities + Hard Questions** 1. Lead with site brief (weather, yesterday recap, today's plan) 2. Tell him the 2-3 most important things to do today (from TODO.md priorities) 3. Ask ONE hard question from /home/ccuser/rateright-growth/rivet/memory/todays-hard-questions.md - Morning questions focus on: what he's avoiding, what would make today a win, what he should prioritize 4. Get his instructions for the day **EVENING CALL (5:30 PM): Business Debrief + Accountability + Hard Questions** 1. Brief business update (what Rivet did today, overnight plan) 2. Ask what HE did today on RateRight (hold him accountable) 3. Ask ONE hard question from todays-hard-questions.md - Evening questions focus on: what didn't get done, why, what he's kicking down the road 4. Capture his handoff tasks for overnight work **SUNDAY CALLS:** - Weekly review: "What worked this week? What didn't? What's the one thing you need to fix next week?" **RULES:** - Read todays-hard-questions.md before each call (regenerated at 6 AM and 5 PM) - One hard question maximum per call - Make it conversational — "So I've been looking at the numbers..." - Follow up if he deflects - Remember answers for continuity You're not a customer service bot. You're Rivet — Michael's AI COO. Talk like a competent colleague, not a robot.""" class TwilioMediaStreamHandler: """Handle Twilio Media Stream protocol and connect to Pipecat services""" def __init__(self): # Initialize services self.deepgram = DeepgramSTTService( api_key=os.getenv("DEEPGRAM_API_KEY"), model="nova-2-conversationalai", language="en-AU", ) self.llm = AnthropicLLMService( api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-opus-4-5-20251101", ) self.tts = ElevenLabsTTSService( api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID", "XB0fDUnXU5powFXDhCwa"), # Charlotte - warm, professional, calm model="eleven_turbo_v2_5", ) self.conversation_history = [ {"role": "system", "content": SYSTEM_PROMPT} ] self.stream_sid = None self.call_sid = None logger.info("TwilioMediaStreamHandler initialized") async def handle_audio(self, audio_payload: str) -> Optional[bytes]: """ Process incoming audio from Twilio: 1. Decode base64 audio 2. Send to Deepgram for STT 3. Send transcription to Claude 4. Get response from Claude 5. Send to ElevenLabs for TTS 6. Return audio bytes """ try: # Decode audio (Twilio sends mulaw/8000) audio_bytes = base64.b64decode(audio_payload) # TODO: Implement STT -> LLM -> TTS pipeline # This is a simplified version - full implementation would use Pipecat's pipeline return None except Exception as e: logger.error(f"Error processing audio: {e}") return None async def handle_websocket(self, websocket, path): """Handle Twilio Media Stream websocket connection""" logger.info(f"New WebSocket connection: {path}") try: async for message in websocket: data = json.loads(message) event = data.get("event") if event == "start": self.stream_sid = data["start"]["streamSid"] self.call_sid = data["start"]["callSid"] logger.info(f"Stream started - Call: {self.call_sid}, Stream: {self.stream_sid}") # Send initial greeting greeting = "Hello! This is Rivet. How can I help you?" await self.send_text_response(websocket, greeting) elif event == "media": # Incoming audio from caller payload = data["media"]["payload"] response_audio = await self.handle_audio(payload) if response_audio: # Send audio back to Twilio await self.send_audio(websocket, response_audio) elif event == "stop": logger.info(f"Stream stopped: {self.stream_sid}") break except websockets.exceptions.ConnectionClosed: logger.info(f"WebSocket closed: {self.stream_sid}") except Exception as e: logger.error(f"Error in websocket handler: {e}", exc_info=True) async def send_text_response(self, websocket, text: str): """Convert text to speech and send to caller""" try: logger.info(f"Rivet: {text}") # Get audio from ElevenLabs # Note: This is simplified - actual implementation needs proper async handling audio_chunks = [] async for chunk in self.tts.run_tts(text): if hasattr(chunk, 'audio'): audio_chunks.append(chunk.audio) if audio_chunks: # Combine and send audio audio_data = b''.join(audio_chunks) await self.send_audio(websocket, audio_data) except Exception as e: logger.error(f"Error sending response: {e}") async def send_audio(self, websocket, audio_data: bytes): """Send audio to Twilio via Media Stream""" # Encode audio as base64 encoded_audio = base64.b64encode(audio_data).decode('utf-8') # Send in Twilio Media Stream format message = { "event": "media", "streamSid": self.stream_sid, "media": { "payload": encoded_audio } } await websocket.send(json.dumps(message)) async def start_server(): """Start WebSocket server for Twilio Media Streams""" host = os.getenv("HOST", "0.0.0.0") port = int(os.getenv("PORT", "8765")) logger.info("=== Rivet Voice Assistant ===") logger.info(f"Starting WebSocket server on {host}:{port}") logger.info(f"Webhook URL for Twilio: ws://{os.getenv('PUBLIC_URL', f'{host}:{port}').replace('http://', '')}/media-stream") handler = TwilioMediaStreamHandler() async with websockets.serve(handler.handle_websocket, host, port): logger.info("✅ Voice Assistant server running") logger.info("Waiting for incoming calls...") await asyncio.Future() # Run forever def main(): """Main entry point""" # Validate credentials required_keys = [ "ANTHROPIC_API_KEY", "ELEVENLABS_API_KEY", "DEEPGRAM_API_KEY", "TWILIO_ACCOUNT_SID", "TWILIO_AUTH_TOKEN" ] missing = [k for k in required_keys if not os.getenv(k)] if missing: logger.error(f"Missing required environment variables: {missing}") sys.exit(1) try: asyncio.run(start_server()) except KeyboardInterrupt: logger.info("Server stopped by user") except Exception as e: logger.error(f"Fatal error: {e}", exc_info=True) sys.exit(1) if __name__ == "__main__": main()