diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py index 4660b50f0..43a85c2fd 100644 --- a/langfuse/_client/attributes.py +++ b/langfuse/_client/attributes.py @@ -68,6 +68,7 @@ class LangfuseOtelSpanAttributes: EXPERIMENT_METADATA = "langfuse.experiment.metadata" EXPERIMENT_DATASET_ID = "langfuse.experiment.dataset.id" EXPERIMENT_ITEM_ID = "langfuse.experiment.item.id" + EXPERIMENT_ITEM_VERSION = "langfuse.experiment.item.version" EXPERIMENT_ITEM_EXPECTED_OUTPUT = "langfuse.experiment.item.expected_output" EXPERIMENT_ITEM_METADATA = "langfuse.experiment.item.metadata" EXPERIMENT_ITEM_ROOT_OBSERVATION_ID = "langfuse.experiment.item.root_observation_id" diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index adf210a3a..dbcb8c9ee 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -2533,8 +2533,9 @@ def run_experiment( Args: 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. + run_name: Optional exact name for the experiment run. If the `data` contains + Langfuse dataset items, runs with the same `run_name` on the same dataset + share one experiment ID and appear as a single experiment in Langfuse. 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. @@ -2558,17 +2559,22 @@ def run_experiment( 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. + This metadata will be included in every trace created during the experiment + and is shown as the experiment's metadata in Langfuse. Returns: ExperimentResult containing: - - run_name: The experiment run name. This is equal to the dataset run name if experiment was on Langfuse dataset. + - run_name: The experiment run name, shown as the experiment name in Langfuse. - 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) + - run_evaluations: List of aggregate evaluation results for the entire run. + They are stored as scores on the experiment run, also for local data. + - experiment_id: Identifier for the experiment run across all items. For + Langfuse dataset items it is derived from the project ID, dataset ID and + run name; for local data it is random. + - experiment_url: Direct URL to view results in Langfuse UI (None if the + project ID could not be resolved) + - dataset_run_id / dataset_run_url: Deprecated aliases of experiment_id / + experiment_url Raises: ValueError: If required parameters are missing or invalid @@ -2666,14 +2672,16 @@ def average_accuracy(*, item_results, **kwargs): ) # Results automatically linked to dataset in Langfuse UI - print(f"View results: {result['dataset_run_url']}") + print(f"View results: {result.experiment_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 + - Experiment runs are recorded only through the exported traces; no + separate dataset run is created via the API + - When using Langfuse datasets, results are automatically linked to the dataset 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 """ @@ -2717,7 +2725,34 @@ async def _run_experiment_async( "Starting experiment '%s' run '%s' with %s items", name, run_name, len(data) ) - shared_fallback_experiment_id = self._create_observation_id() + project_id: Optional[str] = None + if self._tracing_enabled: + try: + project_id = await asyncio.to_thread(self._get_project_id) + except Exception as e: + langfuse_logger.warning( + "Failed to resolve project id for experiment: %s", e + ) + + # One experiment id per run: mixed-dataset data uses the first dataset item's dataset. + experiment_dataset_id = next( + ( + getattr(item, "dataset_id", None) + for item in data + if not isinstance(item, dict) and getattr(item, "dataset_id", None) + ), + None, + ) + experiment_id = self._create_experiment_id( + project_id=project_id, + dataset_id=experiment_dataset_id, + run_name=run_name, + ) + experiment_item_version = ( + self._format_experiment_item_version(dataset_version) + if dataset_version is not None + else None + ) # Set up concurrency control semaphore = asyncio.Semaphore(max_concurrency) @@ -2730,12 +2765,12 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult: task, evaluators, composite_evaluator, - shared_fallback_experiment_id, + experiment_id, name, run_name, description, metadata, - dataset_version, + experiment_item_version, ) # Run all items concurrently @@ -2761,47 +2796,40 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult: except Exception as e: langfuse_logger.error("Run evaluator failed: %s", e) - # Generate dataset run URL if applicable - dataset_run_id = next( - ( - result.dataset_run_id - for result in valid_results - if result.dataset_run_id - ), - None, + experiment_url = ( + f"{self._base_url}/project/{project_id}/experiments/results?baseline={experiment_id}" + if project_id + else None ) - dataset_run_url = None - if dataset_run_id and data: - try: - # Check if the first item has dataset_id (for DatasetItem objects) - first_item = data[0] - dataset_id = None - if hasattr(first_item, "dataset_id"): - dataset_id = getattr(first_item, "dataset_id", None) - - if dataset_id: - project_id = self._get_project_id() - - if project_id: - dataset_run_url = f"{self._base_url}/project/{project_id}/datasets/{dataset_id}/runs/{dataset_run_id}" - - except Exception: - pass # URL generation is optional + # Without an exported item span the experiment does not exist on the server, + # so a run score would be orphaned. + stored_run_evaluations = ( + run_evaluations + if any( + result.trace_id is not None and self._is_trace_sampled(result.trace_id) + for result in valid_results + ) + else [] + ) + if run_evaluations and not stored_run_evaluations: + langfuse_logger.debug( + "Skipping run-level scores for experiment %s: all items were sampled out.", + experiment_id, + ) - # Store run-level evaluations as scores - for evaluation in run_evaluations: + # Run-level scores attach to the experiment via dataset_run_id == experiment_id. + for evaluation in stored_run_evaluations: try: - if dataset_run_id: - self.create_score( - dataset_run_id=dataset_run_id, - name=evaluation.name or "", - value=evaluation.value, # type: ignore - comment=evaluation.comment, - metadata=evaluation.metadata, - data_type=evaluation.data_type, # type: ignore - config_id=evaluation.config_id, - ) + self.create_score( + dataset_run_id=experiment_id, + name=evaluation.name or "", + value=evaluation.value, # type: ignore + comment=evaluation.comment, + metadata=evaluation.metadata, + data_type=evaluation.data_type, # type: ignore + config_id=evaluation.config_id, + ) except Exception as e: langfuse_logger.error("Failed to store run evaluation: %s", e) @@ -2815,23 +2843,75 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult: description=description, item_results=valid_results, run_evaluations=run_evaluations, - experiment_id=dataset_run_id or shared_fallback_experiment_id, - dataset_run_id=dataset_run_id, - dataset_run_url=dataset_run_url, + experiment_id=experiment_id, + experiment_url=experiment_url, ) + def _create_experiment_id( + self, + *, + project_id: Optional[str], + dataset_id: Optional[str], + run_name: str, + ) -> str: + if dataset_id is None: + return self._create_observation_id() + + if project_id is None: + langfuse_logger.warning( + "Could not resolve the project id; using a random experiment id for run '%s'. " + "Re-runs with the same run name will not be grouped into the same experiment.", + run_name, + ) + return self._create_observation_id() + + import json + + # Must match the Langfuse server's stable experiment id derivation byte for + # byte (JS JSON.stringify of the same array, SHA-256, first 16 hex chars). + payload = json.dumps( + ["langfuse-experiment-v1", project_id, dataset_id, run_name], + separators=(",", ":"), + ensure_ascii=False, + ) + + return sha256(payload.encode("utf-8")).hexdigest()[:16] + + def _is_trace_sampled(self, trace_id: str) -> bool: + from opentelemetry.sdk.trace.sampling import Decision + + tracer_provider = self._resources.tracer_provider if self._resources else None + sampler = getattr(tracer_provider, "sampler", None) + if sampler is None: + return True + + try: + decision = sampler.should_sample( + parent_context=None, trace_id=int(trace_id, 16), name="experiment" + ).decision + except Exception: + return True + + return bool(decision == Decision.RECORD_AND_SAMPLE) + + @staticmethod + def _format_experiment_item_version(dataset_version: datetime) -> str: + from langfuse.api.core.datetime_utils import serialize_datetime + + return serialize_datetime(dataset_version) + async def _process_experiment_item( self, item: ExperimentItem, task: Callable, evaluators: List[Callable], composite_evaluator: Optional[CompositeEvaluatorFunction], - fallback_experiment_id: str, + experiment_id: str, experiment_name: str, experiment_run_name: str, experiment_description: Optional[str], experiment_metadata: Optional[Dict[str, Any]] = None, - dataset_version: Optional[datetime] = None, + experiment_item_version: Optional[str] = None, ) -> ExperimentItemResult: with self.start_as_current_observation(name="experiment-item-run") as span: try: @@ -2866,7 +2946,6 @@ async def _process_experiment_item( trace_id = span.trace_id dataset_id = None dataset_item_id = None - dataset_run_id = None if ( not isinstance(item, dict) @@ -2891,6 +2970,9 @@ async def _process_experiment_item( LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_EXPECTED_OUTPUT: _serialize( expected_output ), + LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_VERSION: ( + experiment_item_version if dataset_id else None + ), }.items() if v is not None } @@ -2904,32 +2986,6 @@ async def _process_experiment_item( ) as task_span: task_span._otel_span.set_attributes(experiment_span_attributes) - # Link dataset runs to the canonical task observation so their - # latency excludes the subsequent evaluator subtree. - if hasattr(item, "id") and hasattr(item, "dataset_id"): - try: - # Use sync API to avoid event loop issues when - # run_async_safely creates multiple event loops across - # different threads. - dataset_run_item = await asyncio.to_thread( - self.api.dataset_run_items.create, - run_name=experiment_run_name, - run_description=experiment_description, - metadata=experiment_metadata, - dataset_item_id=item.id, # type: ignore - trace_id=trace_id, - observation_id=task_span.id, - dataset_version=dataset_version, - ) - - dataset_run_id = dataset_run_item.dataset_run_id - - except Exception as e: - langfuse_logger.error( - "Failed to create dataset run item: %s", e - ) - - experiment_id = dataset_run_id or fallback_experiment_id propagated_experiment_attributes = PropagatedExperimentAttributes( experiment_id=experiment_id, experiment_name=experiment_run_name, @@ -3131,7 +3187,7 @@ async def _process_experiment_item( output=output, evaluations=evaluations, trace_id=trace_id, - dataset_run_id=dataset_run_id, + experiment_id=experiment_id, ) def _create_experiment_run_name( diff --git a/langfuse/_client/datasets.py b/langfuse/_client/datasets.py index e28ddbebf..1070c76d3 100644 --- a/langfuse/_client/datasets.py +++ b/langfuse/_client/datasets.py @@ -94,11 +94,11 @@ def run_experiment( """Run an experiment on this Langfuse dataset with automatic tracking. This is a convenience method that runs an experiment using all items in this - dataset. It automatically creates a dataset run in Langfuse for tracking and - comparison purposes, linking all experiment results to the dataset. + dataset. All results are recorded as one experiment in Langfuse that is linked + to the dataset for tracking and comparison purposes. Key benefits of using dataset.run_experiment(): - - Automatic dataset run creation and linking in Langfuse UI + - Automatic experiment tracking and linking to the dataset in Langfuse UI - Built-in experiment tracking and versioning - Easy comparison between different experiment runs - Direct access to dataset items with their metadata and expected outputs @@ -106,10 +106,11 @@ def run_experiment( Args: name: Human-readable name for the experiment run. This will be used as - the dataset run name in Langfuse for tracking and identification. - run_name: Optional exact name for the dataset run. If provided, this will be - used as the exact dataset run name in Langfuse. If not provided, this will - default to the experiment name appended with an ISO timestamp. + the base of the experiment run name in Langfuse for tracking and identification. + run_name: Optional exact name for the experiment run. Runs with the same + `run_name` on this dataset share one experiment ID and appear as a single + experiment in Langfuse. If not provided, this will default to the + experiment name appended with an ISO timestamp. description: Optional description of the experiment's purpose, methodology, or what you're testing. Appears in the Langfuse UI for context. task: Function that processes each dataset item and returns output. @@ -131,12 +132,14 @@ def run_experiment( Returns: ExperimentResult object containing: - name: The experiment name. - - run_name: The experiment run name (equivalent to the dataset run name). + - run_name: The experiment run name, shown as the experiment name in Langfuse. - description: Optional experiment description. - item_results: Results for each dataset item with outputs and evaluations. - run_evaluations: Aggregate evaluation results for the entire run. - - dataset_run_id: ID of the created dataset run in Langfuse. - - dataset_run_url: Direct URL to view the experiment results in Langfuse UI. + - experiment_id: ID of the experiment run in Langfuse. + - experiment_url: Direct URL to view the experiment results in Langfuse UI. + - dataset_run_id / dataset_run_url: Deprecated aliases of experiment_id / + experiment_url. The result object provides a format() method for human-readable output: ```python @@ -176,8 +179,8 @@ def accuracy_evaluator(*, input, output, expected_output=None, **kwargs): evaluators=[accuracy_evaluator] ) - print(f"Evaluated {len(result['item_results'])} questions") - print(f"View detailed results: {result['dataset_run_url']}") + print(f"Evaluated {len(result.item_results)} questions") + print(f"View detailed results: {result.experiment_url}") ``` Advanced experiment with multiple evaluators and run-level analysis: @@ -244,11 +247,11 @@ def content_diversity(*, item_results, **kwargs): ) # Results are automatically linked to dataset in Langfuse - print(f"Experiment completed! View in Langfuse: {result['dataset_run_url']}") + print(f"Experiment completed! View in Langfuse: {result.experiment_url}") # Access individual results - for i, item_result in enumerate(result["item_results"]): - print(f"Item {i+1}: {item_result['evaluations']}") + for i, item_result in enumerate(result.item_results): + print(f"Item {i+1}: {item_result.evaluations}") ``` Comparing different model versions: @@ -274,15 +277,15 @@ def content_diversity(*, item_results, **kwargs): # Both experiments are now visible in Langfuse for easy comparison print("Compare results in Langfuse:") - print(f"GPT-4: {result_gpt4.dataset_run_url}") - print(f"Custom: {result_custom.dataset_run_url}") + print(f"GPT-4: {result_gpt4.experiment_url}") + print(f"Custom: {result_custom.experiment_url}") ``` Note: - - All experiment results are automatically tracked in Langfuse as dataset runs + - All experiment results are automatically tracked in Langfuse as experiments on this dataset - Dataset items provide .input, .expected_output, and .metadata attributes - Results can be easily compared across different experiment runs in the UI - - The dataset_run_url provides direct access to detailed results and analysis + - The experiment_url provides direct access to detailed results and analysis - Failed items are handled gracefully and logged without stopping the experiment - This method works in both sync and async contexts (Jupyter notebooks, web apps, etc.) - Async execution is handled automatically with smart event loop detection diff --git a/langfuse/experiment.py b/langfuse/experiment.py index aa5481829..ee29d8611 100644 --- a/langfuse/experiment.py +++ b/langfuse/experiment.py @@ -6,6 +6,7 @@ """ import asyncio +import warnings from datetime import datetime from typing import ( TYPE_CHECKING, @@ -216,6 +217,14 @@ def __init__( self.config_id = config_id +def _warn_deprecated_alias(old: str, new: str) -> None: + warnings.warn( + f"{old} is deprecated and will be removed in a future major version. Use {new} instead.", + DeprecationWarning, + stacklevel=3, + ) + + class ExperimentItemResult: """Result structure for individual experiment items. @@ -233,8 +242,8 @@ class ExperimentItemResult: contains a name, value, optional comment, and optional metadata. trace_id: Optional Langfuse trace ID for this item's execution. Used to link the experiment result with the detailed trace in Langfuse UI. - dataset_run_id: Optional dataset run ID if this item was part of a - Langfuse dataset. None for local experiments. + experiment_id: ID of the experiment run this item belongs to. + dataset_run_id: Deprecated alias of `experiment_id`. Examples: Accessing item result data: @@ -275,7 +284,8 @@ def __init__( output: Any, evaluations: List[Evaluation], trace_id: Optional[str], - dataset_run_id: Optional[str], + experiment_id: Optional[str] = None, + dataset_run_id: Optional[str] = None, ): """Initialize an ExperimentItemResult with the provided data. @@ -284,7 +294,9 @@ def __init__( output: The actual output produced by the task function for this item. evaluations: List of evaluation results for this item. trace_id: Optional Langfuse trace ID for this item's execution. - dataset_run_id: Optional dataset run ID if this item was part of a Langfuse dataset. + experiment_id: ID of the experiment run this item belongs to. + dataset_run_id: Deprecated alias of `experiment_id`. Used only when + `experiment_id` is not provided. Note: All arguments must be provided as keywords. Positional arguments will raise a TypeError. @@ -293,7 +305,13 @@ def __init__( self.output = output self.evaluations = evaluations self.trace_id = trace_id - self.dataset_run_id = dataset_run_id + self.experiment_id = experiment_id or dataset_run_id + + @property + def dataset_run_id(self) -> Optional[str]: + """Deprecated alias of `experiment_id`.""" + _warn_deprecated_alias("ExperimentItemResult.dataset_run_id", "experiment_id") + return self.experiment_id class ExperimentResult: @@ -312,10 +330,13 @@ class ExperimentResult: run_evaluations: List of aggregate evaluation results computed across all items, such as average scores, statistical summaries, or cross-item analyses. experiment_id: ID of the experiment run propagated across all items. For - Langfuse datasets, this matches the dataset run ID. For local experiments, - this is a stable SDK-generated identifier for the run. - dataset_run_id: Optional ID of the dataset run in Langfuse (when using Langfuse datasets). - dataset_run_url: Optional direct URL to view the experiment results in Langfuse UI. + Langfuse datasets, it is derived from the project, dataset and run name, + so re-running with the same `run_name` adds to the same experiment. For + local experiments, it is a random SDK-generated identifier for the run. + experiment_url: Optional direct URL to view the experiment results in the + Langfuse UI. None if the project ID could not be resolved. + dataset_run_id: Deprecated alias of `experiment_id`. + dataset_run_url: Deprecated alias of `experiment_url`. Examples: Basic usage with local dataset: @@ -347,8 +368,8 @@ class ExperimentResult: ) # View in Langfuse UI - if result.dataset_run_url: - print(f"View detailed results: {result.dataset_run_url}") + if result.experiment_url: + print(f"View detailed results: {result.experiment_url}") ``` Formatted output: @@ -373,6 +394,7 @@ def __init__( item_results: List[ExperimentItemResult], run_evaluations: List[Evaluation], experiment_id: str, + experiment_url: Optional[str] = None, dataset_run_id: Optional[str] = None, dataset_run_url: Optional[str] = None, ): @@ -385,8 +407,11 @@ def __init__( item_results: List of results from processing individual dataset items. run_evaluations: List of aggregate evaluation results for the entire run. experiment_id: ID of the experiment run. - dataset_run_id: Optional ID of the dataset run (for Langfuse datasets). - dataset_run_url: Optional URL to view results in Langfuse UI. + experiment_url: Optional URL to view results in Langfuse UI. + dataset_run_id: Deprecated and ignored; `dataset_run_id` always + returns `experiment_id`. + dataset_run_url: Deprecated alias of `experiment_url`. Used only when + `experiment_url` is not provided. """ self.name = name self.run_name = run_name @@ -394,8 +419,19 @@ def __init__( self.item_results = item_results self.run_evaluations = run_evaluations self.experiment_id = experiment_id - self.dataset_run_id = dataset_run_id - self.dataset_run_url = dataset_run_url + self.experiment_url = experiment_url or dataset_run_url + + @property + def dataset_run_id(self) -> str: + """Deprecated alias of `experiment_id`.""" + _warn_deprecated_alias("ExperimentResult.dataset_run_id", "experiment_id") + return self.experiment_id + + @property + def dataset_run_url(self) -> Optional[str]: + """Deprecated alias of `experiment_url`.""" + _warn_deprecated_alias("ExperimentResult.dataset_run_url", "experiment_url") + return self.experiment_url def format(self, *, include_item_results: bool = False) -> str: r"""Format the experiment result for human-readable display. @@ -426,7 +462,7 @@ def format(self, *, include_item_results: bool = False) -> str: - List of all evaluation metrics that were applied - Average scores across all items for each numeric metric - Run-level evaluation results with comments - - Dataset run URL for viewing in Langfuse UI (if applicable) + - Experiment URL for viewing in Langfuse UI (if available) - Individual item details including inputs, outputs, and scores (if requested) Examples: @@ -588,9 +624,8 @@ def format(self, *, include_item_results: bool = False) -> str: output += f"\n πŸ’­ {run_eval.comment}" output += "\n" - # Add dataset run URL if available - if self.dataset_run_url: - output += f"\nπŸ”— Dataset Run:\n {self.dataset_run_url}" + if self.experiment_url: + output += f"\nπŸ”— Experiment:\n {self.experiment_url}" return output @@ -845,7 +880,7 @@ def __call__( - output: The task function's output for this item - evaluations: List of item-level evaluation results - trace_id: Langfuse trace ID for this execution - - dataset_run_id: Dataset run ID (if using Langfuse datasets) + - experiment_id: ID of the experiment run Note: This list only includes items that were successfully processed. Failed items are excluded but logged separately. @@ -1106,7 +1141,7 @@ def __init__( dataset_version: Optional pinned dataset version. Injected by the action when ``dataset_version`` is configured. metadata: Default metadata attached to every experiment trace and - the dataset run. The action injects GitHub-sourced tags (SHA, + the experiment run. The action injects GitHub-sourced tags (SHA, PR link, workflow run link, branch, GH user, etc.). Merged with any ``metadata`` passed to :meth:`run_experiment`, with user-supplied keys winning on collision. diff --git a/tests/e2e/test_experiments.py b/tests/e2e/test_experiments.py index 80881f144..ceddba7be 100644 --- a/tests/e2e/test_experiments.py +++ b/tests/e2e/test_experiments.py @@ -1,20 +1,121 @@ """Comprehensive tests for Langfuse experiment functionality matching JS SDK.""" +import os import time -from typing import Any, Dict, List +from datetime import datetime, timedelta, timezone +from typing import Any, Callable, Dict, List, TypeVar +from uuid import uuid4 import pytest from opentelemetry import trace as otel_trace_api from langfuse import get_client from langfuse._client.attributes import LangfuseOtelSpanAttributes +from langfuse.api import LangfuseAPI from langfuse.experiment import ( Evaluation, ExperimentData, ExperimentItem, ExperimentItemResult, ) -from tests.support.utils import create_uuid, get_api, wait_for_trace + +T = TypeVar("T") + +READ_TIMEOUT_SECONDS = float(os.environ.get("LANGFUSE_E2E_READ_TIMEOUT_SECONDS", "30")) +READ_INTERVAL_SECONDS = float( + os.environ.get("LANGFUSE_E2E_READ_INTERVAL_SECONDS", "0.5") +) +ITEM_FIELDS = "core,dataset,io,metadata,experimentMetadata,scores" + + +def create_uuid() -> str: + return str(uuid4()) + + +def get_api() -> LangfuseAPI: + return LangfuseAPI( + username=os.environ.get("LANGFUSE_PUBLIC_KEY"), + password=os.environ.get("LANGFUSE_SECRET_KEY"), + base_url=os.environ.get("LANGFUSE_BASE_URL"), + ) + + +def _read_window_start() -> datetime: + return datetime.now(timezone.utc) - timedelta(hours=1) + + +def poll(operation: Callable[[], T], is_ready: Callable[[T], bool]) -> T: + """Re-read until ready or timeout; returns the last result for the caller to assert.""" + deadline = time.monotonic() + READ_TIMEOUT_SECONDS + while True: + result = operation() + if is_ready(result) or time.monotonic() >= deadline: + return result + time.sleep(READ_INTERVAL_SECONDS) + + +def get_experiment_items( + experiment_id: str, *, expected_count: int, with_scores: bool = False +) -> list: + api = get_api() + return poll( + lambda: ( + api.experiments.list_items( + from_start_time=_read_window_start(), + experiment_id=experiment_id, + fields=ITEM_FIELDS, + limit=100, + ).data + ), + lambda items: ( + len(items) >= expected_count + and (not with_scores or all(item.scores for item in items)) + ), + ) + + +def get_experiment( + experiment_id: str, *, is_ready: Callable[[Any], bool] = lambda _: True +): + api = get_api() + experiments = poll( + lambda: ( + api.experiments.list( + from_start_time=_read_window_start(), + id=experiment_id, + fields="core,metadata,scores", + ).data + ), + lambda data: len(data) == 1 and is_ready(data[0]), + ) + assert len(experiments) == 1, f"Experiment {experiment_id} should exist" + return experiments[0] + + +def get_observations(trace_id: str, *, is_ready: Callable[[list], bool]) -> list: + api = get_api() + return poll( + lambda: ( + api.observations.get_many( + trace_id=trace_id, + fields="core,basic,io,metadata", + from_start_time=_read_window_start(), + ).data + ), + is_ready, + ) + + +def get_trace_scores(trace_id: str, *, expected_count: int) -> list: + api = get_api() + return poll( + lambda: api.scores_v3.get_many_v3(trace_id=trace_id, fields="subject").data, + lambda scores: len(scores) >= expected_count, + ) + + +def has_score(name: str) -> Callable[[Any], bool]: + return lambda experiment: any(s.name == name for s in experiment.scores or []) @pytest.fixture @@ -78,63 +179,53 @@ def test_run_experiment_on_local_dataset(sample_dataset): assert len(result.item_results) == 3 assert len(result.run_evaluations) == 1 assert result.run_evaluations[0].name == "average_length" - assert result.dataset_run_id is None # No dataset_run_id for local datasets + assert len(result.experiment_id) == 16 + assert result.experiment_url is not None + assert result.experiment_url.endswith( + f"/experiments/results?baseline={result.experiment_id}" + ) + with pytest.warns(DeprecationWarning): + assert result.dataset_run_id == result.experiment_id # Validate item results structure for item_result in result.item_results: assert hasattr(item_result, "output") assert hasattr(item_result, "evaluations") assert hasattr(item_result, "trace_id") - assert ( - item_result.dataset_run_id is None - ) # No dataset_run_id for local datasets + assert item_result.experiment_id == result.experiment_id assert len(item_result.evaluations) == 2 # Both evaluators should run - # Flush and wait for server processing langfuse_client.flush() - time.sleep(2) - - # Validate traces are correctly persisted with input/output/metadata - api = get_api() - expected_inputs = ["Germany", "France", "Spain"] - expected_outputs = ["Capital of Germany", "Capital of France", "Capital of Spain"] - - for i, item_result in enumerate(result.item_results): - trace_id = item_result.trace_id - assert trace_id is not None, f"Item {i} should have a trace_id" - - # Fetch trace from API - trace = api.trace.get(trace_id) - assert trace is not None, f"Trace {trace_id} should exist" - - # Validate trace name - assert trace.name == "experiment-item-run", ( - f"Trace {trace_id} should have correct name" - ) - - # Validate trace input - should contain the experiment item - assert trace.input is not None, f"Trace {trace_id} should have input" - expected_input = expected_inputs[i] - # The input should contain the item data in some form - assert expected_input in str(trace.input), ( - f"Trace {trace_id} input should contain '{expected_input}'" - ) - # Validate trace output - should be the task result - assert trace.output is not None, f"Trace {trace_id} should have output" - expected_output = expected_outputs[i] - assert trace.output == expected_output, ( - f"Trace {trace_id} output should be '{expected_output}', got '{trace.output}'" - ) + expected = { + "Germany": ("Capital of Germany", "Berlin"), + "France": ("Capital of France", "Paris"), + "Spain": ("Capital of Spain", "Madrid"), + } + items = get_experiment_items(result.experiment_id, expected_count=3) + assert len(items) == 3 + assert {item.trace_id for item in items} == { + r.trace_id for r in result.item_results + } - # Validate trace metadata contains experiment name - assert trace.metadata is not None, f"Trace {trace_id} should have metadata" - assert "experiment_name" in trace.metadata, ( - f"Trace {trace_id} metadata should contain experiment_name" - ) - assert trace.metadata["experiment_name"] == "Euro capitals", ( - f"Trace {trace_id} metadata should have correct experiment_name" - ) + for item in items: + assert item.experiment_name == result.run_name + assert item.experiment_dataset_id is None + assert item.input in expected + expected_output, expected_answer = expected[item.input] + assert item.output == expected_output + assert item.expected_output == expected_answer + assert item.metadata is not None + assert item.metadata["experiment_name"] == "Euro capitals" + + # Run-level evaluations are persisted for local data, too + experiment = get_experiment( + result.experiment_id, is_ready=has_score("average_length") + ) + assert experiment.name == result.run_name + assert experiment.description == "Country capital experiment" + assert experiment.item_count == 3 + assert [s.name for s in experiment.scores or []] == ["average_length"] def test_run_experiment_flattens_large_metadata_for_server_ingestion(): @@ -168,21 +259,25 @@ def task_with_external_child(*, item: ExperimentItem, **kwargs: Dict[str, Any]): trace_id = result.item_results[0].trace_id assert trace_id is not None - trace = wait_for_trace( + observations = get_observations( trace_id, - is_result_ready=lambda fetched_trace: any( - observation.name == external_span_name - for observation in fetched_trace.observations + is_ready=lambda observations: any( + observation.name == external_span_name for observation in observations ), ) - assert trace.metadata is not None + root_observation = next( + observation + for observation in observations + if observation.name == "experiment-item-run" + ) + assert root_observation.metadata is not None for metadata_key, metadata_value in experiment_metadata.items(): - assert trace.metadata[metadata_key] == metadata_value + assert root_observation.metadata[metadata_key] == metadata_value external_observation = next( observation - for observation in trace.observations + for observation in observations if observation.name == external_span_name ) external_metadata = external_observation.metadata or {} @@ -227,117 +322,109 @@ def test_run_experiment_on_langfuse_dataset(): run_evaluators=[run_evaluator_average_length], ) - # Should have dataset run ID for Langfuse datasets - assert result.dataset_run_id is not None + project_id = langfuse_client._get_project_id() + assert project_id is not None + assert result.experiment_id == langfuse_client._create_experiment_id( + project_id=project_id, dataset_id=dataset.id, run_name=result.run_name + ) + assert result.experiment_url == ( + f"{os.environ['LANGFUSE_BASE_URL']}/project/{project_id}" + f"/experiments/results?baseline={result.experiment_id}" + ) assert len(result.item_results) == 2 - assert all(item.dataset_run_id is not None for item in result.item_results) + assert all( + item.experiment_id == result.experiment_id for item in result.item_results + ) - # Flush and wait for server processing langfuse_client.flush() - time.sleep(3) - - # Verify dataset run exists via API - api = get_api() - dataset_run = api.datasets.get_run( - dataset_name=dataset_name, run_name=result.run_name - ) - # Validate traces are correctly persisted with input/output/metadata expected_data = {"Germany": "Capital of Germany", "France": "Capital of France"} - dataset_run_id = result.dataset_run_id - - # Create a mapping from dataset item ID to dataset item for validation dataset_item_map = {item.id: item for item in dataset.items} - for i, item_result in enumerate(result.item_results): - trace_id = item_result.trace_id - assert trace_id is not None, f"Item {i} should have a trace_id" + items = get_experiment_items( + result.experiment_id, expected_count=2, with_scores=True + ) + assert len(items) == 2, "Experiment should have 2 items" + assert {item.trace_id for item in items} == { + r.trace_id for r in result.item_results + } + assert {item.experiment_item_id for item in items} == set(dataset_item_map) - # Fetch trace from API - trace = api.trace.get(trace_id) - assert trace is not None, f"Trace {trace_id} should exist" + for item in items: + assert item.experiment_name == result.run_name + assert item.experiment_dataset_id == dataset.id + assert item.experiment_description == "Test on Langfuse dataset" + assert item.experiment_item_version is None - # Validate trace name - assert trace.name == "experiment-item-run", ( - f"Trace {trace_id} should have correct name" - ) + dataset_item = dataset_item_map[item.experiment_item_id] + assert item.input == dataset_item.input + assert item.output == expected_data[dataset_item.input] + assert item.expected_output == dataset_item.expected_output - # Validate trace input and output match expected pairs - assert trace.input is not None, f"Trace {trace_id} should have input" - trace_input_str = str(trace.input) + assert item.metadata is not None + assert item.metadata["experiment_name"] == experiment_name + assert item.metadata["dataset_id"] == dataset.id + assert item.metadata["dataset_item_id"] == item.experiment_item_id - # Find which expected input this trace corresponds to - matching_input = None - for expected_input in expected_data.keys(): - if expected_input in trace_input_str: - matching_input = expected_input - break + assert [s.name for s in item.scores or []] == ["factuality"] - assert matching_input is not None, ( - f"Trace {trace_id} input '{trace_input_str}' should contain one of {list(expected_data.keys())}" - ) + experiment = get_experiment( + result.experiment_id, is_ready=has_score("average_length") + ) + assert experiment.name == result.run_name + assert experiment.description == "Test on Langfuse dataset" + assert experiment.dataset_id == dataset.id + assert experiment.item_count == 2 + assert [s.name for s in experiment.scores or []] == ["average_length"] + assert experiment.scores[0].subject.kind == "experiment" + assert experiment.scores[0].subject.id == result.experiment_id - # Validate trace output matches the expected output for this input - assert trace.output is not None, f"Trace {trace_id} should have output" - expected_output = expected_data[matching_input] - assert trace.output == expected_output, ( - f"Trace {trace_id} output should be '{expected_output}', got '{trace.output}'" - ) - # Validate trace metadata contains experiment and dataset info - assert trace.metadata is not None, f"Trace {trace_id} should have metadata" - assert "experiment_name" in trace.metadata, ( - f"Trace {trace_id} metadata should contain experiment_name" - ) - assert trace.metadata["experiment_name"] == experiment_name, ( - f"Trace {trace_id} metadata should have correct experiment_name" - ) +def test_run_experiment_on_versioned_dataset_records_item_version(): + """Pinned dataset versions are recorded as the experiment item version.""" + langfuse_client = get_client() + dataset_name = "versioned-dataset-" + create_uuid() + langfuse_client.create_dataset(name=dataset_name) + langfuse_client.create_dataset_item( + dataset_name=dataset_name, input="Germany", expected_output="Berlin" + ) - # Validate dataset-specific metadata fields - assert "dataset_id" in trace.metadata, ( - f"Trace {trace_id} metadata should contain dataset_id" - ) - assert trace.metadata["dataset_id"] == dataset.id, ( - f"Trace {trace_id} metadata should have correct dataset_id" - ) + version = datetime.now(timezone.utc).replace(microsecond=0) + timedelta(seconds=1) + time.sleep(1.5) + dataset = langfuse_client.get_dataset(dataset_name, version=version) + assert len(dataset.items) == 1 - assert "dataset_item_id" in trace.metadata, ( - f"Trace {trace_id} metadata should contain dataset_item_id" - ) - # Get the dataset item ID from metadata and validate it exists - dataset_item_id = trace.metadata["dataset_item_id"] - assert dataset_item_id in dataset_item_map, ( - f"Trace {trace_id} metadata dataset_item_id should correspond to a valid dataset item" - ) + result = dataset.run_experiment( + name="Versioned " + create_uuid()[:8], task=mock_task + ) + langfuse_client.flush() - # Validate the dataset item input matches the trace input - dataset_item = dataset_item_map[dataset_item_id] - assert dataset_item.input == matching_input, ( - f"Trace {trace_id} should correspond to dataset item with input '{matching_input}'" - ) + items = get_experiment_items(result.experiment_id, expected_count=1) + assert len(items) == 1 + assert items[0].experiment_item_version is not None + assert items[0].experiment_item_version.astimezone(timezone.utc) == version - assert dataset_run is not None, f"Dataset run {dataset_run_id} should exist" - assert dataset_run.name == result.run_name, "Dataset run should have correct name" - assert dataset_run.description == "Test on Langfuse dataset", ( - "Dataset run should have correct description" - ) - # Get dataset run items to verify trace linkage - dataset_run_items = api.dataset_run_items.list( - dataset_id=dataset.id, run_name=result.run_name +def test_same_run_name_on_dataset_reuses_experiment_id(): + """Re-running with the same run name on a dataset groups into one experiment.""" + langfuse_client = get_client() + dataset_name = "rerun-dataset-" + create_uuid() + langfuse_client.create_dataset(name=dataset_name) + langfuse_client.create_dataset_item( + dataset_name=dataset_name, input="Germany", expected_output="Berlin" ) - assert len(dataset_run_items.data) == 2, "Dataset run should have 2 items" + dataset = langfuse_client.get_dataset(dataset_name) + run_name = "Shared run " + create_uuid()[:8] - # Verify each dataset run item links to the correct trace - run_item_trace_ids = { - item.trace_id for item in dataset_run_items.data if item.trace_id - } - result_trace_ids = {item.trace_id for item in result.item_results} + first = dataset.run_experiment(name="Rerun", run_name=run_name, task=mock_task) + second = dataset.run_experiment(name="Rerun", run_name=run_name, task=mock_task) + langfuse_client.flush() - assert run_item_trace_ids == result_trace_ids, ( - f"Dataset run items should link to the same traces as experiment results. " - f"Run items: {run_item_trace_ids}, Results: {result_trace_ids}" - ) + assert first.experiment_id == second.experiment_id + items = get_experiment_items(first.experiment_id, expected_count=2) + assert {item.trace_id for item in items} == { + r.trace_id for r in first.item_results + second.item_results + } # Error Handling Tests @@ -630,20 +717,21 @@ def test_run_evaluator(**kwargs): run_evaluators=[test_run_evaluator], ) - assert result.dataset_run_id is not None assert len(result.item_results) == 1 assert len(result.run_evaluations) == 1 langfuse_client.flush() - time.sleep(3) - # Verify scores are persisted via API - api = get_api() - dataset_run = api.datasets.get_run( - dataset_name=dataset_name, run_name=result.run_name + experiment = get_experiment( + result.experiment_id, is_ready=has_score("persistence_run_test") ) + assert experiment.name == "Score persistence test" + run_scores = {s.name: s for s in experiment.scores or []} + assert run_scores["persistence_run_test"].value == 0.9 - assert dataset_run.name == "Score persistence test" + trace_scores = get_trace_scores(result.item_results[0].trace_id, expected_count=1) + assert [(s.name, s.value) for s in trace_scores] == [("persistence_test", 0.85)] + assert trace_scores[0].subject.kind == "observation" def test_multiple_experiments_on_same_dataset(): @@ -688,21 +776,22 @@ def test_multiple_experiments_on_same_dataset(): ) langfuse_client.flush() - time.sleep(2) - # Both experiments should have different run IDs - assert result1.dataset_run_id is not None - assert result2.dataset_run_id is not None - assert result1.dataset_run_id != result2.dataset_run_id + assert result1.experiment_id != result2.experiment_id - # Verify both runs exist in database api = get_api() - runs = api.datasets.get_runs(dataset_name) - assert len(runs.data) >= 2 - - run_names = [run.name for run in runs.data] - assert "Experiment 1" in run_names - assert "Experiment 2" in run_names + experiments = poll( + lambda: ( + api.experiments.list( + from_start_time=_read_window_start(), dataset_id=dataset.id + ).data + ), + lambda data: len(data) >= 2, + ) + assert {e.id: e.name for e in experiments} == { + result1.experiment_id: "Experiment 1", + result2.experiment_id: "Experiment 2", + } # Result Formatting Tests @@ -840,24 +929,24 @@ def mock_task_with_boolean_results(*, item: ExperimentItem, **kwargs): assert run_eval.value is False # Spain should fail, so not all pass assert run_eval.data_type == ScoreDataType.BOOLEAN - # Flush and wait for server processing langfuse_client.flush() - time.sleep(3) # Verify scores are persisted via API with correct data types for i, item_result in enumerate(result.item_results): trace_id = item_result.trace_id assert trace_id is not None, f"Item {i} should have a trace_id" - # Fetch trace from API to verify score persistence - trace = wait_for_trace( - trace_id, - is_result_ready=lambda trace: len(trace.scores) > 0, - ) - assert trace is not None, f"Trace {trace_id} should exist" + scores = get_trace_scores(trace_id, expected_count=1) + assert len(scores) == 1 + assert scores[0].data_type == "BOOLEAN" + assert scores[0].value is expected_results[i] - for score in trace.scores: - assert score.data_type == "BOOLEAN" + experiment = get_experiment( + result.experiment_id, is_ready=has_score("all_items_pass") + ) + run_score = next(s for s in experiment.scores or [] if s.name == "all_items_pass") + assert run_score.data_type == "BOOLEAN" + assert run_score.value is False def test_experiment_composite_evaluator_weighted_average(): diff --git a/tests/unit/test_experiment.py b/tests/unit/test_experiment.py index 3eb55ad40..02596c242 100644 --- a/tests/unit/test_experiment.py +++ b/tests/unit/test_experiment.py @@ -2,7 +2,7 @@ import inspect import typing -from datetime import datetime +from datetime import datetime, timezone from typing import get_type_hints from unittest.mock import MagicMock @@ -12,7 +12,9 @@ from langfuse import Evaluation, RegressionError, RunnerContext from langfuse._client.attributes import LangfuseOtelSpanAttributes from langfuse._client.client import Langfuse +from langfuse.api import DatasetItem, DatasetStatus from langfuse.batch_evaluation import CompositeEvaluatorFunction +from langfuse.experiment import ExperimentItemResult, ExperimentResult def _noop_task(*, item, **kwargs): # pragma: no cover - never invoked via mock @@ -524,3 +526,268 @@ def failing_evaluator(**kwargs): "failed_evaluator_count": 1, "skipped_evaluator_count": 1, } + + +def _dataset_item(*, item_id: str, dataset_id: str, input: str) -> DatasetItem: + return DatasetItem( + id=item_id, + status=DatasetStatus.ACTIVE, + input=input, + expected_output=f"expected {input}", + metadata=None, + source_trace_id=None, + source_observation_id=None, + dataset_id=dataset_id, + dataset_name="dataset", + media_references=[], + created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + updated_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + + +def _run_scores(create_score: MagicMock) -> list: + return [ + call.kwargs + for call in create_score.call_args_list + if call.kwargs.get("dataset_run_id") is not None + ] + + +class TestExperimentRunIdentity: + @pytest.mark.parametrize( + ("project_id", "dataset_id", "run_name", "expected"), + [ + ("p", "d", "r", "eab007015b1c6f77"), + ("proj-ΓΌ", "ds", 'Run – βœ“ "q"', "8d4425c67052ced3"), + # Shared with the JS SDK and the platform; keep these in sync. + ( + "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a", + "cm9x1dataset0000000000001", + "my-run", + "a1164c0f238e4173", + ), + ( + "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a", + "cm9x1dataset0000000000001", + 'LΓ€ufe "v2" πŸš€ – 2026-10-06T12:00:00.000Z', + "dce21641128b88a8", + ), + ], + ) + def test_dataset_experiment_id_matches_server_derivation( + self, langfuse_memory_client, project_id, dataset_id, run_name, expected + ): + assert ( + langfuse_memory_client._create_experiment_id( + project_id=project_id, dataset_id=dataset_id, run_name=run_name + ) + == expected + ) + + def test_dataset_experiment_without_project_id_uses_random_id( + self, langfuse_memory_client, caplog + ): + ids = { + langfuse_memory_client._create_experiment_id( + project_id=None, dataset_id="d", run_name="r" + ) + for _ in range(2) + } + + assert len(ids) == 2 + assert all(len(i) == 16 for i in ids) + assert "random experiment id" in caplog.text + + def test_dataset_run_uses_stable_id_url_version_and_no_run_item_post( + self, langfuse_memory_client, find_spans, monkeypatch + ): + create_score = MagicMock() + create_run_item = MagicMock() + monkeypatch.setattr(langfuse_memory_client, "create_score", create_score) + monkeypatch.setattr( + langfuse_memory_client.api.dataset_run_items, "create", create_run_item + ) + monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p") + version = datetime(2026, 2, 3, 4, 5, 6, tzinfo=timezone.utc) + + result = langfuse_memory_client.run_experiment( + name="exp", + run_name="r", + data=[ + _dataset_item(item_id="i1", dataset_id="d", input="a"), + _dataset_item(item_id="i2", dataset_id="d", input="b"), + ], + task=lambda *, item, **kwargs: item.input, + run_evaluators=[lambda **kwargs: Evaluation(name="run", value=1.0)], + _dataset_version=version, + ) + langfuse_memory_client.flush() + + create_run_item.assert_not_called() + assert result.experiment_id == "eab007015b1c6f77" + assert ( + result.experiment_url + == "http://test-host/project/p/experiments/results?baseline=eab007015b1c6f77" + ) + assert {r.experiment_id for r in result.item_results} == {result.experiment_id} + + spans = find_spans("experiment-item-run") + find_spans("experiment-item-task") + assert len(spans) == 4 + for span in spans: + assert ( + span.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ID] + == result.experiment_id + ) + assert ( + span.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_VERSION] + == "2026-02-03T04:05:06Z" + ) + + run_scores = _run_scores(create_score) + assert len(run_scores) == 1 + assert run_scores[0]["dataset_run_id"] == result.experiment_id + assert run_scores[0]["name"] == "run" + assert "trace_id" not in run_scores[0] + assert "observation_id" not in run_scores[0] + assert "session_id" not in run_scores[0] + + def test_local_run_persists_run_scores_without_item_version( + self, langfuse_memory_client, find_spans, monkeypatch + ): + create_score = MagicMock() + monkeypatch.setattr(langfuse_memory_client, "create_score", create_score) + monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p") + + result = langfuse_memory_client.run_experiment( + name="exp", + run_name="r", + data=[{"input": "a"}], + task=lambda *, item, **kwargs: item["input"], + run_evaluators=[lambda **kwargs: Evaluation(name="run", value=1.0)], + _dataset_version=datetime(2026, 2, 3, tzinfo=timezone.utc), + ) + langfuse_memory_client.flush() + + assert len(result.experiment_id) == 16 + assert result.experiment_id != "eab007015b1c6f77" + assert result.experiment_url == ( + f"http://test-host/project/p/experiments/results?baseline={result.experiment_id}" + ) + assert [s["dataset_run_id"] for s in _run_scores(create_score)] == [ + result.experiment_id + ] + for span in find_spans("experiment-item-run") + find_spans( + "experiment-item-task" + ): + assert LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_VERSION not in ( + span.attributes or {} + ) + + def test_unresolvable_project_id_yields_no_url( + self, langfuse_memory_client, monkeypatch + ): + def fail(): + raise RuntimeError("no network") + + monkeypatch.setattr(langfuse_memory_client, "create_score", MagicMock()) + monkeypatch.setattr(langfuse_memory_client, "_get_project_id", fail) + + result = langfuse_memory_client.run_experiment( + name="exp", + data=[_dataset_item(item_id="i1", dataset_id="d", input="a")], + task=lambda *, item, **kwargs: item.input, + ) + + assert len(result.item_results) == 1 + assert len(result.experiment_id) == 16 + assert result.experiment_url is None + + +class TestExperimentResultAliases: + def test_dataset_run_fields_are_deprecated_aliases(self): + item_result = ExperimentItemResult( + item={"input": "a"}, + output="a", + evaluations=[], + trace_id="t", + experiment_id="e", + ) + result = ExperimentResult( + name="exp", + run_name="r", + description=None, + item_results=[item_result], + run_evaluations=[], + experiment_id="e", + experiment_url="http://host/project/p/experiments/results?baseline=e", + ) + + with pytest.warns(DeprecationWarning, match="experiment_id"): + assert result.dataset_run_id == "e" + with pytest.warns(DeprecationWarning, match="experiment_url"): + assert result.dataset_run_url == result.experiment_url + with pytest.warns(DeprecationWarning, match="experiment_id"): + assert item_result.dataset_run_id == "e" + assert f"Experiment:\n {result.experiment_url}" in result.format() + + def test_legacy_constructor_arguments_still_populate_new_fields(self): + item_result = ExperimentItemResult( + item={"input": "a"}, + output="a", + evaluations=[], + trace_id="t", + dataset_run_id="legacy", + ) + result = ExperimentResult( + name="exp", + run_name="r", + description=None, + item_results=[item_result], + run_evaluations=[], + experiment_id="e", + dataset_run_url="http://legacy", + ) + + assert item_result.experiment_id == "legacy" + assert result.experiment_url == "http://legacy" + + +class TestExperimentRunScoreGating: + def test_run_scores_skipped_when_all_items_are_sampled_out( + self, langfuse_memory_client, monkeypatch + ): + from opentelemetry.sdk.trace.sampling import ALWAYS_OFF + + create_score = MagicMock() + monkeypatch.setattr(langfuse_memory_client, "create_score", create_score) + monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p") + monkeypatch.setattr( + langfuse_memory_client._resources.tracer_provider, "sampler", ALWAYS_OFF + ) + monkeypatch.setattr(langfuse_memory_client._otel_tracer, "sampler", ALWAYS_OFF) + + result = langfuse_memory_client.run_experiment( + name="exp", + data=[{"input": "a"}, {"input": "b"}], + task=lambda *, item, **kwargs: item["input"], + run_evaluators=[lambda **kwargs: Evaluation(name="run", value=1.0)], + ) + + assert [e.name for e in result.run_evaluations] == ["run"] + assert _run_scores(create_score) == [] + + def test_tracing_disabled_skips_project_id_lookup(self, monkeypatch): + client = Langfuse( + public_key="pk", secret_key="sk", base_url="http://x", tracing_enabled=False + ) + get_project_id = MagicMock(return_value="p") + monkeypatch.setattr(client, "_get_project_id", get_project_id) + + result = client.run_experiment( + name="exp", + data=[{"input": "a"}], + task=lambda *, item, **kwargs: item["input"], + ) + + get_project_id.assert_not_called() + assert result.experiment_url is None diff --git a/tests/unit/test_propagate_attributes.py b/tests/unit/test_propagate_attributes.py index 2330269f0..77afcd7cf 100644 --- a/tests/unit/test_propagate_attributes.py +++ b/tests/unit/test_propagate_attributes.py @@ -2872,28 +2872,17 @@ def test_experiment_attributes_propagate_with_dataset( self, langfuse_client, memory_exporter, monkeypatch ): """Test experiment attribute propagation with Langfuse dataset.""" - created_run_items = [] - - # Mock the sync API used by run_experiment to create dataset run items - def mock_create_dataset_run_item(*args, **kwargs): - from langfuse.api import DatasetRunItem - - created_run_items.append(kwargs) - return DatasetRunItem( - id="mock-run-item-id", - dataset_run_id="mock-dataset-run-id-123", - dataset_run_name=kwargs.get("run_name", "Dataset Test"), - dataset_item_id=kwargs.get("dataset_item_id", "mock-item-id"), - trace_id="mock-trace-id", - observation_id=kwargs.get("observation_id"), - created_at=datetime.now(), - updated_at=datetime.now(), - ) + + def fail_create_dataset_run_item(*args, **kwargs): + raise AssertionError("run_experiment must not create dataset run items") monkeypatch.setattr( langfuse_client.api.dataset_run_items, "create", - mock_create_dataset_run_item, + fail_create_dataset_run_item, + ) + monkeypatch.setattr( + langfuse_client, "_get_project_id", lambda: "test-project-id" ) # Create a mock dataset with items @@ -2958,9 +2947,16 @@ def task_with_children(*, item, **kwargs): assert len(root_spans) >= 1, "Should have at least 1 root span" first_root = root_spans[0] task_span = self.get_span_by_name(memory_exporter, "experiment-item-task") - assert result.experiment_id == "mock-dataset-run-id-123" - assert len(created_run_items) == 1 - assert created_run_items[0]["observation_id"] == task_span["span_id"] + assert result.experiment_id == langfuse_client._create_experiment_id( + project_id="test-project-id", + dataset_id=dataset_id, + run_name=result.run_name, + ) + self.verify_span_attribute( + first_root, + LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ROOT_OBSERVATION_ID, + task_span["span_id"], + ) # Root-only attributes should be on root self.verify_span_attribute(