feat(experiments): accept a caller-supplied experiment_id in run_experiment

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,23 +2479,39 @@ 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:
             - 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.
               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
         )

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,
         )

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,
         )
 

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,13 +64,15 @@ def test_call_overrides_win(self):
             task=_noop_task,
             run_name="call-run",
             data=override_data,
+            experiment_id="shared",
             _dataset_version=override_version,
         )
 
         kwargs = ctx.client.run_experiment.call_args.kwargs
         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(