Spaces:
Running
Running
| """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 | |