fix(token_counter): count DeepSeek V4 text with DeepSeek's own tokenizer (#47353)

Backport of #47353 to stable/1.104.x.
Cherry-picked from 2d181e6633 (main).
Adapted to the stable compatibility rule: the bundled DeepSeek tokenizer is selected only on the proxy count routes, through select_count_route_tokenizer, so token_counter, encode and decode, budget reservation, ITPM, context-management triggers and failed-request spend rows keep the line's tokenizer and a request admitted on v1.104.4 is still admitted. A DeepSeek tokenizer that fails to load falls back to tiktoken on the count routes, the way every other bundled tokenizer does. The wire tests read the line's proxy_config.yaml, which has no general_settings or router_settings, and use azure_ai/grok-4 as the control model, since the line's azure_ai/grok-4-1 entry rejects tool_choice. The failed-dispatch spend test is left out, and the wire tests that asserted the DeepSeek tokenizer on those paths now assert the line's count is kept.

litellm/litellm_core_utils/prompt_templates/chat_template_tools.py

@@ -24,7 +24,9 @@ def _qwen3_5_tool_schemas(functions: Sequence[Mapping[str, object]]) -> str:
 
 _TOOL_SECTIONS: Final = (
     (
-        re.compile(r"deepseek-v4(?![.\d]|p\d)", re.IGNORECASE),
+        re.compile(
+            r"deepseek-v4(?![.\d]|p\d)|^(?:deepseek/)?deepseek-(?:chat|reasoner|flash)(?::free)?$", re.IGNORECASE
+        ),
         ChatTemplateToolSection(222, 0, _deepseek_v4_tool_schemas),
     ),
     (re.compile(r"qwen3[._]?5(?!\d)", re.IGNORECASE), ChatTemplateToolSection(196, 5, _qwen3_5_tool_schemas)),

litellm/litellm_core_utils/tokenizers/deepseek_tokenizer_LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 DeepSeek
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

litellm/proxy/proxy_server.py

@@ -13788,7 +13788,7 @@ async def _loaded_tokenizer(
     model: str, custom_tokenizer: CustomHuggingfaceTokenizer | None, threads: anyio.CapacityLimiter
 ) -> SelectTokenizerResponse | BaseException:
     try:
-        return await asyncify(litellm.utils._select_tokenizer, limiter=threads)(
+        return await asyncify(litellm.utils.select_count_route_tokenizer, limiter=threads)(
             model=model, custom_tokenizer=custom_tokenizer
         )
     except asyncio.CancelledError:
@@ -13912,7 +13912,7 @@ async def _local_token_estimate(
     provider_failure: ProviderCountFailure | None,
 ) -> _LocalTokenEstimate:
     tokenizer: Final = (
-        await asyncify(litellm.utils._select_tokenizer)(model=model, custom_tokenizer=custom_tokenizer)
+        await asyncify(litellm.utils.select_count_route_tokenizer)(model=model, custom_tokenizer=custom_tokenizer)
         if provider_failure is None
         else await _tokenizer_or_provider_error(model, custom_tokenizer, provider_failure)
     )

litellm/utils.py

@@ -29,6 +29,7 @@ import textwrap
 import threading
 import time
 import traceback
+from contextlib import nullcontext
 from dataclasses import dataclass, field
 from functools import lru_cache, wraps
 from importlib import resources
@@ -2389,12 +2390,12 @@ def _select_custom_tokenizer_helper(
     return create_pretrained_tokenizer(identifier=identifier, revision=revision, auth_token=auth_token)
 
 
-def _select_tokenizer_helper(model: str) -> SelectTokenizerResponse:
+def _select_tokenizer_helper(model: str, *, count_route: bool = False) -> SelectTokenizerResponse:
     if litellm.disable_hf_tokenizer_download is True:
         return _return_openai_tokenizer(model)
 
     try:
-        result: Final = _return_huggingface_tokenizer(model)
+        result: Final = _return_huggingface_tokenizer(model, count_route=count_route)
         if result is not None:
             return result
     except Exception as e:
@@ -2416,7 +2417,15 @@ def uses_anthropic_tokenizer(model: str) -> bool:
     return model in litellm.anthropic_models and "claude-3" not in model
 
 
-HuggingFaceTokenizerKind = Literal["cohere", "anthropic", "llama2", "llama3"]
+HuggingFaceTokenizerKind = Literal["cohere", "anthropic", "deepseek", "llama2", "llama3"]
+
+_DEEPSEEK_TOKENIZER_MODEL: Final = re.compile(
+    r"deepseek[-_.](?:chat[-_.])?v[34]|(?:^|/)deepseek-(?:chat|reasoner|flash)(?::free)?$", re.IGNORECASE
+)
+
+_BUNDLED_TOKENIZER_LOAD_LOCKS: Final[Mapping[HuggingFaceTokenizerKind, threading.Lock]] = MappingProxyType(
+    {"anthropic": threading.Lock(), "deepseek": threading.Lock()}
+)
 
 
 def huggingface_tokenizer_kind(model: str) -> HuggingFaceTokenizerKind | None:
@@ -2432,14 +2441,27 @@ def huggingface_tokenizer_kind(model: str) -> HuggingFaceTokenizerKind | None:
     return None
 
 
-def _return_huggingface_tokenizer(model: str) -> SelectTokenizerResponse | None:
-    kind: Final = huggingface_tokenizer_kind(model)
+def _return_huggingface_tokenizer(model: str, *, count_route: bool = False) -> SelectTokenizerResponse | None:
+    kind: Final = (
+        "deepseek" if count_route and _DEEPSEEK_TOKENIZER_MODEL.search(model) else huggingface_tokenizer_kind(model)
+    )
     if kind is None:
         return None
-    return {
-        "type": "huggingface_tokenizer",
-        "tokenizer": _load_huggingface_tokenizer(kind, _huggingface_tokenizer_backend()),
-    }
+    return _huggingface_tokenizer(kind)
+
+
+def _huggingface_tokenizer(kind: HuggingFaceTokenizerKind) -> SelectTokenizerResponse:
+    with _BUNDLED_TOKENIZER_LOAD_LOCKS.get(kind, nullcontext()):
+        tokenizer: Final = _load_huggingface_tokenizer(kind, _huggingface_tokenizer_backend())
+    return {"type": "huggingface_tokenizer", "tokenizer": tokenizer}
+
+
+def select_count_route_tokenizer(
+    model: str, custom_tokenizer: CustomHuggingfaceTokenizer | None = None
+) -> SelectTokenizerResponse:
+    if custom_tokenizer is None and _DEEPSEEK_TOKENIZER_MODEL.search(model):
+        return _select_tokenizer_helper(model=model, count_route=True)
+    return _select_tokenizer(model=model, custom_tokenizer=custom_tokenizer)
 
 
 @lru_cache(maxsize=DEFAULT_MAX_LRU_CACHE_SIZE)
@@ -2450,6 +2472,12 @@ def _load_huggingface_tokenizer(kind: HuggingFaceTokenizerKind, backend: Decisio
             return tokenizer_dispatch.from_pretrained("Xenova/c4ai-command-r-v01-tokenizer")
         case "anthropic":
             return tokenizer_dispatch.anthropic()
+        case "deepseek":
+            return tokenizer_dispatch.from_str(
+                resources.files("litellm.litellm_core_utils.tokenizers")
+                .joinpath("deepseek_tokenizer.json")
+                .read_text(encoding="utf-8")
+            )
         case "llama2":
             return tokenizer_dispatch.from_pretrained("hf-internal-testing/llama-tokenizer")
         case "llama3":

tests/integration/_support/deepseek_tokens.py

@@ -0,0 +1,35 @@
+from collections.abc import Callable
+from functools import cache
+from importlib import resources
+from typing import Final
+
+from litellm.litellm_core_utils.default_encoding import encoding as cl100k_base
+from tokenizers import Tokenizer
+
+CHINESE_TEXT: Final = "深度求索的分词器会把这段中文切成更少的词元。" * 20
+SHORT_TEXT: Final = "Find record 14."
+
+
+@cache
+def _deepseek_tokenizer() -> Tokenizer:
+    bundled: Final = resources.files("litellm.litellm_core_utils.tokenizers").joinpath("deepseek_tokenizer.json")
+    return Tokenizer.from_str(bundled.read_text(encoding="utf-8"))
+
+
+def deepseek_tokens(text: str) -> int:
+    return len(_deepseek_tokenizer().encode(text).ids)
+
+
+def cl100k_tokens(text: str) -> int:
+    return len(cl100k_base.encode(text))
+
+
+def chinese_minus_short(tokens: Callable[[str], int]) -> int:
+    return tokens(CHINESE_TEXT) - tokens(SHORT_TEXT)
+
+
+def assert_counted_with_deepseek(counted: int, expected: Callable[[Callable[[str], int]], int]) -> None:
+    cl100k_count: Final = expected(cl100k_tokens)
+    assert counted != cl100k_count, f"counted {counted}, the cl100k_base count, not DeepSeek's tokenizer"
+    deepseek_count: Final = expected(deepseek_tokens)
+    assert counted == deepseek_count, f"counted {counted}, DeepSeek's tokenizer counts {deepseek_count}"

tests/integration/providers/test_chat_template_tool_count_wire.py

@@ -1,7 +1,8 @@
 import json
 import re
 import uuid
-from collections.abc import Iterator, Mapping, Sequence
+from collections.abc import Callable, Iterator, Mapping, Sequence
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import ExitStack
 from dataclasses import dataclass
 from itertools import chain
@@ -19,18 +20,28 @@ from integration._support.client import (
     object_value,
     string_value,
 )
+from integration._support.deepseek_tokens import (
+    CHINESE_TEXT,
+    SHORT_TEXT,
+    assert_counted_with_deepseek,
+    chinese_minus_short,
+    cl100k_tokens,
+)
 from integration._support.openai_wire import chat_reply
 from integration._support.process import graceful_stop_seconds, owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
-from litellm.litellm_core_utils.default_encoding import encoding as cl100k_base
 from pydantic import JsonValue
 
+from litellm.litellm_core_utils.default_encoding import encoding as cl100k_base
+
 pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
 
 _API_KEY: Final = "scripted-provider-key"
 _DEEPSEEK_V4: Final = "azure_ai/DeepSeek-V4-Flash-0731"
+_DEEPSEEK_R1: Final = "azure_ai/DeepSeek-R1-0528"
 _QWEN3_5: Final = "databricks/databricks-qwen35-122b-a10b"
 _GROK: Final = "azure_ai/grok-4-1"
+_TOOL_CHOICE_CONTROL: Final = "azure_ai/grok-4"
 _DEEPSEEK_V4_TOOL_PREAMBLE: Final = 222
 _QWEN3_5_TOOL_PREAMBLE: Final = 196
 _QWEN3_5_OPENED_SYSTEM_TURN: Final = 5
@@ -47,7 +58,7 @@ _OPENAI_TOOL: Final[dict[str, JsonValue]] = {"type": "function", "function": _FU
 _RESPONSES_TOOL: Final[dict[str, JsonValue]] = {"type": "function", **_FUNCTION}
 _NAMED_TOOL_CHOICE: Final[dict[str, JsonValue]] = {"type": "function", "function": {"name": "lookup"}}
 _WEB_SEARCH_TOOL: Final[dict[str, JsonValue]] = {"type": "web_search_20250305", "name": "web_search", "max_uses": 3}
-_PROMPT: Final = "Find record 14."
+_PROMPT: Final = SHORT_TEXT
 _MESSAGES: Final[list[JsonValue]] = [{"role": "user", "content": _PROMPT}]
 _INPUT_PRICE: Final = 0.001
 _OUTPUT_PRICE: Final = 0.002
@@ -59,8 +70,8 @@ def _tokens(text: str) -> int:
     return len(cl100k_base.encode(text))
 
 
-def _deepseek_v4_tool_tokens(function: Mapping[str, JsonValue]) -> int:
-    return _DEEPSEEK_V4_TOOL_PREAMBLE + _tokens(json.dumps(function, ensure_ascii=False))
+def _deepseek_v4_tool_tokens(function: Mapping[str, JsonValue], tokens: Callable[[str], int]) -> int:
+    return _DEEPSEEK_V4_TOOL_PREAMBLE + tokens(json.dumps(function, ensure_ascii=False))
 
 
 def _qwen3_5_tool_tokens(function: Mapping[str, JsonValue], *, with_system: bool) -> int:
@@ -93,9 +104,11 @@ class _CountRoute:
     field: str
     tool: dict[str, JsonValue]
 
-    def body(self, model: str, tools: Sequence[JsonValue]) -> dict[str, JsonValue]:
+    def body(self, model: str, tools: Sequence[JsonValue], text: str = _PROMPT) -> dict[str, JsonValue]:
         prompt: Final[dict[str, JsonValue]] = (
-            {"input": _PROMPT} if self.path == "/v1/responses/input_tokens" else {"messages": _MESSAGES}
+            {"input": text}
+            if self.path == "/v1/responses/input_tokens"
+            else {"messages": [{"role": "user", "content": text}]}
         )
         return {"model": model, **prompt, **({"tools": list(tools)} if tools else {})}
 
@@ -116,18 +129,85 @@ def _count(gateway: Gateway, route: _CountRoute, body: Mapping[str, JsonValue])
     return counted
 
 
-@pytest.mark.parametrize("route", sorted(_COUNT_ROUTES))
+_TOOL_COUNTS: Final[Mapping[str, tuple[str, str]]] = {
+    **{f"{route}-azure_ai-v4": (route, _DEEPSEEK_V4) for route in _COUNT_ROUTES},
+    "messages-count-tokens-deepseek-chat": ("messages-count-tokens", "deepseek/deepseek-chat"),
+    "messages-count-tokens-deepseek-reasoner": ("messages-count-tokens", "deepseek/deepseek-reasoner"),
+    "messages-count-tokens-openrouter-deepseek-chat": ("messages-count-tokens", "openrouter/deepseek/deepseek-chat"),
+}
+
+
+@pytest.mark.parametrize("case", sorted(_TOOL_COUNTS))
 def test_count_routes_count_deepseek_v4_tools_the_way_its_chat_template_renders_them(
-    gateway: Gateway, route: str
+    gateway: Gateway, case: str
 ) -> None:
+    route, backend = _TOOL_COUNTS[case]
     selected: Final = _COUNT_ROUTES[route]
     with wire_server(_completion) as wire, gateway.scenario() as scenario:
-        model: Final = _deployment(scenario, "count", _DEEPSEEK_V4, wire.url)
+        model: Final = _deployment(scenario, "count", backend, wire.url)
         without_tools: Final = _count(gateway, selected, selected.body(model, ()))
         with_tool: Final = _count(gateway, selected, selected.body(model, (selected.tool,)))
         upstream: Final = wire.drain()
     assert upstream == ()
-    assert with_tool - without_tools == _deepseek_v4_tool_tokens(_FUNCTION)
+    tool_tokens: Final = with_tool - without_tools
+    assert tool_tokens > _DEEPSEEK_V4_TOOL_PREAMBLE, f"counted {tool_tokens} tool tokens, short of V4's tool preamble"
+    assert_counted_with_deepseek(tool_tokens, lambda tokens: _deepseek_v4_tool_tokens(_FUNCTION, tokens))
+
+
+_TEXT_COUNTS: Final[Mapping[str, tuple[str, str]]] = {
+    "messages-count-tokens-azure_ai-v4": ("messages-count-tokens", _DEEPSEEK_V4),
+    "responses-input-tokens-azure_ai-v4": ("responses-input-tokens", _DEEPSEEK_V4),
+    "messages-count-tokens-deepseek-chat": ("messages-count-tokens", "deepseek/deepseek-chat"),
+}
+
+
+def _text_count_delta(gateway: Gateway, route: _CountRoute, model: str) -> int:
+    chinese: Final = _count(gateway, route, route.body(model, (), CHINESE_TEXT))
+    short: Final = _count(gateway, route, route.body(model, (), SHORT_TEXT))
+    return chinese - short
+
+
+@pytest.mark.parametrize("case", sorted(_TEXT_COUNTS))
+def test_count_routes_count_deepseek_v3_and_v4_text_with_deepseeks_tokenizer(gateway: Gateway, case: str) -> None:
+    route, backend = _TEXT_COUNTS[case]
+    with wire_server(_completion) as wire, gateway.scenario() as scenario:
+        delta: Final = _text_count_delta(
+            gateway, _COUNT_ROUTES[route], _deployment(scenario, "count", backend, wire.url)
+        )
+        upstream: Final = wire.drain()
+    assert upstream == ()
+    assert_counted_with_deepseek(delta, chinese_minus_short)
+
+
+def _counted_with_deepseek(delta: int) -> None:
+    assert_counted_with_deepseek(delta, chinese_minus_short)
+
+
+def _counted_with_cl100k(delta: int) -> None:
+    assert delta == chinese_minus_short(cl100k_tokens)
+
+
+@pytest.mark.parametrize(
+    ("backend", "tokenizer_type", "counted_with"),
+    (
+        (_DEEPSEEK_V4, "huggingface_tokenizer", _counted_with_deepseek),
+        (_DEEPSEEK_R1, "openai_tokenizer", _counted_with_cl100k),
+    ),
+    ids=("deepseek-v4", "deepseek-r1"),
+)
+def test_utils_token_counter_names_the_tokenizer_that_counted_deepseek_text(
+    gateway: Gateway, backend: str, tokenizer_type: str, counted_with: Callable[[int], None]
+) -> None:
+    route: Final = _COUNT_ROUTES["utils-token-counter"]
+    with wire_server(_completion) as wire, gateway.scenario() as scenario:
+        model: Final = _deployment(scenario, "count", backend, wire.url)
+        delta: Final = _text_count_delta(gateway, route, model)
+        response: Final = gateway.request("POST", route.path, route.body(model, (), CHINESE_TEXT))
+        upstream: Final = wire.drain()
+    assert upstream == ()
+    assert response.status_code == 200, response.text
+    assert JSON_OBJECT.validate_json(response.content)["tokenizer_type"] == tokenizer_type, response.text
+    counted_with(delta)
 
 
 @pytest.mark.parametrize("with_system", (False, True), ids=("no-system", "system"))
@@ -162,7 +242,12 @@ def test_messages_count_tokens_skips_web_search_and_counts_unmapped_tool_keys_fo
         upstream: Final = wire.drain()
     assert upstream == ()
     assert with_web_search == plain
-    assert deferred - plain == _deepseek_v4_tool_tokens(deferred_function) - _deepseek_v4_tool_tokens(_FUNCTION)
+    assert_counted_with_deepseek(
+        deferred - plain,
+        lambda tokens: (
+            _deepseek_v4_tool_tokens(deferred_function, tokens) - _deepseek_v4_tool_tokens(_FUNCTION, tokens)
+        ),
+    )
 
 
 def test_messages_count_tokens_keeps_openai_tool_framing_for_other_models(gateway: Gateway) -> None:
@@ -222,6 +307,24 @@ class _BudgetRig:
     grok_group: str
 
 
+def _owned_config(
+    directory: Path,
+    model_list: Sequence[JsonValue],
+    general_settings: Mapping[str, JsonValue],
+    router_settings: Mapping[str, JsonValue],
+) -> Path:
+    base: Final = JSON_OBJECT.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()))
+    config: Final = {
+        **base,
+        "model_list": list(model_list),
+        "general_settings": {**object_value(base.get("general_settings", {})), **general_settings},
+        "router_settings": {**object_value(base.get("router_settings", {})), **router_settings},
+    }
+    path: Final = directory / f"chat-template-tool-count-{uuid.uuid4().hex[:8]}.yaml"
+    path.write_text(yaml.safe_dump(config))
+    return path
+
+
 def _priced_deployment(group: str, model: str, api_base: str) -> dict[str, JsonValue]:
     return {
         "model_name": group,
@@ -235,27 +338,23 @@ def _priced_deployment(group: str, model: str, api_base: str) -> dict[str, JsonV
     }
 
 
-def _strict_budget_config(directory: Path, groups: Mapping[str, str], api_base: str) -> Path:
-    base: Final = JSON_OBJECT.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()))
-    config: Final = {
-        **base,
-        "model_list": [_priced_deployment(group, model, api_base) for group, model in groups.items()],
-        "general_settings": {**object_value(base["general_settings"]), "fail_closed_budget_enforcement": True},
-    }
-    path: Final = directory / f"chat-template-tool-budget-{uuid.uuid4().hex[:8]}.yaml"
-    path.write_text(yaml.safe_dump(config))
-    return path
-
-
 @pytest.fixture(scope="module")
 def strict_budget(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_BudgetRig]:
     directory: Final = tmp_path_factory.mktemp("chat-template-tool-budget")
     group: Final = f"deepseek-v4-flash-{uuid.uuid4().hex}"
-    grok_group: Final = f"grok-4-1-{uuid.uuid4().hex}"
+    grok_group: Final = f"grok-4-{uuid.uuid4().hex}"
     with ExitStack() as stack:
         rig_gateway: Final = stack.enter_context(gateway_from_environment())
         wire: Final = stack.enter_context(wire_server(_completion))
-        config: Final = _strict_budget_config(directory, {group: _DEEPSEEK_V4, grok_group: _GROK}, wire.url)
+        config: Final = _owned_config(
+            directory,
+            [
+                _priced_deployment(group, _DEEPSEEK_V4, wire.url),
+                _priced_deployment(grok_group, _TOOL_CHOICE_CONTROL, wire.url),
+            ],
+            {"fail_closed_budget_enforcement": True},
+            {},
+        )
         owned: Final = stack.enter_context(owned_proxy_process(rig_gateway, directory, {}, config=config))
         yield _BudgetRig(owned.gateway, wire, group, grok_group)
 
@@ -268,11 +367,10 @@ def _estimated_cost(response: httpx.Response) -> float:
     return float(estimate[1])
 
 
-@pytest.mark.parametrize("endpoint", ("chat-completions", "messages"))
 def test_budget_reservation_counts_deepseek_v4_tools_like_models_without_a_chat_template(
-    strict_budget: _BudgetRig, endpoint: str
+    strict_budget: _BudgetRig,
 ) -> None:
-    selected: Final = _ENDPOINTS[endpoint]
+    selected: Final = _ENDPOINTS["chat-completions"]
     with strict_budget.gateway.scenario() as scenario:
         key: Final = scenario.key(models=[strict_budget.group, strict_budget.grok_group], max_budget=_TINY_BUDGET)
         deepseek: Final = _estimated_cost(_send(strict_budget.gateway, selected, strict_budget.group, key, tools=True))
@@ -281,20 +379,48 @@ def test_budget_reservation_counts_deepseek_v4_tools_like_models_without_a_chat_
     assert deepseek == grok
 
 
-def _project_key(scenario: Scenario, group: str) -> str:
+_CODE_TEXT: Final = "    for item in items:\n        total += item.price * item.quantity\n" * 20
+
+
+def test_budget_reservation_counts_deepseek_v4_text_like_models_without_deepseeks_tokenizer(
+    strict_budget: _BudgetRig,
+) -> None:
+    chat: Final = _ENDPOINTS["chat-completions"]
+    with strict_budget.gateway.scenario() as scenario:
+        key: Final = scenario.key(models=[strict_budget.group, strict_budget.grok_group], max_budget=_TINY_BUDGET)
+        estimates: Final = tuple(
+            _estimated_cost(
+                strict_budget.gateway.request(
+                    "POST",
+                    chat.path,
+                    {"model": group, "messages": [{"role": "user", "content": _CODE_TEXT}], "max_tokens": 1},
+                    key=key,
+                )
+            )
+            for group in (strict_budget.group, strict_budget.grok_group)
+        )
+    assert strict_budget.wire.drain() == ()
+    assert estimates[0] == estimates[1]
+
+
+def _project_key(scenario: Scenario, group: str, itpm_limit: int) -> str:
     team: Final = scenario.team()
-    project: Final = scenario.project(team, model_itpm_limit={group: _deepseek_v4_tool_tokens(_FUNCTION)})
+    project: Final = scenario.project(team, model_itpm_limit={group: itpm_limit})
     return scenario.key(team_id=team, project_id=project)
 
 
-@pytest.mark.parametrize(("alias", "model"), (("deepseek-v4-flash", _DEEPSEEK_V4), ("grok-4-1", _GROK)))
+_TOOL_FRAMING_ITPM_LIMIT: Final = _deepseek_v4_tool_tokens(_FUNCTION, _tokens)
+
+
+@pytest.mark.parametrize(("alias", "model"), (("deepseek-v4-flash", _DEEPSEEK_V4), ("grok-4", _TOOL_CHOICE_CONTROL)))
 @pytest.mark.parametrize("endpoint", sorted(_ENDPOINTS))
 def test_project_itpm_admits_tools_under_a_limit_only_chat_template_framing_exceeds(
     gateway: Gateway, endpoint: str, alias: str, model: str
 ) -> None:
     with wire_server(_completion) as wire, gateway.scenario() as scenario:
         group: Final = _deployment(scenario, alias, model, wire.url)
-        response: Final = _send(gateway, _ENDPOINTS[endpoint], group, _project_key(scenario, group), tools=True)
+        key: Final = _project_key(scenario, group, _TOOL_FRAMING_ITPM_LIMIT)
+        response: Final = _send(gateway, _ENDPOINTS[endpoint], group, key, tools=True)
         upstream: Final = wire.drain()
     assert response.status_code == 200, response.text
     assert len(upstream) == 1, upstream
@@ -311,7 +437,7 @@ _READ_FILE_TOOL: Final[dict[str, JsonValue]] = {
 }
 
 
-def _read_file_round(marker: str, index: int) -> tuple[JsonValue, ...]:
+def _read_file_round(marker: str, index: int, file_body: str) -> tuple[JsonValue, ...]:
     return (
         {"role": "user", "content": f"Read file {index}."},
         {
@@ -326,15 +452,15 @@ def _read_file_round(marker: str, index: int) -> tuple[JsonValue, ...]:
                 {
                     "type": "tool_result",
                     "tool_use_id": f"toolu_{index}",
-                    "content": f"<file {marker}-{index}>short file body.",
+                    "content": f"<file {marker}-{index}>{file_body}",
                 }
             ],
         },
     )
 
 
-def _agent_body(group: str, marker: str) -> dict[str, JsonValue]:
-    rounds: Final = tuple(chain.from_iterable(_read_file_round(marker, index) for index in range(_FILES)))
+def _agent_body(group: str, marker: str, file_body: str, trigger: int) -> dict[str, JsonValue]:
+    rounds: Final = tuple(chain.from_iterable(_read_file_round(marker, index, file_body) for index in range(_FILES)))
     return {
         "model": group,
         "max_tokens": 16,
@@ -344,7 +470,7 @@ def _agent_body(group: str, marker: str) -> dict[str, JsonValue]:
             "edits": [
                 {
                     "type": "clear_tool_uses_20250919",
-                    "trigger": {"type": "input_tokens", "value": _deepseek_v4_tool_tokens(_READ_FILE_FUNCTION)},
+                    "trigger": {"type": "input_tokens", "value": trigger},
                     "keep": {"type": "tool_uses", "value": 1},
                 }
             ]
@@ -357,9 +483,46 @@ def test_messages_context_management_trigger_keeps_openai_tool_framing(gateway:
     marker: Final = uuid.uuid4().hex
     with wire_server(_completion) as wire, gateway.scenario() as scenario:
         group: Final = _deployment(scenario, "context", model, wire.url)
-        response: Final = gateway.request("POST", "/v1/messages", _agent_body(group, marker))
+        trigger: Final = _deepseek_v4_tool_tokens(_READ_FILE_FUNCTION, _tokens)
+        response: Final = gateway.request(
+            "POST", "/v1/messages", _agent_body(group, marker, "short file body.", trigger)
+        )
         upstream: Final = wire.drain()
     assert response.status_code == 200, response.text
     assert len(upstream) == 1, upstream
     sent: Final = upstream[0].body.decode()
     assert tuple(f"<file {marker}-{index}>" in sent for index in range(_FILES)) == (True,) * _FILES
+
+
+_BURST: Final = (CHINESE_TEXT, SHORT_TEXT) * 8
+
+
+def test_concurrent_first_deepseek_v4_counts_in_a_fresh_proxy_agree(tmp_path: Path) -> None:
+    group: Final = f"count-{uuid.uuid4().hex}"
+    with ExitStack() as stack:
+        rig_gateway: Final = stack.enter_context(gateway_from_environment())
+        wire: Final = stack.enter_context(wire_server(_completion))
+        config: Final = _owned_config(
+            tmp_path,
+            [
+                {
+                    "model_name": group,
+                    "litellm_params": {"model": _DEEPSEEK_V4, "api_base": wire.url, "api_key": _API_KEY},
+                }
+            ],
+            {},
+            {},
+        )
+        owned: Final = stack.enter_context(owned_proxy_process(rig_gateway, tmp_path, {}, config=config))
+        with ThreadPoolExecutor(max_workers=len(_BURST)) as pool:
+            counts: Final = tuple(
+                pool.map(
+                    lambda text: _count(owned.gateway, _MESSAGES_COUNT, _MESSAGES_COUNT.body(group, (), text)), _BURST
+                )
+            )
+        upstream: Final = wire.drain()
+    assert upstream == ()
+    chinese: Final = frozenset(counts[0::2])
+    short: Final = frozenset(counts[1::2])
+    assert len(chinese) == 1 and len(short) == 1, counts
+    assert_counted_with_deepseek(min(chinese) - min(short), chinese_minus_short)

tests/integration/sdk/test_huggingface_tokenizer_downloads_sdk.py

@@ -0,0 +1,54 @@
+import os
+import subprocess
+import sys
+import threading
+from pathlib import Path
+from queue import SimpleQueue
+from typing import Final
+
+import litellm
+from integration._support.wire import Reply, Request, wire_server
+
+_SCRIPT_SECONDS: Final = 60
+_CONCURRENT_LLAMA_COUNTS: Final = """
+import sys
+from concurrent.futures import ThreadPoolExecutor
+sys.path.insert(0, sys.argv[1])
+from litellm.utils import _select_tokenizer_helper
+with ThreadPoolExecutor(max_workers=2) as pool:
+    tuple(pool.map(lambda _: _select_tokenizer_helper("meta-llama/llama-3-8b"), range(2)))
+"""
+
+
+def test_concurrent_first_llama_counts_download_without_waiting_on_each_other(tmp_path: Path) -> None:
+    both_downloading: Final = threading.Barrier(2, timeout=5)
+    downloads: Final = SimpleQueue[str]()
+
+    def stalled_hub(request: Request) -> Reply:
+        if request.target.endswith("/tokenizer.json"):
+            try:
+                both_downloading.wait()
+                downloads.put("concurrent")
+            except threading.BrokenBarrierError:
+                downloads.put("serialized")
+        return Reply(status=404)
+
+    with wire_server(stalled_hub) as hub:
+        result: Final = subprocess.run(
+            [sys.executable, "-I", "-c", _CONCURRENT_LLAMA_COUNTS, str(Path(litellm.__file__).parent.parent)],
+            capture_output=True,
+            text=True,
+            timeout=_SCRIPT_SECONDS,
+            check=False,
+            env={
+                **os.environ,
+                "HF_ENDPOINT": hub.url,
+                "HF_HOME": str(tmp_path / "home"),
+                "HF_HUB_CACHE": str(tmp_path / "cache"),
+                "HF_HUB_OFFLINE": "0",
+                "LITELLM_LOCAL_MODEL_COST_MAP": "True",
+            },
+        )
+
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert downloads.get_nowait() == "concurrent"

tests/unit/litellm_core_utils/test_token_counter.py

@@ -10,12 +10,12 @@ import sys
 import threading
 import time
 from collections.abc import Callable, Mapping
-from concurrent.futures import Future, wait
+from concurrent.futures import Future, ThreadPoolExecutor, wait
 from copy import deepcopy
 from functools import partial
 from itertools import accumulate, chain
 from pathlib import Path
-from typing import Final
+from typing import Final, NoReturn
 from unittest.mock import MagicMock
 
 import anyio.to_thread
@@ -42,6 +42,7 @@ from litellm.litellm_core_utils.token_counter import (
 )
 from litellm.litellm_core_utils.token_counter import token_counter as token_counter_new
 from litellm.llms.anthropic.experimental_pass_through.adapters.transformation import LiteLLMAnthropicMessagesAdapter
+from litellm.types.utils import CustomHuggingfaceTokenizer
 from tests.large_text import text
 from tests.unit.litellm_core_utils.event_loop_lag import (
     assert_loop_stayed_free,
@@ -519,10 +520,14 @@ def test_token_counter_tools_match_billed_prompt_for_json_tool_chat_templates(
     [
         ("DeepSeek-V4-Pro", "DeepSeek-V4-Flash-0731"),
         ("accounts/fireworks/models/deepseek-v4-flash", "DeepSeek-V4-Flash-0731"),
+        ("deepseek/deepseek-chat", "DeepSeek-V4-Flash-0731"),
+        ("deepseek/deepseek-chat:free", "DeepSeek-V4-Flash-0731"),
+        ("deepseek-reasoner", "DeepSeek-V4-Flash-0731"),
+        ("deepseek/deepseek-flash", "DeepSeek-V4-Flash-0731"),
         ("DeepSeek-V4.1-Flash", "deepseek-v3.2"),
         ("accounts/fireworks/models/deepseek-v4p1-flash", "deepseek-v3.2"),
         ("Qwen/Qwen3.5-122B-A10B", "databricks-qwen35-122b-a10b"),
-        ("qwen3-235b-a22b", "deepseek-v3.2"),
+        ("qwen3-235b-a22b", "mistral-large-latest"),
     ],
 )
 def test_token_counter_counts_tools_by_the_models_chat_template_family(model: str, counted_like: str):
@@ -878,7 +883,14 @@ def test_token_counter():
 
 import unittest
 
-from litellm.utils import _load_huggingface_tokenizer, _select_tokenizer_helper, claude_json_str, encoding
+from litellm.utils import (
+    _load_huggingface_tokenizer,
+    _select_custom_tokenizer_helper,
+    _select_tokenizer_helper,
+    claude_json_str,
+    encoding,
+    select_count_route_tokenizer,
+)
 
 # Clear the cache at module load to ensure clean state
 _load_huggingface_tokenizer.cache_clear()
@@ -977,6 +989,129 @@ class TestTokenizerSelection(unittest.TestCase):
             monkeypatch.undo()
 
 
+_CODE_SAMPLE: Final = "    for item in items:\n        total += item.price * item.quantity\n" * 20
+
+
+def _deepseek_token_ids(text: str) -> list[int]:
+    deepseek_tokenizer: Final = Tokenizer.from_str(
+        importlib.resources.files("litellm.litellm_core_utils.tokenizers")
+        .joinpath("deepseek_tokenizer.json")
+        .read_text(encoding="utf-8")
+    )
+    return deepseek_tokenizer.encode(text).ids
+
+
+def _deepseek_token_count(text: str) -> int:
+    return len(_deepseek_token_ids(text))
+
+
+def _count_route_count(model: str, text: str) -> int:
+    return litellm.token_counter(model=model, text=text, custom_tokenizer=select_count_route_tokenizer(model))
+
+
+@pytest.mark.parametrize(
+    "model",
+    [
+        "azure_ai/DeepSeek-V4-Flash-0731",
+        "deepseek/deepseek-v4-pro",
+        "fireworks_ai/deepseek-v4p1-flash",
+        "bedrock/us-east-1/deepseek.v3.2",
+        "together_ai/deepseek-ai/DeepSeek-V3.1",
+        "databricks/databricks-deepseek-v4-flash-0731",
+        "replicate/deepseek-ai/deepseek-v3",
+        "deepseek-chat",
+        "deepseek/deepseek-reasoner",
+        "openrouter/deepseek/deepseek-chat-v3-0324",
+        "openrouter/deepseek/deepseek-chat:free",
+    ],
+)
+def test_count_routes_count_deepseek_v3_and_v4_with_the_packaged_deepseek_tokenizer(model: str):
+    expected: Final = _deepseek_token_count(_CODE_SAMPLE)
+
+    assert expected != len(tiktoken.get_encoding("cl100k_base").encode(_CODE_SAMPLE))
+    assert _count_route_count(model, _CODE_SAMPLE) == expected
+
+
+@pytest.mark.parametrize(
+    "model",
+    [
+        "azure_ai/DeepSeek-V4-Flash-0731",
+        "deepseek/deepseek-v4-pro",
+        "fireworks_ai/deepseek-v4p1-flash",
+        "bedrock/us-east-1/deepseek.v3.2",
+        "together_ai/deepseek-ai/DeepSeek-V3.1",
+        "databricks/databricks-deepseek-v4-flash-0731",
+        "deepseek-chat",
+        "deepseek/deepseek-reasoner",
+        "openrouter/deepseek/deepseek-chat-v3-0324",
+        "openrouter/deepseek/deepseek-chat:free",
+    ],
+)
+def test_token_counter_keeps_its_tokenizer_for_deepseek_v3_and_v4(model: str):
+    assert litellm.token_counter(model=model, text=_CODE_SAMPLE) == litellm.token_counter(model="", text=_CODE_SAMPLE)
+
+
+def test_count_routes_use_a_deployments_custom_tokenizer_over_the_packaged_deepseek_one():
+    whitespace_tokenizer: Final = Tokenizer.from_str(_word_level_tokenizer_json(pre_tokenizers.WhitespaceSplit()))
+    configured: Final = CustomHuggingfaceTokenizer(
+        identifier="my-org/deepseek-v4-tokenizer", revision="v2", auth_token="private-repo-read-token"
+    )
+    _select_custom_tokenizer_helper.cache_clear()
+
+    with patch.object(litellm.utils.tokenizer_dispatch, "from_pretrained", return_value=whitespace_tokenizer) as load:
+        selected: Final = select_count_route_tokenizer("deepseek/deepseek-v4-flash", configured)
+
+    load.assert_called_once_with("my-org/deepseek-v4-tokenizer", revision="v2", token="private-repo-read-token")
+    counted: Final = litellm.token_counter(model="deepseek-chat", text=_CODE_SAMPLE, custom_tokenizer=selected)
+    assert counted == len(_CODE_SAMPLE.split())
+
+
+@pytest.mark.parametrize(
+    "model",
+    ["together_ai/deepseek-ai/DeepSeek-R1-Distill-Qwen-14B", "fireworks_ai/accounts/fireworks/models/deepseek-v2p5"],
+)
+def test_count_routes_skip_the_deepseek_tokenizer_for_deepseek_models_built_on_other_vocabularies(model: str):
+    assert _count_route_count(model, _CODE_SAMPLE) != _deepseek_token_count(_CODE_SAMPLE)
+
+
+def test_encode_and_decode_keep_their_tokenizer_for_deepseek_v4():
+    token_ids: Final = encode(model="deepseek/deepseek-v4-flash", text=_CODE_SAMPLE)
+
+    assert token_ids == encode(model="", text=_CODE_SAMPLE)
+    assert decode(model="deepseek/deepseek-v4-flash", tokens=token_ids) == _CODE_SAMPLE
+
+
+def test_disable_hf_tokenizer_download_keeps_deepseek_v4_count_routes_on_cl100k(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr(litellm, "disable_hf_tokenizer_download", True)
+
+    assert _count_route_count("deepseek/deepseek-v4-flash", _CODE_SAMPLE) == litellm.token_counter(
+        model="", text=_CODE_SAMPLE
+    )
+
+
+def test_count_routes_fall_back_to_cl100k_when_the_deepseek_tokenizer_fails_to_load(monkeypatch: pytest.MonkeyPatch):
+    def fail_to_load(kind: str, backend: object) -> NoReturn:
+        raise RuntimeError(f"{kind} tokenizer failed to load on {backend}")
+
+    monkeypatch.setattr(litellm.utils, "_load_huggingface_tokenizer", fail_to_load)
+
+    assert _count_route_count("deepseek/deepseek-v4-flash", _CODE_SAMPLE) == _count_route_count("", _CODE_SAMPLE)
+
+
+def test_concurrent_first_deepseek_counts_share_one_loaded_tokenizer():
+    _load_huggingface_tokenizer.cache_clear()
+    start: Final = threading.Barrier(8)
+
+    def first_tokenizer(_: int) -> object:
+        start.wait()
+        return select_count_route_tokenizer("deepseek/deepseek-v4-flash")["tokenizer"]
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        tokenizers: Final = tuple(pool.map(first_tokenizer, range(8)))
+
+    assert len({id(tokenizer) for tokenizer in tokenizers}) == 1
+
+
 def test_token_counter_with_anthropic_tool_use():
     """
     Test that _count_anthropic_content() correctly handles tool_use blocks.