Repository navigation
feat(experiments): retries, timeouts, failure accounting and resume for experiments and batch evaluation - #2115
Cookie0103 wants to merge 2 commits into
Conversation
…or run_experiment and run_batched_evaluation run_experiment dropped failed items from item_results with only a log line, called the task and evaluators once with no timeout, and had no way to resume an interrupted run. run_batched_evaluation had the same gaps for mapper and evaluator calls. Adds an opt-in ExecutionPolicy (private langfuse/_experiment package): - RetryPolicy with jittered backoff honouring Retry-After and a default predicate that reads HTTP status attributes and SDK error class names - task, evaluator and whole-run timeouts, and a failure budget that stops scheduling and cancels in-flight items - ExperimentResult/BatchEvaluationResult record failures, evaluator failures and an ExecutionSummary (item_results is unchanged) - checkpoint and resume via in-memory or file stores (atomic, debounced, off the event loop); completed items are rebuilt without re-running the task or duplicating scores - retry attempts are recorded as child observations of the task span Without a policy the span tree, scores and results are unchanged. Refs langfuse/langfuse#19902 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
|
|
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| by_key: Dict[str, ExperimentItemResult] = {} | ||
| for r in completed_results: | ||
| by_key[_key_for_item(r.item, self.dataset_version)] = r | ||
| for r in resumed_results: | ||
| by_key[_key_for_item(r.item, self.dataset_version)] = r | ||
|
|
||
| ordered: List[ExperimentItemResult] = [] | ||
| for item, key in zip(self.data, self._item_keys): | ||
| if key in by_key: | ||
| ordered.append(by_key[key]) |
There was a problem hiding this comment.
Repeated items lose results even without an execution policy. make_item_keys adds occurrence suffixes, but _key_for_item rebuilds each result's key from a one-item list. Results overwrite each other under the unsuffixed key, and the suffixed entries are omitted. Run evaluators then receive fewer results than successful task calls. Keep the assigned key alongside each result instead of recomputing it.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/_experiment/executor.py
Line: 480-489
Comment:
**Duplicate results disappear**
Repeated items lose results even without an execution policy. `make_item_keys` adds occurrence suffixes, but `_key_for_item` rebuilds each result's key from a one-item list. Results overwrite each other under the unsuffixed key, and the suffixed entries are omitted. Run evaluators then receive fewer results than successful task calls. Keep the assigned key alongside each result instead of recomputing it.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| if not self._aborted and self._failure_budget_exceeded( | ||
| failed, attempted | ||
| ): | ||
| self._aborted = True | ||
| self._abort_reason = "failure budget exceeded" |
There was a problem hiding this comment.
With on_threshold="raise", both runners still stop and return normally when the failure budget trips. The setting is only checked when constructing FailurePolicy. Jobs that rely on the exception to fail CI or trigger recovery will miss it. Honor the setting in both runners and attach the partial experiment result as documented.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/_experiment/executor.py
Line: 450-454
Comment:
**Requested failure exceptions never arrive**
With `on_threshold="raise"`, both runners still stop and return normally when the failure budget trips. The setting is only checked when constructing `FailurePolicy`. Jobs that rely on the exception to fail CI or trigger recovery will miss it. Honor the setting in both runners and attach the partial experiment result as documented.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| return ExperimentFingerprint( | ||
| task_name=task_name, | ||
| evaluator_names=evaluator_names, | ||
| data_size=len(self.data), | ||
| data_digest=compute_data_digest(self._item_keys), |
There was a problem hiding this comment.
The data fingerprint hashes item keys rather than all data passed to tasks and evaluators. Dataset keys contain only the item ID and optional version; local keys omit metadata. Editing an unpinned dataset item or changing local metadata therefore passes the resume check and returns old outputs and evaluations. Hash the relevant item contents separately from the stable item key.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/_experiment/executor.py
Line: 180-184
Comment:
**Resume accepts changed data**
The data fingerprint hashes item keys rather than all data passed to tasks and evaluators. Dataset keys contain only the item ID and optional version; local keys omit `metadata`. Editing an unpinned dataset item or changing local metadata therefore passes the resume check and returns old outputs and evaluations. Hash the relevant item contents separately from the stable item key.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| processor=processor, | ||
| max_concurrency=max_concurrency, | ||
| ) | ||
| exec_result = await executor.run(task=task, evaluators=evaluators) |
There was a problem hiding this comment.
composite_evaluator is missing from the checkpoint fingerprint. Adding or replacing it while keeping the task and regular evaluators unchanged passes the resume check. Completed items are skipped, leaving their composite evaluations missing or stale. Pass this evaluator into the fingerprint too.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/_client/client.py
Line: 2988
Comment:
**Composite scores stay stale**
`composite_evaluator` is missing from the checkpoint fingerprint. Adding or replacing it while keeping the task and regular evaluators unchanged passes the resume check. Completed items are skipped, leaving their composite evaluations missing or stale. Pass this evaluator into the fingerprint too.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| and not any( | ||
| not isinstance(item, dict) and getattr(item, "dataset_id", None) | ||
| for item in data | ||
| ) |
There was a problem hiding this comment.
Dataset runs allow resume=True without a stable run_name or explicit experiment_id. The generated run name contains the current timestamp, and that name feeds the dataset experiment ID used to find the checkpoint. Calling the experiment again creates a different key and repeats completed task calls. Require a stable name or explicit ID for dataset resume too.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/_client/client.py
Line: 2851-2854
Comment:
**Dataset resume repeats completed work**
Dataset runs allow `resume=True` without a stable `run_name` or explicit `experiment_id`. The generated run name contains the current timestamp, and that name feeds the dataset experiment ID used to find the checkpoint. Calling the experiment again creates a different key and repeats completed task calls. Require a stable name or explicit ID for dataset resume too.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| if execution is not None and execution.failures.is_exceeded( | ||
| len(execution_failures), | ||
| total_items_processed + total_items_failed, |
There was a problem hiding this comment.
The batch failure rate divides failure records by items. Several failing evaluators on one item add several records, so the rate can exceed 100% and stop a run below the requested share of affected items. The same mismatch can make execution.failed exceed the item count. Count distinct affected items for item-based limits and summaries, while keeping individual records for detail.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/batch_evaluation.py
Line: 1172-1174
Comment:
**Failure rates count errors twice**
The batch failure rate divides failure records by items. Several failing evaluators on one item add several records, so the rate can exceed 100% and stop a run below the requested share of affected items. The same mismatch can make `execution.failed` exceed the item count. Count distinct affected items for item-based limits and summaries, while keeping individual records for detail.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| except BaseException as e: # noqa: BLE001 | ||
| attempt_exc = e | ||
| last_exc = e |
There was a problem hiding this comment.
With retry_on=lambda exc: True, _run_with_retry catches CancelledError and retries it. An abort can therefore start another paid task attempt after cancellation. Separately, process_one turns other BaseException values such as SystemExit into item failures. Re-raise cancellation and process-exit exceptions before applying retry or item-failure logic.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/_experiment/retrying.py
Line: 56-58
Comment:
**Cancellation starts another attempt**
With `retry_on=lambda exc: True`, `_run_with_retry` catches `CancelledError` and retries it. An abort can therefore start another paid task attempt after cancellation. Separately, `process_one` turns other `BaseException` values such as `SystemExit` into item failures. Re-raise cancellation and process-exit exceptions before applying retry or item-failure logic.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| stage="task", | ||
| error_type=type(e).__name__, | ||
| error_message=str(e), | ||
| attempts=[], |
There was a problem hiding this comment.
_run_with_retry attaches attempt records to the final exception, but both task failure branches and both evaluator failure branches replace them with attempts=[]. Evaluator retries also pass no attempt hook, so execution retry counts omit them. Preserve the attached records and include evaluator attempts in the summary so callers can inspect the work that ran.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/_client/client.py
Line: 3552
Comment:
**Attempt details are lost**
`_run_with_retry` attaches attempt records to the final exception, but both task failure branches and both evaluator failure branches replace them with `attempts=[]`. Evaluator retries also pass no attempt hook, so execution retry counts omit them. Preserve the attached records and include evaluator attempts in the summary so callers can inspect the work that ran.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| if task_attempts is not None: | ||
| self._emit_attempt_observations( | ||
| task_span, task_attempts | ||
| ) |
There was a problem hiding this comment.
Attempt observations are created only after the task succeeds, using that later moment as their time. They do not cover the failed calls or show the backoff gaps, and exhausted retries emit none. Emit observations around each attempt, or set their times from the recorded start and duration on both success and terminal failure.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/_client/client.py
Line: 3288-3291
Comment:
**Attempt spans show wrong times**
Attempt observations are created only after the task succeeds, using that later moment as their time. They do not cover the failed calls or show the backoff gaps, and exhausted retries emit none. Emit observations around each attempt, or set their times from the recorded start and duration on both success and terminal failure.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| for base in base_keys: | ||
| count = seen.get(base, 0) | ||
| seen[base] = count + 1 | ||
| if count == 0 and base not in keys: |
There was a problem hiding this comment.
base not in keys scans the growing result list for every distinct item, making make_item_keys quadratic. Every experiment pays this cost, even without checkpointing. Use the existing seen dictionary or a set for membership checks so large runs do not repeatedly scan all prior keys before starting tasks.
Prompt To Fix With AI
This is a comment left during a code review.
Path: langfuse/_experiment/item_key.py
Line: 87
Comment:
**Key generation slows large runs**
`base not in keys` scans the growing result list for every distinct item, making `make_item_keys` quadratic. Every experiment pays this cost, even without checkpointing. Use the existing `seen` dictionary or a set for membership checks so large runs do not repeatedly scan all prior keys before starting tasks.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.
What does this PR do?
Fixes langfuse/langfuse#19902
Langfuse.run_experimentandrun_batched_evaluationhandle every item as a single attempt and keep failures only as a log line:asyncio.gather(..., return_exceptions=True)logs a failed item and drops it.ExperimentResult.item_resultscontains only the successes, and run evaluators aggregate over that list. A run where half the items crashed therefore reports aggregates over the survivors and gives no indication of what failed._run_taskcalls the task once, so a transient 429 or 5xx from the model provider fails the item for good. A task or LLM-judge evaluator that hangs blocks the whole run indefinitely. Mapper and evaluator calls inrun_batched_evaluationbehave the same way.experiment_id, but an interrupted 2,000-item run still has to start over and pay again for every completed task call.This PR adds an opt-in
ExecutionPolicy. With no policy passed, the span tree, observations,create_scorecalls anditem_resultsare unchanged. A test pins the default-path span names and the score count.Design
The new code lives in the private package
langfuse/_experiment/. Public names are re-exported fromlangfuse.experimentandlangfuse.policy.pyRetryPolicy,TimeoutPolicy,FailurePolicyandExecutionPolicy. The default retry predicate readsstatus_code/response.status_codeand SDK error class names (OpenAI/Anthropic/LiteLLM naming) through the MRO, so no provider SDK is imported. It parsesRetry-Aftergiven in seconds or as an HTTP date, and never retries cancellation.retrying.pyrun_experimentandrun_batched_evaluation.executor.pyExperimentExecutorhandles scheduling undermax_concurrency, the failure budget, the run deadline, and cancellation (in-flight items are cancelled, with a bounded wait so a task that swallowsCancelledErrorcannot hang the run). It also handles checkpointing and resume.checkpoint.pyInMemoryCheckpointStoreandFileCheckpointStorewrite atomically (temp file, fsync,os.replace) with debounced writes performed throughasyncio.to_thread. Filenames are restricted to[A-Za-z0-9_-].item_key.pyexperiment_item_idexpression that used to be inlined inclient.pymoved here, so the trace attribute and the checkpoint key cannot diverge. Duplicate items get deterministic suffixes.results.pyAttemptRecord,ItemFailure,EvaluatorFailureandExecutionSummary.Behaviour and trade-offs
item_resultskeeps its meaning (successes only). Failures are added next to it asfailures/evaluator_failures.failures=only if its signature declares afailuresparameter. Existing**kwargsevaluators receive exactly what they received before.experiment-item-task-attemptchild observation of the existing task span, so backoff shows up as gaps rather than being folded into one duration. Whenmax_attempts == 1(the default), no extra observations are created.CheckpointMismatchError, unlessresume="force"is passed.experiment_idcannot identify a run. The checkpoint is therefore keyed by(name, run_name), andresume=Truewithout an explicitrun_nameraises a clearValueError. The resumed run reuses the experiment id stored in the checkpoint.run_batched_evaluation. The same policies apply to mapper, evaluator and composite calls. When the failure budget is exceeded, paging stops and the method returnscompleted=Falsewith a validresume_tokenandexecution.aborted=True._create_score, themax_itemslogic and flush call sites are untouched, to avoid conflicts with the open fix(evaluations): account for score duplication in evaluator stats and fix max_items deduplication #2103 and fix(evaluations): flush scores without blocking the event loop #2085.Limitations
Type of change
Verification
The 18 errors are all in
tests/unit/test_prompt.py(Langfuse client is not initialized). They are identical on a clean checkout ofmain, so they are environment-dependent and not related to this change.New tests are in
tests/unit/test_experiment_execution.pyandtests/unit/test_batch_evaluation_execution.py. They run exporter-local, with no server or provider calls, and cover:Retry-Afterin seconds and as an HTTP date, class names, and cancellation;item_resultsstays unchanged, including the run-evaluatorfailuresopt-in;experiment_id;force;Checklist
code_review.md..env.templateif needed. (Public docstrings are updated. Happy to add a docs page in langfuse-docs if you want the feature documented there.)AI disclosure: implemented with AI coding assistance (Claude, Cursor). I reviewed the diff and ran the commands above.
🤖 Generated with Claude Code
This PR needs fixes before merging because it can drop successful results, reuse stale checkpoint results, and continue work after cancellation.
Summary
Adds execution policies, retry records, failure summaries, and checkpoint-based resume for experiments and batch evaluation.
Acknowledged behavior:
Cookie0103intentionally checks batch run timeouts between pages, as pinned by the new test.Cookie0103documents that tasks may ignore cancellation.Cookie0103documents that resume cannot detect server-side deletion of existing scores.Diagram
%%{init: {'theme': 'neutral'}}%% flowchart TD A[Run experiment] --> B[Build item keys and fingerprint] B --> C[Load checkpoint] C --> D[Rebuild completed results] C --> E[Schedule remaining items] E --> F[Task retries and timeout] F --> G[Item evaluators] G --> H[Composite evaluator] H --> I[Record result and save checkpoint] E --> J[Check failure budget and run deadline] D --> K[Order results] I --> K J --> K K --> L[Run evaluators outside executor] L --> M[Return experiment result]Reviews (1) · Last reviewed commit: "feat(experiments): export checkpoint sto..." · Reviewed by Greptile