dheraingoud's picture
feat: synchronize proxy with upstream commits up to 5305bd7 and integrate NIM key rotator
0a54372
Raw
History Blame Contribute Delete
1.83 kB
"""DeepSeek provider implementation (OpenAI-compatible Chat Completions)."""
from typing import Any
from free_claude_code.core.anthropic.models import MessagesRequest
from free_claude_code.core.reasoning import DEFAULT_REASONING_POLICY, ReasoningPolicy
from free_claude_code.providers.admission import ProviderAdmissionController
from free_claude_code.providers.base import ProviderConfig
from free_claude_code.providers.openai_chat import (
NO_REASONING,
OpenAIChatProfile,
OpenAIChatProvider,
usage_int,
)
from .compat import DEEPSEEK_REQUEST_POLICY, build_deepseek_request_body
_PROFILE = OpenAIChatProfile(
DEEPSEEK_REQUEST_POLICY,
NO_REASONING,
)
class DeepSeekProvider(OpenAIChatProvider):
"""DeepSeek using ``https://api.deepseek.com`` Chat Completions."""
def __init__(
self, config: ProviderConfig, *, admission: ProviderAdmissionController
):
super().__init__(
config,
profile=_PROFILE,
admission=admission,
)
def _build_request_body(
self,
request: MessagesRequest,
*,
reasoning: ReasoningPolicy = DEFAULT_REASONING_POLICY,
) -> dict:
return build_deepseek_request_body(
request,
reasoning=reasoning,
)
def _anthropic_usage_fields(self, usage_info: Any) -> dict[str, int]:
usage_fields: dict[str, int] = {}
cache_hit_tokens = usage_int(usage_info, "prompt_cache_hit_tokens")
if cache_hit_tokens is not None:
usage_fields["cache_read_input_tokens"] = cache_hit_tokens
cache_miss_tokens = usage_int(usage_info, "prompt_cache_miss_tokens")
if cache_miss_tokens is not None:
usage_fields["cache_creation_input_tokens"] = cache_miss_tokens
return usage_fields