[LiveKit] Fixes race condition and import issues (#90)

- Adds a check for the `agent_session._started`​ flag and waits for it to start up
- Fixes import errors if packages are not present for google
- since we do process operations sequentially, we would process data only when session is available

Author: SamstyleGhost

README.md

@@ -60,6 +60,10 @@ See [cookbook/agno_agent.py](cookbook/agno_agent.py) for an example of tracing a
 
 ## Version changelog
 
+### 3.11.4
+- fix: Fixes race condition in LiveKit realtime tracing
+- fix: Fixes import errors for Gemini and Google realtime session imports
+
 ### 3.11.3
 - fix: Fixed Nested Spans issue for Google ADK
 

maxim/logger/livekit/agent_session.py

@@ -2,6 +2,7 @@
 import inspect
 import traceback
 import time
+from typing import Optional
 import uuid
 import weakref
 from datetime import datetime, timezone
@@ -41,7 +42,15 @@ def intercept_session_start(self: AgentSession, room, room_name, agent: Agent):
     The session info along with room_id, agent_id, etc is stored in the thread-local store.
     """
     maxim_logger = get_maxim_logger()
-    scribe().debug(f"[Internal][{self.__class__.__name__}] Session started")
+
+    # Wait for start signal (max ~5s) before proceeding
+    for _ in range(500):
+        if getattr(self, "_started", False):
+            scribe().debug(f"[Internal][{self.__class__.__name__}] Session started")
+            break
+        time.sleep(0.01)
+    else:
+        scribe().debug(f"[Internal][{self.__class__.__name__}] start not signaled within timeout; continuing")
     # getting the room_id
     if isinstance(room, str):
         room_id = room
@@ -100,10 +109,10 @@ def intercept_session_start(self: AgentSession, room, room_name, agent: Agent):
     )
 
     current_turn_id = str(uuid.uuid4())
-    if self.stt is not None or self._agent.stt is not NOT_GIVEN:
+    if self.stt is not None or agent.stt is not NOT_GIVEN:
         # Only add generation if we are not in realtime session
-        llm_opts: _LLMOptions = self.llm._opts if self.llm is not None else self._agent.llm._opts
-        model = self.llm.model if self.llm is not None else self._agent.llm.model
+        llm_opts: _LLMOptions = self.llm._opts if self.llm is not None else agent.llm._opts
+        model = self.llm.model if self.llm is not None else agent.llm.model
         if llm_opts is not None:
             model_parameters = extract_llm_model_parameters(llm_opts)
         else:

maxim/logger/livekit/gemini/gemini_realtime_session.py

@@ -32,15 +32,21 @@
 from io import BytesIO
 from typing import Union
 
-from google.genai.types import (
-    LiveConnectConfig,
-    LiveServerContent,
-    LiveServerToolCall,
-    UsageMetadata,
-    Content,
-)
+try:
+    from google.genai.types import (
+        LiveConnectConfig,
+        LiveServerContent,
+        LiveServerToolCall,
+        UsageMetadata,
+        Content,
+    )
+except ImportError:
+    pass
 from livekit.agents.llm import InputTranscriptionCompleted
-from livekit.plugins.google.beta.realtime.realtime_api import RealtimeSession
+try:
+    from livekit.plugins.google.realtime.realtime_api import RealtimeSession
+except ImportError:
+    from livekit.plugins.google.beta.realtime.realtime_api import RealtimeSession
 from livekit.rtc import AudioFrame
 
 from maxim.scribe import scribe

maxim/logger/livekit/gemini/instrumenter.py

@@ -13,14 +13,18 @@
   still capturing detailed observability and logging data for each session.
 """
 
-from maxim.logger.livekit.gemini.gemini_realtime_session import instrument_gemini_session
+# Import instrument_gemini_session conditionally to avoid dependency issues
 
 
 def instrument_gemini():
     """Monkey-patch Gemini's `RealtimeSession` methods with instrumentation.
     """
     try:
-        from livekit.plugins.google.beta.realtime.realtime_api import RealtimeSession
+        try:
+            from livekit.plugins.google.realtime.realtime_api import RealtimeSession
+        except ImportError:
+            from livekit.plugins.google.beta.realtime.realtime_api import RealtimeSession
+        from maxim.logger.livekit.gemini.gemini_realtime_session import instrument_gemini_session
 
         for name, orig in [
             (n, getattr(RealtimeSession, n))
@@ -29,5 +33,6 @@ def instrument_gemini():
         ]:
             if name != "__class__" and not name.startswith("__"):
                 setattr(RealtimeSession, name, instrument_gemini_session(orig, name))
-    except ImportError:
+    except (ImportError, NameError):
+        # Gemini dependencies not available, skip instrumentation
         pass

maxim/logger/livekit/instrumenter.py

@@ -5,7 +5,7 @@
 from ...logger import Logger
 from .agent_activity import instrument_agent_activity
 from .agent_session import instrument_agent_session
-from .gemini.instrumenter import instrument_gemini
+# Import instrument_gemini conditionally to avoid dependency issues
 from .llm import instrument_llm_init
 from .realtime_session import instrument_realtime_session
 from .stt import instrument_stt_init
@@ -115,4 +115,9 @@ def instrument_livekit(logger: Logger, callback: MaximLiveKitCallback = None):
             setattr(TTS, name, instrument_tts(orig, name))
 
     # Instrument gemini models if present
-    instrument_gemini()
+    try:
+        from .gemini.instrumenter import instrument_gemini
+        instrument_gemini()
+    except (ImportError, NameError):
+        # Gemini dependencies not available, skip instrumentation
+        pass

maxim/logger/livekit/realtime_session.py

@@ -30,7 +30,14 @@
 from ...scribe import scribe
 from ..components import FileDataAttachment
 from ..utils import pcm16_to_wav_bytes
-from .gemini.gemini_realtime_session import handle_google_input_transcription_completed
+# Import Gemini handler conditionally to avoid dependency issues
+try:
+    from .gemini.gemini_realtime_session import handle_google_input_transcription_completed
+    GEMINI_HANDLER_AVAILABLE = True
+except (ImportError, NameError):
+    # Gemini dependencies not available
+    handle_google_input_transcription_completed = None
+    GEMINI_HANDLER_AVAILABLE = False
 from .openai.realtime.handler import (
     handle_openai_client_event_queued,
     handle_openai_input_transcription_completed,
@@ -73,7 +80,8 @@ def intercept_realtime_session_emit(self: RealtimeSession, event, data):
         if session_info.provider == "openai-realtime":
             handle_openai_input_transcription_completed(session_info, data)
         elif session_info.provider == "google-realtime":
-            handle_google_input_transcription_completed(session_info, data)
+            if GEMINI_HANDLER_AVAILABLE and handle_google_input_transcription_completed:
+                handle_google_input_transcription_completed(session_info, data)
     elif event == "error":
         scribe().debug(f"[Internal][{self.__class__.__name__}] error;")
         if data is not None and isinstance(data, RealtimeModelError):

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "maxim-py"
-version = "3.11.3"
+version = "3.11.4"
 description = "A package that allows you to use the Maxim Python Library to interact with the Maxim Platform."
 readme = "README.md"
 requires-python = ">=3.9.20"