diff --git a/clearwing/llm/native.py b/clearwing/llm/native.py index 6068f9f..4771c0e 100644 --- a/clearwing/llm/native.py +++ b/clearwing/llm/native.py @@ -7,7 +7,7 @@ import re from collections.abc import Callable from dataclasses import dataclass -from typing import Any +from typing import Any, Literal from urllib.parse import urljoin import aiohttp @@ -26,6 +26,34 @@ logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# reasoning_effort auto-detection +# --------------------------------------------------------------------------- +# +# `reasoning_effort` is an OpenAI-reasoning-only parameter. Most non-OpenAI +# OpenAI-compat endpoints reject it with HTTP 400 when the configured model is +# not reasoning-capable (Groq + Llama is the common case). We default to +# omitting it for model families we know reject it, while preserving the +# previous "medium" default for everything else. +# +# Match semantics: case-insensitive substring against the model name. First +# match wins; order is not significant. +_REASONING_EFFORT_UNSUPPORTED_PATTERNS: tuple[str, ...] = ( + "llama-3", + "llama-4", + "mistral", + "mixtral", + "qwen2", + "gemma", +) + +# Explicit re-enable list for model names that contain an unsupported pattern +# above but actually *do* support reasoning_effort (e.g. reasoning-tuned +# variants of a denylisted family). Compared against the full lowercased model +# name. Intentionally empty today; populate as such models ship. +_REASONING_EFFORT_OVERRIDE_ALLOW: frozenset[str] = frozenset() + + def _run_coro_sync(coro): try: asyncio.get_running_loop() @@ -89,6 +117,28 @@ class AsyncLLMClient: bounded concurrency. """ + @staticmethod + def _auto_resolve_reasoning_effort(model_name: str) -> str | None: + """Return the effective reasoning_effort for *model_name*. + + Returns ``None`` (i.e. omit the parameter) when the model name matches + a pattern in :data:`_REASONING_EFFORT_UNSUPPORTED_PATTERNS` and is not + in :data:`_REASONING_EFFORT_OVERRIDE_ALLOW`. Returns ``"medium"`` + otherwise — the previous default for all callers. + """ + lower = model_name.lower() + if lower in _REASONING_EFFORT_OVERRIDE_ALLOW: + return "medium" + for pattern in _REASONING_EFFORT_UNSUPPORTED_PATTERNS: + if pattern in lower: + logger.info( + "reasoning_effort=None: model %r matches non-reasoning pattern %r", + model_name, + pattern, + ) + return None + return "medium" + def __init__( self, *, @@ -101,7 +151,7 @@ def __init__( rate_limit_max_retries: int = 6, rate_limit_initial_backoff_seconds: float = 1.0, rate_limit_max_backoff_seconds: float = 60.0, - reasoning_effort: str | None = "medium", + reasoning_effort: str | None | Literal["auto"] = "auto", ) -> None: self.model_name = model_name self.provider_name = provider_name @@ -183,12 +233,19 @@ def __init__( self._anthropic_oauth_url = "https://api.anthropic.com/v1/messages" self.default_system = default_system - # `reasoning_effort` controls how much reasoning the provider - # runs (for models that support it). "medium" is a sensible - # default — higher for deeper-reasoning tasks, "none"/None to - # opt out entirely. Accepted values: "none" | "minimal" | "low" - # | "medium" | "high" | "xhigh" | "max" | "budget:". - self.reasoning_effort = reasoning_effort + # `reasoning_effort` controls how much reasoning the provider runs + # (for models that support it). Accepted values: "none" | "minimal" + # | "low" | "medium" | "high" | "xhigh" | "max" | "budget:" | None. + # + # The sentinel "auto" (the default) lets the client pick based on the + # model name — most non-OpenAI OpenAI-compat models reject this param + # outright (see _REASONING_EFFORT_UNSUPPORTED_PATTERNS), so we omit it + # for them while keeping "medium" everywhere else. Any explicit value + # (including None) bypasses auto-resolution. + if reasoning_effort == "auto": + self.reasoning_effort = self._auto_resolve_reasoning_effort(model_name) + else: + self.reasoning_effort = reasoning_effort self.rate_limit_max_retries = max(0, rate_limit_max_retries) self.rate_limit_initial_backoff_seconds = max(0.1, rate_limit_initial_backoff_seconds) self.rate_limit_max_backoff_seconds = max( @@ -253,9 +310,23 @@ async def achat( async with self._semaphore: client = self._build_client(Client) - response = await self._with_rate_limit_retries( - lambda: self._achat_with_provider_policy(client, request, options) - ) + try: + response = await self._with_rate_limit_retries( + lambda: self._achat_with_provider_policy(client, request, options) + ) + except Exception as exc: + if not self._is_unsupported_reasoning_effort_error(exc): + raise + logger.warning( + "Provider rejected reasoning_effort for model %r; retrying " + "with reasoning_effort=None and disabling it for this session.", + self.model_name, + ) + self.reasoning_effort = None + options = self._rebuild_options_without_reasoning(options) + response = await self._with_rate_limit_retries( + lambda: self._achat_with_provider_policy(client, request, options) + ) return response async def achat_stream( @@ -310,26 +381,44 @@ async def achat_stream( ) client = self._build_client(Client) - try: - stream = await client.astream_chat(self.model_name, request, options) + + async def _consume(opts: ChatOptions) -> ChatResponse | None: + stream = await client.astream_chat(self.model_name, request, opts) async for event in stream: if event.content: on_text_delta(event.content) if event.end is not None: return event.end + return None + + try: + end_event = await _consume(options) except Exception as exc: - if not self._should_try_openai_http_fallback(exc): + if self._is_unsupported_reasoning_effort_error(exc): + logger.warning( + "Provider rejected reasoning_effort (streaming) for model " + "%r; retrying with reasoning_effort=None and disabling it " + "for this session.", + self.model_name, + ) + self.reasoning_effort = None + options = self._rebuild_options_without_reasoning(options) + end_event = await _consume(options) + elif self._should_try_openai_http_fallback(exc): + logger.debug( + "Native OpenAI async stream failed for model=%s base_url=%s; " + "falling back to aiohttp chat/completions: %s", + self.model_name, + self.base_url, + exc, + ) + return await self._openai_chat_http_fallback( + request, options, on_text_delta=on_text_delta + ) + else: raise - logger.debug( - "Native OpenAI async stream failed for model=%s base_url=%s; " - "falling back to aiohttp chat/completions: %s", - self.model_name, - self.base_url, - exc, - ) - return await self._openai_chat_http_fallback( - request, options, on_text_delta=on_text_delta - ) + if end_event is not None: + return end_event # Fallback if stream ends without an end event return await self.achat( messages=messages, @@ -1011,6 +1100,40 @@ async def _with_rate_limit_retries(self, op) -> ChatResponse: ) await asyncio.sleep(delay) + @staticmethod + def _rebuild_options_without_reasoning(options: ChatOptions) -> ChatOptions: + """Return a copy of *options* with ``reasoning_effort=None``. + + ``ChatOptions`` is a frozen Rust struct from genai-pyo3, so we + reconstruct it from scratch. ``response_json_spec`` is preserved when + present. + """ + return ChatOptions( + temperature=options.temperature, + max_tokens=options.max_tokens, + capture_content=options.capture_content, + capture_usage=options.capture_usage, + capture_tool_calls=options.capture_tool_calls, + capture_reasoning_content=options.capture_reasoning_content, + normalize_reasoning_content=options.normalize_reasoning_content, + reasoning_effort=None, + response_json_spec=getattr(options, "response_json_spec", None), + ) + + @staticmethod + def _is_unsupported_reasoning_effort_error(exc: BaseException) -> bool: + """True when *exc* indicates the provider rejected ``reasoning_effort``. + + Both conditions required: ``"reasoning_effort"`` must appear in the + message AND either ``"400"`` (the HTTP status) or ``"unsupported"``. + This avoids false positives on log lines or unrelated errors that + merely mention the parameter name. + """ + text = str(exc).lower() + if "reasoning_effort" not in text: + return False + return "400" in text or "unsupported" in text + def _is_rate_limit_error(self, exc: Exception) -> bool: text = str(exc).lower() return ( diff --git a/tests/test_native_reasoning_effort.py b/tests/test_native_reasoning_effort.py new file mode 100644 index 0000000..fa9a4a4 --- /dev/null +++ b/tests/test_native_reasoning_effort.py @@ -0,0 +1,319 @@ +"""Tests for the reasoning_effort auto-detection added to AsyncLLMClient. + +Covers Layer 1 (model-name denylist + sentinel-default constructor) and +Layer 2 (retry-on-400 fallback in achat / achat_stream). +""" + +import asyncio +from unittest.mock import patch + +import pytest + +from clearwing.llm.native import ( + _REASONING_EFFORT_OVERRIDE_ALLOW, + _REASONING_EFFORT_UNSUPPORTED_PATTERNS, + AsyncLLMClient, +) + + +class TestAutoResolveReasoningEffort: + """Layer 1: model-name-based denylist.""" + + def test_groq_llama_3_3_resolves_to_none(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("llama-3.3-70b-versatile") + assert result is None + + def test_match_is_case_insensitive(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("Llama-3-70B-INSTRUCT") + assert result is None + + def test_openai_gpt_4o_keeps_medium(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("gpt-4o") + assert result == "medium" + + def test_openai_o1_keeps_medium(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("o1-preview") + assert result == "medium" + + def test_deepseek_r1_keeps_medium(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("deepseek-r1") + assert result == "medium" + + def test_mistral_resolves_to_none(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("mistral-large-2407") + assert result is None + + def test_mixtral_resolves_to_none(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("mixtral-8x7b-instruct") + assert result is None + + def test_qwen2_resolves_to_none(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("qwen2.5-72b-instruct") + assert result is None + + def test_gemma_resolves_to_none(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("gemma-2-27b-it") + assert result is None + + def test_unknown_model_keeps_medium(self): + result = AsyncLLMClient._auto_resolve_reasoning_effort("some-future-model-2030") + assert result == "medium" + + def test_constants_are_exported(self): + """Sanity check: the public surface for the denylist is the two module-level constants.""" + assert isinstance(_REASONING_EFFORT_UNSUPPORTED_PATTERNS, tuple) + assert "llama-3" in _REASONING_EFFORT_UNSUPPORTED_PATTERNS + assert isinstance(_REASONING_EFFORT_OVERRIDE_ALLOW, frozenset) + + +class TestConstructorAutoBehavior: + """Layer 1, wired into __init__. + + These tests verify that with the new sentinel default ('auto'), the + constructor calls _auto_resolve_reasoning_effort and stores the result. + Explicit values (including None and "medium") must pass through untouched. + """ + + def _kwargs(self, **overrides): + """Minimal kwargs to satisfy AsyncLLMClient.__init__.""" + base = dict( + model_name="llama-3.3-70b-versatile", + provider_name="openai_compat", + api_key="sk-test", + ) + base.update(overrides) + return base + + def test_default_for_groq_llama_resolves_to_none(self): + client = AsyncLLMClient(**self._kwargs(model_name="llama-3.3-70b-versatile")) + assert client.reasoning_effort is None + + def test_default_for_gpt_4o_resolves_to_medium(self): + client = AsyncLLMClient(**self._kwargs(model_name="gpt-4o")) + assert client.reasoning_effort == "medium" + + def test_explicit_medium_passes_through_on_denylist_model(self): + client = AsyncLLMClient( + **self._kwargs( + model_name="llama-3.3-70b-versatile", + reasoning_effort="medium", + ) + ) + assert client.reasoning_effort == "medium" + + def test_explicit_none_passes_through_on_allowed_model(self): + client = AsyncLLMClient(**self._kwargs(model_name="gpt-4o", reasoning_effort=None)) + assert client.reasoning_effort is None + + def test_explicit_high_passes_through(self): + client = AsyncLLMClient(**self._kwargs(model_name="o1-preview", reasoning_effort="high")) + assert client.reasoning_effort == "high" + + +class TestIsUnsupportedReasoningEffortError: + """Layer 2 helper: classifying the exception so we only retry the right ones.""" + + def test_matches_real_groq_400_body(self): + # Verbatim from the original session log + exc = RuntimeError( + "Web stream error for model 'llama-3.3-70b-versatile (adapter: OpenAI)'. " + "Status: 400 Bad Request. " + 'Body: {"error":{"message":"`reasoning_effort` is not supported with ' + 'this model","type":"invalid_request_error","code":"unknown_parameter"}}' + ) + assert AsyncLLMClient._is_unsupported_reasoning_effort_error(exc) is True + + def test_matches_unsupported_word_without_400(self): + exc = RuntimeError("reasoning_effort: unsupported parameter") + assert AsyncLLMClient._is_unsupported_reasoning_effort_error(exc) is True + + def test_rejects_429_rate_limit(self): + exc = RuntimeError("Status: 429 Too Many Requests. rate limit exceeded") + assert AsyncLLMClient._is_unsupported_reasoning_effort_error(exc) is False + + def test_rejects_400_for_other_param(self): + exc = RuntimeError("Status: 400 Bad Request. unknown parameter 'frobnicator'") + assert AsyncLLMClient._is_unsupported_reasoning_effort_error(exc) is False + + def test_rejects_500_server_error(self): + exc = RuntimeError("Status: 500 Internal Server Error") + assert AsyncLLMClient._is_unsupported_reasoning_effort_error(exc) is False + + def test_rejects_non_400_message_mentioning_param(self): + # Defensive: a stringified exception that mentions reasoning_effort but + # isn't actually a 400 or "unsupported" message + exc = RuntimeError("reasoning_effort defaulted to medium for unknown model") + assert AsyncLLMClient._is_unsupported_reasoning_effort_error(exc) is False + + +class TestRebuildOptionsWithoutReasoning: + """Layer 2 helper: reconstruct ChatOptions with reasoning_effort dropped.""" + + def test_drops_reasoning_effort_preserves_everything_else(self): + from genai_pyo3 import ChatOptions + + original = ChatOptions( + temperature=0.7, + max_tokens=2048, + capture_content=True, + capture_usage=True, + capture_tool_calls=True, + capture_reasoning_content=True, + normalize_reasoning_content=True, + reasoning_effort="medium", + ) + + rebuilt = AsyncLLMClient._rebuild_options_without_reasoning(original) + + assert rebuilt.reasoning_effort is None + assert rebuilt.temperature == 0.7 + assert rebuilt.max_tokens == 2048 + assert rebuilt.capture_content is True + assert rebuilt.capture_usage is True + assert rebuilt.capture_tool_calls is True + assert rebuilt.capture_reasoning_content is True + assert rebuilt.normalize_reasoning_content is True + + +class TestAchatRetryOnUnsupportedReasoning: + """Layer 2 in the non-streaming path: catch the 400, retry once, mutate state.""" + + def _make_client(self): + # Pass explicit reasoning_effort="medium" to bypass auto-detection, + # so we can prove the retry path runs even when Layer 1 didn't catch it. + return AsyncLLMClient( + model_name="some-future-model-2030", + provider_name="openai_compat", + api_key="sk-test", + reasoning_effort="medium", + ) + + def test_retries_once_and_mutates_self(self): + client = self._make_client() + first_exc = RuntimeError( + "Status: 400 Bad Request. " + 'Body: {"error":{"message":"`reasoning_effort` is not supported"}}' + ) + success_response = object() # Sentinel — we don't assert its shape + + # Replace the private call helper. _achat_with_provider_policy is the + # narrowest seam: raises on first call, returns on second. + call_log: list[str] = [] + + async def fake_policy(self_, client_obj, request, options): + call_log.append("called") + if len(call_log) == 1: + raise first_exc + assert options.reasoning_effort is None, "Second call must drop reasoning_effort" + return success_response + + with ( + patch.object( + AsyncLLMClient, + "_achat_with_provider_policy", + new=fake_policy, + ), + patch.object( + AsyncLLMClient, + "_build_client", + new=lambda self, cls: object(), + ), + ): + result = asyncio.run(client.achat(messages=[], system=None, tools=None)) + + assert result is success_response + assert len(call_log) == 2, "Retry should have happened exactly once" + assert client.reasoning_effort is None, ( + "Instance state must be mutated so future calls skip the param" + ) + + def test_unrelated_runtime_error_propagates(self): + client = self._make_client() + unrelated = RuntimeError("Status: 500 Internal Server Error") + + async def always_raise(self_, client_obj, request, options): + raise unrelated + + with ( + patch.object( + AsyncLLMClient, + "_achat_with_provider_policy", + new=always_raise, + ), + patch.object( + AsyncLLMClient, + "_build_client", + new=lambda self, cls: object(), + ), + ): + with pytest.raises(RuntimeError) as exc_info: + asyncio.run(client.achat(messages=[], system=None, tools=None)) + + assert exc_info.value is unrelated + assert client.reasoning_effort == "medium", ( + "Instance state must not be touched on unrelated errors" + ) + + +class TestAchatStreamRetryOnUnsupportedReasoning: + """Same Layer 2 behavior, on the streaming path.""" + + def _make_client(self): + return AsyncLLMClient( + model_name="some-future-model-2030", + provider_name="openai_compat", + api_key="sk-test", + reasoning_effort="medium", + ) + + def test_streaming_path_retries_once(self): + client = self._make_client() + first_exc = RuntimeError( + "Status: 400 Bad Request. " + 'Body: {"error":{"message":"`reasoning_effort` is not supported"}}' + ) + success_end = object() + on_text = lambda chunk: None # noqa: E731 + + call_log: list[str] = [] + + class FakeEvent: + content = None + end = success_end + + async def fake_stream(model_name, request, options): + call_log.append("called") + if len(call_log) == 1: + raise first_exc + assert options.reasoning_effort is None, ( + "Second streaming call must drop reasoning_effort" + ) + + async def gen(): + yield FakeEvent() + + return gen() + + fake_client = type( + "FakeClient", + (), + {"astream_chat": staticmethod(fake_stream)}, + )() + + with patch.object( + AsyncLLMClient, + "_build_client", + new=lambda self, cls: fake_client, + ): + result = asyncio.run( + client.achat_stream( + messages=[], + system=None, + tools=None, + on_text_delta=on_text, + ) + ) + + assert result is success_end + assert len(call_log) == 2 + assert client.reasoning_effort is None