mirror of
https://github.com/usestrix/strix.git
synced 2026-10-12 03:47:56 +00:00
refactor(llm): use None, not 0, for an unbounded stream limit
This commit is contained in:
parent
a5d28dc09e
commit
5a69f8073e
2 changed files with 21 additions and 19 deletions
|
|
@ -272,9 +272,9 @@ class _TurnGuardModel(Model):
|
|||
inner: Model,
|
||||
*,
|
||||
max_tool_calls_per_turn: int = 0,
|
||||
stream_idle_timeout: float = 0.0,
|
||||
first_event_timeout: float = 0.0,
|
||||
total_timeout: float = 0.0,
|
||||
stream_idle_timeout: float | None = None,
|
||||
first_event_timeout: float | None = None,
|
||||
total_timeout: float | None = None,
|
||||
) -> None:
|
||||
self._inner = inner
|
||||
self._max_tool_calls_per_turn = max_tool_calls_per_turn
|
||||
|
|
@ -415,23 +415,24 @@ class StreamTimeoutError(TimeoutError):
|
|||
async def _with_timeouts(
|
||||
stream: AsyncIterator[TResponseStreamEvent],
|
||||
*,
|
||||
idle: float,
|
||||
idle: float | None = None,
|
||||
first_event: float | None = None,
|
||||
total: float = 0.0,
|
||||
total: float | None = None,
|
||||
) -> AsyncIterator[TResponseStreamEvent]:
|
||||
"""Bound the first event, the gap between events and the whole stream; 0 is no bound."""
|
||||
"""Bound the first event, the gap between events and the whole stream; None is no bound."""
|
||||
iterator = stream.__aiter__()
|
||||
deadline = time.monotonic() + total
|
||||
started = time.monotonic()
|
||||
while True:
|
||||
limits: list[tuple[float, float, str]] = []
|
||||
# Wait until the soonest limit expires: first-event (else idle) until the
|
||||
# first event arrives, then idle; plus whatever is left of the total.
|
||||
if first_event:
|
||||
if first_event is not None:
|
||||
limits.append((first_event, first_event, "stream_first_event_timeout"))
|
||||
elif idle:
|
||||
elif idle is not None:
|
||||
limits.append((idle, idle, "stream_idle_timeout"))
|
||||
if total:
|
||||
limits.append((max(0.0, deadline - time.monotonic()), total, "stream_total_timeout"))
|
||||
if total is not None:
|
||||
left = max(0.0, started + total - time.monotonic())
|
||||
limits.append((left, total, "stream_total_timeout"))
|
||||
try:
|
||||
if limits:
|
||||
event = await _next_event(iterator, *min(limits))
|
||||
|
|
@ -610,9 +611,10 @@ class StrixProvider(MultiProvider):
|
|||
def get_model(self, model_name: str | None) -> Model:
|
||||
llm = load_settings().llm
|
||||
slug = codex.subscription_model(model_name)
|
||||
idle_timeout = float(llm.stream_idle_timeout)
|
||||
first_event_timeout = float(llm.stream_first_event_timeout)
|
||||
total_timeout = float(llm.stream_total_timeout)
|
||||
# The settings use 0 for no bound.
|
||||
idle_timeout = float(llm.stream_idle_timeout) or None
|
||||
first_event_timeout = float(llm.stream_first_event_timeout) or None
|
||||
total_timeout = float(llm.stream_total_timeout) or None
|
||||
if slug:
|
||||
# The ChatGPT subscription backend is always streamed; it has no
|
||||
# non-streaming mode to fall back to, so LLM_DISABLE_STREAMING
|
||||
|
|
@ -653,7 +655,7 @@ class StrixProvider(MultiProvider):
|
|||
# The wrapper emits its single event only once the whole request
|
||||
# is done, so an idle gap is meaningless here; the request
|
||||
# timeout bounds it instead.
|
||||
idle_timeout = first_event_timeout = total_timeout = 0.0
|
||||
idle_timeout = first_event_timeout = total_timeout = None
|
||||
return _TurnGuardModel(
|
||||
model,
|
||||
max_tool_calls_per_turn=llm.max_tool_calls_per_turn,
|
||||
|
|
|
|||
|
|
@ -79,7 +79,7 @@ def stalling_gateway() -> Iterator[str]:
|
|||
server.server_close()
|
||||
|
||||
|
||||
def _stream(base_url: str, *, idle_timeout: float) -> AsyncIterator[Any]:
|
||||
def _stream(base_url: str, *, idle_timeout: float | None) -> AsyncIterator[Any]:
|
||||
client = AsyncOpenAI(api_key="tok", base_url=base_url, max_retries=0, timeout=_STALL_SECONDS)
|
||||
inner: Model = request_log.RequestLoggingModel(
|
||||
OpenAIChatCompletionsModel(model="gw-model", openai_client=client),
|
||||
|
|
@ -102,7 +102,7 @@ def _stream(base_url: str, *, idle_timeout: float) -> AsyncIterator[Any]:
|
|||
)
|
||||
|
||||
|
||||
async def _drain(base_url: str, *, idle_timeout: float) -> list[Any]:
|
||||
async def _drain(base_url: str, *, idle_timeout: float | None) -> list[Any]:
|
||||
return [event async for event in _stream(base_url, idle_timeout=idle_timeout)]
|
||||
|
||||
|
||||
|
|
@ -111,7 +111,7 @@ async def test_stalled_stream_hangs_without_the_watchdog(stalling_gateway: str)
|
|||
# Repro: tokens arrive, then nothing. Un-watched, the turn just sits there;
|
||||
# the request timeout is far away and would reset on any keepalive byte.
|
||||
with pytest.raises(TimeoutError):
|
||||
await asyncio.wait_for(_drain(stalling_gateway, idle_timeout=0), timeout=2)
|
||||
await asyncio.wait_for(_drain(stalling_gateway, idle_timeout=None), timeout=2)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -216,4 +216,4 @@ def test_idle_timeout_is_off_without_streaming(
|
|||
|
||||
model = StrixProvider().get_model("openai/gpt-4o-mini")
|
||||
assert isinstance(model, _TurnGuardModel)
|
||||
assert model._stream_idle_timeout == model._first_event_timeout == model._total_timeout == 0
|
||||
assert model._stream_idle_timeout is model._first_event_timeout is model._total_timeout is None
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue