added telemetry

Author: sethjuarez

.github/workflows/azure-container-api.yml

@@ -61,7 +61,9 @@ jobs:
                                 "AZURE_OPENAI_ENDPOINT=secretref:azure-openai-endpoint" \
                                 "AZURE_OPENAI_API_KEY=secretref:azure-openai-api-key" \
                                 "AZURE_VOICE_ENDPOINT=secretref:azure-voice-endpoint" \
-                                "AZURE_VOICE_KEY=secretref:azure-voice-key"
+                                "AZURE_VOICE_KEY=secretref:azure-voice-key" \
+                                "APPINSIGHTS_CONNECTIONSTRING=secretref:appinsights-connectionstring" \
+                                "LOCAL_TRACING_ENABLED=false"
 
       - name: Logout
         run: |

api/__init__.py

@@ -0,0 +1,22 @@
+import asyncio
+from functools import wraps
+
+
+def repeat(*, seconds: float):
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            async def loop(*args, **kwargs):
+                while True:
+                    try:
+                        await func(*args, **kwargs)
+                    except Exception as e:
+                        print(f"Error: {e}")
+                    await asyncio.sleep(seconds)
+            asyncio.ensure_future(loop(*args, **kwargs))
+        return wrapper
+    return decorator
+
+
+# Compare this snippet from .venv/Lib/site-packages/fastapi_utilities/repeat/repeat_every.py:
+# slimmed down version of the repeat_every decorator

api/main.py

@@ -6,6 +6,7 @@
 from fastapi.responses import StreamingResponse
 from jinja2 import Environment, FileSystemLoader
 from rtclient import RTLowLevelClient
+from api import repeat
 from api.realtime import RealtimeVoiceClient
 from api.session import Message, RealtimeSession, SessionManager
 from contextlib import asynccontextmanager
@@ -14,13 +15,20 @@
 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
+
+from api.telemetry import init_tracing
 
 load_dotenv()
 
 AZURE_VOICE_ENDPOINT = os.getenv("AZURE_VOICE_ENDPOINT")
 AZURE_VOICE_KEY = os.getenv("AZURE_VOICE_KEY")
 
+LOCAL_TRACING_ENABLED = os.getenv("LOCAL_TRACING_ENABLED", "true") == "true"
+init_tracing(local_tracing=LOCAL_TRACING_ENABLED)
+
 base_path = Path(__file__).parent
 
 # Load products and purchases
@@ -38,7 +46,6 @@
 async def lifespan(app: FastAPI):
     try:
         # manage lifetime scope
-        pass
         yield
     finally:
         # remove all stray sockets
@@ -107,12 +114,13 @@ async def voice_endpoint(websocket: WebSocket):
             key_credential=AzureKeyCredential(AZURE_VOICE_KEY),
             azure_deployment="gpt-4o-realtime-preview",
         ) as rt:
-            
+
             # get current messages for instructions
             chat_items = await websocket.receive_json()
             message = Message(**chat_items)
 
             # create voice system message
+            # TODO: retrieve context from chat messages via thread id
             system_message = env.get_template("script.jinja2").render(
                 customer="Seth",
                 purchases=purchases,
@@ -130,3 +138,12 @@ async def voice_endpoint(websocket: WebSocket):
 
     except WebSocketDisconnect as e:
         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/realtime.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+import json
 from typing import Any, AsyncGenerator
 from rtclient import (
     InputAudioBufferAppendMessage,
@@ -47,6 +48,7 @@ async def close(self):
         if self.client is not None:
             await self.client.close()
 
+    @trace
     async def send_user_message(self, message: str):
         if self.client is None:
             raise Exception("Client not set")
@@ -68,6 +70,7 @@ async def send_user_message(self, message: str):
             )
         )
 
+    @trace
     async def send_user_message_with_response(self, message: str):
         if self.client is None:
             raise Exception("Client not set")
@@ -95,11 +98,13 @@ async def send_user_message_with_response(self, message: str):
 
         await self.client.send(ResponseCreateMessage(response=response))
 
+    @trace
     async def trigger_response(self):
         if self.client is None:
             raise Exception("Client not set")
         await self.client.send(ResponseCreateMessage())
 
+    @trace
     async def send_system_message(self, message: str):
         if self.client is None:
             raise Exception("Client not set")
@@ -121,6 +126,7 @@ async def send_system_message(self, message: str):
             )
         )
 
+    @trace
     async def send_session_update(self, instructions: str = None):
         if self.client is None:
             raise Exception("Client not set")
@@ -149,7 +155,6 @@ async def send_audio_message(self, audio_data: Any):
 
         await self.client.send(InputAudioBufferAppendMessage(audio=audio_data))
 
-    @trace(name="realtime_loop")
     async def receive_message(self) -> AsyncGenerator[RealTimeItem, None]:
         if self.client is None:
             raise Exception("Client not set")
@@ -164,9 +169,14 @@ async def receive_message(self) -> AsyncGenerator[RealTimeItem, None]:
                     print("Session Created Message")
                     print(f"  Model: {message.session.model}")
                     print(f"  Session Id: {message.session.id}")
-                    with Tracer.start("session.created") as tr:
-                        tr("model", message.session.model)
-                        tr("session_id", message.session.id)
+
+                    yield RealTimeItem(
+                        type="session.created",
+                        content={
+                            "model": message.session.model,
+                            "session": message.session.id,
+                        },
+                    )
 
             case "error":
                 if self.verbose:

api/semantic-mapper.json

@@ -0,0 +1,13 @@
+{
+  "result.model.api": "gen_ai.operation.name",
+  "result.object": "gen_ai.operation.name",
+  "result.model.api.config.azure_deployment": "gen_ai.request.modelfile",
+  "inputs.prompt.model.configuration.type": "gen_ai.system",
+  "result.model": "gen_ai.response.model",
+  "result.usage.total_tokens": "gen_ai.usage.input_tokens",
+  "result.usage.completion_tokens": "gen_ai.usage.output_tokens",
+  "result.usage.prompt_tokens": "gen_ai.usage.prompt_tokens",
+  "inputs.data.id": "gen_ai.request.id",
+  "result.id": "gen_ai.response.id",
+  "create.attributes.0": "gen_ai.choice"
+}