Upload 2 files

app.py

@@ -0,0 +1,219 @@
+#!/usr/bin/env python3
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import datetime
+import io
+import json
+import os
+import sys
+import wave
+
+import aiofiles
+import uvicorn
+from dotenv import load_dotenv
+from fastapi import FastAPI, WebSocket
+from fastapi.middleware.cors import CORSMiddleware
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
+from pipecat.serializers.exotel import ExotelFrameSerializer
+from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.network.fastapi_websocket import (
+    FastAPIWebsocketParams,
+    FastAPIWebsocketTransport,
+)
+
+# Load environment variables
+load_dotenv(override=True)
+
+# Configure logger
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+# Create FastAPI app
+app = FastAPI(title="Pipecat Exotel Voice Chatbot", version="1.0.0")
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Allow all origins for testing
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+async def save_audio(server_name: str, audio: bytes, sample_rate: int, num_channels: int):
+    """Save audio recording to file"""
+    if len(audio) > 0:
+        filename = (
+            f"{server_name}_recording_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav"
+        )
+        with io.BytesIO() as buffer:
+            with wave.open(buffer, "wb") as wf:
+                wf.setsampwidth(2)
+                wf.setnchannels(num_channels)
+                wf.setframerate(sample_rate)
+                wf.writeframes(audio)
+            async with aiofiles.open(filename, "wb") as file:
+                await file.write(buffer.getvalue())
+        logger.info(f"Audio saved to {filename}")
+    else:
+        logger.info("No audio data to save")
+
+
+async def run_bot(websocket_client: WebSocket, call_id: str, testing: bool):
+    """Run the voice chatbot pipeline"""
+    transport = FastAPIWebsocketTransport(
+        websocket=websocket_client,
+        params=FastAPIWebsocketParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            add_wav_header=False,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer(),
+            vad_audio_passthrough=True,
+            serializer=ExotelFrameSerializer(stream_sid=call_id),
+        ),
+    )
+
+    # Initialize services
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"), audio_passthrough=True)
+    tts = ElevenLabsTTSService(
+        api_key=os.getenv("ELEVEN_API_KEY"),
+        voice_id=os.getenv("ELEVEN_VOICE_ID"),
+    )
+
+    # System message
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful assistant named Tasha. Your output will be converted to audio so don't include special characters in your answers. Respond with a short sentence.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+    audiobuffer = AudioBufferProcessor(user_continuous_stream=not testing)
+
+    # Create pipeline
+    pipeline = Pipeline(
+        [
+            transport.input(),  # WebSocket input from client
+            stt,  # Speech-To-Text
+            context_aggregator.user(),
+            llm,  # LLM
+            tts,  # Text-To-Speech
+            transport.output(),  # WebSocket output to client
+            audiobuffer,  # Used to buffer the audio in the pipeline
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            audio_in_sample_rate=8000,
+            audio_out_sample_rate=8000,
+            allow_interruptions=True,
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        # Start recording
+        await audiobuffer.start_recording()
+        # Kick off the conversation
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        await task.cancel()
+
+    @audiobuffer.event_handler("on_audio_data")
+    async def on_audio_data(buffer, audio, sample_rate, num_channels):
+        server_name = f"server_{websocket_client.client.port}"
+        await save_audio(server_name, audio, sample_rate, num_channels)
+
+    # Run the pipeline
+    runner = PipelineRunner(handle_sigint=False, force_gc=True)
+    await runner.run(task)
+
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    return {"status": "healthy", "message": "Pipecat Exotel Voice Chatbot is running"}
+
+
+@app.websocket("/media")
+async def media_endpoint(websocket: WebSocket):
+    """WebSocket endpoint for Exotel media streaming"""
+    await websocket.accept()
+    print("[*] Exotel WebSocket connection accepted")
+    
+    try:
+        while True:
+            data = await websocket.receive_text()
+            message = json.loads(data)
+            print(f"\n[*] Received Exotel message: {json.dumps(message, indent=2)}")
+            
+            # Handle start event to get call information
+            if message.get('event') == 'start':
+                stream_sid = message.get('streamSid', 'default')
+                print(f"Starting bot for stream SID: {stream_sid}")
+                await run_bot(websocket, stream_sid, app.state.testing)
+                break
+                
+    except Exception as e:
+        print(f"[!] Error in media endpoint: {str(e)}")
+
+
+def main():
+    """Main function to run the server"""
+    parser = argparse.ArgumentParser(description="Pipecat Exotel Voice Chatbot Server")
+    parser.add_argument(
+        "-t", "--test", action="store_true", default=False, help="set the server in testing mode"
+    )
+    parser.add_argument(
+        "-p", "--port", type=int, default=8005, help="port to run the server on"
+    )
+    args, _ = parser.parse_known_args()
+
+    app.state.testing = args.test
+
+    print(f"[*] Pipecat Exotel Voice Chatbot running on port {args.port}")
+    print(f"[*] Health check: http://localhost:{args.port}/health")
+    print(f"[*] WebSocket endpoint: ws://localhost:{args.port}/media")
+    print(f"[*] Testing mode: {'enabled' if args.test else 'disabled'}")
+    print("")
+    print("Required environment variables:")
+    print("  OPENAI_API_KEY - Your OpenAI API key")
+    print("  DEEPGRAM_API_KEY - Your Deepgram API key")
+    print("  ELEVEN_API_KEY - Your ElevenLabs API key")
+    print("  ELEVEN_VOICE_ID - Your ElevenLabs voice ID")
+    print("")
+    
+    uvicorn.run(
+        app,
+        host="0.0.0.0",
+        port=args.port
+    )
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,6 @@
+pipecat-ai[elevenlabs,openai,silero,deepgram]
+fastapi
+uvicorn
+python-dotenv
+loguru
+aiofiles