mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-03 02:22:24 +00:00
perf(anthropic_messages): remove thread hop in aanthropic_messages dispatch
The async /messages entry point dispatched anthropic_messages_handler via loop.run_in_executor(None, ...) wrapped in a copied contextvars context. That handler is @client-decorated; with is_async=True its async wrapper returns a coroutine immediately rather than blocking, so the executor only ever ran a function that returned a coroutine which was then awaited on the event loop. The thread hop did no blocking work; it was pure overhead and contended for the default ThreadPoolExecutor that is shared with sync logging callbacks and blocking SDK calls, hurting throughput under concurrency. Dispatch the handler directly on the running loop and await the result. The asyncio.iscoroutine guard is retained because the mock_response shortcut returns a materialized dict regardless of is_async, so a flat await would crash that path.
This commit is contained in:
parent
9ccda11919
commit
b04381957e
2 changed files with 37 additions and 13 deletions
|
|
@ -6,8 +6,6 @@
|
|||
"""
|
||||
|
||||
import asyncio
|
||||
import contextvars
|
||||
from functools import partial
|
||||
from typing import (
|
||||
Any,
|
||||
AsyncIterator,
|
||||
|
|
@ -300,11 +298,9 @@ async def anthropic_messages(
|
|||
**kwargs,
|
||||
)
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
kwargs["is_async"] = True
|
||||
|
||||
func = partial(
|
||||
anthropic_messages_handler,
|
||||
response = anthropic_messages_handler(
|
||||
max_tokens=max_tokens,
|
||||
messages=messages,
|
||||
model=model,
|
||||
|
|
@ -330,14 +326,11 @@ async def anthropic_messages(
|
|||
_litellm_messages_presanitized=True,
|
||||
**kwargs,
|
||||
)
|
||||
ctx = contextvars.copy_context()
|
||||
func_with_context = partial(ctx.run, func)
|
||||
init_response = await loop.run_in_executor(None, func_with_context)
|
||||
|
||||
if asyncio.iscoroutine(init_response):
|
||||
response = await init_response
|
||||
else:
|
||||
response = init_response
|
||||
# ``anthropic_messages_handler`` returns a coroutine on the async backend
|
||||
# paths but a materialized response on the ``mock_response`` shortcut, so
|
||||
# only await when there is something to await.
|
||||
if asyncio.iscoroutine(response):
|
||||
return await response
|
||||
return response
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
|
@ -673,3 +674,33 @@ async def test_async_wrapper_sets_presanitized_and_sanitizes_once():
|
|||
assert spy.call_count == 1
|
||||
assert captured["presanitized"] is True
|
||||
assert [b["type"] for b in captured["messages"][0]["content"]] == ["tool_use"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aanthropic_messages_does_not_use_thread_executor():
|
||||
"""The async Anthropic /messages entry must dispatch the handler directly on
|
||||
the running event loop, never through ``loop.run_in_executor``.
|
||||
|
||||
Mutation control: revert handler.py to
|
||||
``await loop.run_in_executor(None, func_with_context)`` and this test must
|
||||
fail with AssertionError('thread hop fired').
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages import handler
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
def fail_on_thread_hop(*args, **kwargs):
|
||||
raise AssertionError("thread hop fired")
|
||||
|
||||
with patch.object(loop, "run_in_executor", side_effect=fail_on_thread_hop):
|
||||
response = await handler.anthropic_messages(
|
||||
max_tokens=8,
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
model="anthropic/claude-3-5-haiku-20241022",
|
||||
custom_llm_provider="anthropic",
|
||||
api_key="sk-test",
|
||||
stream=False,
|
||||
mock_response="pong",
|
||||
)
|
||||
|
||||
assert response["content"][0]["text"] == "pong"
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue