Spaces:
Running
Running
| """Zen/OpenCode provider implementation.""" | |
| from typing import Any | |
| from core.anthropic import ReasoningReplayMode, build_base_request_body | |
| from config.settings import Settings | |
| from providers.base import ProviderConfig | |
| from providers.defaults import ZEN_DEFAULT_BASE | |
| from providers.openai_compat import OpenAIChatTransport | |
| class ZenProvider(OpenAIChatTransport): | |
| """Zen/OpenCode provider using OpenAI-compatible API.""" | |
| def __init__( | |
| self, | |
| config: ProviderConfig, | |
| *, | |
| settings: Settings, | |
| ): | |
| # Zen uses /v1/chat/completions so append /v1 to base URL | |
| base_url = (config.base_url or ZEN_DEFAULT_BASE).rstrip("/") | |
| if not base_url.endswith("/v1"): | |
| base_url = base_url + "/v1" | |
| super().__init__( | |
| config, | |
| provider_name="Zen", | |
| base_url=base_url, | |
| api_key=config.api_key, | |
| ) | |
| self._settings = settings | |
| def _build_request_body( | |
| self, request: Any, thinking_enabled: bool | None = None | |
| ) -> dict: | |
| """Build request body for Zen API.""" | |
| thinking = self._is_thinking_enabled(request, thinking_enabled) | |
| reasoning_replay = ( | |
| ReasoningReplayMode.REASONING_CONTENT | |
| if thinking | |
| else ReasoningReplayMode.DISABLED | |
| ) | |
| return build_base_request_body( | |
| request, | |
| reasoning_replay=reasoning_replay, | |
| ) | |