mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-10 22:41:41 +00:00
Merge pull request #40186 from BerriAI/litellm_lit_5546_count_tokens_offload
fix(token_counter): release the GIL for HuggingFace counts and cap exact counting per string
This commit is contained in:
commit
261aa2f16f
18 changed files with 515 additions and 33 deletions
|
|
@ -398,6 +398,18 @@ TIKTOKEN_ENCODE_CHUNK_SIZE_CHARS: Final = get_env_int_in_range(
|
|||
minimum=1,
|
||||
maximum=TIKTOKEN_ENCODE_MAX_CHUNK_SIZE_CHARS,
|
||||
)
|
||||
TOKEN_COUNTER_MAX_EXACT_CHARS: Final = get_env_int_in_range(
|
||||
"TOKEN_COUNTER_MAX_EXACT_CHARS",
|
||||
default=4_000_000,
|
||||
minimum=1,
|
||||
maximum=1_000_000_000,
|
||||
)
|
||||
TOKEN_COUNTER_MAX_CONCURRENT_COUNTS: Final = get_env_int_in_range(
|
||||
"TOKEN_COUNTER_MAX_CONCURRENT_COUNTS",
|
||||
default=4,
|
||||
minimum=1,
|
||||
maximum=256,
|
||||
)
|
||||
MAX_TILE_WIDTH: Final = int(os.getenv("MAX_TILE_WIDTH", 512))
|
||||
MAX_TILE_HEIGHT: Final = int(os.getenv("MAX_TILE_HEIGHT", 512))
|
||||
OPENAI_FILE_SEARCH_COST_PER_1K_CALLS: Final = float(os.getenv("OPENAI_FILE_SEARCH_COST_PER_1K_CALLS", 2.5 / 1000))
|
||||
|
|
|
|||
|
|
@ -3,11 +3,15 @@
|
|||
import base64
|
||||
import io
|
||||
import struct
|
||||
from collections.abc import Callable, Iterable, Mapping, Sequence
|
||||
from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
|
||||
from typing import Final, Literal, cast
|
||||
|
||||
import anyio
|
||||
import anyio.lowlevel
|
||||
import httpx
|
||||
import tiktoken
|
||||
from tokenizers import Tokenizer
|
||||
from typing_extensions import ParamSpec, TypeVar
|
||||
|
||||
import litellm
|
||||
from litellm import verbose_logger
|
||||
|
|
@ -21,7 +25,10 @@ from litellm.constants import (
|
|||
MAX_TILE_HEIGHT,
|
||||
MAX_TILE_WIDTH,
|
||||
TIKTOKEN_ENCODE_CHUNK_SIZE_CHARS,
|
||||
TOKEN_COUNTER_MAX_CONCURRENT_COUNTS,
|
||||
TOKEN_COUNTER_MAX_EXACT_CHARS,
|
||||
)
|
||||
from litellm.litellm_core_utils.asyncify import asyncify
|
||||
from litellm.litellm_core_utils.default_encoding import encoding as default_encoding
|
||||
from litellm.litellm_core_utils.url_utils import safe_get
|
||||
from litellm.llms.custom_httpx.http_handler import _get_httpx_client
|
||||
|
|
@ -317,6 +324,32 @@ TokenCounterFunction = Callable[[str], int]
|
|||
Type for a function that counts tokens in a string.
|
||||
"""
|
||||
|
||||
EXTRAPOLATION_SAMPLES: Final = 16
|
||||
T_ParamSpec: Final = ParamSpec("T_ParamSpec")
|
||||
T_Retval = TypeVar("T_Retval")
|
||||
_COUNT_OFFLOAD_LIMITER: Final = anyio.lowlevel.RunVar[anyio.CapacityLimiter]("litellm_count_offload_limiter")
|
||||
|
||||
|
||||
def _count_offload_limiter_for_this_loop() -> anyio.CapacityLimiter:
|
||||
existing: Final = _COUNT_OFFLOAD_LIMITER.get(None)
|
||||
if existing is not None:
|
||||
return existing
|
||||
created: Final = anyio.CapacityLimiter(TOKEN_COUNTER_MAX_CONCURRENT_COUNTS)
|
||||
_COUNT_OFFLOAD_LIMITER.set(created)
|
||||
return created
|
||||
|
||||
|
||||
def offload_token_count(
|
||||
function: Callable[T_ParamSpec, T_Retval],
|
||||
) -> Callable[T_ParamSpec, Awaitable[T_Retval]]:
|
||||
async def offloaded(
|
||||
*args: T_ParamSpec.args,
|
||||
**kwargs: T_ParamSpec.kwargs, # kwargs-ok: ParamSpec keeps the wrapped function's own keyword contract
|
||||
) -> T_Retval:
|
||||
return await asyncify(function, limiter=_count_offload_limiter_for_this_loop())(*args, **kwargs)
|
||||
|
||||
return offloaded
|
||||
|
||||
|
||||
def _get_tiktoken_count_function(
|
||||
encode_length: Callable[[str], int],
|
||||
|
|
@ -538,9 +571,40 @@ def _count_extra(
|
|||
return num_tokens
|
||||
|
||||
|
||||
def _get_extrapolating_count_function(
|
||||
count_exactly: TokenCounterFunction,
|
||||
max_exact_chars: int = TOKEN_COUNTER_MAX_EXACT_CHARS,
|
||||
) -> TokenCounterFunction:
|
||||
def count_tokens(text: str) -> int:
|
||||
if len(text) <= max_exact_chars:
|
||||
return count_exactly(text)
|
||||
samples: Final = _evenly_spaced_samples(text, max_exact_chars)
|
||||
sampled_chars: Final = sum(len(sample) for sample in samples)
|
||||
return round(sum(count_exactly(sample) for sample in samples) * len(text) / sampled_chars)
|
||||
|
||||
return count_tokens
|
||||
|
||||
|
||||
def _evenly_spaced_samples(text: str, total_chars: int) -> tuple[str, ...]:
|
||||
sample_count: Final = min(EXTRAPOLATION_SAMPLES, total_chars)
|
||||
sample_chars: Final = total_chars // sample_count
|
||||
last_start: Final = len(text) - sample_chars
|
||||
return tuple(
|
||||
text[start : start + sample_chars]
|
||||
for start in (last_start * index // max(sample_count - 1, 1) for index in range(sample_count))
|
||||
)
|
||||
|
||||
|
||||
def _get_count_function(
|
||||
model: str | None,
|
||||
custom_tokenizer: dict | SelectTokenizerResponse | None = None,
|
||||
) -> TokenCounterFunction:
|
||||
return _get_extrapolating_count_function(_get_exact_count_function(model, custom_tokenizer))
|
||||
|
||||
|
||||
def _get_exact_count_function(
|
||||
model: str | None,
|
||||
custom_tokenizer: dict | SelectTokenizerResponse | None = None,
|
||||
) -> TokenCounterFunction:
|
||||
"""
|
||||
Get the function to count tokens based on the model and custom tokenizer."""
|
||||
|
|
@ -549,10 +613,10 @@ def _get_count_function(
|
|||
if model is not None or custom_tokenizer is not None:
|
||||
tokenizer_json: Final = custom_tokenizer or _select_tokenizer(model)
|
||||
if tokenizer_json["type"] == "huggingface_tokenizer":
|
||||
tokenizer: Final[Tokenizer] = tokenizer_json["tokenizer"]
|
||||
|
||||
def count_tokens(text: str) -> int:
|
||||
enc: Final = tokenizer_json["tokenizer"].encode(text)
|
||||
return len(enc.ids)
|
||||
return len(tokenizer.encode_batch_fast([text])[0])
|
||||
|
||||
return count_tokens
|
||||
elif tokenizer_json["type"] == "openai_tokenizer":
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ from litellm.integrations.custom_logger import CustomLogger
|
|||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
get_str_from_messages,
|
||||
)
|
||||
from litellm.litellm_core_utils.token_counter import offload_token_count
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
from litellm.proxy.auth.auth_utils import (
|
||||
ESTIMATED_OUTPUT_TOKENS_FIELD,
|
||||
|
|
@ -3307,7 +3308,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
min_configured_tpm_limit=min_configured_otpm_limit,
|
||||
call_type=call_type,
|
||||
)
|
||||
raw_estimated_input_tokens: Final = self._estimate_precise_input_tokens(
|
||||
raw_estimated_input_tokens: Final = await offload_token_count(self._estimate_precise_input_tokens)(
|
||||
data=data, model=requested_model, call_type=call_type
|
||||
)
|
||||
estimated_input_tokens: Final = max(raw_estimated_input_tokens, 1)
|
||||
|
|
|
|||
|
|
@ -63,11 +63,13 @@ from litellm.constants import (
|
|||
LITELLM_UI_SESSION_DURATION,
|
||||
RUNTIME_UPDATABLE_ROUTER_SETTINGS,
|
||||
)
|
||||
from litellm.litellm_core_utils.asyncify import asyncify
|
||||
from litellm.litellm_core_utils.litellm_logging import (
|
||||
_init_custom_logger_compatible_class,
|
||||
)
|
||||
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
||||
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
|
||||
from litellm.litellm_core_utils.token_counter import offload_token_count
|
||||
from litellm.proxy._types import (
|
||||
UI_TEAM_ID,
|
||||
CallbackDelete,
|
||||
|
|
@ -272,7 +274,6 @@ from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting
|
|||
from litellm.litellm_core_utils.agentic_loop_settings import (
|
||||
validated_max_agentic_loops,
|
||||
)
|
||||
from litellm.litellm_core_utils.asyncify import asyncify
|
||||
from litellm.litellm_core_utils.audio_utils.utils import resolve_speech_media_type
|
||||
from litellm.litellm_core_utils.core_helpers import (
|
||||
_get_parent_otel_span_from_kwargs,
|
||||
|
|
@ -12816,7 +12817,9 @@ async def token_counter(request: TokenCountRequest, call_endpoint: bool = False)
|
|||
CustomHuggingfaceTokenizer | None,
|
||||
model_info.get("custom_tokenizer", None),
|
||||
)
|
||||
_tokenizer_used: Final = litellm.utils._select_tokenizer(model=model_to_use, custom_tokenizer=custom_tokenizer)
|
||||
_tokenizer_used: Final = await asyncify(litellm.utils._select_tokenizer)(
|
||||
model=model_to_use, custom_tokenizer=custom_tokenizer
|
||||
)
|
||||
|
||||
tokenizer_used: Final = str(_tokenizer_used["type"])
|
||||
system_message: Final = _system_message(system)
|
||||
|
|
@ -12829,7 +12832,7 @@ async def token_counter(request: TokenCountRequest, call_endpoint: bool = False)
|
|||
counted_tools: Final = cast( # cast-ok: raw OpenAI or Anthropic tool dicts, both of which token_counter formats
|
||||
list[ChatCompletionToolParam] | None, tools if counted_messages is not None else None
|
||||
)
|
||||
total_tokens: Final = await asyncify(litellm.token_counter)(
|
||||
total_tokens: Final = await offload_token_count(litellm.token_counter)(
|
||||
model=model_to_use,
|
||||
text=prompt,
|
||||
messages=counted_messages,
|
||||
|
|
|
|||
|
|
@ -101,6 +101,7 @@ from litellm.litellm_core_utils.core_helpers import (
|
|||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
||||
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
|
||||
from litellm.litellm_core_utils.token_counter import offload_token_count
|
||||
from litellm.llms import load_guardrail_translation_mappings
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
|
||||
from litellm.proxy._types import (
|
||||
|
|
@ -2900,7 +2901,7 @@ class ProxyLogging:
|
|||
original_exception=original_exception,
|
||||
)
|
||||
|
||||
request_data.update(_failure_fields_to_lift(request_data))
|
||||
request_data.update(await offload_token_count(_failure_fields_to_lift)(request_data))
|
||||
|
||||
# Remove before callbacks iterate — not serialisable
|
||||
request_data.pop("litellm_logging_obj", None)
|
||||
|
|
|
|||
|
|
@ -67,7 +67,7 @@ from litellm.constants import (
|
|||
SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY,
|
||||
)
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.litellm_core_utils.asyncify import asyncify, run_async_function
|
||||
from litellm.litellm_core_utils.asyncify import run_async_function
|
||||
from litellm.litellm_core_utils.core_helpers import (
|
||||
_get_parent_otel_span_from_kwargs,
|
||||
coerce_token_limit,
|
||||
|
|
@ -98,6 +98,7 @@ from litellm.litellm_core_utils.sensitive_data_masker import (
|
|||
mask_credentials_in_payload,
|
||||
mask_sensitive_structure,
|
||||
)
|
||||
from litellm.litellm_core_utils.token_counter import offload_token_count
|
||||
from litellm.llms.base_llm.vector_store.transformation import (
|
||||
RouterVectorStoreEmbeddingExecutor,
|
||||
vector_store_request_metadata,
|
||||
|
|
@ -12113,7 +12114,7 @@ class Router:
|
|||
try:
|
||||
if not self._pre_call_checks_need_token_count(model, healthy_deployments):
|
||||
return None
|
||||
return await asyncify(self._count_pre_call_check_tokens)(
|
||||
return await offload_token_count(self._count_pre_call_check_tokens)(
|
||||
messages=cast(list[dict[str, str]] | None, messages), # cast-ok: forwarded to the sync counter
|
||||
input=cast(str | list | None, input), # cast-ok: forwarded to the sync counter
|
||||
request_kwargs=request_kwargs,
|
||||
|
|
|
|||
|
|
@ -2568,14 +2568,14 @@ class ComplexityRouter(CustomLogger):
|
|||
"""Real-tokenizer count of the resolved messages plus the out-of-band carriers, off the
|
||||
event loop; None when counting fails, and the gate then leaves the placement alone."""
|
||||
import litellm
|
||||
from litellm.litellm_core_utils.asyncify import asyncify
|
||||
from litellm.litellm_core_utils.token_counter import offload_token_count
|
||||
|
||||
out_of_band: Final = self._out_of_band_request_text(request_kwargs)
|
||||
try:
|
||||
counted: Final = await asyncify(litellm.token_counter)(
|
||||
counted: Final = await offload_token_count(litellm.token_counter)(
|
||||
messages=cast(list, resolved_messages) # cast-ok: token_counter only iterates the sequence
|
||||
)
|
||||
return counted + (await asyncify(litellm.token_counter)(text=out_of_band) if out_of_band else 0)
|
||||
return counted + (await offload_token_count(litellm.token_counter)(text=out_of_band) if out_of_band else 0)
|
||||
except Exception as e: # noqa: BLE001 # best-effort: an uncountable prompt must not fail the request
|
||||
verbose_router_logger.debug("ComplexityRouter: context-window token count failed. Got - %s", e)
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ import litellm
|
|||
from litellm import token_counter
|
||||
from litellm._logging import verbose_router_logger
|
||||
from litellm.caching.dual_cache import DualCache
|
||||
from litellm.litellm_core_utils.token_counter import offload_token_count
|
||||
from litellm.types.router import RouterCacheEnum, RouterErrors
|
||||
from litellm.utils import get_utc_datetime
|
||||
|
||||
|
|
@ -466,7 +467,7 @@ async def async_io_token_pre_call_check(
|
|||
|
||||
request_kwargs: Final = get_io_token_rate_limit_request_kwargs()
|
||||
_model: Final = (deployment.get("litellm_params") or {}).get("model") or ""
|
||||
estimated_input: Final = _estimate_input_tokens(request_kwargs, model=_model)
|
||||
estimated_input: Final = await offload_token_count(_estimate_input_tokens)(request_kwargs, model=_model)
|
||||
max_tokens: Final = _resolve_max_tokens(request_kwargs, deployment)
|
||||
|
||||
dt: Final = get_utc_datetime()
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ from litellm.integrations.anthropic_cache_control_hook import (
|
|||
AnthropicCacheControlHook,
|
||||
)
|
||||
from litellm.integrations.custom_logger import CustomLogger, Span
|
||||
from litellm.litellm_core_utils.token_counter import offload_token_count
|
||||
from litellm.types.llms.openai import AllMessageValues
|
||||
from litellm.types.utils import CallTypes, StandardLoggingPayload
|
||||
from litellm.utils import get_prompt_cache_min_tokens, is_prompt_caching_valid_prompt
|
||||
|
|
@ -61,7 +62,7 @@ class PromptCachingDeploymentCheck(CustomLogger):
|
|||
if request_kwargs is not None and request_kwargs.get("_target_order") is not None:
|
||||
return healthy_deployments
|
||||
|
||||
if messages is not None and is_prompt_caching_valid_prompt(
|
||||
if messages is not None and await offload_token_count(is_prompt_caching_valid_prompt)(
|
||||
messages=messages,
|
||||
model=model,
|
||||
min_token_count=_get_min_token_count_for_deployments(healthy_deployments),
|
||||
|
|
@ -139,7 +140,7 @@ class PromptCachingDeploymentCheck(CustomLogger):
|
|||
return
|
||||
|
||||
## PROMPT CACHING - cache model id, if prompt caching valid prompt + provider
|
||||
if is_prompt_caching_valid_prompt(
|
||||
if await offload_token_count(is_prompt_caching_valid_prompt)(
|
||||
model=model,
|
||||
messages=cast(list[AllMessageValues], messages),
|
||||
):
|
||||
|
|
|
|||
|
|
@ -2293,15 +2293,7 @@ def create_pretrained_tokenizer(identifier: str, revision="main", auth_token: st
|
|||
dict: A dictionary with the tokenizer and its type.
|
||||
"""
|
||||
|
||||
try:
|
||||
tokenizer = Tokenizer.from_pretrained(
|
||||
identifier,
|
||||
revision=revision,
|
||||
auth_token=auth_token,
|
||||
)
|
||||
except Exception as e:
|
||||
verbose_logger.error("Error creating pretrained tokenizer: %s. Defaulting to version without 'auth_token'.", e)
|
||||
tokenizer = Tokenizer.from_pretrained(identifier, revision=revision)
|
||||
tokenizer: Final = Tokenizer.from_pretrained(identifier, revision=revision, token=auth_token)
|
||||
return {"type": "huggingface_tokenizer", "tokenizer": tokenizer}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -23,9 +23,9 @@ from litellm.proxy.proxy_server import token_counter
|
|||
|
||||
def _fake_hf_tokenizer(num_tokens: int) -> MagicMock:
|
||||
encoding = MagicMock()
|
||||
encoding.ids = list(range(num_tokens))
|
||||
encoding.__len__.return_value = num_tokens
|
||||
tokenizer = MagicMock()
|
||||
tokenizer.encode.return_value = encoding
|
||||
tokenizer.encode_batch_fast.return_value = [encoding]
|
||||
return tokenizer
|
||||
|
||||
|
||||
|
|
@ -68,13 +68,11 @@ async def test_custom_tokenizer_from_model_info_is_used(monkeypatch):
|
|||
)
|
||||
)
|
||||
|
||||
mock_tokenizer_cls.from_pretrained.assert_called_once_with(
|
||||
"my-org/custom-tokenizer", revision="v2", auth_token=None
|
||||
)
|
||||
mock_tokenizer_cls.from_pretrained.assert_called_once_with("my-org/custom-tokenizer", revision="v2", token=None)
|
||||
assert response.tokenizer_type == "huggingface_tokenizer"
|
||||
assert response.request_model == "my-embedding-model"
|
||||
assert response.model_used == "self-hosted-embedder"
|
||||
assert response.total_tokens > 0
|
||||
assert response.total_tokens >= 7
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
40
tests/test_litellm/litellm_core_utils/event_loop_lag.py
Normal file
40
tests/test_litellm/litellm_core_utils/event_loop_lag.py
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
import asyncio
|
||||
import time
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Final, TypeVar
|
||||
|
||||
import litellm
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
def warm_tokenizer(model: str) -> None:
|
||||
litellm.token_counter(model=model, text="load the tokenizer before anything is timed")
|
||||
|
||||
|
||||
async def loop_wake_lags(until: asyncio.Event) -> tuple[float, ...]:
|
||||
async def wake_lag() -> float:
|
||||
started: Final = time.perf_counter()
|
||||
await asyncio.sleep(0.001)
|
||||
return time.perf_counter() - started - 0.001
|
||||
|
||||
return tuple([await wake_lag() for _ in iter(until.is_set, True)])
|
||||
|
||||
|
||||
async def timed_with_loop_lags(run: Callable[[], Awaitable[T]]) -> tuple[T, float, tuple[float, ...]]:
|
||||
finished: Final = asyncio.Event()
|
||||
|
||||
async def timed() -> tuple[T, float]:
|
||||
await asyncio.sleep(0)
|
||||
started: Final = time.perf_counter()
|
||||
try:
|
||||
return await run(), time.perf_counter() - started
|
||||
finally:
|
||||
finished.set()
|
||||
|
||||
(result, took), lags = await asyncio.gather(timed(), loop_wake_lags(finished))
|
||||
return result, took, lags
|
||||
|
||||
|
||||
def assert_loop_stayed_free(took: float, lags: tuple[float, ...]) -> None:
|
||||
assert max(lags) < took / 4, f"the event loop stalled {max(lags):.3f}s during a {took:.3f}s count"
|
||||
|
|
@ -1,10 +1,15 @@
|
|||
#### What this tests ####
|
||||
# This tests litellm.token_counter.token_counter() function
|
||||
import asyncio
|
||||
import importlib
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
from concurrent.futures import Future, wait
|
||||
from typing import Final
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import anyio.to_thread
|
||||
import pytest
|
||||
import tiktoken
|
||||
|
||||
|
|
@ -14,9 +19,21 @@ import litellm
|
|||
from litellm import create_pretrained_tokenizer, decode, encode, get_modified_max_tokens
|
||||
from litellm import token_counter as token_counter_old
|
||||
import litellm.constants
|
||||
from litellm.litellm_core_utils.token_counter import _get_tiktoken_count_function
|
||||
from litellm.constants import TOKEN_COUNTER_MAX_CONCURRENT_COUNTS
|
||||
from litellm.litellm_core_utils.asyncify import asyncify
|
||||
from litellm.litellm_core_utils.token_counter import (
|
||||
_get_exact_count_function,
|
||||
_get_extrapolating_count_function,
|
||||
_get_tiktoken_count_function,
|
||||
offload_token_count,
|
||||
)
|
||||
from litellm.litellm_core_utils.token_counter import token_counter as token_counter_new
|
||||
from tests.large_text import text
|
||||
from tests.test_litellm.litellm_core_utils.event_loop_lag import (
|
||||
assert_loop_stayed_free,
|
||||
timed_with_loop_lags,
|
||||
warm_tokenizer,
|
||||
)
|
||||
from tests.test_litellm.litellm_core_utils.messages_with_counts import (
|
||||
MESSAGES_TEXT,
|
||||
MESSAGES_WITH_IMAGES,
|
||||
|
|
@ -120,6 +137,135 @@ def test_valid_chunk_size_config_is_honoured(monkeypatch):
|
|||
importlib.reload(litellm.constants)
|
||||
|
||||
|
||||
async def test_huggingface_count_in_a_worker_thread_leaves_the_event_loop_free():
|
||||
warm_tokenizer("claude-fable-5")
|
||||
|
||||
tokens, took, lags = await timed_with_loop_lags(
|
||||
lambda: asyncify(token_counter_new)(model="claude-fable-5", text=text * 100)
|
||||
)
|
||||
|
||||
assert tokens > 0
|
||||
assert_loop_stayed_free(took, lags)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("max_exact_chars", [64, 1_000, 2_500])
|
||||
def test_count_above_the_cap_samples_the_whole_string_and_scales(max_exact_chars: int):
|
||||
count_exactly: Final = MagicMock(side_effect=lambda chunk: chunk.count("a") + len(chunk))
|
||||
front_heavy: Final = "a" * 1_000 + "b" * 4_000
|
||||
exact: Final = 1_000 + len(front_heavy)
|
||||
|
||||
estimate: Final = _get_extrapolating_count_function(count_exactly, max_exact_chars=max_exact_chars)(front_heavy)
|
||||
|
||||
assert abs(estimate - exact) <= exact // 100
|
||||
assert sum(len(call.args[0]) for call in count_exactly.call_args_list) <= max_exact_chars
|
||||
|
||||
|
||||
def test_count_at_or_below_the_cap_is_exact():
|
||||
count_exactly: Final = MagicMock(side_effect=len)
|
||||
|
||||
assert _get_extrapolating_count_function(count_exactly, max_exact_chars=5_000)("a" * 5_000) == 5_000
|
||||
assert count_exactly.call_args_list == [(("a" * 5_000,),)]
|
||||
|
||||
|
||||
class _SlowEncoder:
|
||||
def __init__(self) -> None:
|
||||
self._lock: Final = threading.Lock()
|
||||
self.in_flight = 0
|
||||
self.peak_in_flight = 0
|
||||
|
||||
def encode_batch_fast(self, texts: list[str]) -> list[list[int]]:
|
||||
with self._lock:
|
||||
self.in_flight += 1
|
||||
self.peak_in_flight = max(self.peak_in_flight, self.in_flight)
|
||||
time.sleep(0.1)
|
||||
with self._lock:
|
||||
self.in_flight -= 1
|
||||
return [[0] * len(text) for text in texts]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_offloaded_counts_do_not_borrow_from_the_shared_thread_pool():
|
||||
encoder: Final = _SlowEncoder()
|
||||
count: Final = _get_exact_count_function(None, {"type": "huggingface_tokenizer", "tokenizer": encoder})
|
||||
shared_pool: Final = anyio.to_thread.current_default_thread_limiter()
|
||||
burst: Final = 2 * TOKEN_COUNTER_MAX_CONCURRENT_COUNTS
|
||||
|
||||
async def shared_pool_borrowed_until_done(counting: asyncio.Future[list[int]]) -> tuple[int, ...]:
|
||||
if counting.done():
|
||||
return ()
|
||||
await asyncio.sleep(0.01)
|
||||
return (shared_pool.borrowed_tokens, *await shared_pool_borrowed_until_done(counting))
|
||||
|
||||
counting: Final = asyncio.ensure_future(asyncio.gather(*(offload_token_count(count)("abc") for _ in range(burst))))
|
||||
borrowed: Final = await shared_pool_borrowed_until_done(counting)
|
||||
|
||||
assert await counting == [3] * burst
|
||||
assert len(borrowed) > 1 and max(borrowed) == 0
|
||||
assert 1 < encoder.peak_in_flight <= TOKEN_COUNTER_MAX_CONCURRENT_COUNTS
|
||||
|
||||
|
||||
def _count_in_a_fresh_event_loop(text: str, result: Future[int]) -> None:
|
||||
def slow_count(counted: str) -> int:
|
||||
time.sleep(0.1)
|
||||
return len(counted)
|
||||
|
||||
result.set_result(asyncio.run(offload_token_count(slow_count)(text)))
|
||||
|
||||
|
||||
def test_offloaded_counts_finish_in_every_event_loop_that_shares_the_process():
|
||||
loops: Final = 2 * TOKEN_COUNTER_MAX_CONCURRENT_COUNTS
|
||||
results: Final = tuple(Future[int]() for _ in range(loops))
|
||||
threads: Final = tuple(
|
||||
threading.Thread(target=_count_in_a_fresh_event_loop, args=("a" * size, result), daemon=True)
|
||||
for size, result in enumerate(results, start=1)
|
||||
)
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
|
||||
_, pending = wait(results, timeout=5)
|
||||
|
||||
assert not pending
|
||||
assert tuple(result.result() for result in results) == tuple(range(1, loops + 1))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("configured", "expected"),
|
||||
[("8", 8), ("0", 4), ("not-an-int", 4)],
|
||||
)
|
||||
def test_max_concurrent_counts_config_is_honoured(monkeypatch: pytest.MonkeyPatch, configured: str, expected: int):
|
||||
monkeypatch.setenv("TOKEN_COUNTER_MAX_CONCURRENT_COUNTS", configured)
|
||||
try:
|
||||
assert importlib.reload(litellm.constants).TOKEN_COUNTER_MAX_CONCURRENT_COUNTS == expected
|
||||
finally:
|
||||
monkeypatch.delenv("TOKEN_COUNTER_MAX_CONCURRENT_COUNTS")
|
||||
importlib.reload(litellm.constants)
|
||||
|
||||
|
||||
def test_token_counter_applies_the_default_cap():
|
||||
max_exact_chars: Final = litellm.constants.TOKEN_COUNTER_MAX_EXACT_CHARS
|
||||
prose: Final = ("The quick brown fox jumps over the lazy dog. " * (max_exact_chars // 45 + 1))[:max_exact_chars]
|
||||
over_the_cap: Final = prose + "a" * 200_000
|
||||
exact: Final = _get_exact_count_function("gpt-5.6")(over_the_cap)
|
||||
|
||||
estimate: Final = token_counter_new(model="gpt-5.6", text=over_the_cap)
|
||||
|
||||
assert estimate != exact
|
||||
assert abs(estimate - exact) <= exact // 100
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("configured", "expected"),
|
||||
[("2048", 2048), ("0", 4_000_000), ("not-an-int", 4_000_000)],
|
||||
)
|
||||
def test_max_exact_chars_config_is_honoured(monkeypatch: pytest.MonkeyPatch, configured: str, expected: int):
|
||||
monkeypatch.setenv("TOKEN_COUNTER_MAX_EXACT_CHARS", configured)
|
||||
try:
|
||||
assert importlib.reload(litellm.constants).TOKEN_COUNTER_MAX_EXACT_CHARS == expected
|
||||
finally:
|
||||
monkeypatch.delenv("TOKEN_COUNTER_MAX_EXACT_CHARS")
|
||||
importlib.reload(litellm.constants)
|
||||
|
||||
|
||||
def test_token_counter_with_prefix():
|
||||
messages = [
|
||||
{"role": "user", "content": "Who won the world cup in 2022?"},
|
||||
|
|
|
|||
|
|
@ -3670,5 +3670,42 @@ async def test_post_call_success_hook_contains_header_merge_failures(
|
|||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_project_itpm_reservation_counts_the_request_off_the_event_loop(rate_limiter):
|
||||
from tests.large_text import text
|
||||
from tests.test_litellm.litellm_core_utils.event_loop_lag import (
|
||||
assert_loop_stayed_free,
|
||||
timed_with_loop_lags,
|
||||
warm_tokenizer,
|
||||
)
|
||||
|
||||
handler, _cache = rate_limiter
|
||||
stash = get_or_create_request_stash()
|
||||
warm_tokenizer("claude-fable-5")
|
||||
data: dict[str, object] = {
|
||||
"model": "claude-fable-5",
|
||||
"messages": [{"role": "user", "content": text * 100}],
|
||||
}
|
||||
itpm_descriptor = {
|
||||
"key": PROJECT_ITPM_DESCRIPTOR_KEY,
|
||||
"value": "proj-loop:claude-fable-5",
|
||||
"rate_limit": {"tokens_per_unit": 10_000_000, "window_size": 60},
|
||||
}
|
||||
|
||||
_, took, lags = await timed_with_loop_lags(
|
||||
lambda: handler._reserve_project_io_tokens_or_raise(
|
||||
descriptors=[itpm_descriptor],
|
||||
data=data,
|
||||
requested_model="claude-fable-5",
|
||||
user_api_key_dict=UserAPIKeyAuth(api_key=hash_token("sk-itpm-loop"), project_id="proj-loop"),
|
||||
tpm_reservation_scopes=[],
|
||||
tpm_reservation_amount=0,
|
||||
)
|
||||
)
|
||||
|
||||
assert stash.rate_limit_response is not None
|
||||
assert_loop_stayed_free(took, lags)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v", "-s"])
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import os
|
|||
import re
|
||||
import socket
|
||||
import subprocess
|
||||
import time
|
||||
import types
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
|
@ -28,7 +29,7 @@ from litellm.caching.caching import RedisCache
|
|||
from litellm.caching.redis_cluster_cache import RedisClusterCache
|
||||
from litellm.litellm_core_utils.get_model_cost_map import ModelCostMapReloaded
|
||||
from litellm.caching.dual_cache import DualCache
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
from litellm.proxy._types import LitellmUserRoles, TokenCountRequest, UserAPIKeyAuth
|
||||
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
|
||||
from litellm.proxy.proxy_server import app, initialize
|
||||
from litellm.utils import _invalidate_model_cost_lowercase_map
|
||||
|
|
@ -12954,3 +12955,59 @@ async def test_update_general_settings_keeps_yaml_openai_websocket_passthrough()
|
|||
import litellm.proxy.proxy_server as ps
|
||||
|
||||
assert ps.general_settings["enable_openai_websocket_passthrough"] is False
|
||||
|
||||
|
||||
async def test_token_counter_keeps_the_event_loop_free_during_a_huggingface_count(monkeypatch):
|
||||
from tests.large_text import text
|
||||
from tests.test_litellm.litellm_core_utils.event_loop_lag import (
|
||||
assert_loop_stayed_free,
|
||||
timed_with_loop_lags,
|
||||
warm_tokenizer,
|
||||
)
|
||||
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None)
|
||||
warm_tokenizer("claude-fable-5")
|
||||
|
||||
response, took, lags = await timed_with_loop_lags(
|
||||
lambda: proxy_server_module.token_counter(TokenCountRequest(model="claude-fable-5", prompt=text * 100))
|
||||
)
|
||||
|
||||
assert response.total_tokens > 0
|
||||
assert_loop_stayed_free(took, lags)
|
||||
|
||||
|
||||
async def test_token_counter_loads_a_custom_tokenizer_off_the_event_loop(monkeypatch):
|
||||
from tokenizers import Tokenizer
|
||||
|
||||
from litellm import Router
|
||||
from tests.test_litellm.litellm_core_utils.event_loop_lag import assert_loop_stayed_free, timed_with_loop_lags
|
||||
|
||||
claude_tokenizer: Final = litellm.utils._select_tokenizer("claude-fable-5")["tokenizer"]
|
||||
|
||||
class SlowHubTokenizer:
|
||||
@staticmethod
|
||||
def from_pretrained(identifier: str, revision: str = "main", token: str | None = None) -> Tokenizer:
|
||||
time.sleep(0.3)
|
||||
return claude_tokenizer
|
||||
|
||||
monkeypatch.setattr(litellm.utils, "Tokenizer", SlowHubTokenizer)
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy.proxy_server.llm_router",
|
||||
Router(
|
||||
model_list=[
|
||||
{
|
||||
"model_name": "self-hosted",
|
||||
"litellm_params": {"model": "openai/self-hosted-model", "api_base": "http://localhost:8080/v1"},
|
||||
"model_info": {"custom_tokenizer": {"identifier": "my-org/tokenizer", "revision": "main", "auth_token": None}},
|
||||
}
|
||||
]
|
||||
),
|
||||
)
|
||||
|
||||
response, took, lags = await timed_with_loop_lags(
|
||||
lambda: proxy_server_module.token_counter(TokenCountRequest(model="self-hosted", prompt="count me off the loop"))
|
||||
)
|
||||
|
||||
assert response.tokenizer_type == "huggingface_tokenizer"
|
||||
assert response.total_tokens > 0
|
||||
assert_loop_stayed_free(took, lags)
|
||||
|
|
|
|||
|
|
@ -1823,6 +1823,44 @@ def test_a_dispatched_failure_lifts_the_four_fields_the_spend_log_needs():
|
|||
assert lifted["standard_logging_object"] == {"id": "log-1"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_dispatched_failure_is_counted_off_the_event_loop():
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from tests.large_text import text
|
||||
from tests.test_litellm.litellm_core_utils.event_loop_lag import (
|
||||
assert_loop_stayed_free,
|
||||
timed_with_loop_lags,
|
||||
warm_tokenizer,
|
||||
)
|
||||
|
||||
warm_tokenizer("claude-fable-5")
|
||||
request_data = {
|
||||
"litellm_logging_obj": _LoggingObj(
|
||||
{
|
||||
"first_api_call_start_time": 1700000000.0,
|
||||
"call_type": "acompletion",
|
||||
"model": "claude-fable-5",
|
||||
"messages": [{"role": "user", "content": text * 100}],
|
||||
}
|
||||
),
|
||||
"metadata": {},
|
||||
}
|
||||
proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache())
|
||||
proxy_logging_obj.alert_types = []
|
||||
with patch.object(proxy_logging_obj, "update_request_status", new=AsyncMock()):
|
||||
_, took, lags = await timed_with_loop_lags(
|
||||
lambda: proxy_logging_obj.post_call_failure_hook(
|
||||
request_data=request_data,
|
||||
original_exception=Exception("boom"),
|
||||
user_api_key_dict=UserAPIKeyAuth(),
|
||||
)
|
||||
)
|
||||
|
||||
assert request_data["combined_usage_object"].prompt_tokens > 0
|
||||
assert_loop_stayed_free(took, lags)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_proxy_only_error_expected_4xx_skips_traceback_for_both_handlers(monkeypatch):
|
||||
"""Regression for LIT-6043: an expected 4xx must not format a traceback for
|
||||
|
|
|
|||
|
|
@ -477,3 +477,65 @@ async def test_wildcard_route_resolves_underlying_model_minimum(local_model_cost
|
|||
|
||||
assert deployments[0]["litellm_params"]["model"] == "anthropic/claude-opus-4-6"
|
||||
assert _get_min_token_count_for_deployments(deployments) == 4096
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_filter_deployments_counts_the_prompt_off_the_event_loop():
|
||||
from tests.large_text import text
|
||||
from tests.test_litellm.litellm_core_utils.event_loop_lag import (
|
||||
assert_loop_stayed_free,
|
||||
timed_with_loop_lags,
|
||||
warm_tokenizer,
|
||||
)
|
||||
|
||||
warm_tokenizer("anthropic/claude-fable-5")
|
||||
check = PromptCachingDeploymentCheck(cache=DualCache())
|
||||
deployments = _deployments("anthropic/claude-fable-5")
|
||||
messages = cast(List[AllMessageValues], [{"role": "user", "content": text * 100}])
|
||||
|
||||
result, took, lags = await timed_with_loop_lags(
|
||||
lambda: check.async_filter_deployments(
|
||||
model=MODEL_GROUP_ALIAS, healthy_deployments=deployments, messages=messages
|
||||
)
|
||||
)
|
||||
|
||||
assert result == deployments
|
||||
assert_loop_stayed_free(took, lags)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_log_success_event_counts_the_prompt_off_the_event_loop():
|
||||
from tests.large_text import text
|
||||
from tests.test_litellm.litellm_core_utils.event_loop_lag import (
|
||||
assert_loop_stayed_free,
|
||||
timed_with_loop_lags,
|
||||
warm_tokenizer,
|
||||
)
|
||||
|
||||
warm_tokenizer("anthropic/claude-fable-5")
|
||||
cache = DualCache()
|
||||
check = PromptCachingDeploymentCheck(cache=cache)
|
||||
messages = cast(
|
||||
List[AllMessageValues],
|
||||
[{"role": "user", "content": [{"type": "text", "text": text * 100, "cache_control": {"type": "ephemeral"}}]}],
|
||||
)
|
||||
standard_logging_object = {
|
||||
"call_type": "acompletion",
|
||||
"model": "anthropic/claude-fable-5",
|
||||
"messages": messages,
|
||||
"model_id": "dep-1",
|
||||
}
|
||||
|
||||
_, took, lags = await timed_with_loop_lags(
|
||||
lambda: check.async_log_success_event(
|
||||
kwargs={"standard_logging_object": standard_logging_object},
|
||||
response_obj=None,
|
||||
start_time=None,
|
||||
end_time=None,
|
||||
)
|
||||
)
|
||||
|
||||
assert await PromptCachingCache(cache=cache).async_get_model_id(messages=messages, tools=None) == {
|
||||
"model_id": "dep-1"
|
||||
}
|
||||
assert_loop_stayed_free(took, lags)
|
||||
|
|
|
|||
|
|
@ -1039,3 +1039,31 @@ class TestContextSlotRetention:
|
|||
assert deployment is not None
|
||||
router._update_kwargs_with_deployment(deployment=deployment.model_dump(), kwargs=kwargs)
|
||||
assert get_io_token_rate_limit_request_kwargs() is kwargs
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_deployment_itpm_reservation_counts_the_request_off_the_event_loop():
|
||||
from litellm.utils import get_utc_datetime
|
||||
from tests.large_text import text
|
||||
from tests.test_litellm.litellm_core_utils.event_loop_lag import (
|
||||
assert_loop_stayed_free,
|
||||
timed_with_loop_lags,
|
||||
warm_tokenizer,
|
||||
)
|
||||
|
||||
dual_cache = DualCache()
|
||||
check = ModelRateLimitingCheck(dual_cache=dual_cache)
|
||||
warm_tokenizer("anthropic/claude-fable-5")
|
||||
deployment = {
|
||||
"litellm_params": {"model": "anthropic/claude-fable-5", "itpm": 10_000_000},
|
||||
"model_info": {"id": "io-loop-id"},
|
||||
"model_name": "claude",
|
||||
}
|
||||
set_io_token_rate_limit_request_kwargs({"messages": [{"role": "user", "content": text * 100}], "metadata": {}})
|
||||
|
||||
_, took, lags = await timed_with_loop_lags(lambda: check.async_pre_call_check(deployment))
|
||||
|
||||
minute = get_utc_datetime().strftime("%H-%M")
|
||||
reserved = await dual_cache.async_get_cache(key=f"global_router:io-loop-id:anthropic/claude-fable-5:itpm:{minute}")
|
||||
assert reserved > 100_000
|
||||
assert_loop_stayed_free(took, lags)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue