feat(experiments): opt in to concurrent synchronous tasks

langfuse/_client/client.py

@@ -9,6 +9,7 @@
 import re
 import urllib.parse
 import uuid
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 from hashlib import sha256
 from time import time_ns
@@ -2430,6 +2431,7 @@ def run_experiment(
         composite_evaluator: Optional[CompositeEvaluatorFunction] = None,
         run_evaluators: List[RunEvaluatorFunction] = [],
         max_concurrency: int = 50,
+        run_sync_tasks_in_threads: bool = False,
         metadata: Optional[Dict[str, str]] = None,
         _dataset_version: Optional[datetime] = None,
     ) -> ExperimentResult:
@@ -2475,6 +2477,17 @@ def run_experiment(
             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.
+            run_sync_tasks_in_threads: Opt in to running synchronous tasks in a
+                dedicated thread pool with up to `max_concurrency` workers (default:
+                False). This enables concurrency for blocking I/O without changing
+                the default event-loop execution. Async tasks stay on the experiment
+                loop. When enabled, tasks must be thread-safe: create and close
+                thread-bound resources, such as SQLite connections, inside the task.
+                Even `max_concurrency=1` offloads synchronous tasks. Exceptions are
+                still logged and failed items omitted. Cancellation does not stop
+                tasks already running in threads, which can continue after the
+                experiment returns and delay Python process exit. Use finite I/O
+                timeouts. For cooperative cancellation, prefer async tasks.
             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.
@@ -2617,6 +2630,7 @@ def average_accuracy(*, item_results, **kwargs):
                     composite_evaluator=composite_evaluator,
                     run_evaluators=run_evaluators or [],
                     max_concurrency=max_concurrency,
+                    run_sync_tasks_in_threads=run_sync_tasks_in_threads,
                     metadata=metadata,
                     dataset_version=_dataset_version,
                 ),
@@ -2635,6 +2649,7 @@ async def _run_experiment_async(
         composite_evaluator: Optional[CompositeEvaluatorFunction],
         run_evaluators: List[RunEvaluatorFunction],
         max_concurrency: int,
+        run_sync_tasks_in_threads: bool = False,
         metadata: Optional[Dict[str, Any]] = None,
         dataset_version: Optional[datetime] = None,
     ) -> ExperimentResult:
@@ -2673,6 +2688,11 @@ async def _run_experiment_async(
 
         # Set up concurrency control
         semaphore = asyncio.Semaphore(max_concurrency)
+        task_executor = (
+            ThreadPoolExecutor(max_workers=max_concurrency)
+            if run_sync_tasks_in_threads
+            else None
+        )
 
         # Process all items
         async def process_item(item: ExperimentItem) -> ExperimentItemResult:
@@ -2688,11 +2708,18 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult:
                     description,
                     metadata,
                     experiment_item_version,
+                    task_executor=task_executor,
                 )
 
         # Run all items concurrently
-        tasks = [process_item(item) for item in data]
-        item_results = await asyncio.gather(*tasks, return_exceptions=True)
+        try:
+            tasks = [process_item(item) for item in data]
+            item_results = await asyncio.gather(*tasks, return_exceptions=True)
+        finally:
+            if task_executor is not None:
+                # Do not block the event loop on cancellation. Running calls
+                # cannot be stopped, but queued calls must not start afterward.
+                task_executor.shutdown(wait=False, cancel_futures=True)
 
         # Filter out any exceptions and log errors
         valid_results: List[ExperimentItemResult] = []
@@ -2829,6 +2856,8 @@ async def _process_experiment_item(
         experiment_description: Optional[str],
         experiment_metadata: Optional[Dict[str, Any]] = None,
         experiment_item_version: Optional[str] = None,
+        *,
+        task_executor: Optional[ThreadPoolExecutor] = None,
     ) -> ExperimentItemResult:
         with self.start_as_current_observation(name="experiment-item-run") as span:
             try:
@@ -2942,7 +2971,7 @@ async def _process_experiment_item(
                         # metadata instead of leaving no room for the output.
                         task_span.update(metadata=final_observation_metadata)
                         try:
-                            output = await _run_task(task, item)
+                            output = await _run_task(task, item, executor=task_executor)
                         except Exception as e:
                             task_span.update(
                                 output=f"Error: {str(e)}",

langfuse/_client/datasets.py

@@ -89,6 +89,7 @@ def run_experiment(
         composite_evaluator: Optional[CompositeEvaluatorFunction] = None,
         run_evaluators: List[RunEvaluatorFunction] = [],
         max_concurrency: int = 50,
+        run_sync_tasks_in_threads: bool = False,
         metadata: Optional[Dict[str, Any]] = None,
     ) -> ExperimentResult:
         """Run an experiment on this Langfuse dataset with automatic tracking.
@@ -126,6 +127,12 @@ def run_experiment(
                 Useful for computing aggregate statistics across all dataset items.
             max_concurrency: Maximum number of concurrent task executions (default: 50).
                 Adjust based on API rate limits and system resources.
+            run_sync_tasks_in_threads: Opt in to concurrent synchronous tasks in a
+                dedicated pool with up to `max_concurrency` workers (default: False).
+                Tasks must be thread-safe; create thread-bound resources inside the
+                task. Async tasks stay on the experiment loop. Cancellation cannot
+                stop synchronous tasks already running in threads. See
+                `Langfuse.run_experiment` for resource ownership and timeout guidance.
             metadata: Optional metadata to attach to the experiment run and all traces.
                 Will be combined with individual item metadata.
 
@@ -300,6 +307,7 @@ def content_diversity(*, item_results, **kwargs):
             composite_evaluator=composite_evaluator,
             run_evaluators=run_evaluators,
             max_concurrency=max_concurrency,
+            run_sync_tasks_in_threads=run_sync_tasks_in_threads,
             metadata=metadata,
             _dataset_version=self.version,
         )

langfuse/experiment.py

@@ -6,8 +6,12 @@
 """
 
 import asyncio
+import inspect
 import warnings
+from concurrent.futures import Future, ThreadPoolExecutor
+from contextvars import copy_context
 from datetime import datetime
+from functools import partial
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -642,6 +646,18 @@ class TaskFunction(Protocol):
     - Return any type of output (will be passed to evaluators)
     - Can be either synchronous or asynchronous
     - Should handle their own errors gracefully (exceptions will be logged)
+
+    By default, synchronous task functions run on the experiment's event loop,
+    so blocking calls execute serially. Set `run_sync_tasks_in_threads=True` on
+    `run_experiment` to run synchronous tasks concurrently in worker threads.
+    Asynchronous tasks, including decorated wrappers, stay on the event loop.
+
+    When opting in, synchronous tasks must be thread-safe. Create and close
+    thread-bound resources such as sqlite3 connections inside the task, rather
+    than capturing a connection created on the caller's thread. Even
+    `max_concurrency=1` offloads the task. Task exceptions are logged and failed
+    items are omitted from the result. Cancellation cannot stop a synchronous
+    task already running in a thread; use finite timeouts on blocking I/O.
     """
 
     def __call__(
@@ -1056,9 +1072,61 @@ async def _run_evaluator(
         return []
 
 
-async def _run_task(task: TaskFunction, item: ExperimentItem) -> Any:
+def _is_async_task(task: Any) -> bool:
+    """Inspect async task metadata without replacing the callable we invoke."""
+    seen: set[int] = set()
+    check_call = True
+    for _ in range(100):
+        if id(task) in seen:
+            return False
+        seen.add(id(task))
+        if isinstance(task, partial):
+            task = task.func
+            continue
+        try:
+            unwrapped = inspect.unwrap(task, stop=inspect.iscoroutinefunction)
+        except ValueError:
+            return False
+        if inspect.iscoroutinefunction(unwrapped):
+            return True
+        if unwrapped is not task:
+            task = unwrapped
+        elif check_call:
+            task = getattr(task, "__call__", None)
+            check_call = False
+        else:
+            return False
+    return False
+
+
+async def _run_task(
+    task: TaskFunction,
+    item: ExperimentItem,
+    *,
+    executor: Optional[ThreadPoolExecutor] = None,
+) -> Any:
     """Run a task function and handle sync/async."""
-    result = task(item=item)
+    result: Any
+    if executor is None or _is_async_task(task):
+        result = task(item=item)
+    else:
+        # Copy separately for every item so worker spans inherit that item's
+        # tracing context. Keep the source future to dispose of a late coroutine
+        # if cancellation prevents the caller from awaiting it.
+        context = copy_context()
+        future = executor.submit(context.run, partial(task, item=item))
+        try:
+            result = await asyncio.wrap_future(future)
+        except asyncio.CancelledError:
+
+            def discard_result(done: Future[Any]) -> None:
+                if not done.cancelled() and done.exception() is None:
+                    value = done.result()
+                    if asyncio.iscoroutine(value):
+                        value.close()
+
+            future.add_done_callback(discard_result)
+            raise
 
     # Handle async tasks
     if asyncio.iscoroutine(result):
@@ -1163,6 +1231,7 @@ def run_experiment(
         composite_evaluator: Optional["CompositeEvaluatorFunction"] = None,
         run_evaluators: List[RunEvaluatorFunction] = [],
         max_concurrency: int = 50,
+        run_sync_tasks_in_threads: bool = False,
         metadata: Optional[Dict[str, str]] = None,
         _dataset_version: Optional[datetime] = None,
     ) -> ExperimentResult:
@@ -1192,6 +1261,7 @@ def run_experiment(
             composite_evaluator=composite_evaluator,
             run_evaluators=run_evaluators,
             max_concurrency=max_concurrency,
+            run_sync_tasks_in_threads=run_sync_tasks_in_threads,
             metadata=merged_metadata,
             _dataset_version=resolved_dataset_version,
         )