diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 63489e8fb..ba7ae0dd0 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -444,9 +444,34 @@ def api(self) -> LangfuseAPI: user, session, tags and input/output are on its root observation. Read scores with `api.scores.get_many` and experiments with `api.experiments`. + - **Read observation pages until `meta.cursor` is absent.** Pages can be + short or empty before the end. The server scans for up to 45 seconds per + request, so pass `request_options={"timeout_in_seconds": 60}`, and pass + `from_start_time` when you know a lower bound (otherwise the server + scans back to 2020-01-01). A 422 means the request hit a server resource + limit: narrow the time range rather than retrying unchanged. A `startTime` + condition in the JSON `filter` overrides `from_start_time` and + `to_start_time`. + - For large-scale aggregation (usage/cost by model, user, etc.), prefer the v2 Metrics API (`api.metrics.metrics(...)`) over paginating row-level data. + Reading all observations of a trace: + ```python + observations, cursor = [], None + while True: + page = langfuse.api.observations.get_many( + trace_id=trace_id, + from_start_time=since, + cursor=cursor, + request_options={"timeout_in_seconds": 60}, + ) + observations.extend(page.data) + cursor = page.meta.cursor + if cursor is None: + break + ``` + See also: `async_api`, https://langfuse.com/docs/api-and-data-platform/features/query-via-sdk @@ -2226,8 +2251,8 @@ def flush(self) -> None: *readable* yet: server-side ingestion is asynchronous, so flushed data may not be queryable for 15-30 seconds — `api.observations.get_many(trace_id=...)` may return empty results right - after a successful flush. See the `api` property docs for a bounded retry - pattern, or + after a successful flush. Poll with a deadline instead of a fixed sleep; + see https://langfuse.com/docs/api-and-data-platform/features/query-via-sdk#ingestion-lag """ if self._resources is not None: @@ -3196,7 +3221,9 @@ def run_batched_evaluation( - '[{"type": "arrayOptions", "column": "tags", "operator": "any of", "value": ["production"]}]' - '[{"type": "string", "column": "traceName", "operator": "=", "value": "chat"}]' - '[{"type": "datetime", "column": "startTime", "operator": ">=", "value": "2026-01-01T00:00:00Z"}]' - Default: None (fetches all observations). + Without a `startTime` lower bound, the server only reads back to + 2020-01-01, and a selective filter can take many slow requests to get + there. Add a `startTime` lower bound when you can. Default: None. fetch_batch_size: Number of items to fetch per API call and hold in memory. Larger values may be faster but use more memory. Maximum 1000. Default: 50. fields: Comma-separated list of observation field groups to fetch. Available @@ -3216,8 +3243,9 @@ def run_batched_evaluation( combined metrics. Default: None. metadata: Optional metadata dict to add to all created scores. Useful for tracking evaluation runs, versions, or other context. Default: None. - max_retries: Maximum number of retry attempts for failed batch fetches. - Default: 3. + max_retries: Maximum retries of a batch fetch on HTTP 408, 409, 429 and + 5xx responses. Other fetch errors, such as timeouts or a 422, stop + the run with `completed=False` and a resume token. Default: 3. verbose: If True, logs progress information to console. Useful for monitoring long-running evaluations. Default: False. resume_from: Optional resume token from a previous run that stopped early @@ -3236,12 +3264,12 @@ def run_batched_evaluation( - total_evaluations_failed: Individual evaluator failures - evaluator_stats: Per-evaluator statistics (success rate, scores created) - resume_token: Token for continuing the run (set after a fetch failure - or when max_items was reached while more items exist) + or when max_items was reached while more items may exist) - completed: False if the run stopped because a fetch failed - duration_seconds: Total execution time - failed_item_ids: Observation IDs of items that failed - error_summary: Error types and counts - - has_more_items: True if max_items reached but more exist + - has_more_items: True if max_items was reached and more items may exist - item_evaluations: Evaluations per observation ID Raises: @@ -3317,7 +3345,7 @@ def composite_evaluator(*, input, output, expected_output, metadata, evaluations ) ``` - Continuing a run that stopped early: + Processing in chunks of `max_items`: ```python result = client.run_batched_evaluation( mapper=generation_mapper, @@ -3325,19 +3353,26 @@ def composite_evaluator(*, input, output, expected_output, metadata, evaluations max_items=10000, ) - while result.resume_token: + while result.has_more_items: result = client.run_batched_evaluation( - mapper=generation_mapper, + mapper=generation_mapper, evaluators=[accuracy_evaluator], max_items=10000, resume_from=result.resume_token, ) + + if not result.completed: + # resume_from=result.resume_token retries the failed page. A 422 + # usually repeats, and a token only resumes its own filter, so + # start a new run whose filter adds a narrower startTime range. + ... ``` Note: - Evaluator failures are logged but don't stop the batch evaluation - Individual item failures are tracked but don't stop processing - - Fetch failures are retried up to `max_retries` times + - Fetches are retried up to `max_retries` times on HTTP 408, 409, 429 + and 5xx responses; other fetch errors stop the run - All scores are automatically flushed to Langfuse at the end - Resuming uses the pagination cursor, so items are neither skipped nor evaluated twice diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index 1c89d1449..8ef89c79b 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -471,9 +471,9 @@ class BatchEvaluationResumeToken: exactly where the previous one stopped, without re-evaluating or skipping items, even if new observations were ingested in the meantime. - A token is returned when a run stops because a batch fetch failed after all - retries (`completed=False`) or because `max_items` was reached while more - items exist (`has_more_items=True`). + A token is returned when a run stops because a batch fetch failed + (`completed=False`) or because `max_items` was reached while more items may + exist (`has_more_items=True`). Attributes: filter: The original JSON filter string used to query items. Pass the @@ -481,7 +481,8 @@ class BatchEvaluationResumeToken: cursor: Cursor of the next page to fetch. None if no page was fetched yet. last_processed_timestamp: ISO 8601 start time of the oldest processed observation. Only used to resume when `cursor` is None, by fetching - observations that started strictly before this timestamp. + observations that started at or before this timestamp, skipping + `last_processed_id`. last_processed_id: The ID of the last processed observation, for reference. last_page_ids: IDs of the observations on the last processed page. A resumed run skips these, because unmerged duplicate rows of one @@ -572,13 +573,13 @@ class BatchEvaluationResult: evaluator_stats: List of per-evaluator statistics (success/failure rates, scores created). resume_token: Token for continuing the run. Set when a batch fetch failed (`completed=False`) or when `max_items` was reached while more items - exist (`has_more_items=True`); None otherwise. + may exist (`has_more_items=True`); None otherwise. completed: False if the run stopped because a batch fetch failed, True otherwise (including when it stopped at `max_items`). duration_seconds: Total time taken to execute the batch evaluation. failed_item_ids: List of observation IDs for items that failed evaluation. error_summary: Dictionary mapping error types to occurrence counts. - has_more_items: True if max_items limit was reached but more items exist. + has_more_items: True if the run stopped at max_items and more items may exist. item_evaluations: Dictionary mapping observation IDs to their evaluation results (both regular and composite). Examples: @@ -632,7 +633,7 @@ class BatchEvaluationResult: print("Pass it as resume_from to continue") if result.has_more_items: - print(f"ℹ️ More items available beyond max_items limit") + print(f"ℹ️ More items may be available beyond max_items limit") ``` Performance monitoring: @@ -691,7 +692,7 @@ def __init__( duration_seconds: Total execution time. failed_item_ids: IDs of failed items. error_summary: Error types and counts. - has_more_items: Whether more items exist beyond max_items. + has_more_items: Whether more items may exist beyond max_items. item_evaluations: Dictionary mapping item IDs to their evaluation results. Note: @@ -790,7 +791,7 @@ def __str__(self) -> str: lines.append(" Use resume_from parameter to continue") if self.has_more_items: - lines.append("\nNote: More items available beyond max_items limit") + lines.append("\nNote: More items may be available beyond max_items limit") lines.append("=" * 60) return "\n".join(lines) @@ -855,7 +856,8 @@ async def run_async( metadata: Metadata to add to all created scores. _add_observation_scores_to_trace: Private option to duplicate observation-level scores onto the parent trace. - max_retries: Maximum retries for failed batch fetches. + max_retries: Maximum retries of a batch fetch on HTTP 408, 409, 429 + and 5xx responses. Other fetch errors are not retried. verbose: If True, log progress to console. resume_from: Resume token from a previous run. If `filter` is omitted, the token's filter is reused. @@ -1104,15 +1106,15 @@ async def _fetch_batch_with_retry( filter: JSON filter string for querying. cursor: Cursor returned with the previous page; None for the first page. limit: Number of items to request. - max_retries: Maximum number of retry attempts. + max_retries: Maximum retries on HTTP 408, 409, 429 and 5xx responses. fields: Comma-separated v2 field groups to include. Returns: - Tuple of the page's observations and the cursor for the next page - (None when there are no more pages). + Tuple of the page's observations, which may be empty, and the cursor + for the next page (None when there are no more pages). Raises: - Exception: If all retry attempts fail. + Exception: If the fetch fails. """ response = await asyncio.to_thread( self.client.api.observations.get_many, @@ -1481,7 +1483,7 @@ def _build_result( start_time: Start time (unix timestamp). failed_item_ids: IDs of failed items. error_summary: Error type counts. - has_more_items: Whether more items exist beyond max_items. + has_more_items: Whether more items may exist beyond max_items. item_evaluations: Dictionary mapping item IDs to their evaluation results. Returns: