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

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

@@ -19,6 +19,13 @@
 HEADER_INFERENCE_PROVIDER = "X-LiveKit-Inference-Provider"
 HEADER_INFERENCE_PRIORITY = "X-LiveKit-Inference-Priority"
 
+# Inference class forced on every LiveKit Inference request issued from a text
+# simulation. A simulation run dispatches many jobs at once and nobody is waiting
+# on the answers, so it is batch load: it must never compete with live traffic for
+# gateway capacity. Audio simulations are excluded: they run in real time against
+# the audio pipeline, so their latency has to stay representative.
+SIMULATION_INFERENCE_CLASS = "low"
+
 
 def get_default_inference_url() -> str:
     """Get the default inference URL based on the environment.
@@ -39,13 +46,17 @@ 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. This resolves it for every LiveKit Inference model,
+    so it is the one place that decides which class a request goes out with.
     """
     headers: dict[str, str] = {
         HEADER_USER_AGENT: (f"LiveKit Agents/{__version__} (python {platform.python_version()})"),
@@ -70,6 +81,19 @@ def get_inference_headers() -> dict[str, str]:
             headers[HEADER_AGENT_ID] = agent_sid
     except RuntimeError:
         pass
+
+    # A text simulation overrides the configured class rather than falling back to it:
+    # a simulation must never be able to ask for priority capacity.
+    from ..job import current_simulation
+    from ..simulation import SimulationMode
+
+    sim = current_simulation()
+    if sim is not None and sim.simulation_mode == SimulationMode.SIMULATION_MODE_TEXT:
+        inference_class = SIMULATION_INFERENCE_CLASS
+
+    if inference_class:
+        headers[HEADER_INFERENCE_PRIORITY] = inference_class
+
     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"

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

@@ -902,9 +902,16 @@ async def start(
 
             # Under a text simulation the simulated user interacts over text
             # streams only: disable audio I/O here, and STT/TTS/VAD via
-            # AgentActivity (both consult _text_only).
+            # AgentActivity (both consult _text_only). LiveKit Inference requests
+            # are demoted to the low class in inference._utils.get_inference_headers,
+            # which catches models this session never sees (per-agent LLMs, tasks).
             if self._text_only:
-                logger.info("text simulation: disabling STT/TTS/VAD and audio I/O")
+                from ..inference._utils import SIMULATION_INFERENCE_CLASS
+
+                logger.info(
+                    "text simulation: disabling STT/TTS/VAD and audio I/O, "
+                    f"pinning LiveKit Inference to the {SIMULATION_INFERENCE_CLASS} class"
+                )
 
             self._session_span = current_span = tracer.start_span("agent_session")
             # we detach here to avoid context issues since tokens need to be detached
@@ -2020,15 +2027,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:

tests/test_inference_utils.py

@@ -62,6 +62,9 @@ def __init__(self, agent_sid: object) -> None:
         self.room = _FakeLiveRoom(connected=True)
         self.agent = _FakeAgent(agent_sid)
 
+    def simulation_context(self) -> None:
+        return None
+
 
 class _CtxDisconnected:
     """Room not connected: ``isconnected()`` is False and ``agent`` access raises.
@@ -77,9 +80,17 @@ class _CtxDisconnected:
     def agent(self):  # noqa: ANN201 - matches the raising property under test
         raise Exception("cannot access local participant before connecting")
 
+    def simulation_context(self) -> None:
+        return None
+
 
 def _patch_ctx(monkeypatch: pytest.MonkeyPatch, ctx: object) -> None:
-    monkeypatch.setattr("livekit.agents.job.get_job_context", lambda: ctx)
+    # keyword-only `required`, matching the real get_job_context: the header builder
+    # calls it both ways (required=True for the ids, required=False for the sim check).
+    def _get(*, required: bool = True) -> object:
+        return ctx
+
+    monkeypatch.setattr("livekit.agents.job.get_job_context", _get)
 
 
 def test_omits_agent_header_when_room_not_connected(
@@ -141,10 +152,12 @@ def test_omits_agent_header_when_sid_empty(monkeypatch: pytest.MonkeyPatch) -> N
 def test_no_job_context_returns_only_user_agent(monkeypatch: pytest.MonkeyPatch) -> None:
     """Outside a job context (console mode / tests) only User-Agent is set."""
 
-    def _raise() -> object:
-        raise RuntimeError("no job context found")
+    def _no_ctx(*, required: bool = True) -> object:
+        if required:
+            raise RuntimeError("no job context found")
+        return None
 
-    monkeypatch.setattr("livekit.agents.job.get_job_context", _raise)
+    monkeypatch.setattr("livekit.agents.job.get_job_context", _no_ctx)
 
     headers = get_inference_headers()