From 2ecf67c92f7b0e0ce7574355e195cd2a33983775 Mon Sep 17 00:00:00 2001 From: "yusheng.wang" Date: Tue, 29 Sep 2026 16:33:00 +0800 Subject: [PATCH] feat(llm): add request concurrency and delay limits --- docs/advanced/configuration.mdx | 15 +++ strix/config/settings.py | 15 +++ strix/llm/request_log.py | 147 +++++++++++++++++++-- tests/test_llm_request_pacing.py | 218 +++++++++++++++++++++++++++++++ 4 files changed, 381 insertions(+), 14 deletions(-) create mode 100644 tests/test_llm_request_pacing.py diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx index d2108e68..e844963a 100644 --- a/docs/advanced/configuration.mdx +++ b/docs/advanced/configuration.mdx @@ -37,6 +37,21 @@ Configure Strix using environment variables or a config file. Request timeout in seconds for LLM calls. + + Maximum number of LLM requests in flight at once, across every agent in the + process. `0` leaves concurrency unlimited, which is the default. Set a + positive value when a provider rate-limits the account and parallel agents + would otherwise open too many requests together. The cap covers agent turns, + retries, context compaction, and finding deduplication. + + + + Minimum seconds between the starts of successive LLM requests. `0` adds no + delay. A positive value spaces calls out after a concurrency slot is + acquired, so a waiting request cannot start inside another request's gap. + The delay is global, not per agent. + + Maximum number of retries for LLM API calls on transient failures. diff --git a/strix/config/settings.py b/strix/config/settings.py index 7bf1de7f..e350ab68 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -69,6 +69,21 @@ class LlmSettings(BaseSettings): ge=0, alias="LLM_MAX_TOOL_CALLS_PER_TURN", ) + # 0 means unlimited: every in-flight request starts immediately, which is + # the historical behavior. A positive value is a process-wide cap shared by + # every agent, including retries and the dedupe judge. + max_concurrent_requests: int = Field( + default=0, + ge=0, + alias="LLM_MAX_CONCURRENT_REQUESTS", + ) + # 0 means no spacing. A positive value is the minimum gap between the starts + # of successive LLM requests, measured after a concurrency slot is acquired. + request_delay: float = Field( + default=0.0, + ge=0, + alias="LLM_REQUEST_DELAY", + ) class DedupeSettings(BaseSettings): diff --git a/strix/llm/request_log.py b/strix/llm/request_log.py index 84dd890d..a418cc1b 100644 --- a/strix/llm/request_log.py +++ b/strix/llm/request_log.py @@ -30,7 +30,7 @@ import logging import re import time import uuid -from collections.abc import AsyncIterator, Callable, Mapping, Sequence +from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping, Sequence from contextvars import ContextVar, Token from dataclasses import asdict, dataclass, replace from datetime import UTC, date, datetime @@ -864,6 +864,84 @@ def _dispatch( _litellm_logger: Any | None = None +class LlmRequestGate: + """Process-wide limit on in-flight LLM requests, plus an optional start gap. + + The defaults (unlimited concurrency, no delay) acquire nothing and sleep + nothing, so an unconfigured scan behaves exactly as before. A positive + concurrency cap is one semaphore for every route — agent turns, retries, + compaction, and the dedupe judge — because provider rate limits are + account-wide, not per agent. The delay is measured from the previous + request's start and only begins once a concurrency slot is held, so a + waiting caller cannot start inside another caller's gap. + """ + + def __init__(self, *, max_concurrent: int = 0, delay: float = 0.0) -> None: + self.max_concurrent = max_concurrent + self.delay = delay + self._semaphore = asyncio.Semaphore(max_concurrent) if max_concurrent > 0 else None + self._spacing = asyncio.Lock() + self._next_start = 0.0 + + @contextlib.asynccontextmanager + async def slot(self) -> AsyncGenerator[None]: + if self._semaphore is None: + async with self._spacing_delay(): + yield + return + async with self._semaphore, self._spacing_delay(): + yield + + @contextlib.asynccontextmanager + async def _spacing_delay(self) -> AsyncGenerator[None]: + if self.delay <= 0: + yield + return + async with self._spacing: + remaining = self._next_start - time.monotonic() + if remaining > 0: + await asyncio.sleep(remaining) + self._next_start = time.monotonic() + self.delay + yield + + +_request_gate: LlmRequestGate | None = None + + +def request_gate() -> LlmRequestGate: + """The gate currently applied to model calls. + + Built from settings on first use. :func:`configure_request_pacing` replaces + it when a scan starts, so a config reload is picked up without restarting + the process. + """ + if _request_gate is None: + configure_request_pacing() + assert _request_gate is not None + return _request_gate + + +def configure_request_pacing( + *, max_concurrent: int | None = None, delay: float | None = None +) -> LlmRequestGate: + """Install the process-wide request gate from explicit values or settings. + + Callers already waiting on a previous gate keep that gate; only requests + that have not yet acquired a slot observe the replacement. + """ + global _request_gate # noqa: PLW0603 + if max_concurrent is None or delay is None: + from strix.config.loader import load_settings + + llm = load_settings().llm + if max_concurrent is None: + max_concurrent = llm.max_concurrent_requests + if delay is None: + delay = llm.request_delay + _request_gate = LlmRequestGate(max_concurrent=max_concurrent, delay=delay) + return _request_gate + + def install() -> None: """Attach the LiteLLM capture (idempotent), the native-route reply capture and the default log-line sink.""" @@ -871,6 +949,7 @@ def install() -> None: if _log_line_sink not in _sinks: _sinks.insert(0, _log_line_sink) _observe_sdk_shared_http_client() + configure_request_pacing() if _litellm_logger is not None: return import litellm @@ -1165,18 +1244,19 @@ class RequestLoggingModel(Model): reply = HttpReply() token = _http_reply.set(reply) try: - response = await self._inner.get_response( - system_instructions, - input, - model_settings, - tools, - output_schema, - handoffs, - tracing, - previous_response_id=previous_response_id, - conversation_id=conversation_id, - prompt=prompt, - ) + async with request_gate().slot(): + response = await self._inner.get_response( + system_instructions, + input, + model_settings, + tools, + output_schema, + handoffs, + tracing, + previous_response_id=previous_response_id, + conversation_id=conversation_id, + prompt=prompt, + ) except BaseException as exc: if self._should_emit(exc): emit( @@ -1207,6 +1287,42 @@ class RequestLoggingModel(Model): ) return response + async def _paced_stream( + self, + system_instructions: str | None, + input: str | list[TResponseInputItem], # noqa: A002 + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + *, + previous_response_id: str | None, + conversation_id: str | None, + prompt: ResponsePromptParam | None, + ) -> AsyncIterator[TResponseStreamEvent]: + """Yield the inner stream while holding one global request slot. + + The slot lives in this generator, not the caller's ``async for``. + Closing or cancelling the outer generator closes this one too, so the + slot is released when the stream ends, fails, or is abandoned — not + when the first event is handed back. + """ + async with request_gate().slot(): + async for event in self._inner.stream_response( + system_instructions, + input, + model_settings, + tools, + output_schema, + handoffs, + tracing, + previous_response_id=previous_response_id, + conversation_id=conversation_id, + prompt=prompt, + ): + yield event + async def stream_response( self, system_instructions: str | None, @@ -1241,7 +1357,7 @@ class RequestLoggingModel(Model): reply = HttpReply() token = _http_reply.set(reply) try: - async for event in self._inner.stream_response( + async for event in self._paced_stream( system_instructions, input, model_settings, @@ -1388,12 +1504,14 @@ __all__ = [ "ERROR_MESSAGE_MAX_CHARS", "LlmCallContext", "LlmRequestEvent", + "LlmRequestGate", "LlmRequestSink", "RequestLoggingModel", "api_host", "bind_call_context", "bound_details", "clean_error_message", + "configure_request_pacing", "current_call_context", "emit", "event_from_litellm", @@ -1404,6 +1522,7 @@ __all__ = [ "json_text", "merge_details", "register_sink", + "request_gate", "request_id_from_headers", "request_id_from_reply", "request_id_from_text", diff --git a/tests/test_llm_request_pacing.py b/tests/test_llm_request_pacing.py new file mode 100644 index 00000000..79e28542 --- /dev/null +++ b/tests/test_llm_request_pacing.py @@ -0,0 +1,218 @@ +"""Tests for the global LLM request cap and inter-request delay. + +Both settings default to off. A scan that does not set them must start every +request immediately, including requests that overlap. A positive cap is one +semaphore for every route; a positive delay is the minimum gap between the +starts of successive requests. +""" + +from __future__ import annotations + +import asyncio +import time +from typing import TYPE_CHECKING, Any + +import pytest +from agents.items import ModelResponse +from agents.model_settings import ModelSettings +from agents.models.interface import Model, ModelTracing +from agents.usage import Usage +from pydantic import ValidationError + +from strix.config import loader +from strix.config.loader import load_settings +from strix.llm import request_log + + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Iterator + + from agents.items import TResponseStreamEvent + + +class _ProbeModel(Model): + """Records when each call holds the provider, and can stay there on demand. + + The Model methods match the SDK signature and ignore its arguments: pacing + depends on when the call is in flight, not on what it asks the provider. + """ + + def __init__(self) -> None: + self.started: list[float] = [] + self.in_flight = 0 + self.peak = 0 + self.release = asyncio.Event() + self.entered = asyncio.Event() + self._active = 0 + + async def _hold(self) -> ModelResponse: + self._active += 1 + self.in_flight = self._active + self.peak = max(self.peak, self._active) + self.started.append(time.monotonic()) + self.entered.set() + try: + await self.release.wait() + finally: + self._active -= 1 + self.in_flight = self._active + return ModelResponse(output=[], usage=Usage(), response_id="resp-probe") + + async def get_response(self, *_args: Any, **_kwargs: Any) -> ModelResponse: + return await self._hold() + + async def stream_response( + self, *_args: Any, **_kwargs: Any + ) -> AsyncIterator[TResponseStreamEvent]: + await self._hold() + if False: # pragma: no cover - keeps this an async generator + yield None + + +def _logging(inner: Model) -> request_log.RequestLoggingModel: + return request_log.RequestLoggingModel( + inner, + model_name="probe", + provider="openai", + base_url=None, + ) + + +def _call(model: Model) -> Any: + return model.get_response( + None, + "go", + ModelSettings(), + [], + None, + [], + ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ) + + +def _stream(model: Model) -> AsyncIterator[Any]: + return model.stream_response( + None, + "go", + ModelSettings(), + [], + None, + [], + ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ) + + +@pytest.fixture +def _reset_gate(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + for key in ("LLM_MAX_CONCURRENT_REQUESTS", "LLM_REQUEST_DELAY"): + monkeypatch.delenv(key, raising=False) + monkeypatch.setattr(loader, "_cached", None) + monkeypatch.setattr(loader, "_override", None) + monkeypatch.setattr(request_log, "_request_gate", None) + yield + monkeypatch.setattr(request_log, "_request_gate", None) + + +def test_defaults_leave_requests_unpaced(_reset_gate: None) -> None: + llm = load_settings().llm + + gate = request_log.configure_request_pacing() + + assert llm.max_concurrent_requests == 0 + assert llm.request_delay == 0 + assert gate.max_concurrent == 0 + assert gate.delay == 0 + assert gate._semaphore is None + + +def test_settings_install_the_gate(monkeypatch: pytest.MonkeyPatch, _reset_gate: None) -> None: + monkeypatch.setenv("LLM_MAX_CONCURRENT_REQUESTS", "3") + monkeypatch.setenv("LLM_REQUEST_DELAY", "1.5") + monkeypatch.setattr(loader, "_cached", None) + + gate = request_log.configure_request_pacing() + + assert gate.max_concurrent == 3 + assert gate.delay == 1.5 + assert gate._semaphore is not None + + +async def test_default_gate_lets_requests_overlap(_reset_gate: None) -> None: + request_log.configure_request_pacing() + probe = _ProbeModel() + model = _logging(probe) + + calls = [asyncio.create_task(_call(model)) for _ in range(4)] + await asyncio.wait_for(probe.entered.wait(), timeout=1) + await asyncio.sleep(0.05) + assert probe.peak == 4 + probe.release.set() + + await asyncio.wait_for(asyncio.gather(*calls), timeout=1) + + +async def test_concurrency_cap_is_global(_reset_gate: None) -> None: + request_log.configure_request_pacing(max_concurrent=2, delay=0) + first = _ProbeModel() + second = _ProbeModel() + models = (_logging(first), _logging(second)) + + calls = [asyncio.create_task(_call(models[i % 2])) for i in range(4)] + await asyncio.wait_for(first.entered.wait(), timeout=1) + await asyncio.sleep(0.05) + assert first.peak + second.peak == 2 + assert first.in_flight + second.in_flight == 2 + + first.release.set() + second.release.set() + await asyncio.wait_for(asyncio.gather(*calls), timeout=1) + assert first.peak + second.peak == 2 + + +async def test_delay_spaces_request_starts(_reset_gate: None) -> None: + delay = 0.08 + request_log.configure_request_pacing(max_concurrent=0, delay=delay) + probe = _ProbeModel() + probe.release.set() + model = _logging(probe) + + await asyncio.wait_for(asyncio.gather(*[_call(model) for _ in range(3)]), timeout=1) + + gaps = [probe.started[i + 1] - probe.started[i] for i in range(len(probe.started) - 1)] + assert len(gaps) == 2 + assert all(gap >= delay * 0.85 for gap in gaps) + + +async def test_open_stream_holds_its_slot_until_closed(_reset_gate: None) -> None: + request_log.configure_request_pacing(max_concurrent=1, delay=0) + streaming = _ProbeModel() + waiting = _ProbeModel() + waiting.release.set() + + stream = _stream(_logging(streaming)) + reader = asyncio.create_task(stream.__anext__()) + await asyncio.wait_for(streaming.entered.wait(), timeout=1) + + follower = asyncio.create_task(_call(_logging(waiting))) + await asyncio.sleep(0.05) + assert waiting.started == [] + + streaming.release.set() + with pytest.raises(StopAsyncIteration): + await asyncio.wait_for(reader, timeout=1) + await stream.aclose() + await asyncio.wait_for(follower, timeout=1) + assert len(waiting.started) == 1 + + +def test_negative_values_are_rejected() -> None: + with pytest.raises(ValidationError, match="greater than or equal to 0"): + load_settings().llm.__class__.model_validate( + {"LLM_MAX_CONCURRENT_REQUESTS": -1, "LLM_REQUEST_DELAY": -0.1} + )