make api_key required for realtime models

Author: katstankiewicz

src/eva/assistant/pipeline/audio_llm_processor.py

@@ -19,7 +19,6 @@
 import asyncio
 import base64
 import io
-import os
 import time
 import wave
 from collections.abc import Awaitable
@@ -418,7 +417,7 @@ def __init__(
         super().__init__(**kwargs)
         self._audio_collector = audio_collector
         params = params or {}
-        self._api_key = params.get("api_key") or os.getenv("OPENAI_API_KEY")
+        self._api_key = params.get["api_key"]
         self._model = model
         self._system_prompt = system_prompt or self.DEFAULT_SYSTEM_PROMPT
         self._sample_rate = sample_rate

src/eva/assistant/pipeline/services.py

@@ -4,7 +4,6 @@
 """
 
 import datetime
-import os
 from typing import Any, AsyncGenerator, Optional
 
 from deepgram import LiveOptions
@@ -409,12 +408,12 @@ def create_realtime_llm_service(
             return InstrumentedRealtimeLLMService(
                 model=params.get("model"),
                 audit_log=audit_log,
-                api_key=params.get("api_key") or os.getenv("OPENAI_API_KEY"),
+                api_key=params.get["api_key"],
                 session_properties=session_properties,
             )
 
         return OpenAIRealtimeLLMService(
-            api_key=params.get("api_key"),
+            api_key=params.get["api_key"],
             session_properties=session_properties,
         )
     elif model_lower.startswith("azure") or model_lower.startswith("gpt-realtime"):
@@ -431,23 +430,23 @@ def create_realtime_llm_service(
             service = InstrumentedRealtimeLLMService(
                 model=params.get("model"),
                 audit_log=audit_log,
-                api_key=params.get("api_key"),
+                api_key=params.get["api_key"],
                 base_url=url,
                 session_properties=session_properties,
             )
             InstrumentedRealtimeLLMService._connect = override__connect  # azure realtime connect
             return service
 
         return OpenAIRealtimeLLMService(
-            api_key=params.get("api_key"),
+            api_key=params.get["api_key"],
             base_url=url,
             session_properties=session_properties,
         )
     elif model_lower == "ultravox":
         logger.info("Using Ultravox LLM")
         return UltravoxRealtimeLLMService(
             params=OneShotInputParams(
-                api_key=params.get("api_key"),
+                api_key=params.get["api_key"],
                 system_prompt=system_prompt,
                 temperature=0.3,
                 max_duration=datetime.timedelta(minutes=6),

src/eva/models/config.py

@@ -452,28 +452,35 @@ def _warn_deprecated_aliases(cls, data: Any) -> Any:
     @model_validator(mode="after")
     def _check_companion_services(self) -> "RunConfig":
         """Ensure required companion services are set for each pipeline mode."""
+        required_keys = ["api_key", "model"]
         if isinstance(self.model, PipelineConfig):
             if not self.model.stt:
                 raise ValueError("EVA_MODEL__STT is required when using EVA_MODEL__LLM (ASR-LLM-TTS pipeline).")
             if not self.model.tts:
                 raise ValueError("EVA_MODEL__TTS is required when using EVA_MODEL__LLM (ASR-LLM-TTS pipeline).")
-            self._validate_service_params("STT", self.model.stt, self.model.stt_params)
-            self._validate_service_params("TTS", self.model.tts, self.model.tts_params)
+            self._validate_service_params("STT", self.model.stt, required_keys, self.model.stt_params)
+            self._validate_service_params("TTS", self.model.tts, required_keys, self.model.tts_params)
         elif isinstance(self.model, AudioLLMConfig):
             if not self.model.tts:
                 raise ValueError("EVA_MODEL__TTS is required when using EVA_MODEL__AUDIO_LLM (SpeechLM-TTS pipeline).")
-            self._validate_service_params("TTS", self.model.tts, self.model.tts_params)
+            self._validate_service_params("TTS", self.model.tts, required_keys, self.model.tts_params)
+            self._validate_service_params("audio_llm", self.model.audio_llm, required_keys, self.model.audio_llm_params)
+        elif isinstance(self.model, SpeechToSpeechConfig):
+            # api_key is required, some s2s services don't require model
+            self._validate_service_params("S2S", self.model.s2s, ["api_key"], self.model.s2s_params)
         return self
 
     # Providers that manage their own model/key resolution (e.g. WebSocket-based)
     _SKIP_PARAMS_VALIDATION: ClassVar[set[str]] = {"nvidia"}
 
     @classmethod
-    def _validate_service_params(cls, service: str, provider: str, params: dict[str, Any]) -> None:
+    def _validate_service_params(
+        cls, service: str, provider: str, required_keys: list[str], params: dict[str, Any]
+    ) -> None:
         """Validate that STT/TTS params contain required keys."""
         if provider.lower() in cls._SKIP_PARAMS_VALIDATION:
             return
-        missing = [key for key in ("api_key", "model") if key not in params]
+        missing = [key for key in required_keys if key not in params]
         if missing:
             missing_str = " and ".join(f'"{k}"' for k in missing)
             env_var = f"EVA_MODEL__{service}_PARAMS"