diff --git a/AGENTS.md b/AGENTS.md index 999a65e..1cf01ad 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -118,6 +118,12 @@ PY - Write `slug` explicitly as the descriptive human-readable identifier. Slugs must be unique, but may be changed before or after publication when naming conventions improve. - Use `filename_safe_name` when a model-run key needs to become part of a filename; do not hand-roll lossy replacements for characters such as `/` or `*`. - Put every runtime call option in the `ModelRun` declaration; do not add hidden defaults elsewhere. + Transport mechanics are the exception: how a provider class talks to its API (the Anthropic + provider streams; the OpenAI provider uses background mode and polls) is not a model option, + lives in the provider, and is not part of any run's options or fingerprint. + For OpenAI runs the `timeout` option is the budget for the whole background response + (cancelled server-side when it runs out), not an HTTP request timeout; size it for the + slowest expected response, not for a single request. - Use exact provider option names and values as they are passed to `get_response`. - If an option affects performance and should appear in human-readable slugs, filenames, or forecast labels, make the handwritten `slug` reflect it. - Unit tests validate declared top-level provider option names against installed provider SDK diff --git a/README.md b/README.md index dde1632..c9e840b 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,37 @@ human-readable slug `claude-sonnet-4-6-1024`. Use immutable `model_run_key` valu references. Human-readable slugs are available for display and convenience lookups but while the `model_run_key` should be used for stable lookups +## Call metadata, cancellation and fetching by id + +`get_response_with_metadata(prompt, context=..., **options)` makes the same call as +`get_response` but returns a `utils.llm.types.LLMResponse`: the text plus the provider's +response id, final status, token usage (`input_tokens`, `output_tokens`, `reasoning_tokens`, +`tool_calls`) and the number of attempts the retry wrapper made. A `CallContext` carries a +`threading.Event` that cancels the call between retries and, for OpenAI, cancels the +server-side response (or, with `cancel_server_side=False`, stops waiting and leaves the +response running so it can be fetched later), and an `on_response_id` hook that receives +the provider's id as soon as the request is accepted. `model_run.fetch_response(id)` +returns an earlier OpenAI response by id without a new call, and +`model_run.wait_for_response(id)` waits for one that is still running. + +OpenAI calls run in the Responses API background mode: the request returns immediately and +the result is polled, so no connection sits idle while the model works and a finished +response is never lost to a dropped connection. This is how the provider talks to the API, +not a model option, so it is not part of any run's declared options. + +In background mode a run's `timeout` option is not sent to the API as an HTTP timeout. It +is the budget for the whole response, from the request until the response finishes; when +it runs out the response is cancelled server-side and the call raises `TimeoutError`. Runs +without a `timeout` get 30 minutes. A failed poll is polled again for as long as the budget +lasts and never starts a new generation. + +Failed attempts are retried up to five attempts in total for OpenAI (in background mode a +retry only follows a failure that was not billed) and two for every other provider (a +failed attempt there is usually billed and lost), except errors that would only be billed +again: a cancelled call, a call that ran out of its budget, a refused (HTTP 403) +OpenAI request, and an incomplete response (`LLMResponseIncomplete`, usually the +output-token cap). Every failed attempt is logged at WARNING, since it may be billed. + ## LLM Structured Output `get_response` returns the model's text. When you ask a provider for structured output, that diff --git a/tests/unit/test_llm_model_runs.py b/tests/unit/test_llm_model_runs.py index 9c6ac17..e86af51 100644 --- a/tests/unit/test_llm_model_runs.py +++ b/tests/unit/test_llm_model_runs.py @@ -1602,3 +1602,68 @@ def test_get_model_run_by_slug_is_convenience_lookup_for_human_slugs(): assert model_runs.get_model_run_by_slug("gpt-5.5-2026-04-23-high") is run with pytest.raises(KeyError, match="missing-slug"): model_runs.get_model_run_by_slug("missing-slug") + + +def test_model_run_get_response_with_metadata_routes_context_and_returns_metadata(): + """The metadata entry point passes the caller's context through and returns the record.""" + from utils.llm import model_registry, model_runs + from utils.llm.types import CallContext, LLMResponse, LLMUsage + + run = model_runs.ModelRun( + model_run_key="gpt-5.5-2026-04-23-run-variant-02", + slug="gpt-5.5-2026-04-23-high", + model=model_registry.MODELS_BY_KEY["gpt-5.5-2026-04-23"], + options={"reasoning": {"effort": "high"}}, + ) + context = CallContext() + expected = LLMResponse( + text="forecast", + provider_response_id="resp_1", + status="completed", + usage=LLMUsage(input_tokens=10, output_tokens=5, reasoning_tokens=3, tool_calls=0), + attempts=2, + duration_s=1.5, + ) + + with patch( + "utils.llm.model_registry.get_response_with_metadata", return_value=expected + ) as get_response_with_metadata: + response = run.get_response_with_metadata("prompt", context=context, max_output_tokens=10) + + assert response == expected + get_response_with_metadata.assert_called_once_with( + provider=PROVIDERS["OpenAI"], + model_id="gpt-5.5-2026-04-23", + prompt="prompt", + options={"reasoning": {"effort": "high"}, "max_output_tokens": 10}, + context=context, + ) + # The text-only entry point is unchanged for callers that want just the text. + with patch("utils.llm.model_registry.get_response", return_value="forecast"): + assert run.get_response("prompt") == "forecast" + + +def test_model_run_wait_for_response_uses_the_run_timeout_as_budget(): + """Waiting on an earlier response is bounded like a fresh call of the same run.""" + from utils.llm import model_registry, model_runs + from utils.llm.types import LLMResponse + + run = model_runs.ModelRun( + model_run_key="gpt-5.5-2026-04-23-run-variant-02", + slug="gpt-5.5-2026-04-23-high", + model=model_registry.MODELS_BY_KEY["gpt-5.5-2026-04-23"], + options={"reasoning": {"effort": "high"}, "timeout": 900}, + ) + expected = LLMResponse(text="forecast", provider_response_id="resp_old", status="completed") + + with patch( + "utils.llm.model_registry.wait_for_response", return_value=expected + ) as wait_for_response: + assert run.wait_for_response("resp_old") == expected + + wait_for_response.assert_called_once_with( + provider=PROVIDERS["OpenAI"], + provider_response_id="resp_old", + budget=900, + context=None, + ) diff --git a/tests/unit/test_llm_routing.py b/tests/unit/test_llm_routing.py index d8d6890..a645d18 100644 --- a/tests/unit/test_llm_routing.py +++ b/tests/unit/test_llm_routing.py @@ -204,6 +204,7 @@ def test_openai_provider_forwards_options_without_defaults(): input="forecast", reasoning={"effort": "low"}, tools=[{"type": "web_search"}], + background=True, ) @@ -257,6 +258,7 @@ def test_openai_provider_route_fields_override_reserved_options(): model="gpt-5-mini-2025-08-07", input="forecast", temperature=0, + background=True, ) @@ -384,7 +386,8 @@ def test_anthropic_provider_uses_sdk_final_text_helper(): assert text == "*0.25\n*0.50" stream.get_final_text.assert_called_once_with() - stream.get_final_message.assert_not_called() + # The final message is read for its id and usage only, after the text helper ran. + stream.get_final_message.assert_called_once_with() def test_anthropic_provider_logs_raw_response_when_response_has_no_text(caplog): @@ -639,3 +642,415 @@ def failing_call() -> str: error_msg="provider failed", max_retries=1, ) + + +def _openai_background_client(statuses: list[str], **final_fields: Any) -> MagicMock: + """Fake OpenAI client whose background response moves through `statuses` on retrieve.""" + created = MagicMock(id="resp_123", status=statuses[0]) + polled = [MagicMock(id="resp_123", status=status, **final_fields) for status in statuses[1:]] + mock_client = MagicMock() + mock_client.responses.create.return_value = created + mock_client.responses.retrieve.side_effect = polled + return mock_client + + +def test_openai_provider_polls_background_response_and_reports_id_and_usage(): + """A background response is polled until completed; its id goes to the caller at once.""" + from utils.llm.providers.openai import OpenAIProvider + from utils.llm.types import CallContext + + usage = SimpleNamespace( + input_tokens=71_000, + output_tokens=2_300, + output_tokens_details=SimpleNamespace(reasoning_tokens=1_900), + ) + output = [ + SimpleNamespace(type="web_search_call"), + SimpleNamespace(type="reasoning"), + SimpleNamespace(type="web_search_call"), + SimpleNamespace(type="message"), + ] + seen_ids: list[str] = [] + polls_before_id: list[int] = [] + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep") as mock_sleep, + ): + mock_client = _openai_background_client( + ["queued", "in_progress", "completed"], + output_text=" 0.42 ", + usage=usage, + output=output, + ) + mock_openai.return_value = mock_client + + def record_id(response_id: str) -> None: + seen_ids.append(response_id) + polls_before_id.append(mock_client.responses.retrieve.call_count) + + provider = OpenAIProvider(api_key="sk-test") + response = provider.get_response_with_metadata( + model_id="gpt-6-astra", + prompt="forecast", + options={"reasoning": {"effort": "high"}}, + context=CallContext(on_response_id=record_id), + ) + + assert response.text == "0.42" + assert response.provider_response_id == "resp_123" + assert response.status == "completed" + assert response.usage.input_tokens == 71_000 + assert response.usage.output_tokens == 2_300 + assert response.usage.reasoning_tokens == 1_900 + assert response.usage.tool_calls == 2 + assert response.attempts == 1 + assert response.duration_s is not None + assert seen_ids == ["resp_123"] + assert polls_before_id == [0] + assert mock_client.responses.retrieve.call_count == 2 + assert mock_sleep.call_count == 2 + mock_client.responses.cancel.assert_not_called() + + +def test_openai_provider_cancels_background_response_without_retrying(): + """A set cancel event cancels the server-side response and is never retried.""" + import threading + + from utils.llm.providers.openai import OpenAIProvider + from utils.llm.types import CallContext, LLMCallCancelled + + cancel_event = threading.Event() + + with patch("utils.llm.providers.openai.OpenAI") as mock_openai: + mock_client = _openai_background_client(["queued", "completed"], output_text="late") + created = mock_client.responses.create.return_value + + def accept_then_cancel(**kwargs): + del kwargs + cancel_event.set() # the caller gives up right after the request is accepted + return created + + mock_client.responses.create.side_effect = accept_then_cancel + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + with pytest.raises(LLMCallCancelled): + provider.get_response_with_metadata( + model_id="gpt-6-astra", + prompt="forecast", + options={}, + context=CallContext(cancel_event=cancel_event), + ) + + mock_client.responses.create.assert_called_once() + mock_client.responses.cancel.assert_called_once_with("resp_123") + mock_client.responses.retrieve.assert_not_called() + + +def test_openai_provider_budget_cancels_overrunning_response(): + """The run's `timeout` option bounds the whole background response, then cancels it.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + patch("utils.llm.providers.openai.time.monotonic", side_effect=[0.0, 0.0, 1000.0]), + ): + mock_client = _openai_background_client(["queued", "in_progress", "in_progress"]) + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + with pytest.raises(TimeoutError, match="did not finish within 900s"): + provider._call_model( + model_id="gpt-5.6-sol", prompt="forecast", options={"timeout": 900} + ) + + # The budget never reaches the SDK as a per-request timeout. + assert "timeout" not in mock_client.responses.create.call_args.kwargs + mock_client.responses.cancel.assert_called_once_with("resp_123") + + +def test_openai_provider_reports_failed_background_response_with_its_id(): + """A terminal non-completed status raises, naming the response so it can be looked up.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = _openai_background_client( + ["queued", "incomplete"], incomplete_details=SimpleNamespace(reason="max_output_tokens") + ) + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + with pytest.raises(RuntimeError, match="status=incomplete, id=resp_123"): + provider._call_model(model_id="gpt-6-astra", prompt="forecast", options={}) + + +def test_openai_provider_fetches_response_by_id(): + """fetch_response returns a finished response without creating a new one.""" + from utils.llm.providers.openai import OpenAIProvider + + with patch("utils.llm.providers.openai.OpenAI") as mock_openai: + mock_client = MagicMock() + mock_client.responses.retrieve.side_effect = [ + MagicMock(id="resp_9", status="in_progress"), + MagicMock(id="resp_9", status="completed", output_text="0.7", usage=None, output=[]), + ] + mock_openai.return_value = mock_client + provider = OpenAIProvider(api_key="sk-test") + + pending = provider.fetch_response("resp_9") + finished = provider.fetch_response("resp_9") + + assert (pending.status, pending.text) == ("in_progress", "") + assert (finished.status, finished.text, finished.provider_response_id) == ( + "completed", + "0.7", + "resp_9", + ) + assert finished.usage.tool_calls == 0 + mock_client.responses.create.assert_not_called() + + +def test_anthropic_provider_reports_message_id_and_usage(): + """The streamed final message supplies the id, token counts and server tool uses.""" + from utils.llm.providers.anthropic import AnthropicProvider + from utils.llm.types import CallContext + + message = SimpleNamespace( + id="msg_abc", + stop_reason="end_turn", + usage=SimpleNamespace(input_tokens=1200, output_tokens=340), + content=[ + SimpleNamespace(type="server_tool_use"), + SimpleNamespace(type="web_search_tool_result"), + SimpleNamespace(type="text"), + ], + ) + stream = MagicMock() + stream.__enter__.return_value = stream + stream.get_final_text.return_value = "0.3" + stream.get_final_message.return_value = message + seen_ids: list[str] = [] + + with patch("anthropic.Anthropic") as mock_anthropic: + mock_client = MagicMock() + mock_client.messages.stream.return_value = stream + mock_anthropic.return_value = mock_client + + provider = AnthropicProvider(api_key="sk-ant-test") + response = provider.get_response_with_metadata( + model_id="claude-opus-5-5", + prompt="forecast", + options={"max_tokens": 4096}, + context=CallContext(on_response_id=seen_ids.append), + ) + + assert response.text == "0.3" + assert response.provider_response_id == "msg_abc" + assert response.status == "end_turn" + assert (response.usage.input_tokens, response.usage.output_tokens) == (1200, 340) + assert response.usage.tool_calls == 1 + assert seen_ids == ["msg_abc"] + + +def test_openai_provider_retries_a_failed_poll_without_recreating_the_response(): + """A poll that fails is polled again; the generation is never started a second time.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = MagicMock() + mock_client.responses.create.return_value = MagicMock(id="resp_1", status="queued") + mock_client.responses.retrieve.side_effect = [ + ConnectionError("vpn dropped it"), + ConnectionError("still dropping"), + MagicMock(id="resp_1", status="completed", output_text="0.5", usage=None, output=[]), + ] + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + response = provider.get_response_with_metadata( + model_id="gpt-6-astra", prompt="forecast", options={} + ) + + assert response.text == "0.5" + assert mock_client.responses.create.call_count == 1 + assert mock_client.responses.retrieve.call_count == 3 + assert response.attempts == 1 + + +def test_openai_provider_keeps_polling_through_failures_until_the_budget(): + """Failing polls never start a second generation; only the budget ends the wait.""" + import itertools + + from utils.llm.providers.openai import OpenAIProvider + + clock = itertools.count(start=0.0, step=100.0) + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + patch("utils.llm.providers.openai.time.monotonic", side_effect=lambda: next(clock)), + ): + mock_client = MagicMock() + mock_client.responses.create.return_value = MagicMock(id="resp_1", status="queued") + mock_client.responses.retrieve.side_effect = ConnectionError("down") + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test", default_wait_time=0) + with pytest.raises(TimeoutError, match="the last 9 polls failed"): + provider.get_response_with_metadata( + model_id="gpt-5.6-sol", prompt="forecast", options={"timeout": 900} + ) + + assert mock_client.responses.create.call_count == 1 + assert mock_client.responses.retrieve.call_count == 9 + mock_client.responses.cancel.assert_called_once_with("resp_1") + + +def test_openai_provider_does_not_retry_an_incomplete_response(): + """An incomplete response (e.g. the output-token cap) is raised, not paid for again.""" + from utils.llm.providers.openai import OpenAIProvider + from utils.llm.types import LLMResponseIncomplete + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = _openai_background_client( + ["queued", "incomplete"], incomplete_details=SimpleNamespace(reason="max_output_tokens") + ) + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test", default_wait_time=0) + with pytest.raises(LLMResponseIncomplete, match="max_output_tokens"): + provider.get_response_with_metadata( + model_id="gpt-6-astra", prompt="forecast", options={} + ) + + assert mock_client.responses.create.call_count == 1 + + +def test_openai_provider_retries_a_failed_background_response(): + """A response that failed server-side produced nothing, so the call is retried.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = MagicMock() + mock_client.responses.create.side_effect = [ + MagicMock(id="resp_1", status="queued"), + MagicMock(id="resp_2", status="queued"), + ] + mock_client.responses.retrieve.side_effect = [ + MagicMock(id="resp_1", status="failed", incomplete_details=None, error="server_error"), + MagicMock(id="resp_2", status="completed", output_text="0.5", usage=None, output=[]), + ] + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test", default_wait_time=0) + response = provider.get_response_with_metadata( + model_id="gpt-6-astra", prompt="forecast", options={} + ) + + assert (response.text, response.provider_response_id, response.attempts) == ("0.5", "resp_2", 2) + + +def test_openai_provider_can_stop_waiting_without_cancelling_the_response(): + """With cancel_server_side=False a cancelled wait leaves the response to be fetched later.""" + import threading + + from utils.llm.providers.openai import OpenAIProvider + from utils.llm.types import CallContext, LLMCallCancelled + + cancel_event = threading.Event() + seen_ids: list[str] = [] + + with patch("utils.llm.providers.openai.OpenAI") as mock_openai: + mock_client = _openai_background_client(["queued", "completed"], output_text="late") + created = mock_client.responses.create.return_value + + def accept_then_cancel(**kwargs): + del kwargs + cancel_event.set() + return created + + mock_client.responses.create.side_effect = accept_then_cancel + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + with pytest.raises(LLMCallCancelled, match="keeps running"): + provider.get_response_with_metadata( + model_id="gpt-6-astra", + prompt="forecast", + options={}, + context=CallContext( + cancel_event=cancel_event, + on_response_id=seen_ids.append, + cancel_server_side=False, + ), + ) + + assert seen_ids == ["resp_123"] + mock_client.responses.create.assert_called_once() + mock_client.responses.cancel.assert_not_called() + + +def test_openai_provider_waits_for_an_existing_response_by_id(): + """wait_for_response polls a response another process started; no create call.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = MagicMock() + mock_client.responses.retrieve.side_effect = [ + MagicMock(id="resp_old", status="in_progress"), + MagicMock(id="resp_old", status="in_progress"), + MagicMock(id="resp_old", status="completed", output_text="0.8", usage=None, output=[]), + ] + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + response = provider.wait_for_response("resp_old") + + assert (response.text, response.provider_response_id, response.status) == ( + "0.8", + "resp_old", + "completed", + ) + mock_client.responses.create.assert_not_called() + assert mock_client.responses.retrieve.call_count == 3 + + +def test_failed_attempts_are_retried_per_provider_billing(): + """The unbilled OpenAI failures get five attempts; a billed Anthropic stream gets two.""" + from utils.llm.providers.anthropic import AnthropicProvider + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.utils.time.sleep"), + ): + mock_openai.return_value.responses.create.side_effect = ConnectionError("create failed") + provider = OpenAIProvider(api_key="sk-test", default_wait_time=0) + with pytest.raises(ConnectionError): + provider.get_response_with_metadata(model_id="gpt-6-astra", prompt="p", options={}) + assert mock_openai.return_value.responses.create.call_count == 5 + + with patch("anthropic.Anthropic") as mock_anthropic, patch("utils.llm.utils.time.sleep"): + mock_anthropic.return_value.messages.stream.side_effect = ConnectionError("stream dropped") + provider = AnthropicProvider(api_key="sk-ant-test", default_wait_time=0) + with pytest.raises(ConnectionError): + provider.get_response_with_metadata( + model_id="claude-opus-5-5", prompt="p", options={"max_tokens": 16} + ) + assert mock_anthropic.return_value.messages.stream.call_count == 2 diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 1b3f292..57cb1be 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -105,3 +105,64 @@ def model_dump_json(self, indent: int) -> str: return f'{{"indent":{indent}}}' assert response_to_plain_text(Response()) == '{"indent":2}' + + +def test_cancel_event_set_during_wait_stops_the_retry(): + """A cancellation that lands during the wait ends the call instead of the next attempt.""" + import threading + + from utils.llm.types import LLMCallCancelled + + cancel_event = threading.Event() + attempts = 0 + + def failing_call(): + nonlocal attempts + attempts += 1 + cancel_event.set() # arrives while the wrapper is waiting to retry + raise RuntimeError("boom") + + with pytest.raises(LLMCallCancelled, match="Cancelled while waiting"): + get_response_with_retry( + failing_call, + wait_time=60, + error_msg="test", + max_retries=3, + cancel_event=cancel_event, + ) + + assert attempts == 1 + + +def test_cancel_event_set_before_the_call_makes_no_attempt(): + """A call that is already cancelled never reaches the provider.""" + import threading + + from utils.llm.types import LLMCallCancelled + + cancel_event = threading.Event() + cancel_event.set() + + with pytest.raises(LLMCallCancelled, match="before attempt 1"): + get_response_with_retry( + lambda: pytest.fail("provider must not be called"), + wait_time=0, + error_msg="test", + cancel_event=cancel_event, + ) + + +def test_failed_attempts_are_logged_as_warnings(caplog): + """A possibly billed failed attempt must be visible under a default (WARNING) setup.""" + caplog.set_level(logging.WARNING, logger="utils.llm.utils") + + with pytest.raises(RuntimeError, match="boom"): + get_response_with_retry( + lambda: (_ for _ in ()).throw(RuntimeError("boom")), + wait_time=0, + error_msg="OpenAI API request failed.", + max_retries=2, + ) + + attempt_records = [r for r in caplog.records if "(attempt" in r.getMessage()] + assert [r.levelno for r in attempt_records] == [logging.WARNING, logging.WARNING] diff --git a/utils/llm/model_registry.py b/utils/llm/model_registry.py index c22d8b5..cf75a75 100644 --- a/utils/llm/model_registry.py +++ b/utils/llm/model_registry.py @@ -77,6 +77,7 @@ from .providers.openai import OpenAIProvider from .providers.together import TogetherProvider from .providers.xai import XAIProvider +from .types import CallContext, LLMResponse # Registry for API keys by provider class _PROVIDER_API_KEYS: dict[Type[BaseLLMProvider], str] = {} @@ -467,6 +468,43 @@ def get_response( ) +def get_response_with_metadata( + provider: Provider, + model_id: str, + prompt: str, + options: dict[str, Any] | None = None, + context: CallContext | None = None, +) -> LLMResponse: + """Request a response with its metadata (id, usage, attempts) through the provider.""" + provider_cls = _get_provider_class(provider) + provider_instance = _get_provider_instance(provider_cls) + return provider_instance.get_response_with_metadata( + model_id=model_id, + prompt=prompt, + options=options if options is not None else {}, + context=context, + ) + + +def fetch_response(provider: Provider, provider_response_id: str) -> LLMResponse: + """Fetch an earlier response by its provider id, for providers that keep responses.""" + provider_cls = _get_provider_class(provider) + provider_instance = _get_provider_instance(provider_cls) + return provider_instance.fetch_response(provider_response_id) + + +def wait_for_response( + provider: Provider, + provider_response_id: str, + budget: float | None = None, + context: CallContext | None = None, +) -> LLMResponse: + """Wait for an earlier, still-running response by id, without starting a new one.""" + provider_cls = _get_provider_class(provider) + provider_instance = _get_provider_instance(provider_cls) + return provider_instance.wait_for_response(provider_response_id, budget=budget, context=context) + + def _get_provider_class(provider: Provider) -> Type[BaseLLMProvider]: """Return the provider class for a supported API provider route.""" if not isinstance(provider, Provider): diff --git a/utils/llm/model_runs.py b/utils/llm/model_runs.py index 626d0ad..e26dd3d 100644 --- a/utils/llm/model_runs.py +++ b/utils/llm/model_runs.py @@ -23,6 +23,7 @@ from .lab_registry import Lab from .metadata.artificial_analysis import load_artificial_analysis_snapshot from .provider_registry import Provider +from .types import CallContext, LLMResponse logger = logging.getLogger(__name__) AA_MODEL_RUN_KEY_SUFFIX_PATTERN = re.compile(r"^aa-run-variant-[0-9]{2}$") @@ -125,6 +126,54 @@ def get_response(self, prompt: str, **kwargs: Any) -> str: options=merged_options, ) + def get_response_with_metadata( + self, prompt: str, *, context: CallContext | None = None, **kwargs: Any + ) -> LLMResponse: + """Request a response and the provider's metadata for it (id, usage, attempts). + + `context` carries a cancellation event and a hook that receives the provider + response id as soon as the call is accepted; see `utils.llm.types.CallContext`. + """ + merged_options = deepcopy(self.options) + merged_options.update(deepcopy(kwargs)) + logger.info( + "Requesting LLM response provider=%s provider_model_id=%s options=%s", + self.provider.name, + self.provider_model_id, + merged_options, + ) + return model_registry.get_response_with_metadata( + provider=self.provider, + model_id=self.provider_model_id, + prompt=prompt, + options=merged_options, + context=context, + ) + + def fetch_response(self, provider_response_id: str) -> LLMResponse: + """Fetch a response this run started earlier, by the id the provider gave it.""" + return model_registry.fetch_response( + provider=self.provider, provider_response_id=provider_response_id + ) + + def wait_for_response( + self, + provider_response_id: str, + *, + budget: float | None = None, + context: CallContext | None = None, + ) -> LLMResponse: + """Wait for a still-running response this run started earlier; never starts a new one. + + `budget` defaults to the run's `timeout` option, as a fresh call's would. + """ + return model_registry.wait_for_response( + provider=self.provider, + provider_response_id=provider_response_id, + budget=budget if budget is not None else self.options.get("timeout"), + context=context, + ) + def _model_run_options_fingerprint(model_key: str, options: dict[str, Any]) -> str: """Return a stable fingerprint for a model key plus provider options.""" diff --git a/utils/llm/providers/anthropic.py b/utils/llm/providers/anthropic.py index d74fb18..90765d0 100644 --- a/utils/llm/providers/anthropic.py +++ b/utils/llm/providers/anthropic.py @@ -7,6 +7,7 @@ import anthropic +from ..types import CallContext, LLMResponse, LLMUsage, int_or_none from ..utils import response_to_plain_text from .base import BaseLLMProvider @@ -43,6 +44,20 @@ def __init__(self, *, api_key: str | None = None, default_wait_time: int | None self._anthropic_console = anthropic.Anthropic(api_key=api_key) def _call_model(self, *, model_id: str, prompt: str, options: dict[str, Any]) -> str: + return self._call_model_with_metadata( + model_id=model_id, prompt=prompt, options=options, context=CallContext() + ).text + + def _call_model_with_metadata( + self, + *, + model_id: str, + prompt: str, + options: dict[str, Any], + context: CallContext, + ) -> LLMResponse: + # The Messages API has no cancel or retrieve-by-id for a streamed call, so the + # context's cancellation is honoured between retries only (see BaseLLMProvider). call_args: dict[str, Any] = { **options, "model": model_id, @@ -64,10 +79,41 @@ def _call_model(self, *, model_id: str, prompt: str, options: dict[str, Any]) -> stream.until_done() try: - return stream.get_final_text().strip() + text = stream.get_final_text().strip() except RuntimeError: logger.error( "LLM provider response did not include text content. response=%s", response_to_plain_text(stream.get_final_message()), ) raise + message = stream.get_final_message() + + message_id = getattr(message, "id", None) + context.report_response_id(message_id if isinstance(message_id, str) else None) + return LLMResponse( + text=text, + provider_response_id=message_id if isinstance(message_id, str) else None, + status=_str_or_none(getattr(message, "stop_reason", None)), + usage=_usage_from_message(message), + ) + + +def _str_or_none(value: Any) -> str | None: + return value if isinstance(value, str) else None + + +def _usage_from_message(message: Any) -> LLMUsage: + """Read token counts and server-side tool uses from a final Messages API message.""" + usage = getattr(message, "usage", None) + content = getattr(message, "content", None) + tool_calls = None + if isinstance(content, (list, tuple)): + tool_calls = sum( + 1 for block in content if getattr(block, "type", None) == "server_tool_use" + ) + return LLMUsage( + input_tokens=int_or_none(getattr(usage, "input_tokens", None)), + output_tokens=int_or_none(getattr(usage, "output_tokens", None)), + reasoning_tokens=None, + tool_calls=tool_calls, + ) diff --git a/utils/llm/providers/base.py b/utils/llm/providers/base.py index 2ec3f70..e2cf68a 100644 --- a/utils/llm/providers/base.py +++ b/utils/llm/providers/base.py @@ -2,9 +2,12 @@ from __future__ import annotations +import time from abc import ABC, abstractmethod +from dataclasses import replace from typing import Any, Final +from ..types import CallContext, LLMCallCancelled, LLMResponse, LLMResponseIncomplete from ..utils import get_response_with_retry _DEFAULT_WAIT_TIME_SECONDS: Final[int] = 30 @@ -16,6 +19,10 @@ class BaseLLMProvider(ABC): retry_message: str = "LLM provider request failed." # Provider errors that a retry cannot fix, such as a request the provider refused. non_retryable_exceptions: tuple[type[BaseException], ...] = () + # Attempts in total, the first one included. A failed attempt is usually billed + # and lost (a dropped stream has no id to fetch it by), so it is retried once. + # Providers whose failed attempts are not billed may allow more. + max_attempts: int = 2 def __init__(self, *, default_wait_time: int | None = None) -> None: """Initialize the provider with an optional custom backoff interval.""" @@ -28,17 +35,89 @@ def get_response( prompt: str, options: dict[str, Any], ) -> str: - """Return the provider response for a prompt, retrying on provider errors.""" + """Return the provider response text for a prompt, retrying on provider errors.""" + return self.get_response_with_metadata( + model_id=model_id, prompt=prompt, options=options + ).text - def api_call() -> str: - return self._call_model(model_id=model_id, prompt=prompt, options=options) + def get_response_with_metadata( + self, + *, + model_id: str, + prompt: str, + options: dict[str, Any], + context: CallContext | None = None, + ) -> LLMResponse: + """Return the provider response with its metadata, retrying on provider errors. + + `context` carries the caller's cancellation event and response-id hook; both + are optional and a plain call behaves exactly like `get_response`. + """ + call_context = context if context is not None else CallContext() + attempts = 0 + started = time.monotonic() - return get_response_with_retry( + def api_call() -> LLMResponse: + nonlocal attempts + attempts += 1 + return self._call_model_with_metadata( + model_id=model_id, prompt=prompt, options=options, context=call_context + ) + + response = get_response_with_retry( api_call, self._default_wait_time, self.retry_message, - non_retryable=self.non_retryable_exceptions, + max_retries=self.max_attempts, + non_retryable=( + *self.non_retryable_exceptions, + LLMCallCancelled, + LLMResponseIncomplete, + ), + cancel_event=call_context.cancel_event, ) + return replace(response, attempts=attempts, duration_s=time.monotonic() - started) + + def fetch_response(self, provider_response_id: str) -> LLMResponse: + """Return a previously started response by its provider id, without a new call. + + Only providers whose API keeps responses addressable support this. + """ + raise NotImplementedError( + f"{type(self).__name__} cannot fetch responses by id; the provider API has no " + "retrieve-by-id route for this call type." + ) + + def wait_for_response( + self, + provider_response_id: str, + *, + budget: float | None = None, + context: CallContext | None = None, + ) -> LLMResponse: + """Wait for a response started earlier, by its provider id, and return it. + + Only providers whose API keeps responses addressable support this; the + default refuses, like `fetch_response`. + """ + del budget, context + return self.fetch_response(provider_response_id) + + def _call_model_with_metadata( + self, + *, + model_id: str, + prompt: str, + options: dict[str, Any], + context: CallContext, + ) -> LLMResponse: + """Execute one request and return the text plus whatever metadata the provider reports. + + The default wraps `_call_model`, so a provider that reports no metadata needs + no change. Providers override this to attach ids, usage, or cancellation. + """ + del context + return LLMResponse(text=self._call_model(model_id=model_id, prompt=prompt, options=options)) @abstractmethod def _call_model(self, *, model_id: str, prompt: str, options: dict[str, Any]) -> str: diff --git a/utils/llm/providers/meta.py b/utils/llm/providers/meta.py index 86b75f9..1ad243e 100644 --- a/utils/llm/providers/meta.py +++ b/utils/llm/providers/meta.py @@ -13,6 +13,11 @@ class MetaProvider(OpenAIProvider): """LLM provider that wraps the Meta Model API through its OpenAI-compatible Responses route.""" retry_message = "Meta Model API request failed." + # The Meta route is OpenAI-compatible for `responses.create` only; it has no + # background mode, so calls stay synchronous. + background_responses = False + # Synchronous calls: a dropped connection is a billed response lost. + max_attempts = 2 def _create_client(self, api_key: str) -> OpenAI: """Return an OpenAI SDK client pointed at the Meta Model API.""" diff --git a/utils/llm/providers/openai.py b/utils/llm/providers/openai.py index a2856c5..08e997c 100644 --- a/utils/llm/providers/openai.py +++ b/utils/llm/providers/openai.py @@ -2,19 +2,60 @@ from __future__ import annotations -from typing import Any, Dict +import logging +import time +from typing import Any, Dict, Final from openai import OpenAI, PermissionDeniedError # type: ignore[import] +from ..types import ( + CallContext, + LLMCallCancelled, + LLMResponse, + LLMResponseIncomplete, + LLMUsage, + int_or_none, +) from .base import BaseLLMProvider +# A background response is polled, starting often and backing off, so a short +# answer is picked up quickly and a long one does not hammer the API. +_POLL_INTERVAL_START_SECONDS: Final[float] = 2.0 +_POLL_INTERVAL_MAX_SECONDS: Final[float] = 10.0 +# How long a background response may run before it is cancelled and the call +# fails, when the run options carry no `timeout`. +_DEFAULT_RESPONSE_BUDGET_SECONDS: Final[float] = 1800.0 + +_PENDING_STATUSES: Final[frozenset[str]] = frozenset({"queued", "in_progress"}) +# While polls keep failing, log the first failure and then every Nth one in a row. +_POLL_FAILURE_LOG_EVERY: Final[int] = 10 + +logger = logging.getLogger(__name__) + class OpenAIProvider(BaseLLMProvider): - """LLM provider that wraps the OpenAI Responses API.""" + """LLM provider that wraps the OpenAI Responses API. + + Responses are requested in background mode: the request returns at once with a + response id, the model runs server-side, and the result is polled with short + requests. No connection stays idle while the model works, so a network path + that drops idle connections cannot lose a finished (and billed) response; the + id is reported to the caller as soon as it exists, so a response can be + cancelled or fetched later. Background mode is transport, not a model option, + which is why it is not part of any run's declared options. + """ retry_message = "OpenAI API request failed." # HTTP 403, including misalignment_policy_violation, which OpenAI says not to retry. - non_retryable_exceptions = (PermissionDeniedError,) + # A cancelled call and a call that outran its budget are the caller's decision. + non_retryable_exceptions = (PermissionDeniedError, LLMCallCancelled, TimeoutError) + # Subclasses for OpenAI-compatible APIs without background mode turn this off and + # get the synchronous `responses.create` call instead. + background_responses: bool = True + # In background mode a retry only follows a failed `create` or a response that + # ended `failed`, neither of which is billed: a failed poll, a budget overrun + # and an incomplete response never retry. So five attempts cost nothing extra. + max_attempts: int = 5 def __init__(self, *, api_key: str | None = None, default_wait_time: int | None = None) -> None: """Instantiate the OpenAI client using the provided API key. @@ -39,26 +80,179 @@ def _create_client(self, api_key: str) -> OpenAI: return OpenAI(api_key=api_key) def _call_model(self, *, model_id: str, prompt: str, options: dict[str, Any]) -> str: + return self._call_model_with_metadata( + model_id=model_id, prompt=prompt, options=options, context=CallContext() + ).text + + def _call_model_with_metadata( + self, + *, + model_id: str, + prompt: str, + options: dict[str, Any], + context: CallContext, + ) -> LLMResponse: request_payload: Dict[str, Any] = { **options, "model": model_id, "input": prompt, } - response = self._openai_client.responses.create(**request_payload) + if not self.background_responses: + response = self._openai_client.responses.create(**request_payload) + context.report_response_id(getattr(response, "id", None)) + return _finished_response(response) - # Get status text (this is useful for catching errors in reasoning models) + # A run's `timeout` option bounds the whole response here, not one HTTP + # request: the create and retrieve calls are short and keep the SDK default. + budget = request_payload.pop("timeout", None) or _DEFAULT_RESPONSE_BUDGET_SECONDS + response = self._openai_client.responses.create(**request_payload, background=True) + context.report_response_id(getattr(response, "id", None)) + response = self._wait_for_response(response, budget=float(budget), context=context) + return _finished_response(response) + + def fetch_response(self, provider_response_id: str) -> LLMResponse: + """Return a background response by id. + + A response that is still running comes back with empty text and its + pending status; a finished one is returned like a fresh call's result. + """ + response = self._openai_client.responses.retrieve(provider_response_id) status = getattr(response, "status", None) - if status != "completed": - reason = getattr(response, "incomplete_details", None) - status_text = f"OpenAI response incomplete (status={status})" - if reason: - status_text += f", reason={reason}" - raise RuntimeError(status_text) - - # output_text is the text of the response in reasoning models - output_text = getattr(response, "output_text", "") - if isinstance(output_text, str): - return output_text.strip() - - return str(output_text) + if status in _PENDING_STATUSES: + return LLMResponse(text="", provider_response_id=provider_response_id, status=status) + return _finished_response(response) + + def wait_for_response( + self, + provider_response_id: str, + *, + budget: float | None = None, + context: CallContext | None = None, + ) -> LLMResponse: + """Wait for a background response started earlier, by id, and return it. + + Same polling, cancellation and budget behaviour as a fresh call, but no new + generation is ever created: this is how a caller picks up a response whose + first process died while it was still running. + """ + response = self._openai_client.responses.retrieve(provider_response_id) + response = self._wait_for_response( + response, + budget=float(budget) if budget else _DEFAULT_RESPONSE_BUDGET_SECONDS, + context=context if context is not None else CallContext(), + ) + return _finished_response(response) + + def _wait_for_response(self, response: Any, *, budget: float, context: CallContext) -> Any: + """Poll a background response until it leaves the pending statuses. + + The budget cancels the response server-side, so no billed work outlives the + caller's interest in it; so does cancellation, unless the context asks to + leave the response running. A failed poll is polled again for as long as the + budget lasts: the generation it watches is already paid for and is never + started again because a short status request failed. A budget that runs out + while polls fail raises `TimeoutError`, which is not retried either. + """ + response_id = getattr(response, "id", None) + started = time.monotonic() + interval = _POLL_INTERVAL_START_SECONDS + failed_polls = 0 + while getattr(response, "status", None) in _PENDING_STATUSES: + if context.cancelled(): + if context.cancel_server_side: + self._cancel_quietly(response_id) + raise LLMCallCancelled( + f"OpenAI response {response_id} cancelled by the caller." + ) + raise LLMCallCancelled( + f"Stopped waiting for OpenAI response {response_id}; it keeps running " + "server-side and can be fetched by id." + ) + if time.monotonic() - started > budget: + self._cancel_quietly(response_id) + raise TimeoutError( + f"OpenAI response {response_id} did not finish within {budget:.0f}s " + "and was cancelled" + + (f" (the last {failed_polls} polls failed)." if failed_polls else ".") + ) + if context.cancel_event is not None: + context.cancel_event.wait(interval) + else: + time.sleep(interval) + interval = min(interval * 2, _POLL_INTERVAL_MAX_SECONDS) + try: + response = self._openai_client.responses.retrieve(response_id) + except Exception as exc: # noqa: BLE001 - a poll failure is retried, not escalated + failed_polls += 1 + if failed_polls == 1 or failed_polls % _POLL_FAILURE_LOG_EVERY == 0: + logger.warning( + "Polling OpenAI response %s failed (%d in a row): %s: %s; polling " + "again until the %.0fs budget runs out", + response_id, + failed_polls, + type(exc).__name__, + exc, + budget, + ) + continue + if failed_polls: + logger.info( + "Polling OpenAI response %s works again after %d failed polls", + response_id, + failed_polls, + ) + failed_polls = 0 + return response + + def _cancel_quietly(self, response_id: str | None) -> None: + """Cancel a background response; a failed cancel must not mask the real error.""" + if not response_id: + return + try: + self._openai_client.responses.cancel(response_id) + except Exception: # noqa: BLE001 - best effort, the caller is already giving up + pass + + +def _finished_response(response: Any) -> LLMResponse: + """Turn a terminal Responses API object into an `LLMResponse`, or raise if it failed.""" + status = getattr(response, "status", None) + response_id = getattr(response, "id", None) + if status != "completed": + reason = getattr(response, "incomplete_details", None) or getattr(response, "error", None) + status_text = f"OpenAI response incomplete (status={status}, id={response_id})" + if reason: + status_text += f", reason={reason}" + if status == "incomplete": + # Usually the output-token cap: the same request would stop there again. + raise LLMResponseIncomplete(status_text) + raise RuntimeError(status_text) + + # output_text is the text of the response in reasoning models + output_text = getattr(response, "output_text", "") + text = output_text.strip() if isinstance(output_text, str) else str(output_text) + return LLMResponse( + text=text, + provider_response_id=response_id if isinstance(response_id, str) else None, + status=status, + usage=_usage_from_response(response), + ) + + +def _usage_from_response(response: Any) -> LLMUsage: + """Read token counts and the number of built-in tool calls from a Responses object.""" + usage = getattr(response, "usage", None) + output_details = getattr(usage, "output_tokens_details", None) + output_items = getattr(response, "output", None) + tool_calls = None + if isinstance(output_items, (list, tuple)): + tool_calls = sum( + 1 for item in output_items if str(getattr(item, "type", "")).endswith("_call") + ) + return LLMUsage( + input_tokens=int_or_none(getattr(usage, "input_tokens", None)), + output_tokens=int_or_none(getattr(usage, "output_tokens", None)), + reasoning_tokens=int_or_none(getattr(output_details, "reasoning_tokens", None)), + tool_calls=tool_calls, + ) diff --git a/utils/llm/types.py b/utils/llm/types.py new file mode 100644 index 0000000..b50680f --- /dev/null +++ b/utils/llm/types.py @@ -0,0 +1,80 @@ +"""Call-level types shared by the LLM providers and their callers.""" + +from __future__ import annotations + +import threading +from dataclasses import dataclass +from typing import Callable + + +class LLMCallCancelled(Exception): + """Raised when a call is abandoned because its `CallContext` was cancelled.""" + + +class LLMResponseIncomplete(RuntimeError): + """Raised when the provider ended a response early, e.g. at the output-token cap. + + Never retried: the same request would most likely stop at the same limit, and + every attempt is billed. + """ + + +@dataclass(frozen=True) +class LLMUsage: + """Token and tool-call counts reported by the provider for one call.""" + + input_tokens: int | None = None + output_tokens: int | None = None + reasoning_tokens: int | None = None + tool_calls: int | None = None + + +@dataclass(frozen=True) +class LLMResponse: + """One model response with the metadata the provider reported alongside the text.""" + + text: str + provider_response_id: str | None = None + status: str | None = None + usage: LLMUsage | None = None + # Attempts made by the retry wrapper, including the one that produced this response. + attempts: int = 1 + # Wall-clock seconds spent in the retry wrapper, waits between attempts included. + duration_s: float | None = None + + +@dataclass +class CallContext: + """Caller-side hooks for one call: cancellation and the provider response id. + + `cancel_event`, when set, makes the provider abandon the call at its next + checkpoint (between retries, between polls). With `cancel_server_side` (the + default) it also cancels server-side work it can; without it, a response that + is addressable by id keeps running, so the caller can fetch it later. + `on_response_id` is invoked with the provider's response id as soon as it is + known, before the response has finished, so the caller can record it. + """ + + cancel_event: threading.Event | None = None + on_response_id: Callable[[str], None] | None = None + cancel_server_side: bool = True + + def cancelled(self) -> bool: + """Return whether the caller asked for the call to be abandoned.""" + return self.cancel_event is not None and self.cancel_event.is_set() + + def report_response_id(self, provider_response_id: str | None) -> None: + """Hand the provider response id to the caller, if there is one to hand over.""" + if provider_response_id and self.on_response_id is not None: + self.on_response_id(provider_response_id) + + +def int_or_none(value: object) -> int | None: + """Return `value` as an int when it is a real integer count, else None. + + Provider SDK objects expose counts as ints; anything else (a missing field, a + mock, a float placeholder) is reported as unknown rather than guessed. + """ + if isinstance(value, bool) or not isinstance(value, int): + return None + return value diff --git a/utils/llm/utils.py b/utils/llm/utils.py index 3b6b9de..e439c29 100644 --- a/utils/llm/utils.py +++ b/utils/llm/utils.py @@ -3,9 +3,12 @@ from __future__ import annotations import logging +import threading import time from typing import Any, Callable +from .types import LLMCallCancelled + logger = logging.getLogger(__name__) @@ -35,6 +38,7 @@ def get_response_with_retry( error_msg: str, max_retries: int = _DEFAULT_MAX_RETRIES, non_retryable: tuple[type[BaseException], ...] = (), + cancel_event: threading.Event | None = None, ) -> str | Any: """Execute an API call, retrying with a delay when errors occur. @@ -44,14 +48,22 @@ def get_response_with_retry( error_msg (str): Message prefix for retry log entries. max_retries (int): Maximum number of attempts before raising. non_retryable (tuple): Exception types that propagate immediately without retrying. + cancel_event (threading.Event | None): When set, no further attempt is started and + the wait between attempts ends early with `LLMCallCancelled`. """ for attempt in range(max_retries): + if cancel_event is not None and cancel_event.is_set(): + raise LLMCallCancelled(f"{error_msg} Cancelled before attempt {attempt + 1}.") try: return api_call() + except LLMCallCancelled: + raise except non_retryable: raise except Exception as exc: # noqa: BLE001 - retries must catch broad exceptions - logger.info( + # WARNING, not INFO: a failed attempt may be billed, so it must show up + # under a caller's default logging configuration. + logger.warning( "%s (attempt %d/%d): %s: %s", error_msg, attempt + 1, @@ -63,5 +75,11 @@ def get_response_with_retry( if attempt + 1 >= max_retries: raise - logger.info("Waiting for %s seconds before retrying...", wait_time) - time.sleep(wait_time) + logger.warning("Waiting for %s seconds before retrying...", wait_time) + if cancel_event is None: + time.sleep(wait_time) + elif cancel_event.wait(wait_time): + raise LLMCallCancelled( + f"{error_msg} Cancelled while waiting to retry after attempt {attempt + 1}." + ) from exc + raise ValueError(f"max_retries must be at least 1, got {max_retries}.")