fix(models): correct retry semantics and Anthropic streaming

evalscope/models/anthropic_compatible.py

@@ -1,8 +1,17 @@
 import os
 import time
-from typing import Any, Dict, List, Optional, Tuple, Union
-
-from anthropic import Anthropic, APIStatusError, AsyncAnthropic, BadRequestError, PermissionDeniedError
+from typing import Any, Dict, List, Optional, Tuple, Type, Union
+
+from anthropic import (
+    Anthropic,
+    APIStatusError,
+    AsyncAnthropic,
+    AuthenticationError,
+    BadRequestError,
+    NotFoundError,
+    PermissionDeniedError,
+    UnprocessableEntityError,
+)
 from anthropic.types import Message
 
 from evalscope.api.messages import ChatMessage
@@ -28,6 +37,18 @@
 
 logger = get_logger()
 
+# Client errors (prompt too long, invalid parameters, bad credentials, ...) are not
+# recoverable by retrying. generate()/generate_async() degrade them gracefully via
+# handle_bad_request(), so retry_call must let them through immediately instead of burning
+# retries * retry_interval on every failing sample.
+NON_RETRYABLE_ANTHROPIC_ERRORS: Tuple[Type[Exception], ...] = (
+    BadRequestError,
+    AuthenticationError,
+    PermissionDeniedError,
+    NotFoundError,
+    UnprocessableEntityError,
+)
+
 
 class AnthropicCompatibleAPI(ModelAPI):
     """Anthropic API compatible model implementation.
@@ -155,18 +176,23 @@ def generate(
             t_start = time.monotonic()
             ttft: Optional[float] = None
 
-            # Generate completion
-            message = retry_call(
-                self.client.messages.create,
+            # A streaming request is not complete when create() returns: the
+            # stream may still fail while its events are being consumed. Retry
+            # the whole request so a partial response is discarded and replaced
+            # by one complete response (mirrors OpenAICompatibleAPI).
+            def _create_and_collect() -> Tuple[Message, Optional[float]]:
+                message = self.client.messages.create(**request)
+                if isinstance(message, Message):
+                    return message, None
+                return collect_stream_response(message, request_start=t_start)
+
+            message, ttft = retry_call(
+                _create_and_collect,
                 retries=config.retries,
                 sleep_interval=config.retry_interval,
-                **request,
+                no_retry_exceptions=NON_RETRYABLE_ANTHROPIC_ERRORS,
             )
 
-            # Handle streaming response
-            if not isinstance(message, Message):
-                message, ttft = collect_stream_response(message, request_start=t_start)
-
             total_time = time.monotonic() - t_start
 
             response = message.model_dump()
@@ -185,7 +211,7 @@ def generate(
             )
             return output
 
-        except (BadRequestError, PermissionDeniedError) as ex:
+        except NON_RETRYABLE_ANTHROPIC_ERRORS as ex:
             return self.handle_bad_request(ex)
 
     async def generate_async(
@@ -239,18 +265,22 @@ async def generate_async(
             t_start = time.monotonic()
             ttft: Optional[float] = None
 
-            # Async generation with retry
-            message = await async_retry_call(
-                self.async_client.messages.create,
+            # Keep stream consumption inside the retry boundary. If an async
+            # stream is interrupted, start a fresh request rather than
+            # returning or persisting its partial response.
+            async def _create_and_collect() -> Tuple[Message, Optional[float]]:
+                message = await self.async_client.messages.create(**request)
+                if isinstance(message, Message):
+                    return message, None
+                return await async_collect_stream_response(message, request_start=t_start)
+
+            message, ttft = await async_retry_call(
+                _create_and_collect,
                 retries=config.retries,
                 sleep_interval=config.retry_interval,
-                **request,
+                no_retry_exceptions=NON_RETRYABLE_ANTHROPIC_ERRORS,
             )
 
-            # Handle streaming response
-            if not isinstance(message, Message):
-                message, ttft = await async_collect_stream_response(message, request_start=t_start)
-
             total_time = time.monotonic() - t_start
 
             response = message.model_dump()
@@ -268,7 +298,7 @@ async def generate_async(
             )
             return output
 
-        except (BadRequestError, PermissionDeniedError) as ex:
+        except NON_RETRYABLE_ANTHROPIC_ERRORS as ex:
             return self.handle_bad_request(ex)
 
     def resolve_tools(self, tools: List[ToolInfo], tool_choice: ToolChoice,

evalscope/models/litellm_compatible.py

@@ -9,7 +9,7 @@
 """
 
 import time
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Type
 
 from openai.types.chat import ChatCompletion
 
@@ -35,6 +35,26 @@
 logger = get_logger()
 
 
+def _litellm_no_retry_exceptions() -> Tuple[Type[Exception], ...]:
+    """LiteLLM 4xx client-error types that must not be retried.
+
+    LiteLLM maps provider client errors (context length exceeded, bad credentials, unknown
+    model, invalid parameters, ...) onto its own exception types. Retrying them only burns
+    retries * retry_interval per failing sample, so pass them as no-retry exceptions.
+    Transient errors (timeouts, connection failures, 429/5xx) stay retryable.
+    The litellm import stays lazy to keep this module importable without litellm.
+    """
+    from litellm.exceptions import (
+        AuthenticationError,
+        BadRequestError,
+        NotFoundError,
+        PermissionDeniedError,
+        UnprocessableEntityError,
+    )
+
+    return (AuthenticationError, BadRequestError, NotFoundError, PermissionDeniedError, UnprocessableEntityError)
+
+
 class LiteLLMAPI(ModelAPI):
     """LiteLLM model API provider.
 
@@ -96,6 +116,7 @@ def generate(
                 litellm.completion,
                 retries=config.retries,
                 sleep_interval=config.retry_interval,
+                no_retry_exceptions=_litellm_no_retry_exceptions(),
                 **request,
             )
 
@@ -167,6 +188,7 @@ async def generate_async(
                 litellm.acompletion,
                 retries=config.retries,
                 sleep_interval=config.retry_interval,
+                no_retry_exceptions=_litellm_no_retry_exceptions(),
                 **request,
             )
 

evalscope/models/openai_compatible.py

@@ -1,8 +1,17 @@
 import os
 import time
-from typing import Any, Dict, List, Optional, Tuple, Union
-
-from openai import APIStatusError, AsyncOpenAI, BadRequestError, OpenAI, PermissionDeniedError, UnprocessableEntityError
+from typing import Any, Dict, List, Optional, Tuple, Type, Union
+
+from openai import (
+    APIStatusError,
+    AsyncOpenAI,
+    AuthenticationError,
+    BadRequestError,
+    NotFoundError,
+    OpenAI,
+    PermissionDeniedError,
+    UnprocessableEntityError,
+)
 from openai._types import NOT_GIVEN
 from openai.types.chat import ChatCompletion
 
@@ -29,6 +38,18 @@
 
 logger = get_logger()
 
+# Client errors (context length exceeded, invalid parameters, bad credentials, ...) are not
+# recoverable by retrying. generate()/generate_async() degrade them gracefully via
+# handle_bad_request(), so retry_call must let them through immediately instead of burning
+# retries * retry_interval on every failing sample.
+NON_RETRYABLE_OPENAI_ERRORS: Tuple[Type[Exception], ...] = (
+    BadRequestError,
+    AuthenticationError,
+    PermissionDeniedError,
+    NotFoundError,
+    UnprocessableEntityError,
+)
+
 
 class OpenAICompatibleAPI(ModelAPI):
 
@@ -136,6 +157,7 @@ def _create_and_collect() -> Tuple[ChatCompletion, Optional[float]]:
                 _create_and_collect,
                 retries=config.retries,
                 sleep_interval=config.retry_interval,
+                no_retry_exceptions=NON_RETRYABLE_OPENAI_ERRORS,
             )
 
             total_time = time.monotonic() - t_start
@@ -158,7 +180,7 @@ def _create_and_collect() -> Tuple[ChatCompletion, Optional[float]]:
             )
             return output
 
-        except (BadRequestError, UnprocessableEntityError, PermissionDeniedError) as ex:
+        except NON_RETRYABLE_OPENAI_ERRORS as ex:
             return self.handle_bad_request(ex)
         except ValueError as ex:
             logger.error(f'Model [{self.model_name}] returned an invalid response: {ex}')
@@ -214,6 +236,7 @@ async def _create_and_collect() -> Tuple[ChatCompletion, Optional[float]]:
                 _create_and_collect,
                 retries=config.retries,
                 sleep_interval=config.retry_interval,
+                no_retry_exceptions=NON_RETRYABLE_OPENAI_ERRORS,
             )
 
             total_time = time.monotonic() - t_start
@@ -235,7 +258,7 @@ async def _create_and_collect() -> Tuple[ChatCompletion, Optional[float]]:
             )
             return output
 
-        except (BadRequestError, UnprocessableEntityError, PermissionDeniedError) as ex:
+        except NON_RETRYABLE_OPENAI_ERRORS as ex:
             return self.handle_bad_request(ex)
         except ValueError as ex:
             logger.error(f'Model [{self.model_name}] returned an invalid response: {ex}')

evalscope/models/openai_responses.py

@@ -2,7 +2,7 @@
 import time
 from typing import Any, Dict, List, Optional, Tuple, Union
 
-from openai import APIStatusError, BadRequestError, PermissionDeniedError, UnprocessableEntityError
+from openai import APIStatusError
 from openai._types import NOT_GIVEN
 
 from evalscope.api.messages import ChatMessage
@@ -13,7 +13,7 @@
 from evalscope.utils.argument_utils import get_supported_params
 from evalscope.utils.function_utils import async_retry_call, retry_call
 
-from .openai_compatible import OpenAICompatibleAPI
+from .openai_compatible import NON_RETRYABLE_OPENAI_ERRORS, OpenAICompatibleAPI
 from .utils.openai import openai_handle_bad_request
 from .utils.openai_responses import (
     async_collect_response_stream,
@@ -84,6 +84,7 @@ def generate(
                 self.client.responses.create,
                 retries=config.retries,
                 sleep_interval=config.retry_interval,
+                no_retry_exceptions=NON_RETRYABLE_OPENAI_ERRORS,
                 **request,
             )
             if not self._is_response_object(response):
@@ -92,7 +93,7 @@ def generate(
             total_time = time.monotonic() - t_start
             return self._build_output(response, tools, total_time, ttft)
 
-        except (BadRequestError, UnprocessableEntityError, PermissionDeniedError) as ex:
+        except NON_RETRYABLE_OPENAI_ERRORS as ex:
             return self.handle_bad_request(ex)
         except ValueError as ex:
             logger.error(f'Model [{self.model_name}] returned an invalid response: {ex}')
@@ -115,6 +116,7 @@ async def generate_async(
                 self.async_client.responses.create,
                 retries=config.retries,
                 sleep_interval=config.retry_interval,
+                no_retry_exceptions=NON_RETRYABLE_OPENAI_ERRORS,
                 **request,
             )
             if not self._is_response_object(response):
@@ -123,7 +125,7 @@ async def generate_async(
             total_time = time.monotonic() - t_start
             return self._build_output(response, tools, total_time, ttft)
 
-        except (BadRequestError, UnprocessableEntityError, PermissionDeniedError) as ex:
+        except NON_RETRYABLE_OPENAI_ERRORS as ex:
             return self.handle_bad_request(ex)
         except ValueError as ex:
             logger.error(f'Model [{self.model_name}] returned an invalid response: {ex}')

evalscope/utils/function_utils.py

@@ -3,7 +3,7 @@
 import time
 from concurrent.futures import Future, ThreadPoolExecutor, wait
 from functools import wraps
-from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, TypeVar, Union
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, Tuple, Type, TypeVar, Union
 
 from evalscope.utils import asyncio_runtime
 from evalscope.utils.logger import get_logger
@@ -68,31 +68,65 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-def retry_call(func, *args, retries=3, sleep_interval=0, **kwargs):
-    """Function that retries a function call up to `retries` times if an exception occurs."""
-    for attempt in range(retries):
+def retry_call(
+    func: Callable[..., T],
+    *args,
+    retries: Optional[int] = 3,
+    sleep_interval: float = 0,
+    no_retry_exceptions: Tuple[Type[Exception], ...] = (),
+    **kwargs,
+) -> T:
+    """Function that retries a function call up to `retries` times if an exception occurs.
+
+    `retries` is the total number of attempts: `None` falls back to the default and values
+    below 1 are clamped to 1, so the wrapped call is always made at least once.
+
+    Exceptions matching `no_retry_exceptions` are re-raised immediately instead of being
+    retried (e.g. non-retryable 4xx client errors handled by the caller itself).
+    """
+    total_attempts = 3 if retries is None else max(1, retries)
+    for attempt in range(total_attempts):
         try:
             return func(*args, **kwargs)
+        except no_retry_exceptions:
+            # Non-retryable errors must reach the caller's own handler without delay.
+            raise
         except Exception as e:
-            if attempt < retries - 1:
+            if attempt < total_attempts - 1:
                 if sleep_interval > 0:
-                    logger.warning(f'Attempt {attempt + 1} / {retries} failed: {e}. Retrying...')
+                    logger.warning(f'Attempt {attempt + 1} / {total_attempts} failed: {e}. Retrying...')
                     time.sleep(sleep_interval)
             else:
                 raise
 
 
 async def async_retry_call(
-    func: Callable[..., Awaitable[T]], *args, retries: int = 3, sleep_interval: float = 0, **kwargs
+    func: Callable[..., Awaitable[T]],
+    *args,
+    retries: Optional[int] = 3,
+    sleep_interval: float = 0,
+    no_retry_exceptions: Tuple[Type[Exception], ...] = (),
+    **kwargs,
 ) -> T:
-    """Async version of retry_call. Retries an async function call up to `retries` times if an exception occurs."""
-    for attempt in range(retries):
+    """Async version of retry_call. Retries an async function call up to `retries` times if an exception occurs.
+
+    `retries` is the total number of attempts: `None` falls back to the default and values
+    below 1 are clamped to 1, so the wrapped call is always made at least once.
+
+    Exceptions matching `no_retry_exceptions` are re-raised immediately instead of being
+    retried (e.g. non-retryable 4xx client errors handled by the caller itself).
+    """
+    total_attempts = 3 if retries is None else max(1, retries)
+    for attempt in range(total_attempts):
         try:
             return await func(*args, **kwargs)
+        except no_retry_exceptions:
+            # Non-retryable errors must reach the caller's own handler without delay.
+            raise
         except Exception as e:
-            if attempt < retries - 1:
+            if attempt < total_attempts - 1:
                 if sleep_interval > 0:
-                    logger.warning(f'Attempt {attempt + 1} / {retries} failed: {e}. Retrying...')
+                    logger.warning(f'Attempt {attempt + 1} / {total_attempts} failed: {e}. Retrying...')
                     await asyncio.sleep(sleep_interval)
             else:
                 raise