From 96809fa5c8ce291a2fc57ea76813e201cb8d61b0 Mon Sep 17 00:00:00 2001 From: wuyang6 Date: Fri, 24 Jul 2026 17:51:10 +0800 Subject: [PATCH] fix: back off evaluation rate limits --- src/evaluation_runtime.py | 156 +++++++++++++++++++++++++------ tests/test_evaluation_runtime.py | 62 ++++++++++++ 2 files changed, 190 insertions(+), 28 deletions(-) diff --git a/src/evaluation_runtime.py b/src/evaluation_runtime.py index 85ce52f..20b1489 100644 --- a/src/evaluation_runtime.py +++ b/src/evaluation_runtime.py @@ -22,6 +22,7 @@ from src.agent_runtime import LocalCodingAgent from src.agent_tools import default_tool_registry from src.agent_types import ( AgentPermissions, + AgentRunResult, AgentRuntimeConfig, ModelConfig, OutputSchemaConfig, @@ -36,6 +37,7 @@ DEFAULT_CONCURRENCY = 8 MAX_CONCURRENCY = 32 MAX_DATASET_ROWS = 20_000 MAX_LABEL_MAPPING_VALUES = 200 +TRANSIENT_MODEL_RETRY_DELAYS_SECONDS = (5.0, 15.0, 30.0, 60.0) VISIBLE_SKILL_FILE_SUFFIXES = {'.json', '.md', '.py', '.txt', '.yaml', '.yml'} EDITABLE_SKILL_FILE_SUFFIXES = {'.json', '.md'} MAX_SKILL_FILE_BYTES = 512 * 1024 @@ -349,6 +351,8 @@ class EvaluationRuntime: thread_name_prefix='evaluation-case', ) self._lock = threading.RLock() + self._model_cooldown_lock = threading.RLock() + self._model_cooldown_until = 0.0 self._controls: dict[str, ExperimentControl] = {} self._scheduler_futures: dict[str, Future[None]] = {} @@ -565,11 +569,16 @@ class EvaluationRuntime: snapshot_root=snapshot_root, raw_labels=raw_labels, ) - run_result = agent.run( + run_result = self._run_agent_with_transient_backoff( + agent, prompt, session_id=f'{run_id}_{case_row_id}', runtime_context=runtime_context, + cancel_event=cancel_event, ) + transient_error = transient_model_error_message(run_result.final_output) + if transient_error: + raise EvaluationError(transient_error) result = normalize_label_mapping_output( run_result.final_output, raw_labels=raw_labels, @@ -1607,6 +1616,75 @@ class EvaluationRuntime: with self._lock: self._controls.pop(experiment_id, None) + def _run_agent_with_transient_backoff( + self, + agent: LocalCodingAgent, + prompt: str, + *, + session_id: str, + runtime_context: str, + cancel_event: threading.Event, + ) -> AgentRunResult: + last_result: AgentRunResult | None = None + attempts = len(TRANSIENT_MODEL_RETRY_DELAYS_SECONDS) + 1 + for attempt in range(attempts): + self._wait_for_model_cooldown(cancel_event) + try: + result = agent.run( + prompt, + session_id=session_id, + runtime_context=runtime_context, + ) + except Exception as exc: + transient_error = transient_model_error_message(str(exc)) + if not transient_error or attempt >= attempts - 1: + if transient_error: + raise EvaluationError(transient_error) from exc + raise + self._extend_model_cooldown( + TRANSIENT_MODEL_RETRY_DELAYS_SECONDS[attempt], + session_id=session_id, + ) + continue + last_result = result + if not transient_model_error_message(result.final_output): + return result + if attempt >= attempts - 1: + return result + self._extend_model_cooldown( + TRANSIENT_MODEL_RETRY_DELAYS_SECONDS[attempt], + session_id=session_id, + ) + assert last_result is not None + return last_result + + def _wait_for_model_cooldown( + self, + cancel_event: threading.Event, + ) -> None: + while True: + if cancel_event.is_set(): + raise EvaluationError('评测任务已取消') + with self._model_cooldown_lock: + remaining = self._model_cooldown_until - time.monotonic() + if remaining <= 0: + return + cancel_event.wait(min(remaining, 1.0)) + + def _extend_model_cooldown( + self, + delay_seconds: float, + *, + session_id: str, + ) -> None: + jitter = (sum(session_id.encode('utf-8')) % 3000) / 1000 + cooldown_until = time.monotonic() + delay_seconds + jitter + with self._model_cooldown_lock: + self._model_cooldown_until = max( + self._model_cooldown_until, + cooldown_until, + ) + def _execute_case_analysis( self, *, @@ -1642,11 +1720,18 @@ class EvaluationRuntime: canonical_case=canonical, ) evaluation_session_id = f'eval_{run_id}_{case_row_id}' - run_result = agent.run( + run_result = self._run_agent_with_transient_backoff( + agent, prompt, session_id=evaluation_session_id, runtime_context=runtime_context, + cancel_event=cancel_event, ) + transient_error = transient_model_error_message( + run_result.final_output + ) + if transient_error: + raise EvaluationError(transient_error) accessed_refs = extract_accessed_refs( run_result.transcript, snapshot_root=snapshot_skill_dir, @@ -1666,7 +1751,8 @@ class EvaluationRuntime: f"prediction `{normalized['prediction']}` 不在 Skill 标签目录中" ) if retry_reason: - run_result = agent.run( + run_result = self._run_agent_with_transient_backoff( + agent, build_evidence_retry_prompt( skill.name, retry_reason, @@ -1674,7 +1760,13 @@ class EvaluationRuntime: ), session_id=evaluation_session_id, runtime_context=runtime_context, + cancel_event=cancel_event, ) + transient_error = transient_model_error_message( + run_result.final_output + ) + if transient_error: + raise EvaluationError(transient_error) accessed_refs = extract_accessed_refs( run_result.transcript, snapshot_root=snapshot_skill_dir, @@ -2812,6 +2904,23 @@ def try_normalize_evaluation_output(raw_output: str) -> dict[str, Any] | None: return None +def transient_model_error_message(value: Any) -> str: + text = str(value or '') + normalized = text.lower() + if 'http 429' in normalized or 'too many requests' in normalized: + return ( + '模型服务持续限流（HTTP 429 Too many requests），' + '已自动退避重试，请稍后重试失败项' + ) + for status in ('http 502', 'http 503', 'http 504'): + if status in normalized: + return ( + f'模型服务暂时不可用（{status.upper()}），' + '已自动退避重试，请稍后重试失败项' + ) + return '' + + def prediction_is_allowed( skill_name: str, prediction: str, @@ -2880,22 +2989,10 @@ def normalize_label_master_result( def normalize_evaluation_output(raw_output: str) -> dict[str, Any]: - text = raw_output.strip() - if text.startswith('```'): - text = re.sub(r'^```(?:json)?\s*', '', text) - text = re.sub(r'\s*```$', '', text) - try: - payload = json.loads(text) - except json.JSONDecodeError: - match = re.search(r'\{.*\}', text, re.DOTALL) - if not match: - raise EvaluationError('模型没有返回合法 JSON') - try: - payload = json.loads(match.group(0)) - except json.JSONDecodeError as exc: - raise EvaluationError('模型返回的评测结果不是合法 JSON') from exc - if not isinstance(payload, dict): - raise EvaluationError('模型评测结果必须是 JSON 对象') + payload = parse_json_object_output( + raw_output, + error_message='模型返回的评测结果不是合法 JSON', + ) prediction = str( payload.get('prediction') or payload.get('label') @@ -2973,17 +3070,20 @@ def parse_json_object_output( text = re.sub(r'\s*```$', '', text) try: payload = json.loads(text) + if isinstance(payload, dict): + return payload except json.JSONDecodeError: - match = re.search(r'\{.*\}', text, re.DOTALL) - if not match: - raise EvaluationError(error_message) + pass + + decoder = json.JSONDecoder() + for match in re.finditer(r'\{', text): try: - payload = json.loads(match.group(0)) - except json.JSONDecodeError as exc: - raise EvaluationError(error_message) from exc - if not isinstance(payload, dict): - raise EvaluationError(error_message) - return payload + payload, _end = decoder.raw_decode(text[match.start() :]) + except json.JSONDecodeError: + continue + if isinstance(payload, dict): + return payload + raise EvaluationError(error_message) def deterministic_label_mapping( diff --git a/tests/test_evaluation_runtime.py b/tests/test_evaluation_runtime.py index f9c1acd..30cfe99 100644 --- a/tests/test_evaluation_runtime.py +++ b/tests/test_evaluation_runtime.py @@ -3,6 +3,7 @@ from __future__ import annotations import base64 import json import tempfile +import threading import time import unittest from pathlib import Path @@ -21,6 +22,7 @@ from src.evaluation_runtime import ( normalize_evaluation_output, parse_dataset_bytes, prediction_is_allowed, + transient_model_error_message, ) @@ -282,6 +284,11 @@ class EvaluationRuntimeTests(unittest.TestCase): ) self.assertEqual(result['prediction'], '地图导航') self.assertEqual(result['confidence'], 1.0) + duplicated = normalize_evaluation_output( + '{"prediction":"慢","complex":false}' + '{"prediction":"慢","complex":false}' + ) + self.assertEqual(duplicated['prediction'], '慢') self.assertTrue(labels_equivalent('Agent(tag="地图导航")', '地图导航')) self.assertTrue( labels_equivalent( @@ -375,6 +382,61 @@ class EvaluationRuntimeTests(unittest.TestCase): self.assertEqual(mapping['label_map']['慢系统'], '慢') self.assertEqual(mapping['label_map']['complex=Ture'], 'complex=true') self.assertEqual(mapping['unmapped'], ['未知']) + self.assertIn( + 'HTTP 429', + transient_model_error_message( + 'HTTP 429 from local model backend: Too many requests' + ), + ) + + def test_transient_model_failure_retries_before_validation(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + runtime = EvaluationRuntime( + root=root / 'evaluations', + cwd_for_account=lambda _account: root, + model_config_for=lambda _account: ModelConfig(model='test-model'), + account_paths_for=lambda _account: {'python_env': root / '.venv'}, + max_workers=1, + ) + transient = AgentRunResult( + final_output='HTTP 429: Too many requests', + turns=0, + tool_calls=0, + transcript=(), + usage=UsageStats(), + ) + success = AgentRunResult( + final_output='{"prediction":"快"}', + turns=1, + tool_calls=0, + transcript=(), + usage=UsageStats(), + ) + + class StubAgent: + def __init__(self) -> None: + self.calls = 0 + + def run(self, *_args: object, **_kwargs: object) -> AgentRunResult: + self.calls += 1 + return transient if self.calls == 1 else success + + agent = StubAgent() + try: + with patch.object(runtime, '_extend_model_cooldown') as cooldown: + result = runtime._run_agent_with_transient_backoff( + agent, # type: ignore[arg-type] + 'prompt', + session_id='session', + runtime_context='context', + cancel_event=threading.Event(), + ) + self.assertIs(result, success) + self.assertEqual(agent.calls, 2) + cooldown.assert_called_once() + finally: + runtime.shutdown() def test_fast_slow_prompt_uses_skill_contract_and_relative_paths(self) -> None: skill = BundledSkill(