diff --git a/litellm/llms/anthropic/common_utils.py b/litellm/llms/anthropic/common_utils.py index c263d903188..3845dc5a46e 100644 --- a/litellm/llms/anthropic/common_utils.py +++ b/litellm/llms/anthropic/common_utils.py @@ -2,7 +2,7 @@ This file contains common utils for anthropic calls. """ -from typing import Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Union import httpx @@ -229,6 +229,60 @@ class AnthropicModelInfo(BaseLLMModelInfo): litellm_model_names.append(litellm_model_name) return litellm_model_names + def get_token_counter(self) -> Optional["AnthropicTokenCounter"]: + """ + Factory method to create an Anthropic token counter. + + Returns: + AnthropicTokenCounter instance for this provider. + """ + return AnthropicTokenCounter() + + +class AnthropicTokenCounter: + """Token counter implementation for Anthropic provider.""" + + def supports_provider( + self, + deployment: Optional[Dict[str, Any]] = None, + from_endpoint: bool = False + ) -> bool: + if not from_endpoint: + return False + + if deployment is None: + return False + + full_model = deployment.get("litellm_params", {}).get("model", "") + is_anthropic_provider = full_model.startswith("anthropic/") or "anthropic" in full_model.lower() + + return is_anthropic_provider + + async def count_tokens( + self, + model_to_use: str, + messages: Optional[List[Dict[str, Any]]], + deployment: Optional[Dict[str, Any]] = None, + request_model: str = "", + ) -> Optional[Dict[str, Any]]: + from litellm.proxy.utils import count_tokens_with_anthropic_api + + result = await count_tokens_with_anthropic_api( + model_to_use=model_to_use, + messages=messages, + deployment=deployment, + ) + + if result is not None: + return { + "total_tokens": result["total_tokens"], + "request_model": request_model, + "model_used": model_to_use, + "tokenizer_type": result["tokenizer_used"], + } + + return None + def process_anthropic_headers(headers: Union[httpx.Headers, dict]) -> dict: openai_headers = {} diff --git a/litellm/llms/base_llm/base_utils.py b/litellm/llms/base_llm/base_utils.py index 35959f0d083..3961ee2b9e9 100644 --- a/litellm/llms/base_llm/base_utils.py +++ b/litellm/llms/base_llm/base_utils.py @@ -70,6 +70,16 @@ class BaseLLMModelInfo(ABC): """ pass + def get_token_counter(self): + """ + Factory method to create a token counter for this provider. + + Returns: + Optional TokenCounterInterface implementation for this provider, + or None if token counting is not supported. + """ + return None + def _convert_tool_response_to_message( tool_calls: List[ChatCompletionToolCallChunk], diff --git a/litellm/proxy/anthropic_endpoints/endpoints.py b/litellm/proxy/anthropic_endpoints/endpoints.py index 8af9e4d1c18..8965ae7715d 100644 --- a/litellm/proxy/anthropic_endpoints/endpoints.py +++ b/litellm/proxy/anthropic_endpoints/endpoints.py @@ -200,3 +200,80 @@ async def anthropic_response( # noqa: PLR0915 param=getattr(e, "param", "None"), code=getattr(e, "status_code", 500), ) + + +@router.post( + "/v1/messages/count_tokens", + tags=["[beta] Anthropic Messages Token Counting"], + dependencies=[Depends(user_api_key_auth)], +) +async def count_tokens( + request: Request, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), # Used for auth +): + """ + Count tokens for Anthropic Messages API format. + + This endpoint follows the Anthropic Messages API token counting specification. + It accepts the same parameters as the /v1/messages endpoint but returns + token counts instead of generating a response. + + Example usage: + ``` + curl -X POST "http://localhost:4000/v1/messages/count_tokens?beta=true" \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer your-key" \ + -d '{ + "model": "claude-3-sonnet-20240229", + "messages": [{"role": "user", "content": "Hello Claude!"}] + }' + ``` + + Returns: {"input_tokens": } + """ + from litellm.proxy.proxy_server import token_counter as internal_token_counter + + try: + request_data = await _read_request_body(request=request) + data: dict = {**request_data} + + # Extract required fields + model_name = data.get("model") + messages = data.get("messages", []) + + if not model_name: + raise HTTPException( + status_code=400, + detail={"error": "model parameter is required"} + ) + + if not messages: + raise HTTPException( + status_code=400, + detail={"error": "messages parameter is required"} + ) + + # Create TokenCountRequest for the internal endpoint + from litellm.proxy._types import TokenCountRequest + + token_request = TokenCountRequest( + model=model_name, + messages=messages + ) + + # Call the internal token counter function with direct request flag set to False + token_response = await internal_token_counter(token_request, is_direct_request=False) + + # Convert the internal response to Anthropic API format + return {"input_tokens": token_response.total_tokens} + + except HTTPException: + raise + except Exception as e: + verbose_proxy_logger.exception( + "litellm.proxy.anthropic_endpoints.count_tokens(): Exception occurred - {}".format(str(e)) + ) + raise HTTPException( + status_code=500, + detail={"error": f"Internal server error: {str(e)}"} + ) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index c77f4f5317c..3676420bab0 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -679,6 +679,7 @@ app = FastAPI( ) + ### CUSTOM API DOCS [ENTERPRISE FEATURE] ### # Custom OpenAPI schema generator to include only selected routes from fastapi.routing import APIWebSocketRoute @@ -5603,13 +5604,57 @@ async def run_thread( # async def get_available_routes(user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth)): +def _get_provider_token_counter(deployment: dict, model_to_use: str): + """ + Auto-route to the correct provider's token counter based on model/deployment. + Uses the existing get_provider_model_info infrastructure with switch-case pattern. + """ + if deployment is None: + return None + + from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider + + full_model = deployment.get("litellm_params", {}).get("model", "") + + try: + # Use existing LiteLLM logic to determine provider + model, provider, dynamic_api_key, api_base = get_llm_provider( + model=full_model, + custom_llm_provider=deployment.get("litellm_params", {}).get("custom_llm_provider"), + api_base=deployment.get("litellm_params", {}).get("api_base"), + api_key=deployment.get("litellm_params", {}).get("api_key") + ) + + # Switch case pattern using existing get_provider_model_info + from litellm.utils import ProviderConfigManager + from litellm.types.utils import LlmProviders + + # Convert string provider to LlmProviders enum + llm_provider_enum = LlmProviders(provider) + # Add more provider mappings as needed + + if llm_provider_enum: + provider_model_info = ProviderConfigManager.get_provider_model_info(model=full_model, provider=llm_provider_enum) + if provider_model_info is not None: + return provider_model_info.get_token_counter() + + except Exception: + # If provider detection fails, fall back to manual checks + if full_model.startswith("anthropic/") or "anthropic" in full_model.lower(): + from litellm.llms.anthropic.common_utils import AnthropicModelInfo + anthropic_model_info = AnthropicModelInfo() + return anthropic_model_info.get_token_counter() + + return None + + @router.post( "/utils/token_counter", tags=["llm utils"], dependencies=[Depends(user_api_key_auth)], response_model=TokenCountResponse, ) -async def token_counter(request: TokenCountRequest): +async def token_counter(request: TokenCountRequest, is_direct_request: bool = True): """ """ from litellm import token_counter @@ -5634,7 +5679,7 @@ async def token_counter(request: TokenCountRequest): break if deployment is not None: litellm_model_name = deployment.get("litellm_params", {}).get("model") - # remove the custom_llm_provider_prefix in the litellm_model_name + # remove the custom_llm_provider_prefix in the litellm_model_name if "/" in litellm_model_name: litellm_model_name = litellm_model_name.split("/", 1)[1] @@ -5642,6 +5687,29 @@ async def token_counter(request: TokenCountRequest): litellm_model_name or request.model ) # use litellm model name, if it's not avalable then fallback to request.model + # Try provider-specific token counting first - only for non-direct requests (from provider endpoints) + provider_counter = None + if deployment is not None and not is_direct_request: + # Auto-route to the correct provider based on model + provider_counter = _get_provider_token_counter(deployment, model_to_use) + + if provider_counter is not None and provider_counter.supports_provider(deployment=deployment, from_endpoint=not is_direct_request): + result = await provider_counter.count_tokens( + model_to_use=model_to_use, + messages=messages, # type: ignore + deployment=deployment, + request_model=request.model, + ) + + if result is not None: + return TokenCountResponse( + total_tokens=result["total_tokens"], + request_model=result["request_model"], + model_used=result["model_used"], + tokenizer_type=result["tokenizer_type"], + ) + + # Default LiteLLM token counting custom_tokenizer: Optional[CustomHuggingfaceTokenizer] = None if model_info is not None: custom_tokenizer = cast( diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 16e151e5e19..14156608278 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -3802,7 +3802,6 @@ def is_valid_api_key(key: str) -> bool: def construct_database_url_from_env_vars() -> Optional[str]: """ Construct a DATABASE_URL from individual environment variables. - Returns: Optional[str]: The constructed DATABASE_URL or None if required variables are missing """ @@ -3829,9 +3828,65 @@ def construct_database_url_from_env_vars() -> Optional[str]: database_url = f"postgresql://{database_username_enc}@{database_host}/{database_name_enc}" return database_url - + return None +async def count_tokens_with_anthropic_api( + model_to_use: str, + messages: Optional[List[Dict[str, Any]]], + deployment: Optional[Dict[str, Any]] = None, +) -> Optional[Dict[str, Any]]: + """ + Helper function to count tokens using Anthropic API directly. + + Args: + model_to_use: The model name to use for token counting + messages: The messages to count tokens for + deployment: Optional deployment configuration containing API key + + Returns: + Optional dict with token count and tokenizer info, or None if failed + """ + if not messages: + return None + + try: + import anthropic + import os + + # Get Anthropic API key from deployment config + anthropic_api_key = None + if deployment is not None: + anthropic_api_key = deployment.get("litellm_params", {}).get("api_key") + + # Fallback to environment variable + if not anthropic_api_key: + anthropic_api_key = os.getenv("ANTHROPIC_API_KEY") + + if anthropic_api_key and messages: + # Call Anthropic API directly for more accurate token counting + client = anthropic.Anthropic(api_key=anthropic_api_key) + + # Call with explicit parameters to satisfy type checking + # Type ignore for now since messages come from generic dict input + response = client.beta.messages.count_tokens( + model=model_to_use, + messages=messages, # type: ignore + betas=["token-counting-2024-11-01"] + ) + total_tokens = response.input_tokens + tokenizer_used = "anthropic_api" + + return { + "total_tokens": total_tokens, + "tokenizer_used": tokenizer_used, + } + + except ImportError: + verbose_proxy_logger.warning("Anthropic library not available, falling back to LiteLLM tokenizer") + except Exception as e: + verbose_proxy_logger.warning(f"Error calling Anthropic API: {e}, falling back to LiteLLM tokenizer") + return None async def get_available_models_for_user( user_api_key_dict: "UserAPIKeyAuth", @@ -4003,4 +4058,4 @@ def validate_model_access( raise HTTPException( status_code=404, detail="The model `{}` does not exist or is not accessible".format(model_id) - ) \ No newline at end of file + ) diff --git a/tests/proxy_unit_tests/test_proxy_token_counter.py b/tests/proxy_unit_tests/test_proxy_token_counter.py index 11dededd6ca..eefe5059b0b 100644 --- a/tests/proxy_unit_tests/test_proxy_token_counter.py +++ b/tests/proxy_unit_tests/test_proxy_token_counter.py @@ -136,3 +136,427 @@ async def test_gpt_token_counting(): response.tokenizer_type == "openai_tokenizer" ) # SHOULD use the OpenAI tokenizer assert response.request_model == "gpt-4" + + +@pytest.mark.asyncio +async def test_anthropic_messages_count_tokens_endpoint(): + """ + Test /v1/messages/count_tokens endpoint with Anthropic model + - Should return response in Anthropic format: {"input_tokens": } + - Should work as wrapper around internal token_counter function + """ + from litellm.proxy.anthropic_endpoints.endpoints import count_tokens + from fastapi import Request + from unittest.mock import AsyncMock, MagicMock + + # Mock request object + mock_request = MagicMock(spec=Request) + mock_request_data = { + "model": "claude-3-sonnet-20240229", + "messages": [{"role": "user", "content": "Hello Claude!"}] + } + + # Mock the _read_request_body function + async def mock_read_request_body(request): + return mock_request_data + + # Mock UserAPIKeyAuth + mock_user_api_key_dict = MagicMock() + + # Patch the _read_request_body function + import litellm.proxy.anthropic_endpoints.endpoints as anthropic_endpoints + original_read_request_body = anthropic_endpoints._read_request_body + anthropic_endpoints._read_request_body = mock_read_request_body + + # Mock the internal token_counter function to return a controlled response + async def mock_token_counter(request, is_direct_request=True): + assert is_direct_request == False, "Should be called with is_direct_request=False for Anthropic endpoint" + assert request.model == "claude-3-sonnet-20240229" + assert request.messages == [{"role": "user", "content": "Hello Claude!"}] + + from litellm.proxy._types import TokenCountResponse + return TokenCountResponse( + total_tokens=15, + request_model="claude-3-sonnet-20240229", + model_used="claude-3-sonnet-20240229", + tokenizer_type="openai_tokenizer" + ) + + # Patch the imported token_counter function from proxy_server + import litellm.proxy.proxy_server as proxy_server + original_token_counter = proxy_server.token_counter + proxy_server.token_counter = mock_token_counter + + try: + # Call the endpoint + response = await count_tokens(mock_request, mock_user_api_key_dict) + + # Verify response format matches Anthropic spec + assert isinstance(response, dict) + assert "input_tokens" in response + assert response["input_tokens"] == 15 + assert len(response) == 1 # Should only contain input_tokens + + print("✅ Anthropic endpoint test passed!") + + finally: + # Restore original functions + anthropic_endpoints._read_request_body = original_read_request_body + proxy_server.token_counter = original_token_counter + + +@pytest.mark.asyncio +async def test_anthropic_messages_count_tokens_with_non_anthropic_model(): + """ + Test /v1/messages/count_tokens endpoint with non-Anthropic model (GPT-4) + - Should still work and return Anthropic format + - Should call internal token_counter with from_anthropic_endpoint=True + """ + from litellm.proxy.anthropic_endpoints.endpoints import count_tokens + from fastapi import Request + from unittest.mock import AsyncMock, MagicMock + + # Mock request object + mock_request = MagicMock(spec=Request) + mock_request_data = { + "model": "gpt-4", + "messages": [{"role": "user", "content": "Hello GPT!"}] + } + + # Mock the _read_request_body function + async def mock_read_request_body(request): + return mock_request_data + + # Mock UserAPIKeyAuth + mock_user_api_key_dict = MagicMock() + + # Patch the _read_request_body function + import litellm.proxy.anthropic_endpoints.endpoints as anthropic_endpoints + original_read_request_body = anthropic_endpoints._read_request_body + anthropic_endpoints._read_request_body = mock_read_request_body + + # Mock the internal token_counter function to return a controlled response + async def mock_token_counter(request, is_direct_request=True): + assert is_direct_request == False, "Should be called with is_direct_request=False for Anthropic endpoint" + assert request.model == "gpt-4" + assert request.messages == [{"role": "user", "content": "Hello GPT!"}] + + from litellm.proxy._types import TokenCountResponse + return TokenCountResponse( + total_tokens=12, + request_model="gpt-4", + model_used="gpt-4", + tokenizer_type="openai_tokenizer" + ) + + # Patch the imported token_counter function from proxy_server + import litellm.proxy.proxy_server as proxy_server + original_token_counter = proxy_server.token_counter + proxy_server.token_counter = mock_token_counter + + try: + # Call the endpoint + response = await count_tokens(mock_request, mock_user_api_key_dict) + + # Verify response format matches Anthropic spec + assert isinstance(response, dict) + assert "input_tokens" in response + assert response["input_tokens"] == 12 + assert len(response) == 1 # Should only contain input_tokens + + print("✅ Non-Anthropic model test passed!") + + finally: + # Restore original functions + anthropic_endpoints._read_request_body = original_read_request_body + proxy_server.token_counter = original_token_counter + + +@pytest.mark.asyncio +async def test_internal_token_counter_anthropic_provider_detection(): + """ + Test that the internal token_counter correctly detects Anthropic providers + and handles the from_anthropic_endpoint flag appropriately + """ + + # Test with Anthropic provider + llm_router = Router( + model_list=[ + { + "model_name": "claude-test", + "litellm_params": { + "model": "anthropic/claude-3-sonnet-20240229", + "api_key": "test-key" + }, + } + ] + ) + + setattr(litellm.proxy.proxy_server, "llm_router", llm_router) + + # Test with is_direct_request=False (simulating call from Anthropic endpoint) + response = await token_counter( + request=TokenCountRequest( + model="claude-test", + messages=[{"role": "user", "content": "hello"}], + ), + is_direct_request=False + ) + + print("Anthropic provider test response:", response) + + # Verify response structure + assert response.request_model == "claude-test" + assert response.model_used == "claude-3-sonnet-20240229" + assert response.total_tokens > 0 + + # Test with non-Anthropic provider + llm_router = Router( + model_list=[ + { + "model_name": "gpt-test", + "litellm_params": { + "model": "gpt-4", + }, + } + ] + ) + + setattr(litellm.proxy.proxy_server, "llm_router", llm_router) + + # Test with is_direct_request=False but non-Anthropic provider + response = await token_counter( + request=TokenCountRequest( + model="gpt-test", + messages=[{"role": "user", "content": "hello"}], + ), + is_direct_request=False + ) + + print("Non-Anthropic provider test response:", response) + + # Verify response structure + assert response.request_model == "gpt-test" + assert response.model_used == "gpt-4" + assert response.total_tokens > 0 + assert response.tokenizer_type == "openai_tokenizer" # Should use LiteLLM tokenizer + + +@pytest.mark.asyncio +async def test_anthropic_endpoint_error_handling(): + """ + Test error handling in the /v1/messages/count_tokens endpoint + """ + from litellm.proxy.anthropic_endpoints.endpoints import count_tokens + from fastapi import Request, HTTPException + from unittest.mock import MagicMock + + # Mock request object + mock_request = MagicMock(spec=Request) + mock_user_api_key_dict = MagicMock() + + # Test missing model parameter + mock_request_data = { + "messages": [{"role": "user", "content": "Hello!"}] + # Missing "model" key + } + + async def mock_read_request_body(request): + return mock_request_data + + import litellm.proxy.anthropic_endpoints.endpoints as anthropic_endpoints + original_read_request_body = anthropic_endpoints._read_request_body + anthropic_endpoints._read_request_body = mock_read_request_body + + try: + # Should raise HTTPException for missing model + with pytest.raises(HTTPException) as exc_info: + await count_tokens(mock_request, mock_user_api_key_dict) + + assert exc_info.value.status_code == 400 + assert "model parameter is required" in str(exc_info.value.detail) + + print("✅ Error handling test passed!") + + finally: + anthropic_endpoints._read_request_body = original_read_request_body + + +@pytest.mark.asyncio +async def test_factory_anthropic_endpoint_calls_anthropic_counter(): + """Test that /v1/messages/count_tokens with Anthropic model uses Anthropic counter.""" + from unittest.mock import patch + from fastapi.testclient import TestClient + from litellm.proxy.proxy_server import app + + # Mock the anthropic token counting function + with patch('litellm.proxy.utils.count_tokens_with_anthropic_api') as mock_anthropic_count: + mock_anthropic_count.return_value = { + "total_tokens": 42, + "tokenizer_used": "anthropic" + } + + # Mock router to return Anthropic deployment + with patch('litellm.proxy.proxy_server.llm_router') as mock_router: + mock_router.model_list = [{ + "model_name": "claude-3-5-sonnet", + "litellm_params": {"model": "anthropic/claude-3-5-sonnet-20241022"}, + "model_info": {} + }] + + client = TestClient(app) + + response = client.post( + "/v1/messages/count_tokens", + json={ + "model": "claude-3-5-sonnet", + "messages": [{"role": "user", "content": "Hello"}] + }, + headers={"Authorization": "Bearer test-key"} + ) + + assert response.status_code == 200 + data = response.json() + assert data["input_tokens"] == 42 + + # Verify that Anthropic API was called + mock_anthropic_count.assert_called_once() + + +@pytest.mark.asyncio +async def test_factory_gpt4_endpoint_does_not_call_anthropic_counter(): + """Test that /v1/messages/count_tokens with GPT-4 does NOT use Anthropic counter.""" + from unittest.mock import patch + from fastapi.testclient import TestClient + from litellm.proxy.proxy_server import app + + # Mock the anthropic token counting function + with patch('litellm.proxy.utils.count_tokens_with_anthropic_api') as mock_anthropic_count: + # Mock litellm token counter + with patch('litellm.token_counter') as mock_litellm_counter: + mock_litellm_counter.return_value = 50 + + # Mock router to return GPT-4 deployment + with patch('litellm.proxy.proxy_server.llm_router') as mock_router: + mock_router.model_list = [{ + "model_name": "gpt-4", + "litellm_params": {"model": "openai/gpt-4"}, + "model_info": {} + }] + + client = TestClient(app) + + response = client.post( + "/v1/messages/count_tokens", + json={ + "model": "gpt-4", + "messages": [{"role": "user", "content": "Hello"}] + }, + headers={"Authorization": "Bearer test-key"} + ) + + assert response.status_code == 200 + data = response.json() + assert data["input_tokens"] == 50 + + # Verify that Anthropic API was NOT called + mock_anthropic_count.assert_not_called() + + +@pytest.mark.asyncio +async def test_factory_normal_token_counter_endpoint_does_not_call_anthropic(): + """Test that /utils/token_counter does NOT use Anthropic counter even with Anthropic model.""" + from unittest.mock import patch + from fastapi.testclient import TestClient + from litellm.proxy.proxy_server import app + + # Mock the anthropic token counting function + with patch('litellm.proxy.utils.count_tokens_with_anthropic_api') as mock_anthropic_count: + # Mock litellm token counter + with patch('litellm.token_counter') as mock_litellm_counter: + mock_litellm_counter.return_value = 35 + + # Mock router to return Anthropic deployment + with patch('litellm.proxy.proxy_server.llm_router') as mock_router: + mock_router.model_list = [{ + "model_name": "claude-3-5-sonnet", + "litellm_params": {"model": "anthropic/claude-3-5-sonnet-20241022"}, + "model_info": {} + }] + + client = TestClient(app) + + response = client.post( + "/utils/token_counter", + json={ + "model": "claude-3-5-sonnet", + "messages": [{"role": "user", "content": "Hello"}] + }, + headers={"Authorization": "Bearer test-key"} + ) + + assert response.status_code == 200 + data = response.json() + assert data["total_tokens"] == 35 + + # Verify that Anthropic API was NOT called (since is_direct_request=True) + mock_anthropic_count.assert_not_called() + + +@pytest.mark.asyncio +async def test_factory_registration(): + """Test that the new factory pattern correctly provides counters.""" + from litellm.llms.anthropic.common_utils import AnthropicModelInfo + + # Test Anthropic ModelInfo provides token counter + anthropic_model_info = AnthropicModelInfo() + counter = anthropic_model_info.get_token_counter() + assert counter is not None + + # Create test deployments + anthropic_deployment = { + "litellm_params": {"model": "anthropic/claude-3-5-sonnet-20241022"} + } + + non_anthropic_deployment = { + "litellm_params": {"model": "openai/gpt-4"} + } + + # Test Anthropic counter supports provider + assert counter.supports_provider(anthropic_deployment, from_endpoint=True) + assert not counter.supports_provider(anthropic_deployment, from_endpoint=False) + + # Test non-Anthropic provider + assert not counter.supports_provider(non_anthropic_deployment, from_endpoint=True) + assert not counter.supports_provider(non_anthropic_deployment, from_endpoint=False) + + # Test None deployment + assert not counter.supports_provider(None, from_endpoint=True) + assert not counter.supports_provider(None, from_endpoint=False) + + +@pytest.mark.asyncio +async def test_factory_anthropic_counter_supports_provider(): + """Test AnthropicTokenCounter provider detection logic.""" + from litellm.llms.anthropic.common_utils import AnthropicModelInfo + + anthropic_model_info = AnthropicModelInfo() + counter = anthropic_model_info.get_token_counter() + + # Test Anthropic provider detection + anthropic_deployment = { + "litellm_params": {"model": "anthropic/claude-3-5-sonnet-20241022"} + } + assert counter.supports_provider(anthropic_deployment, from_endpoint=True) + assert not counter.supports_provider(anthropic_deployment, from_endpoint=False) + + # Test non-Anthropic provider + openai_deployment = { + "litellm_params": {"model": "openai/gpt-4"} + } + assert not counter.supports_provider(openai_deployment, from_endpoint=True) + assert not counter.supports_provider(openai_deployment, from_endpoint=False) + + # Test None deployment + assert not counter.supports_provider(None, from_endpoint=True) + assert not counter.supports_provider(None, from_endpoint=False)