Synchronous client for the full Langfuse REST API (traces, observations, scores, datasets, prompts, ...).

Use this to read or manage data on the Langfuse server; use the tracing methods (start_observation, @observe) to create traces. Use async_api for the asyncio variant.

Semantics that are easy to miss:

• Ingestion is asynchronous. langfuse.flush() only guarantees delivery to the API, not read visibility: reads such as api.trace.get(trace_id) may raise langfuse.api.NotFoundError until processing completes (typically within 15-30 seconds; longer under load). The same applies to scores and dataset run reads. Instead of a fixed sleep, retry with a deadline:

• List endpoints return lightweight views. api.trace.list(...) returns TraceWithDetails, where observations and scores are lists of ID strings. Fetch the full objects with api.trace.get(trace_id) (TraceWithFullDetails), or prefer api.observations.get_many(trace_id=...) for row-level observation queries. The same list-view vs. get-detail pattern applies to other resources.

• Prefer the v2 data APIs — they are the defaults since SDK v4. api.observations and api.metrics map to the high-performance /api/public/v2/... endpoints and are the recommended read path. Their v1 equivalents remain available under api.legacy.observations_v1 / api.legacy.metrics_v1 but are less performant at scale, not recommended for new workflows, and will be deprecated.

• For large-scale aggregation (usage/cost by model, user, etc.), prefer the v2 Metrics API (api.metrics.metrics(...)) over paginating row-level data.

See also: async_api, https://langfuse.com/docs/api-and-data-platform/features/query-via-sdk (ingestion lag: #ingestion-lag, list vs. get: #traces-list-vs-get), https://langfuse.com/docs/api-and-data-platform/features/observations-api, https://langfuse.com/docs/metrics/features/metrics-api

Create a new observation of the specified type.

This method creates a new observation but does not set it as the current span in the context. To create and use an observation within a context, use start_as_current_observation().

Arguments:

• trace_context: Optional context for connecting to an existing trace
• name: Name of the observation
• as_type: Type of observation to create (defaults to "span")
• input: Input data for the operation
• output: Output data from the operation
• metadata: Additional metadata to associate with the observation
• version: Version identifier for the code or component
• level: Importance level of the observation
• status_message: Optional status message for the observation
• completion_start_time: When the model started generating (for generation types)
• model: Name/identifier of the AI model used (for generation types)
• model_parameters: Parameters used for the model (for generation types)
• usage_details: Token usage information (for generation types)
• cost_details: Cost information (for generation types)
• prompt: Associated prompt template (for generation types)

Returns:

An observation object of the appropriate type that must be ended with .end()

Create a new observation and set it as the current span in a context manager.

This method creates a new observation of the specified type and sets it as the current span within a context manager. Use this method with a 'with' statement to automatically handle the observation lifecycle within a code block.

The created observation will be the child of the current span in the context.

Arguments:

• trace_context: Optional context for connecting to an existing trace
• name: Name of the observation (e.g., function or operation name)
• as_type: Type of observation to create (defaults to "span")
• input: Input data for the operation (can be any JSON-serializable object)
• output: Output data from the operation (can be any JSON-serializable object)
• metadata: Additional metadata to associate with the observation
• version: Version identifier for the code or component
• level: Importance level of the observation (info, warning, error)
• status_message: Optional status message for the observation
• end_on_exit (default: True): Whether to end the span automatically when leaving the context manager. If False, the span must be manually ended to avoid memory leaks.
• The following parameters are available when as_type is: "generation" or "embedding".
• completion_start_time: When the model started generating the response
• model: Name/identifier of the AI model used (e.g., "gpt-4")
• model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
• usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
• cost_details: Cost information for the model call
• prompt: Associated prompt template from Langfuse prompt management

Returns:

A context manager that yields the appropriate observation type based on as_type

Example:

# Create a span
with langfuse.start_as_current_observation(name="process-query", as_type="span") as span:
    # Do work
    result = process_data()
    span.update(output=result)

    # Create a child span automatically
    with span.start_as_current_observation(name="sub-operation") as child_span:
        # Do sub-operation work
        child_span.update(output="sub-result")

# Create a tool observation
with langfuse.start_as_current_observation(name="web-search", as_type="tool") as tool:
    # Do tool work
    results = search_web(query)
    tool.update(output=results)

# Create a generation observation
with langfuse.start_as_current_observation(
    name="answer-generation",
    as_type="generation",
    model="gpt-4"
) as generation:
    # Generate answer
    response = llm.generate(...)
    generation.update(output=response)

Update the current active generation span with new information.

This method updates the current generation span in the active context with additional information. It's useful for adding output, usage stats, or other details that become available during or after model generation.

Arguments:

• name: The generation name
• input: Updated input data for the model
• output: Output from the model (e.g., completions)
• metadata: Additional metadata to associate with the generation
• version: Version identifier for the model or component
• level: Importance level of the generation (info, warning, error)
• status_message: Optional status message for the generation
• completion_start_time: When the model started generating the response
• model: Name/identifier of the AI model used (e.g., "gpt-4")
• model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
• usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
• cost_details: Cost information for the model call
• prompt: Associated prompt template from Langfuse prompt management

Example:

with langfuse.start_as_current_generation(name="answer-query") as generation:
    # Initial setup and API call
    response = llm.generate(...)

    # Update with results that weren't available at creation time
    langfuse.update_current_generation(
        output=response.text,
        usage_details={
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens
        }
    )

Update the current active span with new information.

This method updates the current span in the active context with additional information. It's useful for adding outputs or metadata that become available during execution.

Arguments:

• name: The span name
• input: Updated input data for the operation
• output: Output data from the operation
• metadata: Additional metadata to associate with the span
• version: Version identifier for the code or component
• level: Importance level of the span (info, warning, error)
• status_message: Optional status message for the span

Example:

with langfuse.start_as_current_observation(name="process-data") as span:
    # Initial processing
    result = process_first_part()

    # Update with intermediate results
    langfuse.update_current_span(metadata={"intermediate_result": result})

    # Continue processing
    final_result = process_second_part(result)

    # Final update
    langfuse.update_current_span(output=final_result)

Set trace-level input and output for the current span's trace.

Deprecated since version : This is a legacy method for backward compatibility with Langfuse platform features that still rely on trace-level input/output (e.g., legacy LLM-as-a-judge evaluators). It will be removed in a future major version.

For setting other trace attributes (user_id, session_id, metadata, tags, version), use langfuse.propagate_attributes() (top-level import) instead.

Arguments:

• input: Input data to associate with the trace.
• output: Output data to associate with the trace.

Make the current trace publicly accessible via its URL.

When a trace is published, anyone with the trace link can view the full trace without needing to be logged in to Langfuse. This action cannot be undone programmatically - once published, the entire trace becomes public.

This is a convenience method that publishes the trace from the currently active span context. Use this when you want to make a trace public from within a traced function without needing direct access to the span object.

Create a new Langfuse observation of type 'EVENT'.

The created Langfuse Event observation will be the child of the current span in the context.

Arguments:

• trace_context: Optional context for connecting to an existing trace
• name: Name of the span (e.g., function or operation name)
• input: Input data for the operation (can be any JSON-serializable object)
• output: Output data from the operation (can be any JSON-serializable object)
• metadata: Additional metadata to associate with the span
• version: Version identifier for the code or component
• level: Importance level of the span (info, warning, error)
• status_message: Optional status message for the span

Returns:

The Langfuse Event object

Example:

event = langfuse.create_event(name="process-event")

Create a unique trace ID for use with Langfuse.

This method generates a unique trace ID for use with various Langfuse APIs. It can either generate a random ID or create a deterministic ID based on a seed string.

Trace IDs must be 32 lowercase hexadecimal characters, representing 16 bytes. This method ensures the generated ID meets this requirement. If you need to correlate an external ID with a Langfuse trace ID, use the external ID as the seed to get a valid, deterministic Langfuse trace ID.

Arguments:

• seed: Optional string to use as a seed for deterministic ID generation. If provided, the same seed will always produce the same ID. If not provided, a random ID will be generated.

Returns:

A 32-character lowercase hexadecimal string representing the Langfuse trace ID.

Example:

# Generate a random trace ID
trace_id = langfuse.create_trace_id()

# Generate a deterministic ID based on a seed
session_trace_id = langfuse.create_trace_id(seed="session-456")

# Correlate an external ID with a Langfuse trace ID
external_id = "external-system-123456"
correlated_trace_id = langfuse.create_trace_id(seed=external_id)

# Use the ID with trace context
with langfuse.start_as_current_observation(
    name="process-request",
    trace_context={"trace_id": trace_id}
) as span:
    # Operation will be part of the specific trace
    pass

Create a score for a specific trace or observation.

This method creates a score for evaluating a Langfuse trace or observation. Scores can be used to track quality metrics, user feedback, or automated evaluations.

Arguments:

• name: Name of the score (e.g., "relevance", "accuracy")
• value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL/TEXT/CORRECTION)
• session_id: ID of the Langfuse session to associate the score with
• dataset_run_id: ID of the Langfuse dataset run to associate the score with
• trace_id: ID of the Langfuse trace to associate the score with
• observation_id: Optional ID of the specific observation to score. Trace ID must be provided too.
• score_id: Optional custom ID for the score (auto-generated if not provided)
• data_type: Type of score (NUMERIC, BOOLEAN, CATEGORICAL, TEXT, or CORRECTION)
• comment: Optional comment or explanation for the score
• config_id: Optional ID of a score config defined in Langfuse
• metadata: Optional metadata to be attached to the score
• timestamp: Optional timestamp for the score (defaults to current UTC time)
• environment: Optional environment override for this score. If omitted, the score uses the client-level environment from Langfuse(environment=...) or LANGFUSE_TRACING_ENVIRONMENT. Langfuse observation wrapper methods pass their resolved span environment here so scores created via span.score() or span.score_trace() stay grouped with the scored observation or trace, including request-scoped environments propagated with propagate_attributes(environment=...).

Example:

# Create a numeric score for accuracy
langfuse.create_score(
    name="accuracy",
    value=0.92,
    trace_id="abcdef1234567890abcdef1234567890",
    data_type="NUMERIC",
    comment="High accuracy with minor irrelevant details"
)

# Create a categorical score for sentiment
langfuse.create_score(
    name="sentiment",
    value="positive",
    trace_id="abcdef1234567890abcdef1234567890",
    observation_id="abcdef1234567890",
    data_type="CATEGORICAL"
)

Create a score for the current active span.

This method scores the currently active span in the context. It's a convenient way to score the current operation without needing to know its trace and span IDs. If the active span has a langfuse.environment attribute, including one set by propagate_attributes(environment=...), the score uses that environment. Otherwise it uses the client-level environment.

Arguments:

• name: Name of the score (e.g., "relevance", "accuracy")
• value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL/TEXT/CORRECTION)
• score_id: Optional custom ID for the score (auto-generated if not provided)
• data_type: Type of score (NUMERIC, BOOLEAN, CATEGORICAL, TEXT, or CORRECTION)
• comment: Optional comment or explanation for the score
• config_id: Optional ID of a score config defined in Langfuse
• metadata: Optional metadata to be attached to the score

Example:

with langfuse.start_as_current_generation(name="answer-query") as generation:
    # Generate answer
    response = generate_answer(...)
    generation.update(output=response)

    # Score the generation
    langfuse.score_current_span(
        name="relevance",
        value=0.85,
        data_type="NUMERIC",
        comment="Mostly relevant but contains some tangential information",
        metadata={"model": "gpt-4", "prompt_version": "v2"}
    )

Create a score for the current trace.

This method scores the trace of the currently active span. Unlike score_current_span, this method associates the score with the entire trace rather than a specific span. It's useful for scoring overall performance or quality of the entire operation. If the active span has a langfuse.environment attribute, including one set by propagate_attributes(environment=...), the score uses that environment. Otherwise it uses the client-level environment.

Arguments:

• name: Name of the score (e.g., "user_satisfaction", "overall_quality")
• value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL/TEXT/CORRECTION)
• score_id: Optional custom ID for the score (auto-generated if not provided)
• data_type: Type of score (NUMERIC, BOOLEAN, CATEGORICAL, TEXT, or CORRECTION)
• comment: Optional comment or explanation for the score
• config_id: Optional ID of a score config defined in Langfuse
• metadata: Optional metadata to be attached to the score

Example:

with langfuse.start_as_current_observation(name="process-user-request") as span:
    # Process request
    result = process_complete_request()
    span.update(output=result)

    # Score the overall trace
    langfuse.score_current_trace(
        name="overall_quality",
        value=0.95,
        data_type="NUMERIC",
        comment="High quality end-to-end response",
        metadata={"evaluator": "gpt-4", "criteria": "comprehensive"}
    )

Force flush all pending spans and events to the Langfuse API.

This method manually flushes any pending spans, scores, and other events to the Langfuse API. It's useful in scenarios where you want to ensure all data is sent before proceeding, without waiting for the automatic flush interval.

Example:

# Record some spans and scores
with langfuse.start_as_current_observation(name="operation") as span:
    # Do work...
    pass

# Ensure all data is sent to Langfuse before proceeding
langfuse.flush()

# Continue with other work

Note:

flush() guarantees data was delivered to the API, not that it is 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 and api.trace.get() may raise langfuse.api.NotFoundError right after a successful flush. See the api property docs for a bounded retry pattern, or https://langfuse.com/docs/api-and-data-platform/features/query-via-sdk#ingestion-lag

Shut down the Langfuse client and flush all pending data.

This method cleanly shuts down the Langfuse client, ensuring all pending data is flushed to the API and all background threads are properly terminated.

It's important to call this method when your application is shutting down to prevent data loss and resource leaks. For most applications, using the client as a context manager or relying on the automatic shutdown via atexit is sufficient.

Example:

# Initialize Langfuse
langfuse = Langfuse(public_key="...", secret_key="...")

# Use Langfuse throughout your application
# ...

# When application is shutting down
langfuse.shutdown()

Get the trace ID of the current active span.

This method retrieves the trace ID from the currently active span in the context. It can be used to get the trace ID for referencing in logs, external systems, or for creating related operations.

Returns:

The current trace ID as a 32-character lowercase hexadecimal string, or None if there is no active span.

Example:

with langfuse.start_as_current_observation(name="process-request") as span:
    # Get the current trace ID for reference
    trace_id = langfuse.get_current_trace_id()

    # Use it for external correlation
    log.info(f"Processing request with trace_id: {trace_id}")

    # Or pass to another system
    external_system.process(data, trace_id=trace_id)

Get the observation ID (span ID) of the current active span.

This method retrieves the observation ID from the currently active span in the context. It can be used to get the observation ID for referencing in logs, external systems, or for creating scores or other related operations.

Returns:

The current observation ID as a 16-character lowercase hexadecimal string, or None if there is no active span.

Example:

with langfuse.start_as_current_observation(name="process-user-query") as span:
    # Get the current observation ID
    observation_id = langfuse.get_current_observation_id()

    # Store it for later reference
    cache.set(f"query_{query_id}_observation", observation_id)

    # Process the query...

Get the URL to view a trace in the Langfuse UI.

This method generates a URL that links directly to a trace in the Langfuse UI. It's useful for providing links in logs, notifications, or debugging tools.

Arguments:

• trace_id: Optional trace ID to generate a URL for. If not provided, the trace ID of the current active span will be used.

Returns:

A URL string pointing to the trace in the Langfuse UI, or None if the project ID couldn't be retrieved or no trace ID is available.

Example:

# Get URL for the current trace
with langfuse.start_as_current_observation(name="process-request") as span:
    trace_url = langfuse.get_trace_url()
    log.info(f"Processing trace: {trace_url}")

# Get URL for a specific trace
specific_trace_url = langfuse.get_trace_url(trace_id="1234567890abcdef1234567890abcdef")
send_notification(f"Review needed for trace: {specific_trace_url}")

Fetch a dataset by its name.

Arguments:

• name: The name of the dataset to fetch.
• fetch_items_page_size: All items of the dataset will be fetched in chunks of this size. Defaults to 50.
• version: Retrieve dataset items as they existed at this specific point in time (UTC). If provided, returns the state of items at the specified UTC timestamp. If not provided, returns the latest version. Must be a timezone-aware datetime object in UTC.

Returns:

DatasetClient: The dataset with the given name.

Fetch a dataset run by dataset name and run name.

Arguments:

• dataset_name (str): The name of the dataset.
• run_name (str): The name of the run.

Returns:

DatasetRunWithItems: The dataset run with its items.

Fetch all runs for a dataset.

Arguments:

• dataset_name (str): The name of the dataset.
• page (Optional[int]): Page number, starts at 1.
• limit (Optional[int]): Limit of items per page.

Returns:

PaginatedDatasetRuns: Paginated list of dataset runs.

Delete a dataset run and all its run items. This action is irreversible.

Arguments:

• dataset_name (str): The name of the dataset.
• run_name (str): The name of the run.

Returns:

DeleteDatasetRunResponse: Confirmation of deletion.

Run an experiment on a dataset with automatic tracing and evaluation.

This method executes a task function on each item in the provided dataset, automatically traces all executions with Langfuse for observability, runs item-level and run-level evaluators on the outputs, and returns comprehensive results with evaluation metrics.

The experiment system provides:

• Automatic tracing of all task executions
• Concurrent processing with configurable limits
• Comprehensive error handling that isolates failures
• Integration with Langfuse datasets for experiment tracking
• Flexible evaluation framework supporting both sync and async evaluators

Arguments:

• name: Human-readable name for the experiment. Used for identification in the Langfuse UI.
• run_name: Optional exact name for the experiment run. If provided, this will be used as the exact dataset run name if the data contains Langfuse dataset items. If not provided, this will default to the experiment name appended with an ISO timestamp.
• description: Optional description explaining the experiment's purpose, methodology, or expected outcomes.
• data: Array of data items to process. Can be either:
  • List of dict-like items with 'input', 'expected_output', 'metadata' keys
  • List of Langfuse DatasetItem objects from dataset.items
• task: Function that processes each data item and returns output. Must accept 'item' as keyword argument and can return sync or async results. The task function signature should be: task(, item, *kwargs) -> Any
• evaluators: List of functions to evaluate each item's output individually. Each evaluator receives input, output, expected_output, and metadata. Can return single Evaluation dict or list of Evaluation dicts.
• composite_evaluator: Optional function that creates composite scores from item-level evaluations. Receives the same inputs as item-level evaluators (input, output, expected_output, metadata) plus the list of evaluations from item-level evaluators. Useful for weighted averages, pass/fail decisions based on multiple criteria, or custom scoring logic combining multiple metrics.
• run_evaluators: List of functions to evaluate the entire experiment run. Each run evaluator receives all item_results and can compute aggregate metrics. Useful for calculating averages, distributions, or cross-item comparisons.
• max_concurrency: Maximum number of concurrent task executions (default: 50). Controls the number of items processed simultaneously. Adjust based on API rate limits and system resources.
• metadata: Optional metadata dictionary to attach to all experiment traces. This metadata will be included in every trace created during the experiment. If data are Langfuse dataset items, the metadata will be attached to the dataset run, too.

Returns:

ExperimentResult containing:

• run_name: The experiment run name. This is equal to the dataset run name if experiment was on Langfuse dataset.
• item_results: List of results for each processed item with outputs and evaluations
• run_evaluations: List of aggregate evaluation results for the entire run
• experiment_id: Stable identifier for the experiment run across all items
• dataset_run_id: ID of the dataset run (if using Langfuse datasets)
• dataset_run_url: Direct URL to view results in Langfuse UI (if applicable)

Raises:

• ValueError: If required parameters are missing or invalid
• Exception: If experiment setup fails (individual item failures are handled gracefully)

Examples:

Basic experiment with local data:

def summarize_text(*, item, **kwargs):
    return f"Summary: {item['input'][:50]}..."

def length_evaluator(*, input, output, expected_output=None, **kwargs):
    return {
        "name": "output_length",
        "value": len(output),
        "comment": f"Output contains {len(output)} characters"
    }

result = langfuse.run_experiment(
    name="Text Summarization Test",
    description="Evaluate summarization quality and length",
    data=[
        {"input": "Long article text...", "expected_output": "Expected summary"},
        {"input": "Another article...", "expected_output": "Another summary"}
    ],
    task=summarize_text,
    evaluators=[length_evaluator]
)

print(f"Processed {len(result.item_results)} items")
for item_result in result.item_results:
    print(f"Input: {item_result.item['input']}")
    print(f"Output: {item_result.output}")
    print(f"Evaluations: {item_result.evaluations}")

Advanced experiment with async task and multiple evaluators:

async def llm_task(*, item, **kwargs):
    # Simulate async LLM call
    response = await openai_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": item["input"]}]
    )
    return response.choices[0].message.content

def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):
    if expected_output and expected_output.lower() in output.lower():
        return {"name": "accuracy", "value": 1.0, "comment": "Correct answer"}
    return {"name": "accuracy", "value": 0.0, "comment": "Incorrect answer"}

def toxicity_evaluator(*, input, output, expected_output=None, **kwargs):
    # Simulate toxicity check
    toxicity_score = check_toxicity(output)  # Your toxicity checker
    return {
        "name": "toxicity",
        "value": toxicity_score,
        "comment": f"Toxicity level: {'high' if toxicity_score > 0.7 else 'low'}"
    }

def average_accuracy(*, item_results, **kwargs):
    accuracies = [
        eval.value for result in item_results
        for eval in result.evaluations
        if eval.name == "accuracy"
    ]
    return {
        "name": "average_accuracy",
        "value": sum(accuracies) / len(accuracies) if accuracies else 0,
        "comment": f"Average accuracy across {len(accuracies)} items"
    }

result = langfuse.run_experiment(
    name="LLM Safety and Accuracy Test",
    description="Evaluate model accuracy and safety across diverse prompts",
    data=test_dataset,  # Your dataset items
    task=llm_task,
    evaluators=[accuracy_evaluator, toxicity_evaluator],
    run_evaluators=[average_accuracy],
    max_concurrency=5,  # Limit concurrent API calls
    metadata={"model": "gpt-4", "temperature": 0.7}
)

Using with Langfuse datasets:

# Get dataset from Langfuse
dataset = langfuse.get_dataset("my-eval-dataset")

result = dataset.run_experiment(
    name="Production Model Evaluation",
    description="Monthly evaluation of production model performance",
    task=my_production_task,
    evaluators=[accuracy_evaluator, latency_evaluator]
)

# Results automatically linked to dataset in Langfuse UI
print(f"View results: {result['dataset_run_url']}")

Note:

• Task and evaluator functions can be either synchronous or asynchronous
• Individual item failures are logged but don't stop the experiment
• All executions are automatically traced and visible in Langfuse UI
• When using Langfuse datasets, results are automatically linked for easy comparison
• This method works in both sync and async contexts (Jupyter notebooks, web apps, etc.)
• Async execution is handled automatically with smart event loop detection

Fetch traces or observations and run evaluations on each item.

This method provides a powerful way to evaluate existing data in Langfuse at scale. It fetches items based on filters, transforms them using a mapper function, runs evaluators on each item, and creates scores that are linked back to the original entities. This is ideal for:

• Running evaluations on production traces after deployment
• Backtesting new evaluation metrics on historical data
• Batch scoring of observations for quality monitoring
• Periodic evaluation runs on recent data

The method uses a streaming/pipeline approach to process items in batches, making it memory-efficient for large datasets. It includes comprehensive error handling, retry logic, and resume capability for long-running evaluations.

Arguments:

• scope: The type of items to evaluate. Must be one of:
  • "traces": Evaluate complete traces with all their observations
  • "observations": Evaluate individual observations (spans, generations, events)
• mapper: Function that transforms API response objects into evaluator inputs. Receives a trace/observation object and returns an EvaluatorInputs instance with input, output, expected_output, and metadata fields. Can be sync or async.
• evaluators: List of evaluation functions to run on each item. Each evaluator receives the mapped inputs and returns Evaluation object(s). Evaluator failures are logged but don't stop the batch evaluation.
• filter: Optional JSON filter string for querying items (same format as Langfuse API). Examples:
  • '{"tags": ["production"]}'
  • '{"user_id": "user123", "timestamp": {"operator": ">", "value": "2024-01-01"}}' Default: None (fetches all items).
• fetch_batch_size: Number of items to fetch per API call and hold in memory. Larger values may be faster but use more memory. Default: 50.
• fetch_trace_fields: Comma-separated list of fields to include when fetching traces. Available field groups: 'core' (always included), 'io' (input, output, metadata), 'scores', 'observations', 'metrics'. If not specified, all fields are returned. Example: 'core,scores,metrics'. Note: Excluded 'observations' or 'scores' fields return empty arrays; excluded 'metrics' returns -1 for 'totalCost' and 'latency'. Only relevant if scope is 'traces'.
• max_items: Maximum total number of items to process. If None, processes all items matching the filter. Useful for testing or limiting evaluation runs. Default: None (process all).
• max_concurrency: Maximum number of items to evaluate concurrently. Controls parallelism and resource usage. Default: 5.
• composite_evaluator: Optional function that creates a composite score from item-level evaluations. Receives the original item and its evaluations, returns a single Evaluation. Useful for weighted averages or 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. Uses exponential backoff (1s, 2s, 4s). 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 incomplete run. Allows continuing evaluation after interruption or failure. Default: None.

Returns:

BatchEvaluationResult containing: - total_items_fetched: Number of items fetched from API - total_items_processed: Number of items successfully evaluated - total_items_failed: Number of items that failed evaluation - total_scores_created: Scores created by item-level evaluators - total_composite_scores_created: Scores created by composite evaluator - total_evaluations_failed: Individual evaluator failures - evaluator_stats: Per-evaluator statistics (success rate, scores created) - resume_token: Token for resuming if incomplete (None if completed) - completed: True if all items processed - duration_seconds: Total execution time - failed_item_ids: IDs of items that failed - error_summary: Error types and counts - has_more_items: True if max_items reached but more exist

Raises:

• ValueError: If invalid scope is provided.

Examples:

Basic trace evaluation:

from langfuse import Langfuse, EvaluatorInputs, Evaluation

client = Langfuse()

# Define mapper to extract fields from traces
def trace_mapper(trace):
    return EvaluatorInputs(
        input=trace.input,
        output=trace.output,
        expected_output=None,
        metadata={"trace_id": trace.id}
    )

# Define evaluator
def length_evaluator(*, input, output, expected_output, metadata):
    return Evaluation(
        name="output_length",
        value=len(output) if output else 0
    )

# Run batch evaluation
result = client.run_batched_evaluation(
    scope="traces",
    mapper=trace_mapper,
    evaluators=[length_evaluator],
    filter='{"tags": ["production"]}',
    max_items=1000,
    verbose=True
)

print(f"Processed {result.total_items_processed} traces")
print(f"Created {result.total_scores_created} scores")

Evaluation with composite scorer:

def accuracy_evaluator(*, input, output, expected_output, metadata):
    # ... evaluation logic
    return Evaluation(name="accuracy", value=0.85)

def relevance_evaluator(*, input, output, expected_output, metadata):
    # ... evaluation logic
    return Evaluation(name="relevance", value=0.92)

def composite_evaluator(*, item, evaluations):
    # Weighted average of evaluations
    weights = {"accuracy": 0.6, "relevance": 0.4}
    total = sum(
        e.value * weights.get(e.name, 0)
        for e in evaluations
        if isinstance(e.value, (int, float))
    )
    return Evaluation(
        name="composite_score",
        value=total,
        comment=f"Weighted average of {len(evaluations)} metrics"
    )

result = client.run_batched_evaluation(
    scope="traces",
    mapper=trace_mapper,
    evaluators=[accuracy_evaluator, relevance_evaluator],
    composite_evaluator=composite_evaluator,
    filter='{"user_id": "important_user"}',
    verbose=True
)

Handling incomplete runs with resume:

# Initial run that may fail or timeout
result = client.run_batched_evaluation(
    scope="observations",
    mapper=obs_mapper,
    evaluators=[my_evaluator],
    max_items=10000,
    verbose=True
)

# Check if incomplete
if not result.completed and result.resume_token:
    print(f"Processed {result.resume_token.items_processed} items before interruption")

    # Resume from where it left off
    result = client.run_batched_evaluation(
        scope="observations",
        mapper=obs_mapper,
        evaluators=[my_evaluator],
        resume_from=result.resume_token,
        verbose=True
    )

print(f"Total items processed: {result.total_items_processed}")

Monitoring evaluator performance:

result = client.run_batched_evaluation(...)

for stats in result.evaluator_stats:
    success_rate = stats.successful_runs / stats.total_runs
    print(f"{stats.name}:")
    print(f"  Success rate: {success_rate:.1%}")
    print(f"  Scores created: {stats.total_scores_created}")

    if stats.failed_runs > 0:
        print(f"  ⚠️  Failed {stats.failed_runs} times")

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 with exponential backoff
• All scores are automatically flushed to Langfuse at the end
• The resume mechanism uses timestamp-based filtering to avoid duplicates

Check if the provided credentials (public and secret key) are valid.

Raises:

• Exception: If no projects were found for the provided credentials.

Note:

This method is blocking. It is discouraged to use it in production code.

Create a dataset with the given name on Langfuse.

Arguments:

• name: Name of the dataset to create.
• description: Description of the dataset. Defaults to None.
• metadata: Additional metadata. Defaults to None.
• input_schema: JSON Schema for validating dataset item inputs. When set, all new items will be validated against this schema.
• expected_output_schema: JSON Schema for validating dataset item expected outputs. When set, all new items will be validated against this schema.

Returns:

Dataset: The created dataset as returned by the Langfuse API.

Create a dataset item.

Upserts if an item with id already exists.

Arguments:

• dataset_name: Name of the dataset in which the dataset item should be created.
• input: Input data. Defaults to None. Can contain any dict, list or scalar.
• expected_output: Expected output data. Defaults to None. Can contain any dict, list or scalar.
• metadata: Additional metadata. Defaults to None. Can contain any dict, list or scalar.
• source_trace_id: Id of the source trace. Defaults to None.
• source_observation_id: Id of the source observation. Defaults to None.
• status: Status of the dataset item. Defaults to ACTIVE for newly created items.
• id: Id of the dataset item. Defaults to None. Provide your own id if you want to dedupe dataset items. Id needs to be globally unique and cannot be reused across datasets.

Returns:

DatasetItem: The created dataset item as returned by the Langfuse API.

Example:

from langfuse import Langfuse

langfuse = Langfuse()

# Uploading items to the Langfuse dataset named "capital_cities"
langfuse.create_dataset_item(
    dataset_name="capital_cities",
    input={"input": {"country": "Italy"}},
    expected_output={"expected_output": "Rome"},
    metadata={"foo": "bar"}
)

Replace media reference strings in an object with base64 data URIs.

This method recursively traverses an object (up to max_depth) looking for media reference strings in the format "@@@langfuseMedia:...@@@". When found, it (synchronously) fetches the actual media content using the provided Langfuse client and replaces the reference string with a base64 data URI.

If fetching media content fails for a reference string, a warning is logged and the reference string is left unchanged.

Arguments:

• obj: The object to process. Can be a primitive value, array, or nested object. If the object has a __dict__ attribute, a dict will be returned instead of the original object type.
• resolve_with: The representation of the media content to replace the media reference string with. Currently only "base64_data_uri" is supported.
• max_depth: int: The maximum depth to traverse the object. Default is 10.
• content_fetch_timeout_seconds: int: The timeout in seconds for fetching media content. Default is 5.

Returns:

A deep copy of the input object with all media references replaced with base64 data URIs where possible. If the input object has a __dict__ attribute, a dict will be returned instead of the original object type.

Example:

obj = { "image": "@@@langfuseMedia:type=image/jpeg|id=123|source=bytes@@@", "nested": { "pdf": "@@@langfuseMedia:type=application/pdf|id=456|source=bytes@@@" } }

result = await LangfuseMedia.resolve_media_references(obj, langfuse_client)

Result:

{

"image": "data:image/jpeg;base64,/9j/4AAQSkZJRg...",

"nested": {

"pdf": "data:application/pdf;base64,JVBERi0xLjcK..."

}

}

Get a prompt.

This method attempts to fetch the requested prompt from the local cache. If the prompt is not found in the cache or if the cached prompt has expired, it will try to fetch the prompt from the server again and update the cache. If fetching the new prompt fails, and there is an expired prompt in the cache, it will return the expired prompt as a fallback.

Arguments:

• name (str): The name of the prompt to retrieve.

Keyword Args:

version (Optional[int]): The version of the prompt to retrieve. If no label and version is specified, the production label is returned. Specify either version or label, not both. label: Optional[str]: The label of the prompt to retrieve. If no label and version is specified, the production label is returned. Specify either version or label, not both. cache_ttl_seconds: Optional[int]: Time-to-live in seconds for caching the prompt. Must be specified as a keyword argument. If not set, defaults to 60 seconds. Disables caching if set to 0. type: Literal["chat", "text"]: The type of the prompt to retrieve. Defaults to "text". fallback: Union[Optional[List[ChatMessageDict]], Optional[str]]: The prompt string to return if fetching the prompt fails. Important on the first call where no cached prompt is available. Follows Langfuse prompt formatting with double curly braces for variables. Defaults to None. max_retries: Optional[int]: The maximum number of retries in case of API/network errors. Defaults to 2. The maximum value is 4. Retries have an exponential backoff with a maximum delay of 10 seconds. fetch_timeout_seconds: Optional[int]: The timeout in milliseconds for fetching the prompt. Defaults to the default timeout set on the SDK, which is 5 seconds per default.

Returns:

The prompt object retrieved from the cache or directly fetched if not cached or expired of type

• TextPromptClient, if type argument is 'text'.
• ChatPromptClient, if type argument is 'chat'.

Raises:

• Exception: Propagates any exceptions raised during the fetching of a new prompt, unless there is an
• expired prompt in the cache, in which case it logs a warning and returns the expired prompt.

Create a new prompt in Langfuse.

Keyword Args:

name : The name of the prompt to be created. prompt : The content of the prompt to be created. is_active [DEPRECATED] : A flag indicating whether the prompt is active or not. This is deprecated and will be removed in a future release. Please use the 'production' label instead. labels: The labels of the prompt. Defaults to None. To create a default-served prompt, add the 'production' label. tags: The tags of the prompt. Defaults to None. Will be applied to all versions of the prompt. config: Additional structured data to be saved with the prompt. Defaults to None. type: The type of the prompt to be created. "chat" vs. "text". Defaults to "text". commit_message: Optional string describing the change.

Returns:

TextPromptClient: The prompt if type argument is 'text'. ChatPromptClient: The prompt if type argument is 'chat'.

Update an existing prompt version in Langfuse. The Langfuse SDK prompt cache is invalidated for all prompts witht he specified name.

Arguments:

• name (str): The name of the prompt to update.
• version (int): The version number of the prompt to update.
• new_labels (List[str], optional): New labels to assign to the prompt version. Labels are unique across versions. The "latest" label is reserved and managed by Langfuse. Defaults to [].

Returns:

Prompt: The updated prompt from the Langfuse API.

Clear the entire prompt cache, removing all cached prompts.

This method is useful when you want to force a complete refresh of all cached prompts, for example after major updates or when you need to ensure the latest versions are fetched from the server.

Initialize a LangfuseMedia object.

Arguments:

• obj: The object to wrap.
• base64_data_uri: A base64-encoded data URI containing the media content and content type (e.g., "data:image/jpeg;base64,/9j/4AAQ...").
• content_type: The MIME type of the media content when providing raw bytes or reading from a file.
• content_bytes: Raw bytes of the media content.
• file_path: The path to the file containing the media content. For relative paths, the current working directory is used.

Parse a media reference string into a ParsedMediaReference.

Example reference string:

"@@@langfuseMedia:type=image/jpeg|id=some-uuid|source=base64_data_uri@@@"

Arguments:

• reference_string: The reference string to parse.

Returns:

A TypedDict with the media_id, source, and content_type.

Raises:

• ValueError: If the reference string is empty or not a string.
• ValueError: If the reference string does not start with "@@@langfuseMedia:type=".
• ValueError: If the reference string does not end with "@@@".
• ValueError: If the reference string is missing required fields.

Replace media reference strings in an object with base64 data URIs.

This method recursively traverses an object (up to max_depth) looking for media reference strings in the format "@@@langfuseMedia:...@@@". When found, it (synchronously) fetches the actual media content using the provided Langfuse client and replaces the reference string with a base64 data URI.

If fetching media content fails for a reference string, a warning is logged and the reference string is left unchanged.

Arguments:

• obj: The object to process. Can be a primitive value, array, or nested object. If the object has a __dict__ attribute, a dict will be returned instead of the original object type.
• langfuse_client: Langfuse client instance used to fetch media content.
• resolve_with: The representation of the media content to replace the media reference string with. Currently only "base64_data_uri" is supported.
• max_depth: Optional. Default is 10. The maximum depth to traverse the object.

Returns:

A deep copy of the input object with all media references replaced with base64 data URIs where possible. If the input object has a __dict__ attribute, a dict will be returned instead of the original object type.

Example:

obj = { "image": "@@@langfuseMedia:type=image/jpeg|id=123|source=bytes@@@", "nested": { "pdf": "@@@langfuseMedia:type=application/pdf|id=456|source=bytes@@@" } }

result = await LangfuseMedia.resolve_media_references(obj, langfuse_client)

Result:

{

"image": "data:image/jpeg;base64,/9j/4AAQSkZJRg...",

"nested": {

"pdf": "data:application/pdf;base64,JVBERi0xLjcK..."

}

}

Return whether the signed URL is already expired.

Fetch the media content from the signed URL.

Arguments:

• timeout: Request timeout in seconds.
• client: Optional httpx client to use for the request. Pass this to honor custom transport settings (proxy, CA bundle, mTLS) — in particular when multiple Langfuse clients are configured, since the SDK cannot otherwise tell which client produced this reference. When omitted, the single configured client is used, falling back to a default httpx client.

Fetch media and return raw base64 without a data URI prefix.

See fetch_bytes() for the client argument.

Fetch media and return it as a data URI.

See fetch_bytes() for the client argument.

Initialize a new LangfuseSpan.

Arguments:

• otel_span: The OpenTelemetry span to wrap
• langfuse_client: Reference to the parent Langfuse client
• input: Input data for the span (any JSON-serializable object)
• output: Output data from the span (any JSON-serializable object)
• metadata: Additional metadata to associate with the span
• environment: The tracing environment
• release: Release identifier for the application
• version: Version identifier for the code or component
• level: Importance level of the span (info, warning, error)
• status_message: Optional status message for the span

Initialize a new LangfuseGeneration span.

Arguments:

• otel_span: The OpenTelemetry span to wrap
• langfuse_client: Reference to the parent Langfuse client
• input: Input data for the generation (e.g., prompts)
• output: Output from the generation (e.g., completions)
• metadata: Additional metadata to associate with the generation
• environment: The tracing environment
• release: Release identifier for the application
• version: Version identifier for the model or component
• level: Importance level of the generation (info, warning, error)
• status_message: Optional status message for the generation
• completion_start_time: When the model started generating the response
• model: Name/identifier of the AI model used (e.g., "gpt-4")
• model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
• usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
• cost_details: Cost information for the model call
• prompt: Associated prompt template from Langfuse prompt management

Initialize a new LangfuseEvent span.

Arguments:

• otel_span: The OpenTelemetry span to wrap
• langfuse_client: Reference to the parent Langfuse client
• input: Input data for the event
• output: Output from the event
• metadata: Additional metadata to associate with the generation
• environment: The tracing environment
• release: Release identifier for the application
• version: Version identifier for the model or component
• level: Importance level of the generation (info, warning, error)
• status_message: Optional status message for the generation

Update is not allowed for LangfuseEvent because events cannot be updated.

This method logs a warning and returns self without making changes.

Returns:

self: Returns the unchanged LangfuseEvent instance

Initialize a new LangfuseAgent span.

Initialize a new LangfuseTool span.

Initialize a new LangfuseChain span.

Initialize a new LangfuseEmbedding span.

Initialize a new LangfuseEvaluator span.

Initialize a new LangfuseRetriever span.

Initialize a new LangfuseGuardrail span.