13274243a0
Keep K3 suite selection and report-schema scoring in bash, merge K3/vision dataset_args into dpv4 yamls, and pin EvalScope at 735d920ee911 with local patches. Co-authored-by: Cursor <cursoragent@cursor.com>
59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
from evalscope.api.messages.request_stats import RequestStats, is_client_error
|
|
from evalscope.utils.function_utils import retry_call
|
|
|
|
|
|
class _FakeHTTPError(Exception):
|
|
|
|
def __init__(self, code=None, message=''):
|
|
super().__init__(message)
|
|
self.code = code
|
|
self.message = message
|
|
|
|
|
|
def test_is_client_error_context_length_and_content_filter():
|
|
assert is_client_error(_FakeHTTPError(code='context_length_exceeded', message='too long'))
|
|
assert is_client_error(_FakeHTTPError(code='content_filter', message='blocked'))
|
|
assert is_client_error(_FakeHTTPError(message='prompt is too long for the context window'))
|
|
assert not is_client_error(_FakeHTTPError(code='internal_error', message='upstream 500'))
|
|
assert not is_client_error(TimeoutError('timed out'))
|
|
|
|
|
|
def test_request_stats_success_rate_excludes_client_errors():
|
|
stats = RequestStats()
|
|
stats.on_attempt(True, None)
|
|
stats.on_attempt(True, None)
|
|
stats.on_attempt(False, TimeoutError('timeout'))
|
|
stats.on_attempt(False, _FakeHTTPError(code='context_length_exceeded', message='overflow'))
|
|
stats.record_logical(True)
|
|
stats.record_logical(True)
|
|
stats.record_logical(False)
|
|
|
|
snap = stats.snapshot()
|
|
assert snap['total_attempts'] == 4
|
|
assert snap['success_attempts'] == 2
|
|
assert snap['failed_attempts'] == 1
|
|
assert snap['client_errors'] == 1
|
|
# denominator is HTTP attempts that are vendor-channel (success + failed)
|
|
assert snap['success_rate'] == 0.6667
|
|
assert snap['logical_calls'] == 3
|
|
assert snap['logical_success'] == 2
|
|
assert snap['client_error_types'].get('_FakeHTTPError') == 1
|
|
|
|
|
|
def test_retry_call_on_attempt_counts_retries():
|
|
stats = RequestStats()
|
|
calls = {'n': 0}
|
|
|
|
def flaky():
|
|
calls['n'] += 1
|
|
if calls['n'] < 3:
|
|
raise TimeoutError('again')
|
|
return 'ok'
|
|
|
|
assert retry_call(flaky, retries=3, sleep_interval=0, on_attempt=stats.on_attempt) == 'ok'
|
|
snap = stats.snapshot()
|
|
assert snap['total_attempts'] == 3
|
|
assert snap['success_attempts'] == 1
|
|
assert snap['failed_attempts'] == 2
|
|
assert snap['success_rate'] == round(1 / 3, 4)
|