mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-03 07:23:33 +00:00
16645cc963
* fix(router): bound fake-streamed calls by the request timeout, not stream_timeout A stream=True call that the provider serves as one non-streaming request (Bedrock Converse for json_schema on models without native structured output, tools on some Llama models, ai21) gets no byte until generation ends, so stream_timeout acted as a total-duration cap. The Router now passes timeout and stream_timeout separately to chat completion calls, and completion() applies stream_timeout only when it reads a live stream. * test(timeout): cover stream_timeout on live streams only and the router's timeout split * fix(router): cover every fake-stream site and keep the target group's stream_timeout under prompt management * fix(router): never shorten a fake stream below stream_timeout, and split timeouts on every surface A faked stream now takes the longer of timeout and stream_timeout, so a caller who raised stream_timeout to cover a slow fake stream keeps that headroom. Azure o-series dispatch is mirrored when deciding whether a stream is faked. text_completion, /v1/messages and /v1/responses through the Router now get timeout and stream_timeout separately, like chat calls. * fix(router): report the effective stream timeout, split adapter streams, and cover M365 and Rust messages * fix(router): report and log the timeout each stream was bounded by Native /v1/messages and /v1/responses now record the timeout they applied, so x-litellm-timeout and the cooldown guard read the real limit instead of the raw request value. A prompt-managed model also passes on only the caller's or its own timeout, so the target deployment's timeout still wins. * fix(router): keep the caller's request_timeout on a prompt-managed handoff and report the applied stream limit * fix(router): give streams answered in one piece the longer limit and accept text timeouts on responses * fix(router): give /v1/messages and /v1/responses streams answered in one piece the longer limit * test(integration): cover faked, live and converted stream timeout limits per endpoint * test(router): regression tests for faked-stream timeouts
52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
"""An upstream whose first byte takes longer than the request allows, the way a slow model does.
|
|
|
|
httpx hands every transport the request's timeout in ``request.extensions["timeout"]``, so this one honours it
|
|
in process the way a socket would: a read timeout shorter than the first byte's latency times out, a longer one
|
|
gets the answer.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Callable
|
|
from typing import Final
|
|
|
|
import httpx
|
|
from pydantic import TypeAdapter
|
|
|
|
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
|
|
|
|
STREAM_TIMEOUT_SECONDS: Final = 0.5
|
|
UPSTREAM_FIRST_BYTE_SECONDS: Final = 2.0
|
|
CONVERSE_ANSWER: Final = json.dumps(
|
|
{
|
|
"output": {"message": {"role": "assistant", "content": [{"text": "hi"}]}},
|
|
"stopReason": "end_turn",
|
|
"usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2},
|
|
"metrics": {"latencyMs": 1},
|
|
}
|
|
).encode()
|
|
|
|
_TIMEOUT_EXTENSION: Final = TypeAdapter(dict[str, float | None])
|
|
|
|
|
|
def slow_upstream_answer(body: bytes = b"") -> Callable[[httpx.Request], httpx.Response]:
|
|
def answer(request: httpx.Request) -> httpx.Response:
|
|
read_timeout: Final = _TIMEOUT_EXTENSION.validate_python(request.extensions["timeout"])["read"]
|
|
if read_timeout is not None and read_timeout < UPSTREAM_FIRST_BYTE_SECONDS:
|
|
raise httpx.ReadTimeout(f"no byte arrived within {read_timeout}s", request=request)
|
|
return httpx.Response(200, content=body, request=request)
|
|
|
|
return answer
|
|
|
|
|
|
def slow_upstream_transport(body: bytes = b"") -> httpx.MockTransport:
|
|
return httpx.MockTransport(slow_upstream_answer(body))
|
|
|
|
|
|
def slow_upstream_async_client(body: bytes = b"") -> AsyncHTTPHandler:
|
|
return AsyncHTTPHandler(transport=slow_upstream_transport(body))
|
|
|
|
|
|
def slow_upstream_sync_client(body: bytes = b"") -> HTTPHandler:
|
|
return HTTPHandler(client=httpx.Client(transport=slow_upstream_transport(body)))
|