added voice settings

Author: sethjuarez

api/chat/__init__.py

@@ -1,5 +1,5 @@
 import json
-from typing import List
+from typing import List, Union
 import prompty
 import prompty.azure
 from prompty.tracer import trace
@@ -10,7 +10,10 @@
 
 @trace
 async def create_response(
-    customer: str, question: str, context: List[str] = [], image: str = None
+    customer: str,
+    question: str,
+    context: List[str] = [],
+    image: Union[str, None] = None,
 ):
     inputs = {"customer": customer, "question": question, "context": context}
     if image:
@@ -26,7 +29,7 @@ async def create_response(
 
     customer = "Seth Juarez"
     question = "My friend just sent me this and I'm worried I don't have the right gear for my camping trip. Can you help me? CALL ME"
-    context = []
+    context: List[str] = []
     image = "winter.jpg"
 
     asyncio.run(create_response(customer, question, context, image))

api/main.py

@@ -5,17 +5,15 @@
 from typing import List
 from fastapi.responses import StreamingResponse
 from jinja2 import Environment, FileSystemLoader
-from rtclient import RTLowLevelClient
-from api import repeat
+
+from rtclient import RTLowLevelClient  # type: ignore
 from api.realtime import RealtimeVoiceClient
 from api.session import Message, RealtimeSession, SessionManager
 from contextlib import asynccontextmanager
 from fastapi.middleware.cors import CORSMiddleware
 from azure.core.credentials import AzureKeyCredential
 from fastapi import FastAPI, WebSocket, WebSocketDisconnect
-from api.session import SessionManager
 from api.suggestions import SimpleMessage, create_suggestion, suggestion_requested
-from prompty.tracer import Tracer, trace
 from dotenv import load_dotenv
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 
@@ -24,7 +22,7 @@
 load_dotenv()
 
 AZURE_VOICE_ENDPOINT = os.getenv("AZURE_VOICE_ENDPOINT")
-AZURE_VOICE_KEY = os.getenv("AZURE_VOICE_KEY")
+AZURE_VOICE_KEY = os.getenv("AZURE_VOICE_KEY", "fake_key")
 
 LOCAL_TRACING_ENABLED = os.getenv("LOCAL_TRACING_ENABLED", "true") == "true"
 init_tracing(local_tracing=LOCAL_TRACING_ENABLED)
@@ -120,20 +118,34 @@ async def voice_endpoint(websocket: WebSocket):
             message = Message(**chat_items)
 
             # get current username
+            # and receive any parameters
             user_message = await websocket.receive_json()
             user = Message(**user_message)
 
+            settings = json.loads(user.payload)
+            print(
+                "Starting voice session with settings:\n",
+                json.dumps(settings, indent=2),
+            )
+
             # create voice system message
             # TODO: retrieve context from chat messages via thread id
             system_message = env.get_template("script.jinja2").render(
-                customer=user.payload,
+                customer=settings["user"] if "user" in settings else "Seth",
                 purchases=purchases,
                 context=json.loads(message.payload),
                 products=products,
             )
 
             session = RealtimeSession(RealtimeVoiceClient(rt, verbose=False), websocket)
-            await session.send_realtime_instructions(system_message)
+            await session.send_realtime_instructions(
+                system_message,
+                threshold=settings["threshold"] if "threshold" in settings else 0.8,
+                silence_duration_ms=(
+                    settings["silence"] if "silence" in settings else 500
+                ),
+                prefix_padding_ms=(settings["prefix"] if "prefix" in settings else 300),
+            )
             tasks = [
                 asyncio.create_task(session.receive_realtime()),
                 asyncio.create_task(session.receive_client()),
@@ -144,10 +156,4 @@ async def voice_endpoint(websocket: WebSocket):
         print("Voice Socket Disconnected", e)
 
 
-@repeat(seconds=60)
-@trace
-async def cleanup_sessions():
-    await SessionManager.clear_closed_sessions()
-
-
 FastAPIInstrumentor.instrument_app(app, exclude_spans=["send", "receive"])

api/models.py

@@ -1,5 +1,5 @@
 from pydantic import BaseModel
-from typing import List, Literal, Optional
+from typing import Literal, Optional
 
 
 class Action(BaseModel):

api/realtime.py

@@ -1,7 +1,8 @@
 from dataclasses import dataclass
-import json
-from typing import Any, AsyncGenerator
-from rtclient import (
+from typing import Any, AsyncGenerator, Union
+from prompty.tracer import trace
+
+from rtclient import (  # type: ignore
     InputAudioBufferAppendMessage,
     InputAudioBufferClearMessage,
     InputAudioTranscription,
@@ -14,8 +15,7 @@
     SessionUpdateParams,
     SystemMessageItem,
     UserMessageItem,
-)
-from prompty.tracer import trace, Tracer
+) 
 
 # things to have in here
 # - send user message
@@ -125,16 +125,22 @@ async def send_system_message(self, message: str):
         )
 
     @trace
-    async def send_session_update(self, instructions: str = None):
+    async def send_session_update(
+        self,
+        instructions: Union[str | None] = None,
+        threshold: float = 0.8,
+        silence_duration_ms: int = 500,
+        prefix_padding_ms: int = 300,
+    ):
         if self.client is None:
             raise Exception("Client not set")
 
         session = SessionUpdateParams(
             turn_detection=ServerVAD(
                 type="server_vad",
-                threshold=0.8,
-                silence_duration_ms=500,
-                prefix_padding_ms=300,
+                threshold=threshold,
+                silence_duration_ms=silence_duration_ms,
+                prefix_padding_ms=prefix_padding_ms,
             ),
             input_audio_transcription=InputAudioTranscription(model="whisper-1"),
             voice="shimmer",

api/requirements.txt

@@ -1,7 +1,7 @@
 fastapi[standard]
 websockets
 python-dotenv
-prompty[azure,serverless]==0.1.39
+prompty[azure,serverless]==0.1.47
 opentelemetry-instrumentation
 azure-monitor-opentelemetry-exporter
 opentelemetry-instrumentation-fastapi

api/session.py

@@ -1,8 +1,9 @@
 import asyncio
 import json
+from typing import Dict, List, Literal, Union
+from fastapi import WebSocket
 from prompty.tracer import trace
-from typing import Dict, List, Literal
-from fastapi import WebSocket, WebSocketDisconnect
+from fastapi import WebSocketDisconnect
 from pydantic import BaseModel
 from prompty.tracer import Tracer
 from fastapi.websockets import WebSocketState
@@ -26,26 +27,38 @@ class Message(BaseModel):
 class RealtimeSession:
 
     def __init__(self, realtime: RealtimeVoiceClient, client: WebSocket):
-        self.realtime: RealtimeVoiceClient = realtime
-        self.client: WebSocket = client
+        self.realtime: Union[RealtimeVoiceClient, None] = realtime
+        self.client: Union[WebSocket, None] = client
 
     async def send_message(self, message: Message):
-        await self.client.send_json(message.model_dump())
+        if self.client is not None:
+            await self.client.send_json(message.model_dump())
 
     async def send_audio(self, audio: Message):
         # send audio to client, format into bytes
-        await self.client.send_json(audio.model_dump())
+        if self.client is not None:
+            await self.client.send_json(audio.model_dump())
 
     async def send_console(self, message: Message):
-        await self.client.send_json(message.model_dump())
-
-    async def send_realtime_instructions(self, instructions: str):
-        await self.realtime.send_session_update(instructions)
+        if self.client is not None:
+            await self.client.send_json(message.model_dump())
+
+    async def send_realtime_instructions(
+        self,
+        instructions: Union[str | None] = None,
+        threshold: float = 0.8,
+        silence_duration_ms: int = 500,
+        prefix_padding_ms: int = 300,
+    ):
+        if self.realtime is not None:
+            await self.realtime.send_session_update(
+                instructions, threshold, silence_duration_ms, prefix_padding_ms
+            )
 
     @trace
     async def receive_realtime(self):
         signature = "api.session.RealtimeSession.receive_realtime"
-        while self.realtime != None and not self.realtime.closed:
+        while self.realtime is not None and not self.realtime.closed:
             async for message in self.realtime.receive_message():
                 # print("received message", message.type)
                 if message is None:
@@ -57,7 +70,9 @@ async def receive_realtime(self):
                             t(Tracer.SIGNATURE, signature)
                             t(Tracer.INPUTS, message.content)
                             await self.send_console(
-                                Message(type="console", payload=json.dumps(message.content))
+                                Message(
+                                    type="console", payload=json.dumps(message.content)
+                                )
                             )
                     case "conversation.item.input_audio_transcription.completed":
                         with Tracer.start("receive_user_transcript") as t:
@@ -70,9 +85,14 @@ async def receive_realtime(self):
                                     "content": message.content,
                                 },
                             )
-                            await self.send_message(
-                                Message(type="user", payload=message.content)
-                            )
+                            if (
+                                message.content is not None
+                                and isinstance(message.content, str)
+                                and message.content != ""
+                            ):
+                                await self.send_message(
+                                    Message(type="user", payload=message.content)
+                                )
 
                     case "response.audio_transcript.done":
                         with Tracer.start("receive_assistant_transcript") as t:
@@ -85,15 +105,25 @@ async def receive_realtime(self):
                                     "content": message.content,
                                 },
                             )
-                            # audio stream
-                            await self.send_message(
-                                Message(type="assistant", payload=message.content)
-                            )
+                            if (
+                                message.content is not None
+                                and isinstance(message.content, str)
+                                and message.content != ""
+                            ):
+                                # audio stream
+                                await self.send_message(
+                                    Message(type="assistant", payload=message.content)
+                                )
 
                     case "response.audio.delta":
-                        await self.send_audio(
-                            Message(type="audio", payload=message.content)
-                        )
+                        if (
+                            message.content is not None
+                            and isinstance(message.content, str)
+                            and message.content != ""
+                        ):
+                            await self.send_audio(
+                                Message(type="audio", payload=message.content)
+                            )
 
                     case "response.failed":
                         with Tracer.start("realtime_failure") as t:
@@ -139,6 +169,8 @@ async def receive_realtime(self):
     @trace
     async def receive_client(self):
         signature = "api.session.RealtimeSession.receive_client"
+        if self.client is None or self.realtime is None:
+            return
         try:
             while self.client.client_state != WebSocketState.DISCONNECTED:
                 message = await self.client.receive_text()
@@ -170,6 +202,8 @@ async def receive_client(self):
             print("Realtime Socket Disconnected")
 
     async def close(self):
+        if self.client is None or self.realtime is None:
+            return
         try:
             await self.client.close()
             await self.realtime.close()
@@ -182,7 +216,7 @@ async def close(self):
 class ChatSession:
     def __init__(self, client: WebSocket):
         self.client = client
-        self.realtime: RealtimeSession = None
+        self.realtime: Union[RealtimeSession, None] = None
         self.context: List[str] = []
 
     async def send_message(self, message: Message):
@@ -193,16 +227,20 @@ def add_realtime(self, realtime: RealtimeSession):
 
     def is_closed(self):
         client_closed = (
-            self.client == None
+            self.client is None
             or self.client.client_state == WebSocketState.DISCONNECTED
         )
-        realtime_closed = self.realtime == None or self.realtime.realtime.closed
+        realtime_closed = (
+            self.realtime is None
+            or self.realtime.realtime is None
+            or self.realtime.realtime.closed
+        )
         return client_closed and realtime_closed
 
     @trace
     async def receive_chat(self):
         while (
-            self.client != None
+            self.client is not None
             and self.client.client_state != WebSocketState.DISCONNECTED
         ):
             with Tracer.start("chat_turn") as t:
@@ -214,7 +252,7 @@ async def receive_chat(self):
                     Tracer.INPUTS,
                     {
                         "request": msg.text,
-                        "image": msg.image != None,
+                        "image": msg.image is not None,
                     },
                 )
 
@@ -250,9 +288,10 @@ async def receive_chat(self):
                     },
                 )
 
-
-
     async def start_realtime(self, prompt: str):
+        if self.realtime is None:
+            raise Exception("Realtime session not available")
+
         await self.realtime.send_realtime_instructions(prompt)
         tasks = [
             asyncio.create_task(self.realtime.receive_realtime()),