Merge branch 'main' into fix/control-flow-exception-check-in-error-handlers

Author: vishnumishra

langfuse/_client/client.py

@@ -2483,7 +2483,7 @@ def get_dataset(
 
             items = [DatasetItemClient(i, langfuse=self) for i in dataset_items]
 
-            return DatasetClient(dataset, items=items)
+            return DatasetClient(dataset, items=items, version=version)
 
         except Error as e:
             handle_fern_exception(e)
@@ -2574,6 +2574,7 @@ def run_experiment(
         run_evaluators: List[RunEvaluatorFunction] = [],
         max_concurrency: int = 50,
         metadata: Optional[Dict[str, str]] = None,
+        _dataset_version: Optional[datetime] = None,
     ) -> ExperimentResult:
         """Run an experiment on a dataset with automatic tracing and evaluation.
 
@@ -2751,6 +2752,7 @@ def average_accuracy(*, item_results, **kwargs):
                     run_evaluators=run_evaluators or [],
                     max_concurrency=max_concurrency,
                     metadata=metadata,
+                    dataset_version=_dataset_version,
                 ),
             ),
         )
@@ -2768,6 +2770,7 @@ async def _run_experiment_async(
         run_evaluators: List[RunEvaluatorFunction],
         max_concurrency: int,
         metadata: Optional[Dict[str, Any]] = None,
+        dataset_version: Optional[datetime] = None,
     ) -> ExperimentResult:
         langfuse_logger.debug(
             f"Starting experiment '{name}' run '{run_name}' with {len(data)} items"
@@ -2788,6 +2791,7 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult:
                     run_name,
                     description,
                     metadata,
+                    dataset_version,
                 )
 
         # Run all items concurrently
@@ -2874,6 +2878,7 @@ async def _process_experiment_item(
         experiment_run_name: str,
         experiment_description: Optional[str],
         experiment_metadata: Optional[Dict[str, Any]] = None,
+        dataset_version: Optional[datetime] = None,
     ) -> ExperimentItemResult:
         span_name = "experiment-item-run"
 
@@ -2925,6 +2930,7 @@ async def _process_experiment_item(
                                 datasetItemId=item.id,  # type: ignore
                                 traceId=trace_id,
                                 observationId=span.id,
+                                datasetVersion=dataset_version,
                             ),
                         )
 

langfuse/_client/datasets.py

@@ -155,7 +155,7 @@ class DatasetClient:
         created_at (datetime): Timestamp of dataset creation.
         updated_at (datetime): Timestamp of the last update to the dataset.
         items (List[DatasetItemClient]): List of dataset items associated with the dataset.
-
+        version (Optional[datetime]): Timestamp of the dataset version.
     Example:
         Print the input of each dataset item in a dataset.
         ```python
@@ -178,8 +178,14 @@ class DatasetClient:
     created_at: dt.datetime
     updated_at: dt.datetime
     items: List[DatasetItemClient]
+    version: Optional[dt.datetime]
 
-    def __init__(self, dataset: Dataset, items: List[DatasetItemClient]):
+    def __init__(
+        self,
+        dataset: Dataset,
+        items: List[DatasetItemClient],
+        version: Optional[dt.datetime] = None,
+    ):
         """Initialize the DatasetClient."""
         self.id = dataset.id
         self.name = dataset.name
@@ -189,6 +195,7 @@ def __init__(self, dataset: Dataset, items: List[DatasetItemClient]):
         self.created_at = dataset.created_at
         self.updated_at = dataset.updated_at
         self.items = items
+        self.version = version
         self._langfuse: Optional["Langfuse"] = None
 
     def _get_langfuse_client(self) -> Optional["Langfuse"]:
@@ -279,7 +286,7 @@ def answer_questions(*, item, **kwargs):
 
             def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):
                 if not expected_output:
-                    return {"name": "accuracy", "value": None, "comment": "No expected output"}
+                    return {"name": "accuracy", "value": 0, "comment": "No expected output"}
 
                 is_correct = output.strip().lower() == expected_output.strip().lower()
                 return {
@@ -421,4 +428,5 @@ def content_diversity(*, item_results, **kwargs):
             run_evaluators=run_evaluators,
             max_concurrency=max_concurrency,
             metadata=metadata,
+            _dataset_version=self.version,
         )

langfuse/api/reference.md

@@ -7167,6 +7167,14 @@ client.score_v_2.get()
 <dl>
 <dd>
 
+**observation_id:** `typing.Optional[str]` — Comma-separated list of observation IDs to filter scores by.
+    
+</dd>
+</dl>
+
+<dl>
+<dd>
+
 **queue_id:** `typing.Optional[str]` — Retrieve only scores with a specific annotation queueId.
 
 </dd>

langfuse/api/resources/score_v_2/client.py

@@ -43,6 +43,7 @@ def get(
         session_id: typing.Optional[str] = None,
         dataset_run_id: typing.Optional[str] = None,
         trace_id: typing.Optional[str] = None,
+        observation_id: typing.Optional[str] = None,
         queue_id: typing.Optional[str] = None,
         data_type: typing.Optional[ScoreDataType] = None,
         trace_tags: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None,
@@ -99,6 +100,9 @@ def get(
         trace_id : typing.Optional[str]
             Retrieve only scores with a specific traceId.
 
+        observation_id : typing.Optional[str]
+            Comma-separated list of observation IDs to filter scores by.
+
         queue_id : typing.Optional[str]
             Retrieve only scores with a specific annotation queueId.
 
@@ -155,6 +159,7 @@ def get(
                 "sessionId": session_id,
                 "datasetRunId": dataset_run_id,
                 "traceId": trace_id,
+                "observationId": observation_id,
                 "queueId": queue_id,
                 "dataType": data_type,
                 "traceTags": trace_tags,
@@ -276,6 +281,7 @@ async def get(
         session_id: typing.Optional[str] = None,
         dataset_run_id: typing.Optional[str] = None,
         trace_id: typing.Optional[str] = None,
+        observation_id: typing.Optional[str] = None,
         queue_id: typing.Optional[str] = None,
         data_type: typing.Optional[ScoreDataType] = None,
         trace_tags: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None,
@@ -332,6 +338,9 @@ async def get(
         trace_id : typing.Optional[str]
             Retrieve only scores with a specific traceId.
 
+        observation_id : typing.Optional[str]
+            Comma-separated list of observation IDs to filter scores by.
+
         queue_id : typing.Optional[str]
             Retrieve only scores with a specific annotation queueId.
 
@@ -396,6 +405,7 @@ async def main() -> None:
                 "sessionId": session_id,
                 "datasetRunId": dataset_run_id,
                 "traceId": trace_id,
+                "observationId": observation_id,
                 "queueId": queue_id,
                 "dataType": data_type,
                 "traceTags": trace_tags,

langfuse/experiment.py

@@ -719,7 +719,7 @@ def __call__(
             ```python
             def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):
                 if expected_output is None:
-                    return {"name": "accuracy", "value": None, "comment": "No expected output"}
+                    return {"name": "accuracy", "value": 0, "comment": "No expected output"}
 
                 is_correct = output.strip().lower() == expected_output.strip().lower()
                 return {
@@ -773,7 +773,7 @@ async def llm_judge_evaluator(*, input, output, expected_output=None, **kwargs):
                 except ValueError:
                     return {
                         "name": "llm_judge_quality",
-                        "value": None,
+                        "value": 0,
                         "comment": "Could not parse LLM judge score"
                     }
             ```
@@ -867,7 +867,7 @@ def average_accuracy(*, item_results, **kwargs):
                             accuracy_values.append(evaluation.value)
 
                 if not accuracy_values:
-                    return {"name": "avg_accuracy", "value": None, "comment": "No accuracy evaluations found"}
+                    return {"name": "avg_accuracy", "value": 0, "comment": "No accuracy evaluations found"}
 
                 avg = sum(accuracy_values) / len(accuracy_values)
                 return {

langfuse/version.py

@@ -1,3 +1,3 @@
 """@private"""
 
-__version__ = "3.14.0"
+__version__ = "3.14.1"

poetry.lock

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "langfuse"
 
-version = "3.14.0"
+version = "3.14.1"
 description = "A client library for accessing langfuse"
 authors = ["langfuse <developers@langfuse.com>"]
 license = "MIT"

pyproject.toml

@@ -569,3 +569,74 @@ def test_get_dataset_with_version():
     # Verify fetching without version returns both items (latest)
     dataset_latest = langfuse.get_dataset(name)
     assert len(dataset_latest.items) == 2
+
+
+def test_run_experiment_with_versioned_dataset():
+    """Test that running an experiment on a versioned dataset works correctly."""
+    from datetime import timedelta
+    import time
+
+    langfuse = Langfuse(debug=False)
+
+    # Create dataset
+    name = create_uuid()
+    langfuse.create_dataset(name=name)
+
+    # Create first item
+    langfuse.create_dataset_item(
+        dataset_name=name, input={"question": "What is 2+2?"}, expected_output=4
+    )
+    langfuse.flush()
+    time.sleep(3)
+
+    # Fetch dataset to get the actual server-assigned timestamp of item1
+    dataset_after_item1 = langfuse.get_dataset(name)
+    assert len(dataset_after_item1.items) == 1
+    item1_id = dataset_after_item1.items[0].id
+    item1_created_at = dataset_after_item1.items[0].created_at
+
+    # Use a timestamp 1 second after item1's creation
+    version_timestamp = item1_created_at + timedelta(seconds=1)
+    time.sleep(3)
+
+    # Update item1 after the version timestamp (this should not affect versioned query)
+    langfuse.create_dataset_item(
+        id=item1_id,
+        dataset_name=name,
+        input={"question": "What is 4+4?"},
+        expected_output=8,
+    )
+    langfuse.flush()
+    time.sleep(3)
+
+    # Create second item (after version timestamp)
+    langfuse.create_dataset_item(
+        dataset_name=name, input={"question": "What is 3+3?"}, expected_output=6
+    )
+    langfuse.flush()
+    time.sleep(3)
+
+    # Get versioned dataset (should only have first item with ORIGINAL state)
+    versioned_dataset = langfuse.get_dataset(name, version=version_timestamp)
+    assert len(versioned_dataset.items) == 1
+    assert versioned_dataset.version == version_timestamp
+    # Verify it returns the ORIGINAL version of item1 (before the update)
+    assert versioned_dataset.items[0].input == {"question": "What is 2+2?"}
+    assert versioned_dataset.items[0].expected_output == 4
+    assert versioned_dataset.items[0].id == item1_id
+
+    # Run a simple experiment on the versioned dataset
+    def simple_task(*, item, **kwargs):
+        # Just return a static answer
+        return item.expected_output
+
+    result = versioned_dataset.run_experiment(
+        name="Versioned Dataset Test",
+        description="Testing experiment with versioned dataset",
+        task=simple_task,
+    )
+
+    # Verify experiment ran successfully
+    assert result.name == "Versioned Dataset Test"
+    assert len(result.item_results) == 1  # Only one item in versioned dataset
+    assert result.item_results[0].output == 4