"""Silicon Flow provider using OpenAI-compatible API.""" from typing import Any from config.settings import Settings from core.anthropic import ReasoningReplayMode, build_base_request_body from providers.base import ProviderConfig from providers.defaults import SILICON_DEFAULT_BASE from providers.openai_compat import OpenAIChatTransport class SiliconProvider(OpenAIChatTransport): """Silicon Flow provider using OpenAI-compatible /chat/completions.""" def __init__(self, config: ProviderConfig, *, settings: Settings): base_url = (config.base_url or SILICON_DEFAULT_BASE).rstrip("/") if not base_url.endswith("/v1"): base_url = base_url + "/v1" # Silicon Flow has generous rate limits super().__init__( config, provider_name="Silicon", base_url=base_url, api_key=config.api_key, nim_rate_limit=300, nim_max_concurrency=80, ) self._settings = settings def _build_request_body( self, request: Any, thinking_enabled: bool | None = None ) -> dict: thinking = self._is_thinking_enabled(request, thinking_enabled) reasoning_replay = ( ReasoningReplayMode.REASONING_CONTENT if thinking else ReasoningReplayMode.DISABLED ) body = build_base_request_body(request, reasoning_replay=reasoning_replay) # Strip silicon/ prefix so the API gets the bare model ID model = body.get("model", "") if model.startswith("silicon/"): body["model"] = model[len("silicon/") :] return body