Spaces:
Running
Running
| """ | |
| Additional provider adapters for OpenAI-compatible APIs. | |
| Includes SiliconFlow, GLM, Kimi, Nvidia, MiniMax, and ModelScope adapters. | |
| """ | |
| from typing import Any | |
| from agno.models.openai.like import OpenAILike | |
| from agno.run.agent import RunContentEvent | |
| from .base import ProviderConfig | |
| from .openai import OpenAIAdapter | |
| class SiliconFlowAdapter(OpenAIAdapter): | |
| """Adapter for SiliconFlow (DeepSeek models).""" | |
| def __init__(self): | |
| self.config = ProviderConfig( | |
| name="siliconflow", | |
| base_url="https://api.siliconflow.cn/v1", | |
| default_model="Qwen/Qwen2.5-7B-Instruct", | |
| supports_streaming=True, | |
| supports_tools=True, | |
| supports_streaming_tool_calls=False, | |
| supports_json_schema=True, | |
| supports_thinking=True, # DeepSeek has reasoning_content | |
| supports_vision=False, | |
| ) | |
| def build_model( | |
| self, | |
| api_key: str, | |
| model: str | None = None, | |
| base_url: str | None = None, | |
| thinking: dict[str, Any] | bool | None = None, | |
| tools: list[dict[str, Any]] | None = None, | |
| tool_choice: Any = None, | |
| **kwargs | |
| ) -> OpenAILike: | |
| """Build SiliconFlow model with thinking support.""" | |
| resolved_base = base_url or self.config.base_url | |
| resolved_model = model or self.config.default_model | |
| extra_body: dict[str, Any] = {} | |
| # Thinking mode support (DeepSeek models) | |
| if thinking: | |
| # Extract budget_tokens from thinking dict | |
| if isinstance(thinking, dict): | |
| budget = thinking.get("budget_tokens") or thinking.get("budgetTokens") or 1024 | |
| else: | |
| budget = 1024 | |
| model_id_lower = resolved_model.lower() | |
| is_kimi_thinking_model = "kimi" in model_id_lower and "thinking" in model_id_lower | |
| extra_body["thinking_budget"] = budget | |
| if not is_kimi_thinking_model: | |
| extra_body["enable_thinking"] = True | |
| # Add tools support | |
| if tools: | |
| extra_body["tools"] = tools | |
| if tool_choice: | |
| extra_body["tool_choice"] = tool_choice | |
| return OpenAILike( | |
| id=resolved_model, | |
| api_key=api_key, | |
| base_url=resolved_base, | |
| extra_body=extra_body if extra_body else None, | |
| ) | |
| def _extract_thinking_from_event(self, event: RunContentEvent) -> str | None: | |
| """ | |
| Extract thinking content for SiliconFlow (DeepSeek models). | |
| DeepSeek uses reasoning_content field when thinking is enabled. | |
| """ | |
| # First try parent method | |
| thinking = super()._extract_thinking_from_event(event) | |
| if thinking: | |
| return thinking | |
| # SiliconFlow-specific: Check for reasoning_content in model_provider_data | |
| if hasattr(event, "model_provider_data") and event.model_provider_data: | |
| data = event.model_provider_data | |
| if isinstance(data, dict): | |
| choices = data.get("choices", []) | |
| if choices and len(choices) > 0: | |
| delta = choices[0].get("delta", {}) | |
| reasoning = delta.get("reasoning_content") | |
| if reasoning: | |
| return str(reasoning) | |
| return None | |
| class GLMAdapter(OpenAIAdapter): | |
| """Adapter for GLM (Zhipu AI).""" | |
| def __init__(self): | |
| self.config = ProviderConfig( | |
| name="glm", | |
| base_url="https://open.bigmodel.cn/api/paas/v4", | |
| default_model="glm-4-flash", | |
| supports_streaming=True, | |
| supports_tools=True, | |
| supports_streaming_tool_calls=True, # glm-4.6+ supports tool streaming | |
| supports_json_schema=True, | |
| supports_thinking=True, | |
| supports_vision=False, | |
| ) | |
| def build_model( | |
| self, | |
| api_key: str, | |
| model: str | None = None, | |
| base_url: str | None = None, | |
| thinking: dict[str, Any] | bool | None = None, | |
| tools: list[dict[str, Any]] | None = None, | |
| tool_choice: Any = None, | |
| **kwargs | |
| ) -> OpenAILike: | |
| """Build GLM model with thinking support.""" | |
| resolved_base = base_url or self.config.base_url | |
| resolved_model = model or self.config.default_model | |
| extra_body: dict[str, Any] = {} | |
| # Thinking mode configuration for GLM | |
| # Only set if explicitly provided - don't set to 'disabled' by default, | |
| # as it prevents reasoning_content in tool_stream | |
| if thinking: | |
| if isinstance(thinking, bool): | |
| # Boolean true -> enable thinking with default config | |
| extra_body["thinking"] = {"type": "enabled"} | |
| elif isinstance(thinking, dict): | |
| # Dict format - extract type field if present | |
| if "type" in thinking: | |
| extra_body["thinking"] = {"type": thinking["type"]} | |
| else: | |
| # No type specified, default to enabled | |
| extra_body["thinking"] = {"type": "enabled"} | |
| # Add tools support | |
| if tools: | |
| extra_body["tools"] = tools | |
| if tool_choice: | |
| extra_body["tool_choice"] = tool_choice | |
| return OpenAILike( | |
| id=resolved_model, | |
| api_key=api_key, | |
| base_url=resolved_base, | |
| extra_body=extra_body if extra_body else None, | |
| ) | |
| def _extract_thinking_from_event(self, event: RunContentEvent) -> str | None: | |
| """ | |
| Extract thinking content for GLM (Zhipu AI). | |
| GLM uses reasoning_content field when thinking type is "enabled". | |
| """ | |
| # First try parent method | |
| thinking = super()._extract_thinking_from_event(event) | |
| if thinking: | |
| return thinking | |
| # GLM-specific: Check for reasoning_content in model_provider_data | |
| if hasattr(event, "model_provider_data") and event.model_provider_data: | |
| data = event.model_provider_data | |
| if isinstance(data, dict): | |
| choices = data.get("choices", []) | |
| if choices and len(choices) > 0: | |
| delta = choices[0].get("delta", {}) | |
| reasoning = delta.get("reasoning_content") | |
| if reasoning: | |
| return str(reasoning) | |
| return None | |
| class KimiAdapter(OpenAIAdapter): | |
| """Adapter for Kimi (Moonshot AI).""" | |
| def __init__(self): | |
| self.config = ProviderConfig( | |
| name="kimi", | |
| base_url="https://api.moonshot.cn/v1", | |
| default_model="moonshot-v1-8k", | |
| supports_streaming=True, | |
| supports_tools=True, | |
| supports_streaming_tool_calls=False, | |
| supports_json_schema=True, | |
| supports_thinking=False, | |
| supports_vision=False, | |
| ) | |
| class NvidiaAdapter(OpenAIAdapter): | |
| """Adapter for Nvidia NIM.""" | |
| def __init__(self): | |
| self.config = ProviderConfig( | |
| name="nvidia", | |
| base_url="https://integrate.api.nvidia.com/v1", | |
| default_model="deepseek-ai/deepseek-r1", | |
| supports_streaming=True, | |
| supports_tools=True, | |
| supports_streaming_tool_calls=True, | |
| supports_json_schema=True, | |
| supports_thinking=True, | |
| supports_vision=True, | |
| ) | |
| def build_model( | |
| self, | |
| api_key: str, | |
| model: str | None = None, | |
| base_url: str | None = None, | |
| thinking: dict[str, Any] | bool | None = None, | |
| tools: list[dict[str, Any]] | None = None, | |
| tool_choice: Any = None, | |
| **kwargs | |
| ) -> OpenAILike: | |
| """Build Nvidia NIM model with thinking support.""" | |
| resolved_base = base_url or self.config.base_url | |
| resolved_model = model or self.config.default_model | |
| extra_body: dict[str, Any] = {} | |
| # Thinking mode support - use chat_template_kwargs for NVIDIA | |
| if thinking: | |
| extra_body["chat_template_kwargs"] = {"thinking": True} | |
| # Add tools support | |
| if tools: | |
| extra_body["tools"] = tools | |
| if tool_choice: | |
| extra_body["tool_choice"] = tool_choice | |
| return OpenAILike( | |
| id=resolved_model, | |
| api_key=api_key, | |
| base_url=resolved_base, | |
| extra_body=extra_body if extra_body else None, | |
| ) | |
| def _extract_thinking_from_event(self, event: RunContentEvent) -> str | None: | |
| """ | |
| Extract thinking content for Nvidia NIM. | |
| Nvidia DeepSeek-R1: reasoning_content in delta or direct access. | |
| Similar to Node.js: messageChunk?.choices?.[0]?.delta?.reasoning_content | |
| """ | |
| # First try parent method | |
| thinking = super()._extract_thinking_from_event(event) | |
| if thinking: | |
| return thinking | |
| # Nvidia-specific: Check direct model_provider_data.choices[0].delta.reasoning_content | |
| if hasattr(event, "model_provider_data") and event.model_provider_data: | |
| data = event.model_provider_data | |
| if isinstance(data, dict): | |
| choices = data.get("choices", []) | |
| if choices and len(choices) > 0: | |
| delta = choices[0].get("delta", {}) | |
| reasoning = delta.get("reasoning_content") | |
| if reasoning: | |
| return str(reasoning) | |
| return None | |
| class MinimaxAdapter(OpenAIAdapter): | |
| """Adapter for MiniMax.""" | |
| def __init__(self): | |
| self.config = ProviderConfig( | |
| name="minimax", | |
| base_url="https://api.minimax.io/v1", | |
| default_model="minimax-m2", | |
| supports_streaming=True, | |
| supports_tools=True, | |
| supports_streaming_tool_calls=True, | |
| supports_json_schema=True, | |
| supports_thinking=True, # Interleaved Thinking via reasoning_split | |
| supports_vision=False, | |
| ) | |
| def build_model( | |
| self, | |
| api_key: str, | |
| model: str | None = None, | |
| base_url: str | None = None, | |
| thinking: dict[str, Any] | bool | None = None, | |
| tools: list[dict[str, Any]] | None = None, | |
| tool_choice: Any = None, | |
| **kwargs | |
| ) -> OpenAILike: | |
| """Build MiniMax model with thinking support.""" | |
| resolved_base = base_url or self.config.base_url | |
| resolved_model = model or self.config.default_model | |
| extra_body: dict[str, Any] = {} | |
| # MiniMax Thinking mode configuration | |
| # Use reasoning_split=true to separate thinking content into reasoning_details field | |
| if thinking: | |
| if isinstance(thinking, bool): | |
| # Boolean true -> enable reasoning_split | |
| extra_body["reasoning_split"] = True | |
| elif isinstance(thinking, dict): | |
| # Check if thinking type is not 'disabled' | |
| thinking_type = thinking.get("type", "enabled") | |
| if thinking_type != "disabled": | |
| extra_body["reasoning_split"] = True | |
| # Add tools support | |
| if tools: | |
| extra_body["tools"] = tools | |
| if tool_choice: | |
| extra_body["tool_choice"] = tool_choice | |
| return OpenAILike( | |
| id=resolved_model, | |
| api_key=api_key, | |
| base_url=resolved_base, | |
| extra_body=extra_body if extra_body else None, | |
| ) | |
| def _extract_thinking_from_event(self, event: RunContentEvent) -> str | None: | |
| """ | |
| Extract thinking content for MiniMax. | |
| MiniMax uses reasoning_details field when reasoning_split is enabled. | |
| """ | |
| # First try parent method | |
| thinking = super()._extract_thinking_from_event(event) | |
| if thinking: | |
| return thinking | |
| # MiniMax-specific: Check for reasoning_details field | |
| if hasattr(event, "model_provider_data") and event.model_provider_data: | |
| data = event.model_provider_data | |
| if isinstance(data, dict): | |
| choices = data.get("choices", []) | |
| if choices and len(choices) > 0: | |
| delta = choices[0].get("delta", {}) | |
| # MiniMax uses reasoning_details when reasoning_split=true | |
| reasoning = delta.get("reasoning_details") or delta.get("reasoning_content") | |
| if reasoning: | |
| return str(reasoning) | |
| return None | |
| class ModelScopeAdapter(OpenAIAdapter): | |
| """Adapter for ModelScope (Chinese models).""" | |
| def __init__(self): | |
| self.config = ProviderConfig( | |
| name="modelscope", | |
| base_url="https://api-inference.modelscope.cn/v1", | |
| default_model="AI-ModelScope/glm-4-9b-chat", | |
| supports_streaming=True, | |
| supports_tools=True, | |
| supports_streaming_tool_calls=False, # API limitation: tools + stream not supported together | |
| supports_json_schema=True, | |
| supports_thinking=True, | |
| supports_vision=False, | |
| ) | |
| def build_model( | |
| self, | |
| api_key: str, | |
| model: str | None = None, | |
| base_url: str | None = None, | |
| thinking: dict[str, Any] | bool | None = None, | |
| stream: bool = True, | |
| tools: list[dict[str, Any]] | None = None, | |
| tool_choice: Any = None, | |
| **kwargs | |
| ) -> OpenAILike: | |
| """Build ModelScope model with thinking support.""" | |
| resolved_base = base_url or self.config.base_url | |
| resolved_model = model or self.config.default_model | |
| extra_body: dict[str, Any] = {} | |
| # Thinking mode configuration for ModelScope | |
| if thinking and stream: | |
| # Extract budget_tokens from thinking dict | |
| if isinstance(thinking, dict): | |
| budget = thinking.get("budget_tokens") or thinking.get("budgetTokens") or 1024 | |
| else: | |
| budget = 1024 | |
| extra_body["enable_thinking"] = True | |
| extra_body["thinking_budget"] = budget | |
| elif not stream: | |
| # Disable thinking when not streaming | |
| extra_body["enable_thinking"] = False | |
| # Add tools support | |
| if tools: | |
| extra_body["tools"] = tools | |
| if tool_choice: | |
| extra_body["tool_choice"] = tool_choice | |
| return OpenAILike( | |
| id=resolved_model, | |
| api_key=api_key, | |
| base_url=resolved_base, | |
| extra_body=extra_body if extra_body else None, | |
| ) | |
| def _extract_thinking_from_event(self, event: RunContentEvent) -> str | None: | |
| """ | |
| Extract thinking content for ModelScope. | |
| ModelScope uses reasoning_content field similar to GLM. | |
| """ | |
| # First try parent method | |
| thinking = super()._extract_thinking_from_event(event) | |
| if thinking: | |
| return thinking | |
| # ModelScope-specific: Check for reasoning_content in model_provider_data | |
| if hasattr(event, "model_provider_data") and event.model_provider_data: | |
| data = event.model_provider_data | |
| if isinstance(data, dict): | |
| choices = data.get("choices", []) | |
| if choices and len(choices) > 0: | |
| delta = choices[0].get("delta", {}) | |
| reasoning = delta.get("reasoning_content") | |
| if reasoning: | |
| return str(reasoning) | |
| return None | |
| class GeminiAdapter(OpenAIAdapter): | |
| """ | |
| Adapter for Google Gemini. | |
| Note: Gemini has some differences but can be accessed via OpenAI-compatible endpoint. | |
| For native Gemini features, use the dedicated Gemini SDK. | |
| """ | |
| def __init__(self): | |
| self.config = ProviderConfig( | |
| name="gemini", | |
| base_url="https://generativelanguage.googleapis.com/v1beta", | |
| default_model="gemini-2.0-flash-exp", | |
| supports_streaming=True, | |
| supports_tools=True, | |
| supports_streaming_tool_calls=True, | |
| supports_json_schema=False, # Uses different format | |
| supports_thinking=True, | |
| supports_vision=True, | |
| ) | |