Spaces:
Sleeping
Sleeping
File size: 16,308 Bytes
4ef118d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 | """
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,
)
|