Create main.py

main.py

@@ -0,0 +1,216 @@
+import asyncio
+import os
+import sys
+import time
+import logging
+from pipecat.frames.frames import (
+    TextFrame, AudioFrame, CallEndEvent, UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame, LLMResponseStartFrame, LLMResponseEndFrame,
+    TTSStartedFrame, TTSEndFrame
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams
+from pipecat.processors.frame_processor import FrameProcessor, FrameDirection
+from pipecat.services.elevenlabs import ElevenLabsTTSService
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+from azure_openai import AzureOpenAILLMService
+from elevenlabs import ElevenLabs
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+# Configuration constants
+SILENCE_TIMEOUT_SECONDS = float(os.environ.get("SILENCE_TIMEOUT_SECONDS", 10))
+MAX_SILENCE_PROMPTS = int(os.environ.get("MAX_SILENCE_PROMPTS", 3))
+SILENCE_PROMPT_TEXT = "Are you still there?"
+GOODBYE_PROMPT_TEXT = "It seems you're no longer there. I'm hanging up now. Goodbye."
+
+class SilenceAndCallLogicProcessor(FrameProcessor):
+    def __init__(self, tts_service, pipeline, app):
+        super().__init__()
+        self.tts_service = tts_service
+        self.pipeline = pipeline
+        self.app = app
+        self.last_activity_ts = time.time()
+        self.silence_prompts_count = 0
+        self._bot_is_speaking = False
+        self._silence_check_task = None
+        self.app.current_call_stats["silence_events"] = 0
+
+    async def start(self):
+        self.last_activity_ts = time.time()
+        self.silence_prompts_count = 0
+        self._bot_is_speaking = False
+        if self._silence_check_task:
+            self._silence_check_task.cancel()
+        self._silence_check_task = asyncio.create_task(self._check_silence_loop())
+
+    async def stop(self):
+        if self._silence_check_task:
+            self._silence_check_task.cancel()
+            try:
+                await self._silence_check_task
+            except asyncio.CancelledError:
+                pass
+        await self.tts_service.stop()
+
+    def _reset_activity_timer(self):
+        self.last_activity_ts = time.time()
+        self.silence_prompts_count = 0
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        if isinstance(frame, (UserStartedSpeakingFrame, TextFrame)) and direction == FrameDirection.UPSTREAM:
+            self._reset_activity_timer()
+        if isinstance(frame, (LLMResponseStartFrame, TTSStartedFrame)) and direction == FrameDirection.DOWNSTREAM:
+            self._bot_is_speaking = True
+        elif isinstance(frame, (LLMResponseEndFrame, TTSEndFrame)) and direction == FrameDirection.DOWNSTREAM:
+            self._bot_is_speaking = False
+            self.last_activity_ts = time.time()
+        await self.push_frame(frame, direction)
+
+    async def _check_silence_loop(self):
+        while True:
+            await asyncio.sleep(1)
+            if self._bot_is_speaking:
+                continue
+            if time.time() - self.last_activity_ts > SILENCE_TIMEOUT_SECONDS:
+                self.app.current_call_stats["silence_events"] += 1
+                self.silence_prompts_count += 1
+                self._bot_is_speaking = True
+                if self.silence_prompts_count >= MAX_SILENCE_PROMPTS:
+                    await self.push_frame(TextFrame(GOODBYE_PROMPT_TEXT), FrameDirection.DOWNSTREAM)
+                    await asyncio.sleep(2)
+                    await self.pipeline.stop_when_done()
+                    break
+                else:
+                    await self.push_frame(TextFrame(SILENCE_PROMPT_TEXT), FrameDirection.DOWNSTREAM)
+                    self.last_activity_ts = time.time()
+                    self._bot_is_speaking = False
+
+class PhoneChatbotApp:
+    def __init__(self):
+        self.daily_transport = None
+        self.pipeline = None
+        self.stt_service = None
+        self.tts_service = None
+        self.llm_service = None
+        self.silence_processor = None
+        self.call_start_time = None
+        self.current_call_stats = {
+            "duration_seconds": 0,
+            "silence_events": 0,
+            "start_time": None,
+            "end_time": None,
+            "ended_by_silence": False
+        }
+
+    def _reset_call_stats(self):
+        self.call_start_time = time.time()
+        self.current_call_stats = {
+            "duration_seconds": 0,
+            "silence_events": 0,
+            "start_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.call_start_time)),
+            "end_time": None,
+            "ended_by_silence": False
+        }
+
+    async def _log_call_summary(self):
+        if self.call_start_time:
+            call_end_time = time.time()
+            self.current_call_stats["duration_seconds"] = round(call_end_time - self.call_start_time, 2)
+            self.current_call_stats["end_time"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(call_end_time))
+            if self.silence_processor and self.silence_processor.silence_prompts_count >= MAX_SILENCE_PROMPTS:
+                self.current_call_stats["ended_by_silence"] = True
+            logger.info("--- Post-Call Summary ---")
+            for key, value in self.current_call_stats.items():
+                logger.info(f"  {key.replace('_', ' ').title()}: {value}")
+            logger.info("-------------------------")
+
+    def setup_pipeline_hook(self, pipeline_params: PipelineParams, room_url: str, token: str):
+        self._reset_call_stats()
+        self.pipeline = Pipeline([self.stt_service, self.llm_service, self.tts_service])
+        self.silence_processor = SilenceAndCallLogicProcessor(
+            tts_service=self.tts_service,
+            pipeline=self.pipeline,
+            app=self
+        )
+        self.pipeline.processors.append(self.silence_processor)
+        pipeline_params.pipeline = self.pipeline
+        pipeline_params.params = PipelineParams(allow_interruptions=True, enable_metrics=True)
+        return pipeline_params
+
+    def validate_voice_id(self, voice_id: str) -> bool:
+        try:
+            client = ElevenLabs(api_key=os.environ.get("elevenlabs"))
+            client.voices.get(voice_id=voice_id)
+            return True
+        except Exception as e:
+            logger.error(f"Failed to validate ElevenLabs voice ID {voice_id}: {e}")
+            return False
+
+    async def run(self):
+        # Validate environment variables (Hugging Face Secrets)
+        required_keys = [
+            "deepgram", "elevenlabs", "dailyco", "azure_openai"
+        ]
+        missing_keys = [key for key in required_keys if not os.environ.get(key)]
+        if missing_keys:
+            logger.error(f"Missing Hugging Face Secrets: {', '.join(missing_keys)}")
+            sys.exit(1)
+
+        # Validate ElevenLabs voice ID
+        voice_id = os.environ.get("ELEVENLABS_VOICE_ID", "cgSgspJ2msm6clMCkdW9")
+        if not self.validate_voice_id(voice_id):
+            logger.error(f"Invalid ElevenLabs voice ID: {voice_id}")
+            sys.exit(1)
+
+        self.stt_service = DeepgramSTTService(
+            api_key=os.environ.get("deepgram"),
+            input_audio_format="linear16"
+        )
+        self.tts_service = ElevenLabsTTSService(
+            api_key=os.environ.get("elevenlabs"),
+            voice_id=voice_id
+        )
+        self.llm_service = AzureOpenAILLMService(
+            preprompt="You are a friendly and helpful phone assistant."
+        )
+        self.daily_transport = DailyTransport(
+            os.environ.get("DAILY_DOMAIN", "your-username.daily.co"),
+            os.environ.get("dailyco"),
+            None,
+            None,
+            "Pipecat Phone Demo",
+            vad_analyzer=SileroVADAnalyzer(),
+            daily_params=DailyParams(
+                audio_in_enabled=True,
+                audio_out_enabled=True,
+                transcription_enabled=False
+            )
+        )
+        self.daily_transport.pipeline_params_hook = self.setup_pipeline_hook
+
+        runner = PipelineRunner()
+        try:
+            await runner.run(self.daily_transport)
+        except KeyboardInterrupt:
+            logger.info("Ctrl+C pressed, shutting down")
+        except Exception as e:
+            logger.error(f"An error occurred: {e}", exc_info=True)
+        finally:
+            await self._log_call_summary()
+            if self.pipeline:
+                await self.pipeline.stop_when_done()
+            if self.silence_processor:
+                await self.silence_processor.stop()
+
+async def main():
+    app = PhoneChatbotApp()
+    await app.run()
+
+if __name__ == "__main__":
+    asyncio.run(main())