| 1 | """OpenAI-compatible provider for all non-Anthropic LLM APIs.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import asyncio |
| 6 | import json |
| 7 | import hashlib |
| 8 | import importlib.util |
| 9 | import os |
| 10 | import secrets |
| 11 | import string |
| 12 | import time |
| 13 | import uuid |
| 14 | from collections.abc import Awaitable, Callable |
| 15 | from typing import TYPE_CHECKING, Any |
| 16 | |
| 17 | import json_repair |
| 18 | import httpx |
| 19 | from loguru import logger |
| 20 | |
| 21 | if os.environ.get("LANGFUSE_SECRET_KEY") and importlib.util.find_spec("langfuse"): |
| 22 | from langfuse.openai import AsyncOpenAI |
| 23 | else: |
| 24 | if os.environ.get("LANGFUSE_SECRET_KEY"): |
| 25 | import logging |
| 26 | logging.getLogger(__name__).warning( |
| 27 | "LANGFUSE_SECRET_KEY is set but langfuse is not installed; " |
| 28 | "install with `pip install langfuse` to enable tracing" |
| 29 | ) |
| 30 | from openai import AsyncOpenAI |
| 31 | |
| 32 | from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest |
| 33 | from nanobot.providers.openai_responses import ( |
| 34 | consume_sdk_stream, |
| 35 | convert_messages, |
| 36 | convert_tools, |
| 37 | parse_response_output, |
| 38 | ) |
| 39 | |
| 40 | if TYPE_CHECKING: |
| 41 | from nanobot.providers.registry import ProviderSpec |
| 42 | |
| 43 | _ALLOWED_MSG_KEYS = frozenset({ |
| 44 | "role", "content", "tool_calls", "tool_call_id", "name", |
| 45 | "reasoning_content", "extra_content", |
| 46 | }) |
| 47 | _ALNUM = string.ascii_letters + string.digits |
| 48 | |
| 49 | _STANDARD_TC_KEYS = frozenset({"id", "type", "index", "function"}) |
| 50 | _STANDARD_FN_KEYS = frozenset({"name", "arguments"}) |
| 51 | _DEFAULT_OPENROUTER_HEADERS = { |
| 52 | "HTTP-Referer": "https://github.com/HKUDS/nanobot", |
| 53 | "X-OpenRouter-Title": "nanobot", |
| 54 | "X-OpenRouter-Categories": "cli-agent,personal-agent", |
| 55 | } |
| 56 | _KIMI_THINKING_MODELS: frozenset[str] = frozenset({ |
| 57 | "kimi-k2.5", |
| 58 | "kimi-k2.6", |
| 59 | "k2.6-code-preview", |
| 60 | }) |
| 61 | |
| 62 | # Maps ProviderSpec.thinking_style → extra_body builder. |
| 63 | # Each builder takes a bool (thinking_enabled) and returns the dict to |
| 64 | # merge into extra_body, keeping the style→wire-format mapping in one place. |
| 65 | _THINKING_STYLE_MAP: dict[str, Any] = { |
| 66 | "thinking_type": lambda on: {"thinking": {"type": "enabled" if on else "disabled"}}, |
| 67 | "enable_thinking": lambda on: {"enable_thinking": on}, |
| 68 | "reasoning_split": lambda on: {"reasoning_split": on}, |
| 69 | } |
| 70 | |
| 71 | |
| 72 | def _is_kimi_thinking_model(model_name: str) -> bool: |
| 73 | """Return True if model_name refers to a Kimi thinking-capable model. |
| 74 | |
| 75 | Supports two forms: |
| 76 | - Exact match: e.g. kimi-k2.5 / kimi-k2.6 in _KIMI_THINKING_MODELS |
| 77 | - Slug match: moonshotai/kimi-k2.5 -> the part after the last "/" |
| 78 | is checked against _KIMI_THINKING_MODELS |
| 79 | |
| 80 | This covers both the native Moonshot provider (bare slug) and |
| 81 | OpenRouter-style names (``"publisher/slug"``). |
| 82 | """ |
| 83 | name = model_name.lower() |
| 84 | if name in _KIMI_THINKING_MODELS: |
| 85 | return True |
| 86 | if "/" in name and name.rsplit("/", 1)[1] in _KIMI_THINKING_MODELS: |
| 87 | return True |
| 88 | return False |
| 89 | |
| 90 | |
| 91 | def _short_tool_id() -> str: |
| 92 | """9-char alphanumeric ID compatible with all providers (incl. Mistral).""" |
| 93 | return "".join(secrets.choice(_ALNUM) for _ in range(9)) |
| 94 | |
| 95 | |
| 96 | def _get(obj: Any, key: str) -> Any: |
| 97 | """Get a value from dict or object attribute, returning None if absent.""" |
| 98 | if isinstance(obj, dict): |
| 99 | return obj.get(key) |
| 100 | return getattr(obj, key, None) |
| 101 | |
| 102 | |
| 103 | def _coerce_dict(value: Any) -> dict[str, Any] | None: |
| 104 | """Try to coerce *value* to a dict; return None if not possible or empty.""" |
| 105 | if value is None: |
| 106 | return None |
| 107 | if isinstance(value, dict): |
| 108 | return value if value else None |
| 109 | model_dump = getattr(value, "model_dump", None) |
| 110 | if callable(model_dump): |
| 111 | dumped = model_dump() |
| 112 | if isinstance(dumped, dict) and dumped: |
| 113 | return dumped |
| 114 | return None |
| 115 | |
| 116 | |
| 117 | def _extract_tc_extras(tc: Any) -> tuple[ |
| 118 | dict[str, Any] | None, |
| 119 | dict[str, Any] | None, |
| 120 | dict[str, Any] | None, |
| 121 | ]: |
| 122 | """Extract (extra_content, provider_specific_fields, fn_provider_specific_fields). |
| 123 | |
| 124 | Works for both SDK objects and dicts. Captures Gemini ``extra_content`` |
| 125 | verbatim and any non-standard keys on the tool-call / function. |
| 126 | """ |
| 127 | extra_content = _coerce_dict(_get(tc, "extra_content")) |
| 128 | |
| 129 | tc_dict = _coerce_dict(tc) |
| 130 | prov = None |
| 131 | fn_prov = None |
| 132 | if tc_dict is not None: |
| 133 | leftover = {k: v for k, v in tc_dict.items() |
| 134 | if k not in _STANDARD_TC_KEYS and k != "extra_content" and v is not None} |
| 135 | if leftover: |
| 136 | prov = leftover |
| 137 | fn = _coerce_dict(tc_dict.get("function")) |
| 138 | if fn is not None: |
| 139 | fn_leftover = {k: v for k, v in fn.items() |
| 140 | if k not in _STANDARD_FN_KEYS and v is not None} |
| 141 | if fn_leftover: |
| 142 | fn_prov = fn_leftover |
| 143 | else: |
| 144 | prov = _coerce_dict(_get(tc, "provider_specific_fields")) |
| 145 | fn_obj = _get(tc, "function") |
| 146 | if fn_obj is not None: |
| 147 | fn_prov = _coerce_dict(_get(fn_obj, "provider_specific_fields")) |
| 148 | |
| 149 | return extra_content, prov, fn_prov |
| 150 | |
| 151 | |
| 152 | def _uses_openrouter_attribution(spec: "ProviderSpec | None", api_base: str | None) -> bool: |
| 153 | """Apply Nanobot attribution headers to OpenRouter requests by default.""" |
| 154 | if spec and spec.name == "openrouter": |
| 155 | return True |
| 156 | return bool(api_base and "openrouter" in api_base.lower()) |
| 157 | |
| 158 | |
| 159 | _RESPONSES_FAILURE_THRESHOLD = 3 |
| 160 | _RESPONSES_PROBE_INTERVAL_S = 300 # 5 minutes |
| 161 | |
| 162 | |
| 163 | def _is_direct_openai_base(api_base: str | None) -> bool: |
| 164 | """Return True for direct OpenAI endpoints, not generic OpenAI-compatible gateways.""" |
| 165 | if not api_base: |
| 166 | return True |
| 167 | normalized = api_base.strip().lower().rstrip("/") |
| 168 | return "api.openai.com" in normalized and "openrouter" not in normalized |
| 169 | |
| 170 | |
| 171 | def _responses_circuit_key( |
| 172 | model: str | None, |
| 173 | default_model: str, |
| 174 | reasoning_effort: str | None, |
| 175 | ) -> str: |
| 176 | model_name = (model or default_model).lower() |
| 177 | effort = reasoning_effort.lower() if isinstance(reasoning_effort, str) else "" |
| 178 | return f"{model_name}:{effort}" |
| 179 | |
| 180 | |
| 181 | class OpenAICompatProvider(LLMProvider): |
| 182 | """Unified provider for all OpenAI-compatible APIs. |
| 183 | |
| 184 | Receives a resolved ``ProviderSpec`` from the caller — no internal |
| 185 | registry lookups needed. |
| 186 | """ |
| 187 | |
| 188 | def __init__( |
| 189 | self, |
| 190 | api_key: str | None = None, |
| 191 | api_base: str | None = None, |
| 192 | default_model: str = "gpt-4o", |
| 193 | extra_headers: dict[str, str] | None = None, |
| 194 | spec: ProviderSpec | None = None, |
| 195 | ): |
| 196 | super().__init__(api_key, api_base) |
| 197 | self.default_model = default_model |
| 198 | self.extra_headers = extra_headers or {} |
| 199 | self._spec = spec |
| 200 | |
| 201 | if api_key and spec and spec.env_key: |
| 202 | self._setup_env(api_key, api_base) |
| 203 | |
| 204 | effective_base = api_base or (spec.default_api_base if spec else None) or None |
| 205 | self._effective_base = effective_base |
| 206 | default_headers = {"x-session-affinity": uuid.uuid4().hex} |
| 207 | if _uses_openrouter_attribution(spec, effective_base): |
| 208 | default_headers.update(_DEFAULT_OPENROUTER_HEADERS) |
| 209 | if extra_headers: |
| 210 | default_headers.update(extra_headers) |
| 211 | |
| 212 | self._client = AsyncOpenAI( |
| 213 | api_key=api_key or "no-key", |
| 214 | base_url=effective_base, |
| 215 | default_headers=default_headers, |
| 216 | max_retries=0, |
| 217 | # PE re-caption prompts are large and routinely take 20s+; keep headroom |
| 218 | # above edge/gateway HTTP cutoffs that used to kill the sync path. |
| 219 | timeout=httpx.Timeout(180.0, connect=30.0), |
| 220 | ) |
| 221 | |
| 222 | # Responses API circuit breaker: skip after repeated failures, |
| 223 | # probe again after _RESPONSES_PROBE_INTERVAL_S seconds. |
| 224 | self._responses_failures: dict[str, int] = {} |
| 225 | self._responses_tripped_at: dict[str, float] = {} |
| 226 | |
| 227 | def _setup_env(self, api_key: str, api_base: str | None) -> None: |
| 228 | """Set environment variables based on provider spec.""" |
| 229 | spec = self._spec |
| 230 | if not spec or not spec.env_key: |
| 231 | return |
| 232 | if spec.is_gateway: |
| 233 | os.environ[spec.env_key] = api_key |
| 234 | else: |
| 235 | os.environ.setdefault(spec.env_key, api_key) |
| 236 | effective_base = api_base or spec.default_api_base |
| 237 | for env_name, env_val in spec.env_extras: |
| 238 | resolved = env_val.replace("{api_key}", api_key).replace("{api_base}", effective_base) |
| 239 | os.environ.setdefault(env_name, resolved) |
| 240 | |
| 241 | @classmethod |
| 242 | def _apply_cache_control( |
| 243 | cls, |
| 244 | messages: list[dict[str, Any]], |
| 245 | tools: list[dict[str, Any]] | None, |
| 246 | ) -> tuple[list[dict[str, Any]], list[dict[str, Any]] | None]: |
| 247 | """Inject cache_control markers for prompt caching.""" |
| 248 | cache_marker = {"type": "ephemeral"} |
| 249 | new_messages = list(messages) |
| 250 | |
| 251 | def _mark(msg: dict[str, Any]) -> dict[str, Any]: |
| 252 | content = msg.get("content") |
| 253 | if isinstance(content, str): |
| 254 | return {**msg, "content": [ |
| 255 | {"type": "text", "text": content, "cache_control": cache_marker}, |
| 256 | ]} |
| 257 | if isinstance(content, list) and content: |
| 258 | nc = list(content) |
| 259 | nc[-1] = {**nc[-1], "cache_control": cache_marker} |
| 260 | return {**msg, "content": nc} |
| 261 | return msg |
| 262 | |
| 263 | if new_messages and new_messages[0].get("role") == "system": |
| 264 | new_messages[0] = _mark(new_messages[0]) |
| 265 | if len(new_messages) >= 3: |
| 266 | new_messages[-2] = _mark(new_messages[-2]) |
| 267 | |
| 268 | new_tools = tools |
| 269 | if tools: |
| 270 | new_tools = list(tools) |
| 271 | for idx in cls._tool_cache_marker_indices(new_tools): |
| 272 | new_tools[idx] = {**new_tools[idx], "cache_control": cache_marker} |
| 273 | return new_messages, new_tools |
| 274 | |
| 275 | @staticmethod |
| 276 | def _normalize_tool_call_id(tool_call_id: Any) -> Any: |
| 277 | """Normalize to a provider-safe 9-char alphanumeric form.""" |
| 278 | if not isinstance(tool_call_id, str): |
| 279 | return tool_call_id |
| 280 | if len(tool_call_id) == 9 and tool_call_id.isalnum(): |
| 281 | return tool_call_id |
| 282 | return hashlib.sha1(tool_call_id.encode()).hexdigest()[:9] |
| 283 | |
| 284 | @staticmethod |
| 285 | def _normalize_tool_call_arguments(arguments: Any) -> str: |
| 286 | """Force function.arguments into a valid JSON object string.""" |
| 287 | if isinstance(arguments, str): |
| 288 | stripped = arguments.strip() |
| 289 | if not stripped: |
| 290 | return "{}" |
| 291 | try: |
| 292 | parsed = json_repair.loads(stripped) |
| 293 | except Exception: |
| 294 | return "{}" |
| 295 | if isinstance(parsed, dict): |
| 296 | return json.dumps(parsed, ensure_ascii=False) |
| 297 | return "{}" |
| 298 | if isinstance(arguments, dict): |
| 299 | return json.dumps(arguments, ensure_ascii=False) |
| 300 | return "{}" |
| 301 | |
| 302 | def _sanitize_messages(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| 303 | """Strip non-standard keys, normalize tool_call IDs.""" |
| 304 | sanitized = LLMProvider._sanitize_request_messages(messages, _ALLOWED_MSG_KEYS) |
| 305 | id_map: dict[str, str] = {} |
| 306 | |
| 307 | def map_id(value: Any) -> Any: |
| 308 | if not isinstance(value, str): |
| 309 | return value |
| 310 | return id_map.setdefault(value, self._normalize_tool_call_id(value)) |
| 311 | |
| 312 | for clean in sanitized: |
| 313 | if isinstance(clean.get("tool_calls"), list): |
| 314 | normalized = [] |
| 315 | for tc in clean["tool_calls"]: |
| 316 | if not isinstance(tc, dict): |
| 317 | normalized.append(tc) |
| 318 | continue |
| 319 | tc_clean = dict(tc) |
| 320 | tc_clean["id"] = map_id(tc_clean.get("id")) |
| 321 | function = tc_clean.get("function") |
| 322 | if isinstance(function, dict): |
| 323 | function_clean = dict(function) |
| 324 | if "arguments" in function_clean: |
| 325 | function_clean["arguments"] = self._normalize_tool_call_arguments( |
| 326 | function_clean.get("arguments") |
| 327 | ) |
| 328 | else: |
| 329 | function_clean["arguments"] = "{}" |
| 330 | tc_clean["function"] = function_clean |
| 331 | normalized.append(tc_clean) |
| 332 | clean["tool_calls"] = normalized |
| 333 | if clean.get("role") == "assistant": |
| 334 | # Some OpenAI-compatible gateways reject assistant messages |
| 335 | # that mix non-empty content with tool_calls. |
| 336 | clean["content"] = None |
| 337 | if "tool_call_id" in clean and clean["tool_call_id"]: |
| 338 | clean["tool_call_id"] = map_id(clean["tool_call_id"]) |
| 339 | return self._enforce_role_alternation(sanitized) |
| 340 | |
| 341 | # ------------------------------------------------------------------ |
| 342 | # Build kwargs |
| 343 | # ------------------------------------------------------------------ |
| 344 | |
| 345 | @staticmethod |
| 346 | def _supports_temperature( |
| 347 | model_name: str, |
| 348 | reasoning_effort: str | None = None, |
| 349 | ) -> bool: |
| 350 | """Return True when the model accepts a temperature parameter. |
| 351 | |
| 352 | GPT-5 family and reasoning models (o1/o3/o4) reject temperature |
| 353 | when reasoning_effort is set to anything other than ``"none"``. |
| 354 | """ |
| 355 | if reasoning_effort and reasoning_effort.lower() != "none": |
| 356 | return False |
| 357 | name = model_name.lower() |
| 358 | return not any(token in name for token in ("gpt-5", "o1", "o3", "o4")) |
| 359 | |
| 360 | def _build_kwargs( |
| 361 | self, |
| 362 | messages: list[dict[str, Any]], |
| 363 | tools: list[dict[str, Any]] | None, |
| 364 | model: str | None, |
| 365 | max_tokens: int, |
| 366 | temperature: float, |
| 367 | reasoning_effort: str | None, |
| 368 | tool_choice: str | dict[str, Any] | None, |
| 369 | ) -> dict[str, Any]: |
| 370 | model_name = model or self.default_model |
| 371 | spec = self._spec |
| 372 | |
| 373 | if spec and spec.supports_prompt_caching: |
| 374 | model_name = model or self.default_model |
| 375 | if any(model_name.lower().startswith(k) for k in ("anthropic/", "claude")): |
| 376 | messages, tools = self._apply_cache_control(messages, tools) |
| 377 | |
| 378 | if spec and spec.strip_model_prefix: |
| 379 | model_name = model_name.split("/")[-1] |
| 380 | |
| 381 | kwargs: dict[str, Any] = { |
| 382 | "model": model_name, |
| 383 | "messages": self._sanitize_messages(self._sanitize_empty_content(messages)), |
| 384 | } |
| 385 | |
| 386 | # GPT-5 and reasoning models (o1/o3/o4) reject temperature when |
| 387 | # reasoning_effort is active. Only include it when safe. |
| 388 | if self._supports_temperature(model_name, reasoning_effort): |
| 389 | kwargs["temperature"] = temperature |
| 390 | |
| 391 | if spec and getattr(spec, "supports_max_completion_tokens", False): |
| 392 | kwargs["max_completion_tokens"] = max(1, max_tokens) |
| 393 | else: |
| 394 | kwargs["max_tokens"] = max(1, max_tokens) |
| 395 | |
| 396 | if spec: |
| 397 | model_lower = model_name.lower() |
| 398 | for pattern, overrides in spec.model_overrides: |
| 399 | if pattern in model_lower: |
| 400 | kwargs.update(overrides) |
| 401 | break |
| 402 | |
| 403 | # Normalize reasoning_effort into a semantic form (OpenAI vocab) |
| 404 | # used for internal decisions, and a wire form actually sent out. |
| 405 | # "minimum" is accepted as a DashScope-native alias for "minimal". |
| 406 | semantic_effort: str | None = None |
| 407 | if isinstance(reasoning_effort, str): |
| 408 | semantic_effort = reasoning_effort.lower() |
| 409 | if semantic_effort == "minimum": |
| 410 | semantic_effort = "minimal" |
| 411 | |
| 412 | wire_effort = reasoning_effort |
| 413 | if spec and spec.name == "dashscope" and semantic_effort == "minimal": |
| 414 | # DashScope accepts none/minimum/low/medium/high/xhigh; "minimal" 400s. |
| 415 | wire_effort = "minimum" |
| 416 | |
| 417 | if wire_effort: |
| 418 | kwargs["reasoning_effort"] = wire_effort |
| 419 | |
| 420 | # Provider-specific thinking parameters. |
| 421 | # Only sent when reasoning_effort is explicitly configured so that |
| 422 | # the provider default is preserved otherwise. |
| 423 | # The mapping is driven by ProviderSpec.thinking_style so that adding |
| 424 | # a new provider never requires touching this function. |
| 425 | if spec and spec.thinking_style and reasoning_effort is not None: |
| 426 | thinking_enabled = semantic_effort != "minimal" |
| 427 | extra = _THINKING_STYLE_MAP.get(spec.thinking_style, lambda _: None)(thinking_enabled) |
| 428 | if extra: |
| 429 | kwargs.setdefault("extra_body", {}).update(extra) |
| 430 | |
| 431 | # Model-level thinking injection for Kimi thinking-capable models. |
| 432 | # Strip any provider prefix (e.g. "moonshotai/") before the set lookup |
| 433 | # so that OpenRouter-style names like "moonshotai/kimi-k2.5" are handled |
| 434 | # identically to bare names like "kimi-k2.5". |
| 435 | if reasoning_effort is not None and _is_kimi_thinking_model(model_name): |
| 436 | thinking_enabled = semantic_effort != "minimal" |
| 437 | kwargs.setdefault("extra_body", {}).update( |
| 438 | {"thinking": {"type": "enabled" if thinking_enabled else "disabled"}} |
| 439 | ) |
| 440 | |
| 441 | if tools: |
| 442 | kwargs["tools"] = tools |
| 443 | kwargs["tool_choice"] = tool_choice or "auto" |
| 444 | |
| 445 | # Backfill reasoning_content on legacy assistant messages. |
| 446 | # DeepSeek V4 (and potentially others) rejects thinking-mode |
| 447 | # requests that contain assistant messages without reasoning_content |
| 448 | # — even on turns that had no tool calls. This happens when a |
| 449 | # session was started with a non-thinking model or without |
| 450 | # reasoning_effort, then the user switches thinking mode on |
| 451 | # mid-session. Injecting an empty string satisfies the API |
| 452 | # without altering semantics (the model treats it as "no |
| 453 | # thinking happened on that turn"). |
| 454 | thinking_active = ( |
| 455 | (spec and spec.thinking_style and reasoning_effort is not None |
| 456 | and semantic_effort != "minimal") |
| 457 | or (reasoning_effort is not None and _is_kimi_thinking_model(model_name) |
| 458 | and semantic_effort != "minimal") |
| 459 | ) |
| 460 | if thinking_active: |
| 461 | for msg in kwargs["messages"]: |
| 462 | if msg.get("role") == "assistant" and "reasoning_content" not in msg: |
| 463 | msg["reasoning_content"] = "" |
| 464 | |
| 465 | return kwargs |
| 466 | |
| 467 | def _should_use_responses_api( |
| 468 | self, |
| 469 | model: str | None, |
| 470 | reasoning_effort: str | None, |
| 471 | ) -> bool: |
| 472 | """Use Responses API only for direct OpenAI requests that benefit from it.""" |
| 473 | if self._spec and self._spec.name not in ("openai", "github_copilot"): |
| 474 | return False |
| 475 | if self._spec is None or self._spec.name != "github_copilot": |
| 476 | if not _is_direct_openai_base(self._effective_base): |
| 477 | return False |
| 478 | |
| 479 | model_name = (model or self.default_model).lower() |
| 480 | wants = False |
| 481 | if reasoning_effort and reasoning_effort.lower() != "none": |
| 482 | wants = True |
| 483 | elif any(token in model_name for token in ("gpt-5", "o1", "o3", "o4")): |
| 484 | wants = True |
| 485 | if not wants: |
| 486 | return False |
| 487 | |
| 488 | # Circuit breaker: skip after repeated failures, probe periodically. |
| 489 | key = _responses_circuit_key(model, self.default_model, reasoning_effort) |
| 490 | failures = self._responses_failures.get(key, 0) |
| 491 | if failures >= _RESPONSES_FAILURE_THRESHOLD: |
| 492 | tripped = self._responses_tripped_at.get(key, 0.0) |
| 493 | if (time.monotonic() - tripped) < _RESPONSES_PROBE_INTERVAL_S: |
| 494 | return False |
| 495 | # Half-open: allow one probe attempt |
| 496 | return True |
| 497 | |
| 498 | def _record_responses_failure(self, model: str | None, reasoning_effort: str | None) -> None: |
| 499 | key = _responses_circuit_key(model, self.default_model, reasoning_effort) |
| 500 | count = self._responses_failures.get(key, 0) + 1 |
| 501 | self._responses_failures[key] = count |
| 502 | if count >= _RESPONSES_FAILURE_THRESHOLD: |
| 503 | self._responses_tripped_at[key] = time.monotonic() |
| 504 | logger.warning( |
| 505 | "Responses API circuit open for {} — falling back to Chat Completions", |
| 506 | key, |
| 507 | ) |
| 508 | |
| 509 | def _record_responses_success(self, model: str | None, reasoning_effort: str | None) -> None: |
| 510 | key = _responses_circuit_key(model, self.default_model, reasoning_effort) |
| 511 | self._responses_failures.pop(key, None) |
| 512 | self._responses_tripped_at.pop(key, None) |
| 513 | |
| 514 | @staticmethod |
| 515 | def _should_fallback_from_responses_error(e: Exception) -> bool: |
| 516 | """Fallback only for likely Responses API compatibility errors.""" |
| 517 | response = getattr(e, "response", None) |
| 518 | status_code = getattr(e, "status_code", None) |
| 519 | if status_code is None and response is not None: |
| 520 | status_code = getattr(response, "status_code", None) |
| 521 | if status_code not in {400, 404, 422}: |
| 522 | return False |
| 523 | |
| 524 | body = ( |
| 525 | getattr(e, "body", None) |
| 526 | or getattr(e, "doc", None) |
| 527 | or getattr(response, "text", None) |
| 528 | ) |
| 529 | body_text = str(body).lower() if body is not None else "" |
| 530 | compatibility_markers = ( |
| 531 | "responses", |
| 532 | "response api", |
| 533 | "max_output_tokens", |
| 534 | "instructions", |
| 535 | "previous_response", |
| 536 | "unsupported", |
| 537 | "not supported", |
| 538 | "unknown parameter", |
| 539 | "unrecognized request argument", |
| 540 | ) |
| 541 | return any(marker in body_text for marker in compatibility_markers) |
| 542 | |
| 543 | def _build_responses_body( |
| 544 | self, |
| 545 | messages: list[dict[str, Any]], |
| 546 | tools: list[dict[str, Any]] | None, |
| 547 | model: str | None, |
| 548 | max_tokens: int, |
| 549 | temperature: float, |
| 550 | reasoning_effort: str | None, |
| 551 | tool_choice: str | dict[str, Any] | None, |
| 552 | ) -> dict[str, Any]: |
| 553 | """Build a Responses API body for direct OpenAI requests.""" |
| 554 | model_name = model or self.default_model |
| 555 | if self._spec and self._spec.strip_model_prefix: |
| 556 | model_name = model_name.split("/")[-1] |
| 557 | sanitized_messages = self._sanitize_messages(self._sanitize_empty_content(messages)) |
| 558 | instructions, input_items = convert_messages(sanitized_messages) |
| 559 | |
| 560 | body: dict[str, Any] = { |
| 561 | "model": model_name, |
| 562 | "instructions": instructions or None, |
| 563 | "input": input_items, |
| 564 | "max_output_tokens": max(1, max_tokens), |
| 565 | "store": False, |
| 566 | "stream": False, |
| 567 | } |
| 568 | |
| 569 | if self._supports_temperature(model_name, reasoning_effort): |
| 570 | body["temperature"] = temperature |
| 571 | |
| 572 | if reasoning_effort and reasoning_effort.lower() != "none": |
| 573 | body["reasoning"] = {"effort": reasoning_effort} |
| 574 | body["include"] = ["reasoning.encrypted_content"] |
| 575 | |
| 576 | if tools: |
| 577 | body["tools"] = convert_tools(tools) |
| 578 | body["tool_choice"] = tool_choice or "auto" |
| 579 | |
| 580 | return body |
| 581 | |
| 582 | # ------------------------------------------------------------------ |
| 583 | # Response parsing |
| 584 | # ------------------------------------------------------------------ |
| 585 | |
| 586 | @staticmethod |
| 587 | def _maybe_mapping(value: Any) -> dict[str, Any] | None: |
| 588 | if isinstance(value, dict): |
| 589 | return value |
| 590 | model_dump = getattr(value, "model_dump", None) |
| 591 | if callable(model_dump): |
| 592 | dumped = model_dump() |
| 593 | if isinstance(dumped, dict): |
| 594 | return dumped |
| 595 | return None |
| 596 | |
| 597 | @classmethod |
| 598 | def _extract_text_content(cls, value: Any) -> str | None: |
| 599 | if value is None: |
| 600 | return None |
| 601 | if isinstance(value, str): |
| 602 | return value |
| 603 | if isinstance(value, list): |
| 604 | parts: list[str] = [] |
| 605 | for item in value: |
| 606 | item_map = cls._maybe_mapping(item) |
| 607 | if item_map: |
| 608 | text = item_map.get("text") |
| 609 | if isinstance(text, str): |
| 610 | parts.append(text) |
| 611 | continue |
| 612 | text = getattr(item, "text", None) |
| 613 | if isinstance(text, str): |
| 614 | parts.append(text) |
| 615 | continue |
| 616 | if isinstance(item, str): |
| 617 | parts.append(item) |
| 618 | return "".join(parts) or None |
| 619 | return str(value) |
| 620 | |
| 621 | @classmethod |
| 622 | def _extract_usage(cls, response: Any) -> dict[str, int]: |
| 623 | """Extract token usage from an OpenAI-compatible response. |
| 624 | |
| 625 | Handles both dict-based (raw JSON) and object-based (SDK Pydantic) |
| 626 | responses. Provider-specific ``cached_tokens`` fields are normalised |
| 627 | under a single key; see the priority chain inside for details. |
| 628 | """ |
| 629 | # --- resolve usage object --- |
| 630 | usage_obj = None |
| 631 | response_map = cls._maybe_mapping(response) |
| 632 | if response_map is not None: |
| 633 | usage_obj = response_map.get("usage") |
| 634 | elif hasattr(response, "usage") and response.usage: |
| 635 | usage_obj = response.usage |
| 636 | |
| 637 | usage_map = cls._maybe_mapping(usage_obj) |
| 638 | if usage_map is not None: |
| 639 | result = { |
| 640 | "prompt_tokens": int(usage_map.get("prompt_tokens") or 0), |
| 641 | "completion_tokens": int(usage_map.get("completion_tokens") or 0), |
| 642 | "total_tokens": int(usage_map.get("total_tokens") or 0), |
| 643 | } |
| 644 | elif usage_obj: |
| 645 | result = { |
| 646 | "prompt_tokens": getattr(usage_obj, "prompt_tokens", 0) or 0, |
| 647 | "completion_tokens": getattr(usage_obj, "completion_tokens", 0) or 0, |
| 648 | "total_tokens": getattr(usage_obj, "total_tokens", 0) or 0, |
| 649 | } |
| 650 | else: |
| 651 | return {} |
| 652 | |
| 653 | # --- cached_tokens (normalised across providers) --- |
| 654 | # Try nested paths first (dict), fall back to attribute (SDK object). |
| 655 | # Priority order ensures the most specific field wins. |
| 656 | for path in ( |
| 657 | ("prompt_tokens_details", "cached_tokens"), # OpenAI/Zhipu/MiniMax/Qwen/Mistral/xAI |
| 658 | ("cached_tokens",), # StepFun/Moonshot (top-level) |
| 659 | ("prompt_cache_hit_tokens",), # DeepSeek/SiliconFlow |
| 660 | ): |
| 661 | cached = cls._get_nested_int(usage_map, path) |
| 662 | if not cached and usage_obj: |
| 663 | cached = cls._get_nested_int(usage_obj, path) |
| 664 | if cached: |
| 665 | result["cached_tokens"] = cached |
| 666 | break |
| 667 | |
| 668 | return result |
| 669 | |
| 670 | @staticmethod |
| 671 | def _get_nested_int(obj: Any, path: tuple[str, ...]) -> int: |
| 672 | """Drill into *obj* by *path* segments and return an ``int`` value. |
| 673 | |
| 674 | Supports both dict-key access and attribute access so it works |
| 675 | uniformly with raw JSON dicts **and** SDK Pydantic models. |
| 676 | """ |
| 677 | current = obj |
| 678 | for segment in path: |
| 679 | if current is None: |
| 680 | return 0 |
| 681 | if isinstance(current, dict): |
| 682 | current = current.get(segment) |
| 683 | else: |
| 684 | current = getattr(current, segment, None) |
| 685 | return int(current or 0) if current is not None else 0 |
| 686 | |
| 687 | def _parse(self, response: Any) -> LLMResponse: |
| 688 | if isinstance(response, str): |
| 689 | return LLMResponse(content=response, finish_reason="stop") |
| 690 | |
| 691 | response_map = self._maybe_mapping(response) |
| 692 | if response_map is not None: |
| 693 | choices = response_map.get("choices") or [] |
| 694 | if not choices: |
| 695 | content = self._extract_text_content( |
| 696 | response_map.get("content") or response_map.get("output_text") |
| 697 | ) |
| 698 | reasoning_content = self._extract_text_content( |
| 699 | response_map.get("reasoning_content") |
| 700 | ) |
| 701 | if content is not None: |
| 702 | return LLMResponse( |
| 703 | content=content, |
| 704 | reasoning_content=reasoning_content, |
| 705 | finish_reason=str(response_map.get("finish_reason") or "stop"), |
| 706 | usage=self._extract_usage(response_map), |
| 707 | ) |
| 708 | return LLMResponse(content="Error: API returned empty choices.", finish_reason="error") |
| 709 | |
| 710 | choice0 = self._maybe_mapping(choices[0]) or {} |
| 711 | msg0 = self._maybe_mapping(choice0.get("message")) or {} |
| 712 | content = self._extract_text_content(msg0.get("content")) |
| 713 | finish_reason = str(choice0.get("finish_reason") or "stop") |
| 714 | |
| 715 | raw_tool_calls: list[Any] = [] |
| 716 | # StepFun Plan: fallback to reasoning field when content is empty |
| 717 | if not content and msg0.get("reasoning"): |
| 718 | content = self._extract_text_content(msg0.get("reasoning")) |
| 719 | reasoning_content = msg0.get("reasoning_content") |
| 720 | if not reasoning_content and msg0.get("reasoning"): |
| 721 | reasoning_content = self._extract_text_content(msg0.get("reasoning")) |
| 722 | for ch in choices: |
| 723 | ch_map = self._maybe_mapping(ch) or {} |
| 724 | m = self._maybe_mapping(ch_map.get("message")) or {} |
| 725 | tool_calls = m.get("tool_calls") |
| 726 | if isinstance(tool_calls, list) and tool_calls: |
| 727 | raw_tool_calls.extend(tool_calls) |
| 728 | if ch_map.get("finish_reason") in ("tool_calls", "stop"): |
| 729 | finish_reason = str(ch_map["finish_reason"]) |
| 730 | if not content: |
| 731 | content = self._extract_text_content(m.get("content")) |
| 732 | if not reasoning_content: |
| 733 | reasoning_content = m.get("reasoning_content") |
| 734 | |
| 735 | parsed_tool_calls = [] |
| 736 | for tc in raw_tool_calls: |
| 737 | tc_map = self._maybe_mapping(tc) or {} |
| 738 | fn = self._maybe_mapping(tc_map.get("function")) or {} |
| 739 | args = fn.get("arguments", {}) |
| 740 | if isinstance(args, str): |
| 741 | args = json_repair.loads(args) |
| 742 | ec, prov, fn_prov = _extract_tc_extras(tc) |
| 743 | parsed_tool_calls.append(ToolCallRequest( |
| 744 | id=_short_tool_id(), |
| 745 | name=str(fn.get("name") or ""), |
| 746 | arguments=args if isinstance(args, dict) else {}, |
| 747 | extra_content=ec, |
| 748 | provider_specific_fields=prov, |
| 749 | function_provider_specific_fields=fn_prov, |
| 750 | )) |
| 751 | |
| 752 | return LLMResponse( |
| 753 | content=content, |
| 754 | tool_calls=parsed_tool_calls, |
| 755 | finish_reason=finish_reason, |
| 756 | usage=self._extract_usage(response_map), |
| 757 | reasoning_content=reasoning_content if isinstance(reasoning_content, str) else None, |
| 758 | ) |
| 759 | |
| 760 | if not response.choices: |
| 761 | return LLMResponse(content="Error: API returned empty choices.", finish_reason="error") |
| 762 | |
| 763 | choice = response.choices[0] |
| 764 | msg = choice.message |
| 765 | content = msg.content |
| 766 | finish_reason = choice.finish_reason |
| 767 | |
| 768 | raw_tool_calls: list[Any] = [] |
| 769 | for ch in response.choices: |
| 770 | m = ch.message |
| 771 | if hasattr(m, "tool_calls") and m.tool_calls: |
| 772 | raw_tool_calls.extend(m.tool_calls) |
| 773 | if ch.finish_reason in ("tool_calls", "stop"): |
| 774 | finish_reason = ch.finish_reason |
| 775 | if not content and m.content: |
| 776 | content = m.content |
| 777 | if not content and getattr(m, "reasoning", None): |
| 778 | content = m.reasoning |
| 779 | |
| 780 | tool_calls = [] |
| 781 | for tc in raw_tool_calls: |
| 782 | args = tc.function.arguments |
| 783 | if isinstance(args, str): |
| 784 | args = json_repair.loads(args) |
| 785 | ec, prov, fn_prov = _extract_tc_extras(tc) |
| 786 | tool_calls.append(ToolCallRequest( |
| 787 | id=_short_tool_id(), |
| 788 | name=tc.function.name, |
| 789 | arguments=args, |
| 790 | extra_content=ec, |
| 791 | provider_specific_fields=prov, |
| 792 | function_provider_specific_fields=fn_prov, |
| 793 | )) |
| 794 | |
| 795 | reasoning_content = getattr(msg, "reasoning_content", None) or None |
| 796 | if not reasoning_content and getattr(msg, "reasoning", None): |
| 797 | reasoning_content = msg.reasoning |
| 798 | |
| 799 | return LLMResponse( |
| 800 | content=content, |
| 801 | tool_calls=tool_calls, |
| 802 | finish_reason=finish_reason or "stop", |
| 803 | usage=self._extract_usage(response), |
| 804 | reasoning_content=reasoning_content, |
| 805 | ) |
| 806 | |
| 807 | @classmethod |
| 808 | def _parse_chunks(cls, chunks: list[Any]) -> LLMResponse: |
| 809 | content_parts: list[str] = [] |
| 810 | reasoning_parts: list[str] = [] |
| 811 | tc_bufs: dict[int, dict[str, Any]] = {} |
| 812 | finish_reason = "stop" |
| 813 | usage: dict[str, int] = {} |
| 814 | |
| 815 | def _accum_tc(tc: Any, idx_hint: int) -> None: |
| 816 | """Accumulate one streaming tool-call delta into *tc_bufs*.""" |
| 817 | tc_index: int = _get(tc, "index") if _get(tc, "index") is not None else idx_hint |
| 818 | buf = tc_bufs.setdefault(tc_index, { |
| 819 | "id": "", "name": "", "arguments": "", |
| 820 | "extra_content": None, "prov": None, "fn_prov": None, |
| 821 | }) |
| 822 | tc_id = _get(tc, "id") |
| 823 | if tc_id: |
| 824 | buf["id"] = str(tc_id) |
| 825 | fn = _get(tc, "function") |
| 826 | if fn is not None: |
| 827 | fn_name = _get(fn, "name") |
| 828 | if fn_name: |
| 829 | buf["name"] = str(fn_name) |
| 830 | fn_args = _get(fn, "arguments") |
| 831 | if fn_args: |
| 832 | buf["arguments"] += str(fn_args) |
| 833 | ec, prov, fn_prov = _extract_tc_extras(tc) |
| 834 | if ec: |
| 835 | buf["extra_content"] = ec |
| 836 | if prov: |
| 837 | buf["prov"] = prov |
| 838 | if fn_prov: |
| 839 | buf["fn_prov"] = fn_prov |
| 840 | |
| 841 | for chunk in chunks: |
| 842 | if isinstance(chunk, str): |
| 843 | content_parts.append(chunk) |
| 844 | continue |
| 845 | |
| 846 | chunk_map = cls._maybe_mapping(chunk) |
| 847 | if chunk_map is not None: |
| 848 | choices = chunk_map.get("choices") or [] |
| 849 | if not choices: |
| 850 | usage = cls._extract_usage(chunk_map) or usage |
| 851 | text = cls._extract_text_content( |
| 852 | chunk_map.get("content") or chunk_map.get("output_text") |
| 853 | ) |
| 854 | if text: |
| 855 | content_parts.append(text) |
| 856 | continue |
| 857 | choice = cls._maybe_mapping(choices[0]) or {} |
| 858 | if choice.get("finish_reason"): |
| 859 | finish_reason = str(choice["finish_reason"]) |
| 860 | delta = cls._maybe_mapping(choice.get("delta")) or {} |
| 861 | text = cls._extract_text_content(delta.get("content")) |
| 862 | if text: |
| 863 | content_parts.append(text) |
| 864 | text = cls._extract_text_content(delta.get("reasoning_content")) |
| 865 | if not text: |
| 866 | text = cls._extract_text_content(delta.get("reasoning")) |
| 867 | if text: |
| 868 | reasoning_parts.append(text) |
| 869 | for idx, tc in enumerate(delta.get("tool_calls") or []): |
| 870 | _accum_tc(tc, idx) |
| 871 | usage = cls._extract_usage(chunk_map) or usage |
| 872 | continue |
| 873 | |
| 874 | if not chunk.choices: |
| 875 | usage = cls._extract_usage(chunk) or usage |
| 876 | continue |
| 877 | choice = chunk.choices[0] |
| 878 | if choice.finish_reason: |
| 879 | finish_reason = choice.finish_reason |
| 880 | delta = choice.delta |
| 881 | if delta and delta.content: |
| 882 | content_parts.append(delta.content) |
| 883 | if delta: |
| 884 | reasoning = getattr(delta, "reasoning_content", None) |
| 885 | if not reasoning: |
| 886 | reasoning = getattr(delta, "reasoning", None) |
| 887 | if reasoning: |
| 888 | reasoning_parts.append(reasoning) |
| 889 | for tc in (delta.tool_calls or []) if delta else []: |
| 890 | _accum_tc(tc, getattr(tc, "index", 0)) |
| 891 | |
| 892 | return LLMResponse( |
| 893 | content="".join(content_parts) or None, |
| 894 | tool_calls=[ |
| 895 | ToolCallRequest( |
| 896 | id=b["id"] or _short_tool_id(), |
| 897 | name=b["name"], |
| 898 | arguments=json_repair.loads(b["arguments"]) if b["arguments"] else {}, |
| 899 | extra_content=b.get("extra_content"), |
| 900 | provider_specific_fields=b.get("prov"), |
| 901 | function_provider_specific_fields=b.get("fn_prov"), |
| 902 | ) |
| 903 | for b in tc_bufs.values() |
| 904 | ], |
| 905 | finish_reason=finish_reason, |
| 906 | usage=usage, |
| 907 | reasoning_content="".join(reasoning_parts) or None, |
| 908 | ) |
| 909 | |
| 910 | @classmethod |
| 911 | def _extract_error_metadata(cls, e: Exception) -> dict[str, Any]: |
| 912 | response = getattr(e, "response", None) |
| 913 | headers = getattr(response, "headers", None) |
| 914 | payload = ( |
| 915 | getattr(e, "body", None) |
| 916 | or getattr(e, "doc", None) |
| 917 | or getattr(response, "text", None) |
| 918 | ) |
| 919 | if payload is None and response is not None: |
| 920 | response_json = getattr(response, "json", None) |
| 921 | if callable(response_json): |
| 922 | try: |
| 923 | payload = response_json() |
| 924 | except Exception: |
| 925 | payload = None |
| 926 | error_type, error_code = LLMProvider._extract_error_type_code(payload) |
| 927 | |
| 928 | status_code = getattr(e, "status_code", None) |
| 929 | if status_code is None and response is not None: |
| 930 | status_code = getattr(response, "status_code", None) |
| 931 | |
| 932 | should_retry: bool | None = None |
| 933 | if headers is not None: |
| 934 | raw = headers.get("x-should-retry") |
| 935 | if isinstance(raw, str): |
| 936 | lowered = raw.strip().lower() |
| 937 | if lowered == "true": |
| 938 | should_retry = True |
| 939 | elif lowered == "false": |
| 940 | should_retry = False |
| 941 | |
| 942 | error_kind: str | None = None |
| 943 | error_name = e.__class__.__name__.lower() |
| 944 | if "timeout" in error_name: |
| 945 | error_kind = "timeout" |
| 946 | elif "connection" in error_name: |
| 947 | error_kind = "connection" |
| 948 | |
| 949 | return { |
| 950 | "error_status_code": int(status_code) if status_code is not None else None, |
| 951 | "error_kind": error_kind, |
| 952 | "error_type": error_type, |
| 953 | "error_code": error_code, |
| 954 | "error_retry_after_s": cls._extract_retry_after_from_headers(headers), |
| 955 | "error_should_retry": should_retry, |
| 956 | } |
| 957 | |
| 958 | @staticmethod |
| 959 | def _handle_error( |
| 960 | e: Exception, |
| 961 | *, |
| 962 | spec: ProviderSpec | None = None, |
| 963 | api_base: str | None = None, |
| 964 | ) -> LLMResponse: |
| 965 | body = ( |
| 966 | getattr(e, "doc", None) |
| 967 | or getattr(e, "body", None) |
| 968 | or getattr(getattr(e, "response", None), "text", None) |
| 969 | ) |
| 970 | body_text = body if isinstance(body, str) else str(body) if body is not None else "" |
| 971 | msg = f"Error: {body_text.strip()[:500]}" if body_text.strip() else f"Error calling LLM: {e}" |
| 972 | |
| 973 | text = f"{body_text} {e}".lower() |
| 974 | if spec and spec.is_local and ("502" in text or "connection" in text or "refused" in text): |
| 975 | msg += ( |
| 976 | "\nHint: this is a local model endpoint. Check that the local server is reachable at " |
| 977 | f"{api_base or spec.default_api_base}, and if you are using a proxy/tunnel, make sure it " |
| 978 | "can reach your local Ollama/vLLM service instead of routing localhost through the remote host." |
| 979 | ) |
| 980 | |
| 981 | response = getattr(e, "response", None) |
| 982 | retry_after = LLMProvider._extract_retry_after_from_headers(getattr(response, "headers", None)) |
| 983 | if retry_after is None: |
| 984 | retry_after = LLMProvider._extract_retry_after(msg) |
| 985 | return LLMResponse( |
| 986 | content=msg, |
| 987 | finish_reason="error", |
| 988 | retry_after=retry_after, |
| 989 | **OpenAICompatProvider._extract_error_metadata(e), |
| 990 | ) |
| 991 | |
| 992 | # ------------------------------------------------------------------ |
| 993 | # Public API |
| 994 | # ------------------------------------------------------------------ |
| 995 | |
| 996 | async def chat( |
| 997 | self, |
| 998 | messages: list[dict[str, Any]], |
| 999 | tools: list[dict[str, Any]] | None = None, |
| 1000 | model: str | None = None, |
| 1001 | max_tokens: int = 4096, |
| 1002 | temperature: float = 0.7, |
| 1003 | reasoning_effort: str | None = None, |
| 1004 | tool_choice: str | dict[str, Any] | None = None, |
| 1005 | ) -> LLMResponse: |
| 1006 | try: |
| 1007 | if self._should_use_responses_api(model, reasoning_effort): |
| 1008 | try: |
| 1009 | body = self._build_responses_body( |
| 1010 | messages, tools, model, max_tokens, temperature, |
| 1011 | reasoning_effort, tool_choice, |
| 1012 | ) |
| 1013 | result = parse_response_output(await self._client.responses.create(**body)) |
| 1014 | self._record_responses_success(model, reasoning_effort) |
| 1015 | return result |
| 1016 | except Exception as responses_error: |
| 1017 | if self._spec and self._spec.name == "github_copilot": |
| 1018 | # Copilot gateway exposes GPT-5/o-series only via /responses; |
| 1019 | # falling back to /chat/completions cannot succeed and would |
| 1020 | # hide the real error. |
| 1021 | raise |
| 1022 | if not self._should_fallback_from_responses_error(responses_error): |
| 1023 | raise |
| 1024 | self._record_responses_failure(model, reasoning_effort) |
| 1025 | |
| 1026 | kwargs = self._build_kwargs( |
| 1027 | messages, tools, model, max_tokens, temperature, |
| 1028 | reasoning_effort, tool_choice, |
| 1029 | ) |
| 1030 | return self._parse(await self._client.chat.completions.create(**kwargs)) |
| 1031 | except Exception as e: |
| 1032 | return self._handle_error(e, spec=self._spec, api_base=self.api_base) |
| 1033 | |
| 1034 | async def chat_stream( |
| 1035 | self, |
| 1036 | messages: list[dict[str, Any]], |
| 1037 | tools: list[dict[str, Any]] | None = None, |
| 1038 | model: str | None = None, |
| 1039 | max_tokens: int = 4096, |
| 1040 | temperature: float = 0.7, |
| 1041 | reasoning_effort: str | None = None, |
| 1042 | tool_choice: str | dict[str, Any] | None = None, |
| 1043 | on_content_delta: Callable[[str], Awaitable[None]] | None = None, |
| 1044 | ) -> LLMResponse: |
| 1045 | idle_timeout_s = int(os.environ.get("NANOBOT_STREAM_IDLE_TIMEOUT_S", "90")) |
| 1046 | try: |
| 1047 | if self._should_use_responses_api(model, reasoning_effort): |
| 1048 | try: |
| 1049 | body = self._build_responses_body( |
| 1050 | messages, tools, model, max_tokens, temperature, |
| 1051 | reasoning_effort, tool_choice, |
| 1052 | ) |
| 1053 | body["stream"] = True |
| 1054 | stream = await self._client.responses.create(**body) |
| 1055 | |
| 1056 | async def _timed_stream(): |
| 1057 | stream_iter = stream.__aiter__() |
| 1058 | while True: |
| 1059 | try: |
| 1060 | yield await asyncio.wait_for( |
| 1061 | stream_iter.__anext__(), |
| 1062 | timeout=idle_timeout_s, |
| 1063 | ) |
| 1064 | except StopAsyncIteration: |
| 1065 | break |
| 1066 | |
| 1067 | content, tool_calls, finish_reason, usage, reasoning_content = await consume_sdk_stream( |
| 1068 | _timed_stream(), |
| 1069 | on_content_delta, |
| 1070 | ) |
| 1071 | self._record_responses_success(model, reasoning_effort) |
| 1072 | return LLMResponse( |
| 1073 | content=content or None, |
| 1074 | tool_calls=tool_calls, |
| 1075 | finish_reason=finish_reason, |
| 1076 | usage=usage, |
| 1077 | reasoning_content=reasoning_content, |
| 1078 | ) |
| 1079 | except Exception as responses_error: |
| 1080 | if self._spec and self._spec.name == "github_copilot": |
| 1081 | # Copilot gateway exposes GPT-5/o-series only via /responses; |
| 1082 | # falling back to /chat/completions cannot succeed and would |
| 1083 | # hide the real error. |
| 1084 | raise |
| 1085 | if not self._should_fallback_from_responses_error(responses_error): |
| 1086 | raise |
| 1087 | self._record_responses_failure(model, reasoning_effort) |
| 1088 | |
| 1089 | kwargs = self._build_kwargs( |
| 1090 | messages, tools, model, max_tokens, temperature, |
| 1091 | reasoning_effort, tool_choice, |
| 1092 | ) |
| 1093 | kwargs["stream"] = True |
| 1094 | kwargs["stream_options"] = {"include_usage": True} |
| 1095 | stream = await self._client.chat.completions.create(**kwargs) |
| 1096 | chunks: list[Any] = [] |
| 1097 | stream_iter = stream.__aiter__() |
| 1098 | while True: |
| 1099 | try: |
| 1100 | chunk = await asyncio.wait_for( |
| 1101 | stream_iter.__anext__(), |
| 1102 | timeout=idle_timeout_s, |
| 1103 | ) |
| 1104 | except StopAsyncIteration: |
| 1105 | break |
| 1106 | chunks.append(chunk) |
| 1107 | if on_content_delta and chunk.choices: |
| 1108 | text = getattr(chunk.choices[0].delta, "content", None) |
| 1109 | if text: |
| 1110 | await on_content_delta(text) |
| 1111 | return self._parse_chunks(chunks) |
| 1112 | except asyncio.TimeoutError: |
| 1113 | return LLMResponse( |
| 1114 | content=( |
| 1115 | f"Error calling LLM: stream stalled for more than " |
| 1116 | f"{idle_timeout_s} seconds" |
| 1117 | ), |
| 1118 | finish_reason="error", |
| 1119 | error_kind="timeout", |
| 1120 | ) |
| 1121 | except Exception as e: |
| 1122 | return self._handle_error(e, spec=self._spec, api_base=self.api_base) |
| 1123 | |
| 1124 | def get_default_model(self) -> str: |
| 1125 | return self.default_model |
| 1126 |