Skip to content

WUM-2: bound auto_verified optimize runtime #932

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wuman001 wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

WUM-2: bound auto_verified optimize runtime #932

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 83 additions & 0 deletions aworld-cli/src/aworld_cli/top_level_commands/optimize_cmd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,18 @@ def register_parser(self, subparsers) -> None:
dest="judge_timeout_seconds",
help="Timeout in seconds for each self-evolve judge attempt.",
)
parser.add_argument(
"--judge-failure-retries",
type=int,
dest="judge_failure_retries",
help="Number of failed judge attempts to retry before degrading.",
)
parser.add_argument(
"--judge-concurrency",
type=int,
dest="judge_concurrency",
help="Bounded concurrency for independent self-evolve evaluator calls.",
)
parser.add_argument(
"--baseline-replay-repetitions",
type=int,
Expand All @@ -91,6 +103,25 @@ def register_parser(self, subparsers) -> None:
dest="candidate_replay_repetitions",
help="Number of candidate replay rollouts to aggregate.",
)
parser.add_argument(
"--replay-concurrency",
type=int,
dest="replay_concurrency",
help="Bounded concurrency for independent self-evolve replay subprocesses.",
)
parser.add_argument(
"--max-optimize-seconds",
"--wall-clock-budget",
type=int,
dest="max_optimize_seconds",
help="Best-effort wall-clock budget in seconds for one optimize run.",
)
parser.add_argument(
"--fast-verified",
action="store_true",
dest="fast_verified",
help="Use lower non-explicit auto_verified runtime caps for exploratory verification.",
)
parser.add_argument(
"--drain-pending",
action="store_true",
Expand Down Expand Up @@ -138,8 +169,13 @@ def run(self, args, context) -> int:
judge_backend_ref=getattr(args, "judge_backend_ref", None),
judge_repetitions=getattr(args, "judge_repetitions", None),
judge_timeout_seconds=getattr(args, "judge_timeout_seconds", None),
judge_failure_retries=getattr(args, "judge_failure_retries", None),
judge_concurrency=getattr(args, "judge_concurrency", None),
replay_timeout_seconds=getattr(args, "replay_timeout_seconds", None),
replay_max_steps=getattr(args, "replay_max_steps", None),
replay_concurrency=getattr(args, "replay_concurrency", None),
max_optimize_seconds=getattr(args, "max_optimize_seconds", None),
fast_verified=getattr(args, "fast_verified", False),
baseline_replay_repetitions=getattr(args, "baseline_replay_repetitions", None),
candidate_replay_repetitions=getattr(args, "candidate_replay_repetitions", None),
progress_callback=_print_optimize_progress,
Expand Down Expand Up @@ -224,8 +260,13 @@ def run_optimize_cli(
judge_backend_ref: str | None = None,
judge_repetitions: int | None = None,
judge_timeout_seconds: int | None = None,
judge_failure_retries: int | None = None,
judge_concurrency: int | None = None,
replay_timeout_seconds: int | None = None,
replay_max_steps: int | None = None,
replay_concurrency: int | None = None,
max_optimize_seconds: int | None = None,
fast_verified: bool = False,
baseline_replay_repetitions: int | None = None,
candidate_replay_repetitions: int | None = None,
runtime_registry_refresher: Callable[[Any], Any] | None = None,
Expand All @@ -236,6 +277,19 @@ def run_optimize_cli(
) -> Mapping[str, Any]:
import aworld.self_evolve as self_evolve

if fast_verified and apply == "auto_verified":
if judge_failure_retries is None:
judge_failure_retries = 0
if replay_concurrency is None:
replay_concurrency = 2
if candidate_replay_repetitions is None:
candidate_replay_repetitions = 1
if max_optimize_seconds is None:
max_optimize_seconds = 900
replay_candidate_limit = 1
else:
replay_candidate_limit = None

judge_repetitions = _auto_verified_default(
apply,
judge_repetitions,
Expand Down Expand Up @@ -295,9 +349,16 @@ def run_optimize_cli(
**_replay_options(
replay_timeout_seconds=replay_timeout_seconds,
replay_max_steps=replay_max_steps,
replay_concurrency=replay_concurrency,
replay_candidate_limit=replay_candidate_limit,
baseline_replay_repetitions=baseline_replay_repetitions,
candidate_replay_repetitions=candidate_replay_repetitions,
),
**_runtime_bound_options(
judge_failure_retries=judge_failure_retries,
judge_concurrency=judge_concurrency,
max_optimize_seconds=max_optimize_seconds,
),
)


Expand Down Expand Up @@ -353,6 +414,8 @@ def _replay_options(
*,
replay_timeout_seconds: int | None,
replay_max_steps: int | None,
replay_concurrency: int | None,
replay_candidate_limit: int | None,
baseline_replay_repetitions: int | None,
candidate_replay_repetitions: int | None,
) -> dict[str, int]:
Expand All @@ -361,13 +424,33 @@ def _replay_options(
options["replay_timeout_seconds"] = replay_timeout_seconds
if replay_max_steps is not None:
options["replay_max_steps"] = replay_max_steps
if replay_concurrency is not None:
options["replay_concurrency"] = replay_concurrency
if replay_candidate_limit is not None:
options["replay_candidate_limit"] = replay_candidate_limit
if baseline_replay_repetitions is not None:
options["baseline_replay_repetitions"] = baseline_replay_repetitions
if candidate_replay_repetitions is not None:
options["candidate_replay_repetitions"] = candidate_replay_repetitions
return options


def _runtime_bound_options(
*,
judge_failure_retries: int | None,
judge_concurrency: int | None,
max_optimize_seconds: int | None,
) -> dict[str, int]:
options: dict[str, int] = {}
if judge_failure_retries is not None:
options["judge_failure_retries"] = judge_failure_retries
if judge_concurrency is not None:
options["judge_concurrency"] = judge_concurrency
if max_optimize_seconds is not None:
options["max_optimize_seconds"] = max_optimize_seconds
return options


def drain_pending_self_evolve_jobs(
*,
workspace_root: str,
Expand Down
81 changes: 62 additions & 19 deletions aworld/self_evolve/evaluation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -403,7 +403,7 @@ async def _run_evaluator_source_with_timeout(
runner_kwargs: Mapping[str, Any],
) -> Mapping[str, Any]:
call = self._run_evaluator_source(runner, runner_kwargs=runner_kwargs)
if self.judge_timeout_seconds is None or self.run_evaluator_source is None:
if self.judge_timeout_seconds is None:
return await call
return await asyncio.wait_for(call, timeout=self.judge_timeout_seconds)

Expand Down Expand Up @@ -433,30 +433,73 @@ async def evaluate_baseline_and_candidate(
dataset_split: str = "validation",
baseline_variant_id: str = "baseline",
artifact_namespace: str | None = None,
judge_concurrency: int = 1,
) -> tuple[EvaluationSummary, EvaluationSummary]:
baseline_summary = await backend.evaluate_variant(
EvaluationRequest(
variant_id=baseline_variant_id,
candidate=None,
dataset=dataset,
eval_config=eval_config,
dataset_split=dataset_split,
artifact_namespace=artifact_namespace,
)
if judge_concurrency <= 0:
raise ValueError("judge_concurrency must be positive")
baseline_request = EvaluationRequest(
variant_id=baseline_variant_id,
candidate=None,
dataset=dataset,
eval_config=eval_config,
dataset_split=dataset_split,
artifact_namespace=artifact_namespace,
)
candidate_summary = await backend.evaluate_variant(
EvaluationRequest(
variant_id=candidate.candidate_id,
candidate=candidate,
dataset=dataset,
eval_config=eval_config,
dataset_split=dataset_split,
artifact_namespace=artifact_namespace,
)
candidate_request = EvaluationRequest(
variant_id=candidate.candidate_id,
candidate=candidate,
dataset=dataset,
eval_config=eval_config,
dataset_split=dataset_split,
artifact_namespace=artifact_namespace,
)
if judge_concurrency == 1:
baseline_summary = await backend.evaluate_variant(baseline_request)
candidate_summary = await backend.evaluate_variant(candidate_request)
return baseline_summary, candidate_summary

semaphore = asyncio.Semaphore(judge_concurrency)

async def evaluate(request: EvaluationRequest) -> EvaluationSummary:
async with semaphore:
return await backend.evaluate_variant(request)

baseline_summary, candidate_summary = await asyncio.gather(
evaluate(baseline_request),
evaluate(candidate_request),
)
return baseline_summary, candidate_summary


def estimate_replay_wall_clock_seconds(
*,
estimate: ReplayCostEstimate,
replay_timeout_seconds: float | None,
judge_timeout_seconds: float | None,
judge_failure_retries: int,
replay_concurrency: int = 1,
judge_concurrency: int = 1,
) -> float | None:
if judge_failure_retries < 0:
raise ValueError("judge_failure_retries must be non-negative")
if replay_concurrency <= 0:
raise ValueError("replay_concurrency must be positive")
if judge_concurrency <= 0:
raise ValueError("judge_concurrency must be positive")
replay_seconds = None
if replay_timeout_seconds is not None:
replay_batches = (estimate.total_replay_count + replay_concurrency - 1) // replay_concurrency
replay_seconds = replay_batches * replay_timeout_seconds
judge_seconds = None
if judge_timeout_seconds is not None:
judge_attempts = estimate.judge_call_count + judge_failure_retries
judge_batches = (judge_attempts + judge_concurrency - 1) // judge_concurrency
judge_seconds = judge_batches * judge_timeout_seconds
if replay_seconds is None and judge_seconds is None:
return None
return (replay_seconds or 0.0) + (judge_seconds or 0.0)
Comment on lines +474 to +500

@gemini-code-assist gemini-code-assist Bot Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The current wall-clock time estimation logic in estimate_replay_wall_clock_seconds is highly inaccurate and significantly underestimates the worst-case execution time. This is because:

  1. Sequential Candidate Evaluation: Candidates in the population are evaluated sequentially in a for loop in SelfEvolveRunner.run_explicit_target, not in parallel. Therefore, replays and judge calls cannot be parallelized across different candidates.
  2. Sequential Baseline and Candidate Replays: Within a single candidate's replay, the baseline replays and candidate replays are executed sequentially (first baseline, then candidate), not in parallel.
  3. Sequential Judge Repetitions: Within AWorldTrajectoryEvaluatorBackend, the judge repetitions (attempts) are executed sequentially in a for loop, not in parallel.
  4. Batch Case Evaluation: AWorldTrajectoryEvaluatorBackend evaluates all cases in a single runner call, so the number of sequential evaluator calls does not scale with case_count in the way estimate.judge_call_count (which multiplies by case_count) suggests.

We should update the estimation formula to accurately reflect these sequential/parallel execution boundaries.

def estimate_replay_wall_clock_seconds(
    *,
    replayed_candidate_count: int,
    baseline_repetitions: int,
    candidate_repetitions: int,
    judge_repetitions: int,
    judge_failure_retries: int,
    replay_timeout_seconds: float | None,
    judge_timeout_seconds: float | None,
    replay_concurrency: int = 1,
    judge_concurrency: int = 1,
) -> float | None:
    if judge_failure_retries < 0:
        raise ValueError("judge_failure_retries must be non-negative")
    if replay_concurrency <= 0:
        raise ValueError("replay_concurrency must be positive")
    if judge_concurrency <= 0:
        raise ValueError("judge_concurrency must be positive")
    if replayed_candidate_count < 0:
        raise ValueError("replayed_candidate_count must be non-negative")

    replay_seconds = None
    if replay_timeout_seconds is not None:
        baseline_batches = (baseline_repetitions + replay_concurrency - 1) // replay_concurrency
        candidate_batches = (candidate_repetitions + replay_concurrency - 1) // replay_concurrency
        total_batches = baseline_batches + replayed_candidate_count * candidate_batches
        replay_seconds = total_batches * replay_timeout_seconds

    judge_seconds = None
    if judge_timeout_seconds is not None:
        eval_batches = 1 if judge_concurrency >= 2 else 2
        judge_attempts = judge_repetitions + judge_failure_retries
        judge_seconds = replayed_candidate_count * eval_batches * judge_attempts * judge_timeout_seconds

    if replay_seconds is None and judge_seconds is None:
        return None
    return (replay_seconds or 0.0) + (judge_seconds or 0.0)



def estimate_replay_cost(
*,
dataset: SelfEvolveDataset,
Expand Down
23 changes: 16 additions & 7 deletions aworld/self_evolve/replay.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,8 +130,12 @@ def __init__(
self,
*,
executor: ReplayExecutor | None = None,
replay_concurrency: int = 1,
) -> None:
if replay_concurrency <= 0:
raise ValueError("replay_concurrency must be positive")
self.executor = executor or AWorldCliReplayExecutor()
self.replay_concurrency = replay_concurrency

async def replay_candidate(
self,
Expand Down Expand Up @@ -207,37 +211,42 @@ async def _run_repetitions(
) -> ReplayVariantResult:
if repetitions <= 0:
raise ValueError("replay repetitions must be positive")
results: list[ReplayVariantResult] = []
results: list[ReplayVariantResult | None] = [None] * repetitions
semaphore = asyncio.Semaphore(self.replay_concurrency)
logger.info(
"self_evolve.replay.repetitions.start "
f"run_id={request.run_id} task_id={request.task_id} "
f"variant_id={base_variant_id} repetitions={repetitions}"
)
for index in range(1, repetitions + 1):
async def run_one(index: int) -> None:
variant_id = base_variant_id if repetitions == 1 else f"{base_variant_id}-{index}"
repetition_dir = artifact_dir if repetitions == 1 else artifact_dir / str(index)
logger.info(
"self_evolve.replay.repetition.start "
f"run_id={request.run_id} task_id={request.task_id} "
f"variant_id={variant_id} index={index}/{repetitions}"
)
results.append(
await self._run_variant_with_evidence_retries(
async with semaphore:
result = await self._run_variant_with_evidence_retries(
request,
variant_id=variant_id,
skill_root=skill_root,
artifact_dir=repetition_dir,
)
)
results[index - 1] = result
logger.info(
"self_evolve.replay.repetition.end "
f"run_id={request.run_id} task_id={request.task_id} "
f"variant_id={variant_id} index={index}/{repetitions} "
f"status={results[-1].status}"
f"status={result.status}"
)
await asyncio.gather(*(run_one(index) for index in range(1, repetitions + 1)))
ordered_results = [
result for result in results if result is not None
]
aggregated = _aggregate_variant_results(
base_variant_id=base_variant_id,
results=results,
results=ordered_results,
artifact_dir=artifact_dir,
)
logger.info(
Expand Down
Loading
Loading