feat(simulation): pin text simulations to the low inference class

livekit-agents/livekit/agents/evals/judge.py

@@ -13,6 +13,25 @@
 """The verdict of a judgment: pass, fail, or maybe (uncertain)."""
 
 
+def _judge_chat_kwargs(llm: LLM) -> dict[str, Any]:
+    """Extra ``chat()`` kwargs for a judgment call.
+
+    Judging is batch load: an eval suite fans out many judgments at once, they run
+    after the conversation they grade, and nobody is waiting on the verdict. So they
+    are pinned to the low inference class and must not compete with live traffic for
+    gateway capacity, even when the judge LLM was configured with another class.
+
+    Empty for a plugin LLM, which has no LiveKit Inference class to set.
+    """
+    from ..inference import LLM as InferenceLLM
+    from ..inference._utils import INFERENCE_CLASS_LOW
+
+    if isinstance(llm, InferenceLLM):
+        return {"inference_class": INFERENCE_CLASS_LOW}
+
+    return {}
+
+
 @dataclass
 class JudgmentResult:
     verdict: Verdict
@@ -131,6 +150,7 @@ async def submit_verdict(verdict: Verdict, reasoning: str) -> tuple[Verdict, str
         tool_choice="required",
         conn_options=_JUDGE_CONN_OPTIONS,
         extra_kwargs=extra_kwargs,
+        **_judge_chat_kwargs(llm),
     ).collect()
 
     if not response.tool_calls:

livekit-agents/livekit/agents/inference/_utils.py

@@ -19,6 +19,8 @@
 HEADER_INFERENCE_PROVIDER = "X-LiveKit-Inference-Provider"
 HEADER_INFERENCE_PRIORITY = "X-LiveKit-Inference-Priority"
 
+INFERENCE_CLASS_LOW = "low"
+
 
 def get_default_inference_url() -> str:
     """Get the default inference URL based on the environment.
@@ -39,17 +41,24 @@ def get_default_inference_url() -> str:
     return DEFAULT_INFERENCE_URL
 
 
-def get_inference_headers() -> dict[str, str]:
+def get_inference_headers(*, inference_class: str | None = None) -> dict[str, str]:
     """Build identification headers for inference requests.
 
     Always includes User-Agent with SDK version and Python version.
     Includes X-LiveKit-Room-ID, X-LiveKit-Job-ID, and X-LiveKit-Agent-ID
     when running inside a job context (omitted in console mode or tests).
     Includes X-LiveKit-Worker-Token when LIVEKIT_WORKER_TOKEN is set (hosted agents).
+
+    ``inference_class`` is the class the caller configured, if any; it lands in
+    X-LiveKit-Inference-Priority. A job can override it through
+    :attr:`JobContext.inference_headers`, which is merged last.
     """
     headers: dict[str, str] = {
         HEADER_USER_AGENT: (f"LiveKit Agents/{__version__} (python {platform.python_version()})"),
     }
+    if inference_class:
+        headers[HEADER_INFERENCE_PRIORITY] = inference_class
+
     try:
         from ..job import get_job_context
 
@@ -68,8 +77,12 @@ def get_inference_headers() -> dict[str, str]:
         # cleared, so the access below won't raise once isconnected() is True.
         if ctx.room.isconnected() and isinstance(agent_sid := ctx.agent.sid, str) and agent_sid:
             headers[HEADER_AGENT_ID] = agent_sid
+        # merged last: what the job asserts about itself outranks what an individual
+        # model was configured with.
+        headers.update(ctx.inference_headers)
     except RuntimeError:
         pass
+
     return headers
 
 

livekit-agents/livekit/agents/inference/llm.py

@@ -28,7 +28,6 @@
 from ..types import DEFAULT_API_CONNECT_OPTIONS, NOT_GIVEN, APIConnectOptions, NotGivenOr
 from ..utils import is_given
 from ._utils import (
-    HEADER_INFERENCE_PRIORITY,
     HEADER_INFERENCE_PROVIDER,
     create_access_token,
     get_default_inference_url,
@@ -439,11 +438,9 @@ async def _run(self) -> None:
                 self._extra_kwargs.pop("tool_choice", None)
 
             extra_headers = self._extra_kwargs.setdefault("extra_headers", {})
-            extra_headers.update(get_inference_headers())
+            extra_headers.update(get_inference_headers(inference_class=self._inference_class))
             if self._provider:
                 extra_headers[HEADER_INFERENCE_PROVIDER] = self._provider
-            if self._inference_class:
-                extra_headers[HEADER_INFERENCE_PRIORITY] = self._inference_class
 
             self._oai_stream = stream = await self._client.chat.completions.create(
                 messages=cast(list[ChatCompletionMessageParam], chat_ctx),

livekit-agents/livekit/agents/job.py

@@ -94,6 +94,21 @@ def get_job_context(*, required: bool = True) -> JobContext | None:
 get_current_job_context = get_job_context
 
 
+def current_simulation() -> SimulationContext | None:
+    """The :class:`SimulationContext` of the job running on this task, or ``None``.
+
+    ``None`` covers everything that is not a simulation: a production job, and code
+    running outside a job context at all (console mode, tests). Unlike
+    :meth:`JobContext.simulation_context` this does not need the job context in hand,
+    so it can be called from deep inside the stack.
+    """
+    ctx = get_job_context(required=False)
+    if ctx is None:
+        return None
+
+    return ctx.simulation_context()
+
+
 @unique
 class JobExecutorType(Enum):
     PROCESS = "process"
@@ -489,6 +504,30 @@ def simulation_context(self) -> SimulationContext | None:
         self._simulation_ctx = SimulationContext(dispatch, self)
         return self._simulation_ctx
 
+    @property
+    def inference_headers(self) -> dict[str, str]:
+        """Extra headers this job puts on every LiveKit Inference request it makes.
+
+        Merged last by ``inference.get_inference_headers``, so what the job asserts
+        about itself outranks what an individual model was configured with. Empty for
+        an ordinary job.
+        """
+        from .inference._utils import HEADER_INFERENCE_PRIORITY, INFERENCE_CLASS_LOW
+        from .simulation import SimulationMode
+
+        headers: dict[str, str] = {}
+
+        # A text simulation is batch load: a run fans out many jobs at once and nobody
+        # is waiting on the answers, so it must not compete with live traffic for
+        # gateway capacity, and it must not be able to ask for priority either. Audio
+        # simulations are excluded: they run in real time against the audio pipeline,
+        # so their latency has to stay representative of production.
+        sim = self.simulation_context()
+        if sim is not None and sim.simulation_mode == SimulationMode.SIMULATION_MODE_TEXT:
+            headers[HEADER_INFERENCE_PRIORITY] = INFERENCE_CLASS_LOW
+
+        return headers
+
     @property
     def local_participant_identity(self) -> str:
         if identity := self.token_claims().identity:

livekit-agents/livekit/agents/voice/agent_session.py

@@ -2020,15 +2020,11 @@ def _config_update_added(self, item: llm.AgentConfigUpdate) -> None:
     def _text_only(self) -> bool:
         """True when running under a text simulation: the session uses no audio
         I/O and no audio models (STT/TTS/VAD)."""
-        from ..job import get_job_context
-
-        job_ctx = get_job_context(required=False)
-        if job_ctx is None or (sim_ctx := job_ctx.simulation_context()) is None:
-            return False
-
+        from ..job import current_simulation
         from ..simulation import SimulationMode
 
-        return sim_ctx.simulation_mode == SimulationMode.SIMULATION_MODE_TEXT
+        sim = current_simulation()
+        return sim is not None and sim.simulation_mode == SimulationMode.SIMULATION_MODE_TEXT
 
     @property
     def stt(self) -> stt.STT | None:

livekit-agents/livekit/agents/voice/run_result.py

@@ -1024,12 +1024,15 @@ async def check_intent(success: bool, reason: str) -> tuple[bool, str]:
         if not any(excluded_model in llm_v.model for excluded_model in excluded_models_temperature):
             extra_kwargs["temperature"] = 0.0
 
+        from ..evals.judge import _judge_chat_kwargs
+
         # TODO(theomonnom): LLMStream should provide utilities to make function calling easier.
         async for chunk in llm_v.chat(
             chat_ctx=chat_ctx,
             tools=[check_intent],
             tool_choice="required",
             extra_kwargs=extra_kwargs,
+            **_judge_chat_kwargs(llm_v),
         ):
             if chunk.usage is not None:
                 usage = chunk.usage