diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index fe80d7e85..774950332 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -2431,6 +2431,7 @@ def run_experiment( run_evaluators: List[RunEvaluatorFunction] = [], max_concurrency: int = 50, metadata: Optional[Dict[str, str]] = None, + experiment_id: Optional[str] = None, _dataset_version: Optional[datetime] = None, ) -> ExperimentResult: """Run an experiment on a dataset with automatic tracing and evaluation. @@ -2478,6 +2479,20 @@ def run_experiment( metadata: Optional metadata dictionary to attach to all experiment traces. This metadata will be included in every trace created during the experiment and is shown as the experiment's metadata in Langfuse. + experiment_id: Optional ID of the experiment run to record the items in. + Calls with the same `experiment_id` add their items and scores to one + experiment in Langfuse, so separate workers, batches, or a resumed run + can share an experiment, also for local data. Pass the same `run_name` + too, as it sets the experiment name. If provided, it is used as is, + also for Langfuse dataset items, instead of the ID derived from the + dataset and `run_name`. If not provided, the ID is derived for dataset + items and random for local data. Create a reusable ID with + `Langfuse.create_trace_id()` (random; store it and share it with the + workers) or `Langfuse.create_trace_id(seed="...")` (deterministic; every + worker derives the same ID from the same seed, and the JS SDK's + `createTraceId(seed)` returns the same ID). `run_evaluators` see only + the items of their own call, so each call adds its own run-level + scores to the shared experiment. Returns: ExperimentResult containing: @@ -2485,16 +2500,18 @@ def run_experiment( - item_results: List of results for each processed item with outputs and evaluations - 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_id: Identifier for the experiment run across all items. This + is the `experiment_id` argument if provided. Otherwise, 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 + ValueError: If required parameters are missing or invalid, or if + `experiment_id` is empty or whitespace-only Exception: If experiment setup fails (individual item failures are handled gracefully) Examples: @@ -2592,6 +2609,22 @@ def average_accuracy(*, item_results, **kwargs): print(f"View results: {result.experiment_url}") ``` + Recording one experiment from several workers or resumed jobs: + ```python + # Every worker derives the same ID from the same seed + experiment_id = Langfuse.create_trace_id(seed="nightly-eval/2026-10-20") + + # Each worker or resumed job runs its share of the items + langfuse.run_experiment( + name="Nightly eval", + run_name="nightly-eval/2026-10-20", + experiment_id=experiment_id, + data=items_for_this_worker, + task=my_task, + evaluators=[accuracy_evaluator], + ) + ``` + Note: - Task and evaluator functions can be either synchronous or asynchronous - Individual item failures are logged but don't stop the experiment @@ -2618,6 +2651,7 @@ def average_accuracy(*, item_results, **kwargs): run_evaluators=run_evaluators or [], max_concurrency=max_concurrency, metadata=metadata, + experiment_id=experiment_id, dataset_version=_dataset_version, ), ), @@ -2636,8 +2670,15 @@ async def _run_experiment_async( run_evaluators: List[RunEvaluatorFunction], max_concurrency: int, metadata: Optional[Dict[str, Any]] = None, + experiment_id: Optional[str] = None, dataset_version: Optional[datetime] = None, ) -> ExperimentResult: + if experiment_id is not None and ( + not isinstance(experiment_id, str) or not experiment_id.strip() + ): + message = "experiment_id must be a non-empty string if provided." + raise ValueError(message) + langfuse_logger.debug( "Starting experiment '%s' run '%s' with %s items", name, run_name, len(data) ) @@ -2660,11 +2701,12 @@ async def _run_experiment_async( ), None, ) - experiment_id = self._create_experiment_id( - project_id=project_id, - dataset_id=experiment_dataset_id, - run_name=run_name, - ) + if experiment_id is 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 @@ -2714,7 +2756,7 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult: langfuse_logger.error("Run evaluator failed: %s", e) experiment_url = ( - f"{self._base_url}/project/{project_id}/experiments/results?baseline={experiment_id}" + f"{self._base_url}/project/{project_id}/experiments/results?baseline={urllib.parse.quote(experiment_id, safe='')}" if project_id else None ) diff --git a/langfuse/_client/datasets.py b/langfuse/_client/datasets.py index 1070c76d3..97b2e497d 100644 --- a/langfuse/_client/datasets.py +++ b/langfuse/_client/datasets.py @@ -90,6 +90,7 @@ def run_experiment( run_evaluators: List[RunEvaluatorFunction] = [], max_concurrency: int = 50, metadata: Optional[Dict[str, Any]] = None, + experiment_id: Optional[str] = None, ) -> ExperimentResult: """Run an experiment on this Langfuse dataset with automatic tracking. @@ -128,6 +129,13 @@ def run_experiment( Adjust based on API rate limits and system resources. metadata: Optional metadata to attach to the experiment run and all traces. Will be combined with individual item metadata. + experiment_id: Optional ID of the experiment run to record the items in. + If provided, it is used instead of the ID derived from this dataset and + `run_name`, so calls that pass the same `experiment_id` share one + experiment, for example items split across workers or a resumed run. + Create a reusable ID with `Langfuse.create_trace_id()` (random) or + `Langfuse.create_trace_id(seed="...")` (deterministic). See + `Langfuse.run_experiment` for details. Returns: ExperimentResult object containing: @@ -136,7 +144,8 @@ def run_experiment( - description: Optional experiment description. - item_results: Results for each dataset item with outputs and evaluations. - run_evaluations: Aggregate evaluation results for the entire run. - - experiment_id: ID of the experiment run in Langfuse. + - experiment_id: ID of the experiment run in Langfuse (the + `experiment_id` argument if provided). - 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. @@ -301,5 +310,6 @@ def content_diversity(*, item_results, **kwargs): run_evaluators=run_evaluators, max_concurrency=max_concurrency, metadata=metadata, + experiment_id=experiment_id, _dataset_version=self.version, ) diff --git a/langfuse/experiment.py b/langfuse/experiment.py index ee29d8611..9782ac1a6 100644 --- a/langfuse/experiment.py +++ b/langfuse/experiment.py @@ -1164,6 +1164,7 @@ def run_experiment( run_evaluators: List[RunEvaluatorFunction] = [], max_concurrency: int = 50, metadata: Optional[Dict[str, str]] = None, + experiment_id: Optional[str] = None, _dataset_version: Optional[datetime] = None, ) -> ExperimentResult: resolved_data = data if data is not None else self.data @@ -1193,6 +1194,7 @@ def run_experiment( run_evaluators=run_evaluators, max_concurrency=max_concurrency, metadata=merged_metadata, + experiment_id=experiment_id, _dataset_version=resolved_dataset_version, ) diff --git a/tests/unit/test_experiment.py b/tests/unit/test_experiment.py index fea3587b2..8d8ad5192 100644 --- a/tests/unit/test_experiment.py +++ b/tests/unit/test_experiment.py @@ -1,5 +1,6 @@ """Tests for ``langfuse.experiment`` — ``RunnerContext`` and ``RegressionError``.""" +import asyncio import inspect import json import typing @@ -13,6 +14,7 @@ from langfuse import Evaluation, RegressionError, RunnerContext from langfuse._client.attributes import LangfuseOtelSpanAttributes from langfuse._client.client import Langfuse +from langfuse._client.datasets import DatasetClient from langfuse.api import DatasetItem, DatasetStatus from langfuse.batch_evaluation import CompositeEvaluatorFunction from langfuse.experiment import ExperimentItemResult, ExperimentResult @@ -62,6 +64,7 @@ def test_call_overrides_win(self): task=_noop_task, run_name="call-run", data=override_data, + experiment_id="shared", _dataset_version=override_version, ) @@ -69,6 +72,7 @@ def test_call_overrides_win(self): assert kwargs["name"] == "exp" assert kwargs["run_name"] == "call-run" assert kwargs["data"] is override_data + assert kwargs["experiment_id"] == "shared" assert kwargs["_dataset_version"] == override_version @@ -695,6 +699,151 @@ def fail(): assert result.experiment_url is None +class TestCallerSuppliedExperimentId: + def test_local_runs_with_the_same_id_share_one_experiment( + 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") + experiment_id = Langfuse.create_trace_id(seed="nightly-eval/2026-10-20") + + results = [ + langfuse_memory_client.run_experiment( + name="exp", + run_name="r", + experiment_id=experiment_id, + data=[{"input": item_input}], + task=lambda *, item, **kwargs: item["input"], + run_evaluators=[lambda **kwargs: Evaluation(name="run", value=1.0)], + ) + for item_input in ("a", "b") + ] + langfuse_memory_client.flush() + + for result in results: + assert result.experiment_id == experiment_id + assert result.experiment_url == ( + f"http://test-host/project/p/experiments/results?baseline={experiment_id}" + ) + assert [r.experiment_id for r in result.item_results] == [experiment_id] + + spans = find_spans("experiment-item-run") + find_spans("experiment-item-task") + assert len(spans) == 4 + assert { + span.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ID] for span in spans + } == {experiment_id} + assert [s["dataset_run_id"] for s in _run_scores(create_score)] == [ + experiment_id, + experiment_id, + ] + + def test_supplied_id_wins_over_the_derived_dataset_id( + self, langfuse_memory_client, find_spans, monkeypatch + ): + monkeypatch.setattr(langfuse_memory_client, "create_score", MagicMock()) + monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p") + + result = langfuse_memory_client.run_experiment( + name="exp", + run_name="r", + experiment_id="resumed-run", + data=[_dataset_item(item_id="i1", dataset_id="d", input="a")], + task=lambda *, item, **kwargs: item.input, + ) + langfuse_memory_client.flush() + + # Without experiment_id this run would be "eab007015b1c6f77". + assert result.experiment_id == "resumed-run" + spans = find_spans("experiment-item-run") + find_spans("experiment-item-task") + assert len(spans) == 2 + for span in spans: + assert ( + span.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ID] + == "resumed-run" + ) + assert ( + span.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_DATASET_ID] == "d" + ) + + def test_supplied_id_is_url_encoded_in_the_experiment_url( + self, langfuse_memory_client, monkeypatch + ): + monkeypatch.setattr(langfuse_memory_client, "create_score", MagicMock()) + monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p") + + result = langfuse_memory_client.run_experiment( + name="exp", + experiment_id="team a/run#1", + data=[{"input": "a"}], + task=lambda *, item, **kwargs: item["input"], + ) + + assert result.experiment_id == "team a/run#1" + assert result.experiment_url == ( + "http://test-host/project/p/experiments/results?baseline=team%20a%2Frun%231" + ) + + def test_supplied_id_is_used_when_called_inside_a_running_event_loop( + self, langfuse_memory_client, find_spans, monkeypatch + ): + monkeypatch.setattr(langfuse_memory_client, "create_score", MagicMock()) + monkeypatch.setattr(langfuse_memory_client, "_get_project_id", lambda: "p") + + async def task(*, item, **kwargs): + await asyncio.sleep(0) + return item["input"] + + async def main(): + return langfuse_memory_client.run_experiment( + name="exp", + experiment_id="from-async", + data=[{"input": "a"}], + task=task, + ) + + result = asyncio.run(main()) + langfuse_memory_client.flush() + + assert result.experiment_id == "from-async" + assert [r.output for r in result.item_results] == ["a"] + assert { + span.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ID] + for span in find_spans("experiment-item-task") + } == {"from-async"} + + @pytest.mark.parametrize("experiment_id", ["", " "]) + def test_blank_id_raises_before_running_items( + self, langfuse_memory_client, experiment_id + ): + task = MagicMock() + + with pytest.raises(ValueError, match="experiment_id"): + langfuse_memory_client.run_experiment( + name="exp", + experiment_id=experiment_id, + data=[{"input": "a"}], + task=task, + ) + + task.assert_not_called() + + def test_dataset_client_forwards_the_id(self): + langfuse_client = MagicMock() + item = _dataset_item(item_id="i1", dataset_id="d", input="a") + dataset = DatasetClient( + dataset=MagicMock(), + items=[item], + langfuse_client=langfuse_client, + ) + + dataset.run_experiment(name="exp", task=_noop_task, experiment_id="shared") + + assert langfuse_client.run_experiment.call_args.kwargs["experiment_id"] == ( + "shared" + ) + + class TestExperimentResultAliases: def test_dataset_run_fields_are_deprecated_aliases(self): item_result = ExperimentItemResult(