diff --git a/artemis/agents/explorer/explorer.py b/artemis/agents/explorer/explorer.py index a0af8fc..1c48427 100644 --- a/artemis/agents/explorer/explorer.py +++ b/artemis/agents/explorer/explorer.py @@ -19,7 +19,6 @@ import hashlib import json import os from pathlib import Path -import random import re import time from typing import Any, Literal @@ -27,7 +26,6 @@ from typing import Any, Literal import cv2 from google import genai from google.genai import types -from google.genai.errors import APIError import httpx from langchain_core.messages import ( BaseMessage, @@ -44,7 +42,13 @@ from artemis.context import ArtemisContext from artemis.data_engine.storage import StorageManager from artemis.data_engine.trace import TraceSpan, trace from artemis.graph.state import State -from artemis.services.llm import get_llm +from artemis.llm.reliability import ( + LLMExhaustedError, + LLMPermanentError, + classify_failure, + retry_policy_for, +) +from artemis.services.llm import _record_llm_event, _record_llm_retry, get_llm # Import diagnostic functions directly from artemis.agents.explorer.constants import EXPLORE_DESCRIPTIONS @@ -59,6 +63,84 @@ from artemis.utils.visualization import draw_dots, format_minimal_list_with_poin logger = get_logger(__name__) + +async def _generate_content_with_reliability(operation, *, label: str = "Explorer model call"): + """Run one native google-genai model call under the shared reliability layer. + + Retry decisions are owned by ``classify_failure``/``retry_policy_for`` + (artemis.llm.reliability) instead of a blanket exponential-backoff loop: + non-retryable categories (auth, bad request) are raised immediately as + ``LLMPermanentError`` and exhausted retryable categories surface as + ``LLMExhaustedError``. Only the retry/error-classification policy is + centralized here; the google-genai transport itself stays untouched. + ``operation`` must be a zero-argument callable returning a fresh awaitable + per attempt. + """ + attempt = 0 + while True: + try: + return await operation() + except Exception as call_err: + attempt += 1 + failure = classify_failure(call_err) + if not failure.retryable: + logger.error( + f"{label} permanently failed [{failure.category.value}]: {call_err}" + ) + _record_llm_event( + "llm_gave_up", + { + "source": "explorer", + "error": str(call_err)[:1000], + "category": failure.category.value, + "retryable": False, + }, + status="failed", + ) + raise LLMPermanentError( + f"{label} failed [{failure.category.value}]: {call_err}", + failure=failure, + cause=call_err, + ) from call_err + policy = retry_policy_for(failure.category) + if attempt >= policy.max_attempts: + logger.error( + f"{label} exhausted {attempt} attempt(s)" + f" [{failure.category.value}]: {call_err}" + ) + _record_llm_event( + "llm_gave_up", + { + "source": "explorer", + "error": str(call_err)[:1000], + "category": failure.category.value, + "retryable": True, + "attempts": attempt, + }, + status="failed", + ) + raise LLMExhaustedError( + f"{label} exhausted {attempt} attempt(s)" + f" [{failure.category.value}]: {call_err}", + failure=failure, + cause=call_err, + ) from call_err + delay = policy.delay_for(attempt) + logger.warning( + f"{label} failed [{failure.category.value}] on attempt" + f" {attempt}/{policy.max_attempts}: {call_err}." + f" Retrying in {delay:.2f}s..." + ) + _record_llm_retry( + str(call_err), + delay, + attempt=attempt, + max_retries=policy.max_attempts, + source="explorer", + ) + await asyncio.sleep(delay) + + UNIVERSAL_EXPLORER_TOOLS = [ { "type": "function", @@ -1802,41 +1884,16 @@ Initial marked UI elements list (corresponding to numbers ①, ②... in the ima ), ) - max_call_retries = 5 - for attempt in range(max_call_retries + 1): - try: - response = await asyncio.wait_for( - client.aio.models.generate_content( - model=model_name, - contents=contents, - config=generate_config, - ), - timeout=180, - ) - break - except APIError as api_err: - code = getattr(api_err, "code", None) - if code in [429, 503]: - if attempt >= max_call_retries: - raise api_err - backoff = (2.0**attempt) + random.uniform(0.1, 1.0) - logger.warning( - f"APIError {code} (overload/rate limit) on" - f" attempt {attempt + 1}. Retrying in" - f" {backoff:.2f}s..." - ) - await asyncio.sleep(backoff) - else: - raise api_err - except Exception as e: - if attempt >= max_call_retries: - raise e - backoff = (2.0**attempt) + random.uniform(0.1, 1.0) - logger.warning( - f"Unexpected error on attempt {attempt + 1}:" - f" {e}. Retrying in {backoff:.2f}s..." - ) - await asyncio.sleep(backoff) + response = await _generate_content_with_reliability( + lambda: asyncio.wait_for( + client.aio.models.generate_content( + model=model_name, + contents=contents, + config=generate_config, + ), + timeout=180, + ), + ) end_turn = time.perf_counter() turn_dur = end_turn - start_turn self.turn_latencies.append(turn_dur) diff --git a/artemis/agents/video_analyzer/video_analyzer.py b/artemis/agents/video_analyzer/video_analyzer.py index 6db7d27..13d6833 100644 --- a/artemis/agents/video_analyzer/video_analyzer.py +++ b/artemis/agents/video_analyzer/video_analyzer.py @@ -19,7 +19,6 @@ import glob import json import os from pathlib import Path -import random import re import shutil import tempfile @@ -28,7 +27,6 @@ from typing import Any from google import genai from google.genai import types -from google.genai.errors import APIError from langchain_core.messages import ( BaseMessage, HumanMessage, @@ -49,7 +47,8 @@ from artemis.constants import SAFETY_SETTINGS_BLOCK_NONE from artemis.context import ArtemisContext from artemis.controllers.controller_factory import get_controller from artemis.data_engine.trace import CURRENT_TRACE_ID, TraceSpan, trace -from artemis.services.llm import get_llm +from artemis.llm.reliability import retry_policy_for +from artemis.services.llm import _record_llm_event, _record_llm_retry, get_llm from artemis.utils.logger import get_logger from artemis.utils.video import ( compress_video_for_api, @@ -76,27 +75,50 @@ _CLEANUP_LOCK = asyncio.Lock() async def _invoke_with_retry(operation, label: str, max_attempts: int = 3): - """Retry side-effect-free model coordination calls on transient failures.""" - for attempt in range(1, max_attempts + 1): + """Retry side-effect-free model coordination calls using shared classification. + + Retry decisions come from ``classify_video_failure`` plus the shared + per-category policies in :mod:`artemis.llm.reliability` instead of ad-hoc + status-code sets. ``operation`` must be restartable from scratch: each + retry re-invokes it and any partially accumulated result (e.g. consumed + stream chunks) from the failed attempt is discarded, never surfaced as a + complete output. The original exception is re-raised on giving up so + downstream ``classify_video_failure`` consumers keep seeing the provider's + native error shape. + """ + attempt = 0 + while True: + attempt += 1 try: return await operation() - except APIError as exc: - code = getattr(exc, "code", None) - if code in {400, 401, 403, 404} or attempt >= max_attempts: + except Exception as exc: + failure = classify_video_failure(exc) + if not failure.retryable: + logger.warning( + f"{label} non-retryable failure [{failure.category.value}] " + f"on attempt {attempt}: {exc}" + ) raise - delay = (2.0 ** (attempt - 1)) + random.uniform(0.1, 0.8) + policy = retry_policy_for(failure.category) + allowed_attempts = min(max_attempts, policy.max_attempts) + if attempt >= allowed_attempts: + logger.warning( + f"{label} exhausted {attempt} attempt(s) " + f"[{failure.category.value}]: {exc}" + ) + raise + delay = policy.delay_for(attempt) logger.warning( - f"{label} transient API failure ({attempt}/{max_attempts}): {exc}; " - f"retrying in {delay:.2f}s" + f"{label} retryable failure [{failure.category.value}] " + f"({attempt}/{allowed_attempts}): {exc}; retrying from scratch " + f"in {delay:.2f}s (partial output from the failed attempt is discarded)" ) - await asyncio.sleep(delay) - except (TimeoutError, ConnectionError, OSError) as exc: - if attempt >= max_attempts: - raise - delay = (2.0 ** (attempt - 1)) + random.uniform(0.1, 0.8) - logger.warning( - f"{label} transient failure ({attempt}/{max_attempts}): {exc}; " - f"retrying in {delay:.2f}s" + _record_llm_retry( + str(exc), + delay, + attempt=attempt, + max_retries=allowed_attempts, + source=label, ) await asyncio.sleep(delay) @@ -1094,6 +1116,14 @@ class VideoAnalyzer: logger.warning( "Native video chunk exhausted retries; using universal fallback" ) + _record_llm_event( + "llm_fallback", + { + "reason": "native_video_chunk_failed", + "category": failure.category.value, + "error": str(error)[:500], + }, + ) fallback_result = await self._exec_single_chunk_universal( compressed_path=compressed_path, raw_path=path, @@ -1697,35 +1727,43 @@ class VideoAnalyzer: except Exception as ce: logger.error(f"Failed to delete cloud file {file.name}: {ce}") - except APIError as api_err: - logger.warning(f"Sub-agent APIError attempt {attempt + 1} failed: {api_err}") - code = getattr(api_err, "code", None) - if code == 429: - if attempt >= max_retries: - return await finish_failed(api_err) - - backoff = (2.0**attempt) + random.uniform(0.1, 1.0) - logger.info( - f"Rate limited (429). Backing off for {backoff:.2f}s" - " without expanding boundaries..." - ) - await asyncio.sleep(backoff) - continue - elif code in [400, 401, 403, 404]: - return await finish_failed(api_err, retryable=False) - - if attempt >= max_retries: - return await finish_failed(api_err) - backoff = 2.0**attempt - await asyncio.sleep(backoff) except Exception as e: - logger.warning(f"Sub-agent attempt {attempt + 1} failed: {e}") - if attempt >= max_retries: + # Retry decisions are owned by the shared classification layer. + # Every attempt restarts from scratch (segment re-extracted, + # file re-uploaded, stream re-consumed); partially accumulated + # stream output from the failed attempt is discarded, never + # surfaced as a complete answer. + failure = classify_video_failure(e) + attempts_so_far = attempt + 1 + if not failure.retryable: + logger.warning( + f"Sub-agent non-retryable failure [{failure.category.value}]" + f" on attempt {attempts_so_far}: {e}" + ) + return await finish_failed(e, retryable=False) + policy = retry_policy_for(failure.category) + allowed_attempts = min(max_retries + 1, policy.max_attempts) + if attempts_so_far >= allowed_attempts: + logger.warning( + f"Sub-agent exhausted {attempts_so_far} attempt(s)" + f" [{failure.category.value}]: {e}" + ) return await finish_failed(e) - - backoff = 2.0**attempt - logger.info(f"Backing off for {backoff}s before retry...") - await asyncio.sleep(backoff) + delay = policy.delay_for(attempts_so_far) + logger.warning( + f"Sub-agent retryable failure [{failure.category.value}]" + f" ({attempts_so_far}/{allowed_attempts}): {e}; retrying" + f" from scratch in {delay:.2f}s (partial stream output" + " from the failed attempt is discarded)" + ) + _record_llm_retry( + str(e), + delay, + attempt=attempts_so_far, + max_retries=allowed_attempts, + source="video_sub_agent", + ) + await asyncio.sleep(delay) raise RuntimeError( f"Video sub-agent exhausted retries for {current_start:.1f}s-{current_end:.1f}s" @@ -1803,6 +1841,18 @@ class VideoAnalyzer: and prompt_with_context is not None ): try: + logger.warning( + "Native audio analysis failed" + f" [{failure.category.value}]; using universal fallback" + ) + _record_llm_event( + "llm_fallback", + { + "reason": "native_audio_analysis_failed", + "category": failure.category.value, + "error": str(error)[:500], + }, + ) fallback_result = await self._exec_analyze_audio_universal( audio_path=audio_path, start_time=current_start, @@ -2135,35 +2185,40 @@ class VideoAnalyzer: except Exception as ce: logger.error(f"Failed to delete cloud file {file.name}: {ce}") - except APIError as api_err: - logger.warning(f"Audio APIError attempt {attempt + 1} failed: {api_err}") - code = getattr(api_err, "code", None) - if code == 429: - if attempt >= max_retries: - return await finish_failed_audio(api_err) - - backoff = (2.0**attempt) + random.uniform(0.1, 1.0) - logger.info( - f"Rate limited (429). Backing off for {backoff:.2f}s" - " without expanding boundaries..." - ) - await asyncio.sleep(backoff) - continue - elif code in [400, 401, 403, 404]: - return await finish_failed_audio(api_err, retryable=False) - - if attempt >= max_retries: - return await finish_failed_audio(api_err) - backoff = 2.0**attempt - await asyncio.sleep(backoff) except Exception as e: - logger.warning(f"Audio attempt {attempt + 1} failed: {e}") - if attempt >= max_retries: + # Same contract as the video sub-agent loop: classification + # owns the retry decision and every retry restarts the attempt + # from scratch. + failure = classify_video_failure(e) + attempts_so_far = attempt + 1 + if not failure.retryable: + logger.warning( + f"Audio non-retryable failure [{failure.category.value}]" + f" on attempt {attempts_so_far}: {e}" + ) + return await finish_failed_audio(e, retryable=False) + policy = retry_policy_for(failure.category) + allowed_attempts = min(max_retries + 1, policy.max_attempts) + if attempts_so_far >= allowed_attempts: + logger.warning( + f"Audio analysis exhausted {attempts_so_far} attempt(s)" + f" [{failure.category.value}]: {e}" + ) return await finish_failed_audio(e) - - backoff = 2.0**attempt - logger.info(f"Backing off for {backoff}s before retry...") - await asyncio.sleep(backoff) + delay = policy.delay_for(attempts_so_far) + logger.warning( + f"Audio retryable failure [{failure.category.value}]" + f" ({attempts_so_far}/{allowed_attempts}): {e}; retrying" + f" from scratch in {delay:.2f}s" + ) + _record_llm_retry( + str(e), + delay, + attempt=attempts_so_far, + max_retries=allowed_attempts, + source="video_audio_agent", + ) + await asyncio.sleep(delay) async def _exec_single_chunk_universal( self, diff --git a/artemis/llm/reliability.py b/artemis/llm/reliability.py index beef7cf..c35ba6b 100644 --- a/artemis/llm/reliability.py +++ b/artemis/llm/reliability.py @@ -77,6 +77,9 @@ def extract_status_code(error: BaseException) -> int | None: _RATE_LIMIT_MARKERS = ("rate limit", "quota", "resource_exhausted") _UNAVAILABLE_MARKERS = ("unavailable", "overloaded", "high demand", "service unavailable") _AUTH_MARKERS = ("unauthorized", "forbidden", "api key", "credential") +# "deadline exceeded"/"deadline_exceeded" cover google-genai SDK timeouts that +# surface without an HTTP status code attached. +_TIMEOUT_MARKERS = ("timed out", "timeout", "deadline exceeded", "deadline_exceeded") def classify_failure(error: BaseException) -> Failure: @@ -91,7 +94,7 @@ def classify_failure(error: BaseException) -> Failure: return Failure(FailureCategory.RATE_LIMIT, True, True) if code in {500, 502, 503, 504} or any(marker in message for marker in _UNAVAILABLE_MARKERS): return Failure(FailureCategory.PROVIDER_UNAVAILABLE, True, True) - if isinstance(error, TimeoutError) or "timed out" in message or "timeout" in message: + if isinstance(error, TimeoutError) or any(marker in message for marker in _TIMEOUT_MARKERS): return Failure(FailureCategory.TIMEOUT, True, True) if isinstance(error, (ConnectionError, OSError)): return Failure(FailureCategory.CONNECTION, True, True) @@ -163,7 +166,19 @@ _DEFAULT_RETRY_POLICIES: dict[FailureCategory, RetryPolicy] = { } -def retry_policy_for(category: FailureCategory) -> RetryPolicy: +def retry_policy_for(category: FailureCategory | str) -> RetryPolicy: + """Look up the retry policy for a failure category. + + Accepts the generic ``FailureCategory`` or any string-valued category from + a domain extension (e.g. the video pipeline's ``VideoFailureCategory``). + Extension-only categories fall back to the UNKNOWN policy, so callers must + gate on ``failure.retryable`` before consulting attempt counts. + """ + if not isinstance(category, FailureCategory): + try: + category = FailureCategory(str(category)) + except ValueError: + category = FailureCategory.UNKNOWN return _DEFAULT_RETRY_POLICIES.get(category, _DEFAULT_RETRY_POLICIES[FailureCategory.UNKNOWN]) diff --git a/tests/unit/agents/test_explorer.py b/tests/unit/agents/test_explorer.py index c00b91e..20e983c 100644 --- a/tests/unit/agents/test_explorer.py +++ b/tests/unit/agents/test_explorer.py @@ -1272,3 +1272,93 @@ def test_explorer_prune_historical_images(): assert contents[0].parts[1].text == "Some text" # Second (last) image part should be kept intact assert contents[1].parts[0] is part2 + + +class _FakeGenaiError(Exception): + """Mimics google.genai.errors.APIError's shape (a ``code`` attribute).""" + + def __init__(self, message: str, code: int | None = None): + super().__init__(message) + self.code = code + + +@pytest.mark.asyncio +async def test_generate_content_reliability_permanent_error_is_not_retried(): + from artemis.agents.explorer.explorer import _generate_content_with_reliability + from artemis.llm.reliability import LLMPermanentError + + calls = 0 + + async def op(): + nonlocal calls + calls += 1 + raise _FakeGenaiError("PERMISSION_DENIED: forbidden", code=403) + + with pytest.raises(LLMPermanentError) as exc_info: + await _generate_content_with_reliability(op) + assert calls == 1 + assert exc_info.value.failure.category.value == "authentication" + assert isinstance(exc_info.value.__cause__, _FakeGenaiError) + + +@pytest.mark.asyncio +async def test_generate_content_reliability_bad_request_is_not_retried(): + from artemis.agents.explorer.explorer import _generate_content_with_reliability + from artemis.llm.reliability import LLMPermanentError + + calls = 0 + + async def op(): + nonlocal calls + calls += 1 + raise _FakeGenaiError("invalid request payload", code=400) + + with pytest.raises(LLMPermanentError) as exc_info: + await _generate_content_with_reliability(op) + assert calls == 1 + assert exc_info.value.failure.category.value == "bad_request" + + +@pytest.mark.asyncio +async def test_generate_content_reliability_retries_transient_then_succeeds(): + from artemis.agents.explorer import explorer as explorer_module + + calls = 0 + sentinel = object() + + async def op(): + nonlocal calls + calls += 1 + if calls < 2: + raise _FakeGenaiError("429 rate limit exceeded", code=429) + return sentinel + + with patch.object(explorer_module.asyncio, "sleep", new=AsyncMock()) as sleep_mock: + result = await explorer_module._generate_content_with_reliability(op) + assert result is sentinel + assert calls == 2 + assert sleep_mock.await_count == 1 + + +@pytest.mark.asyncio +async def test_generate_content_reliability_exhausts_to_typed_error(): + from artemis.agents.explorer import explorer as explorer_module + from artemis.llm.reliability import ( + FailureCategory, + LLMExhaustedError, + retry_policy_for, + ) + + calls = 0 + + async def op(): + nonlocal calls + calls += 1 + raise _FakeGenaiError("503 service unavailable", code=503) + + with patch.object(explorer_module.asyncio, "sleep", new=AsyncMock()): + with pytest.raises(LLMExhaustedError) as exc_info: + await explorer_module._generate_content_with_reliability(op) + assert calls == retry_policy_for(FailureCategory.PROVIDER_UNAVAILABLE).max_attempts + assert exc_info.value.failure.category.value == "provider_unavailable" + assert isinstance(exc_info.value.__cause__, _FakeGenaiError) diff --git a/tests/unit/agents/test_video_reliability.py b/tests/unit/agents/test_video_reliability.py index 7dca85e..ef16dc7 100644 --- a/tests/unit/agents/test_video_reliability.py +++ b/tests/unit/agents/test_video_reliability.py @@ -191,3 +191,104 @@ def test_action_timestamps_drive_dense_sampling(): assert offsets[0] == 2.0 assert 4.0 in offsets assert offsets[-1] == 6.0 + + +class _FakeGenaiError(Exception): + """Mimics google.genai.errors.APIError's shape (a ``code`` attribute).""" + + def __init__(self, message: str, code: int | None = None): + super().__init__(message) + self.code = code + + +@pytest.mark.asyncio +async def test_invoke_with_retry_raises_non_retryable_immediately(): + from artemis.agents.video_analyzer.video_analyzer import _invoke_with_retry + + calls = 0 + + async def op(): + nonlocal calls + calls += 1 + raise _FakeGenaiError("invalid api key", code=401) + + with pytest.raises(_FakeGenaiError): + await _invoke_with_retry(op, "test-op") + assert calls == 1 + + +@pytest.mark.asyncio +async def test_invoke_with_retry_bad_request_never_retried(): + from artemis.agents.video_analyzer.video_analyzer import _invoke_with_retry + + calls = 0 + + async def op(): + nonlocal calls + calls += 1 + raise _FakeGenaiError("invalid request payload", code=400) + + with pytest.raises(_FakeGenaiError): + await _invoke_with_retry(op, "test-op") + assert calls == 1 + + +@pytest.mark.asyncio +async def test_invoke_with_retry_restarts_from_scratch_on_transient_failure(): + from artemis.agents.video_analyzer import video_analyzer as va + + calls = 0 + + async def op(): + nonlocal calls + calls += 1 + accumulated = "partial-" # per-attempt accumulator, like a stream loop + if calls < 2: + raise _FakeGenaiError("503 service unavailable", code=503) + return accumulated + "complete" + + with patch.object(va.asyncio, "sleep", new=AsyncMock()) as sleep_mock: + result = await va._invoke_with_retry(op, "test-op") + + # The successful attempt produced the whole output; the failed attempt's + # partial accumulation never leaked into the result. + assert result == "partial-complete" + assert calls == 2 + assert sleep_mock.await_count == 1 + + +@pytest.mark.asyncio +async def test_invoke_with_retry_exhaustion_reraises_original_error(): + from artemis.agents.video_analyzer import video_analyzer as va + + calls = 0 + + async def op(): + nonlocal calls + calls += 1 + raise _FakeGenaiError("503 service unavailable", code=503) + + with patch.object(va.asyncio, "sleep", new=AsyncMock()): + with pytest.raises(_FakeGenaiError): + await va._invoke_with_retry(op, "test-op", max_attempts=2) + assert calls == 2 + + +@pytest.mark.asyncio +async def test_invoke_with_retry_caps_unknown_failures_by_policy(): + from artemis.llm.reliability import FailureCategory, retry_policy_for + + from artemis.agents.video_analyzer import video_analyzer as va + + calls = 0 + + async def op(): + nonlocal calls + calls += 1 + raise RuntimeError("something odd") + + with patch.object(va.asyncio, "sleep", new=AsyncMock()): + with pytest.raises(RuntimeError): + await va._invoke_with_retry(op, "test-op", max_attempts=5) + # UNKNOWN failures follow the shared policy, not the caller's outer cap. + assert calls == retry_policy_for(FailureCategory.UNKNOWN).max_attempts diff --git a/tests/unit/test_llm_reliability.py b/tests/unit/test_llm_reliability.py index c57970b..4cfcf12 100644 --- a/tests/unit/test_llm_reliability.py +++ b/tests/unit/test_llm_reliability.py @@ -66,3 +66,31 @@ def test_breaker_only_trips_on_transient_categories(): assert not breaker.allow("m", now=13.6) breaker.record_success("m") assert breaker.allow("m", now=13.7) + + +def test_retry_policy_accepts_extension_category_values(): + from enum import StrEnum + + class ExtCategory(StrEnum): + RATE_LIMIT = "rate_limit" + MEDIA_PROCESSING = "media_processing" + + assert retry_policy_for(ExtCategory.RATE_LIMIT) is retry_policy_for( + FailureCategory.RATE_LIMIT + ) + assert retry_policy_for("provider_unavailable") is retry_policy_for( + FailureCategory.PROVIDER_UNAVAILABLE + ) + # Extension-only categories fall back to the UNKNOWN policy; callers gate + # on failure.retryable before consulting attempt counts. + assert retry_policy_for(ExtCategory.MEDIA_PROCESSING) is retry_policy_for( + FailureCategory.UNKNOWN + ) + + +def test_classifier_recognizes_genai_deadline_exceeded_shapes(): + # google-genai SDK timeouts can surface without a status code attached. + for message in ("DEADLINE_EXCEEDED", "deadline exceeded while awaiting response"): + failure = classify_failure(RuntimeError(message)) + assert failure.category is FailureCategory.TIMEOUT + assert failure.retryable