mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-11 03:38:38 +00:00
Offload router stream reconstruction
This commit is contained in:
parent
0ece758e72
commit
5b28d82a93
2 changed files with 51 additions and 1 deletions
|
|
@ -3028,7 +3028,11 @@ class Router:
|
|||
finally:
|
||||
await response.aclose()
|
||||
|
||||
result = stream_chunk_builder(chunks, messages=messages)
|
||||
# stream_chunk_builder is CPU-bound and can briefly hold the GIL on large
|
||||
# streams; keep that reconstruction work off the async router event loop.
|
||||
result = await asyncio.to_thread(
|
||||
stream_chunk_builder, chunks, messages=messages
|
||||
)
|
||||
if result is None:
|
||||
raise litellm.APIError(
|
||||
status_code=500,
|
||||
|
|
|
|||
|
|
@ -4975,6 +4975,52 @@ async def test_router_ttft_timeout_returns_non_streaming_response():
|
|||
fake_stream.aclose.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_ttft_timeout_reconstruction_runs_off_event_loop_thread():
|
||||
"""Final stream reconstruction is CPU-bound, so it must not pin the event loop."""
|
||||
import threading
|
||||
from unittest.mock import patch
|
||||
|
||||
from litellm import ModelResponse
|
||||
|
||||
router = litellm.Router(
|
||||
model_list=[
|
||||
{
|
||||
"model_name": "test-model",
|
||||
"litellm_params": {"model": "openai/gpt-4o", "api_key": "fake-key"},
|
||||
}
|
||||
],
|
||||
ttft_timeout=5.0,
|
||||
)
|
||||
|
||||
chunks = [
|
||||
_make_chunk("Hello"),
|
||||
_make_chunk(" world"),
|
||||
_make_chunk("", finish_reason="stop"),
|
||||
]
|
||||
fake_stream = _fake_stream(lambda: _async_chunks(*chunks))
|
||||
reconstructed = MagicMock(spec=ModelResponse)
|
||||
event_loop_thread_id = threading.get_ident()
|
||||
builder_thread_id: dict[str, int] = {}
|
||||
|
||||
def _fake_stream_chunk_builder(*args, **kwargs):
|
||||
builder_thread_id["value"] = threading.get_ident()
|
||||
return reconstructed
|
||||
|
||||
with patch(
|
||||
"litellm.main.stream_chunk_builder", side_effect=_fake_stream_chunk_builder
|
||||
):
|
||||
result = await router._collect_stream_with_ttft_timeout(
|
||||
response=fake_stream,
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
ttft_timeout=5.0,
|
||||
)
|
||||
|
||||
assert result is reconstructed
|
||||
assert builder_thread_id["value"] != event_loop_thread_id
|
||||
fake_stream.aclose.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_ttft_timeout_raises_on_hung_provider():
|
||||
"""Router raises litellm.Timeout when provider never sends a first token."""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue