From e6c39161a184f0212959be77e5ddb2f5cf4238aa Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 7 Sep 2026 10:05:41 +0000 Subject: [PATCH 1/7] fix(evals): use v2 observations for batch evaluation Co-authored-by: Hassieb Pakzad --- langfuse/batch_evaluation.py | 204 +++++++++++++++++++++++----- tests/unit/test_batch_evaluation.py | 167 +++++++++++++++++++++++ 2 files changed, 340 insertions(+), 31 deletions(-) create mode 100644 tests/unit/test_batch_evaluation.py diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index b75f0b3d2..56c86f569 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -24,8 +24,11 @@ ) from langfuse.api import ( + ObservationLevel, ObservationsView, + ObservationV2, TraceWithFullDetails, + Usage, ) from langfuse.experiment import Evaluation, EvaluatorFunction from langfuse.logger import langfuse_logger as logger @@ -918,6 +921,7 @@ async def run_async( # Pagination state page = 1 + cursor: Optional[str] = None has_more = True last_item_timestamp: Optional[str] = None last_item_id: Optional[str] = None @@ -944,10 +948,10 @@ async def run_async( # Fetch next batch with retry logic try: - items = await self._fetch_batch_with_retry( + items, next_cursor = await self._fetch_batch_with_retry( scope=scope, filter=effective_filter, - page=page, + cursor=cursor, limit=fetch_batch_size, max_retries=max_retries, fields=fetch_trace_fields, @@ -1091,10 +1095,10 @@ async def process_item( ) # Check if we should continue to next page - if len(items) < fetch_batch_size: - # Last page - no more items available + if next_cursor is None: has_more = False else: + cursor = next_cursor page += 1 # Check max_items again before next fetch @@ -1148,49 +1152,191 @@ async def _fetch_batch_with_retry( *, scope: str, filter: Optional[str], - page: int, + cursor: Optional[str], limit: int, max_retries: int, fields: Optional[str], - ) -> List[Union[TraceWithFullDetails, ObservationsView]]: + ) -> Tuple[List[Union[TraceWithFullDetails, ObservationsView]], Optional[str]]: """Fetch a batch of items with retry logic. Args: scope: The type of items ("traces", "observations"). filter: JSON filter string for querying. - page: Page number (1-indexed). + cursor: Cursor from the previous response. limit: Number of items per page. max_retries: Maximum number of retry attempts. verbose: Whether to log retry attempts. fields: Trace fields to fetch Returns: - List of items from the API. + A tuple containing the items and the next-page cursor. Raises: Exception: If all retry attempts fail. """ - if scope == "traces": - response = self.client.api.trace.list( - page=page, - limit=limit, - filter=filter, - request_options={"max_retries": max_retries}, - fields=fields, - ) # type: ignore - return list(response.data) # type: ignore - elif scope == "observations": - response = self.client.api.legacy.observations_v1.get_many( - page=page, - limit=limit, - filter=filter, - request_options={"max_retries": max_retries}, - ) # type: ignore - return list(response.data) # type: ignore - else: + if scope not in {"traces", "observations"}: error_message = f"Invalid scope: {scope}" raise ValueError(error_message) + response = self.client.api.observations.get_many( + fields=self._get_v2_observation_fields(scope=scope, trace_fields=fields), + cursor=cursor, + limit=limit, + filter=self._build_v2_filter(filter=filter, scope=scope), + request_options={"max_retries": max_retries}, + ) + + if scope == "traces": + items: List[Union[TraceWithFullDetails, ObservationsView]] = [ + self._observation_to_trace(observation) for observation in response.data + ] + else: + items = [ + self._observation_to_legacy_view(observation) + for observation in response.data + ] + + return items, response.meta.cursor + + @staticmethod + def _get_v2_observation_fields(*, scope: str, trace_fields: Optional[str]) -> str: + """Map legacy trace field groups to v2 observation field groups.""" + all_fields = { + "basic", + "time", + "io", + "metadata", + "model", + "usage", + "prompt", + "metrics", + "trace_context", + } + if scope == "observations" or trace_fields is None: + selected_fields = all_fields + else: + requested_fields = { + field.strip() for field in trace_fields.split(",") if field.strip() + } + selected_fields = {"basic", "time", "trace_context"} + if "io" in requested_fields: + selected_fields.update({"io", "metadata"}) + if "metrics" in requested_fields: + selected_fields.update({"metrics", "usage"}) + + return ",".join(sorted(selected_fields)) + + @staticmethod + def _build_v2_filter(*, filter: Optional[str], scope: str) -> Optional[str]: + """Adapt legacy filter columns and select root observations for traces.""" + try: + filters = json.loads(filter) if filter else [] + except json.JSONDecodeError: + return filter + + if not isinstance(filters, list): + return filter + + column_aliases = { + "timestamp": "startTime", + "start_time": "startTime", + "user_id": "userId", + "session_id": "sessionId", + } + for condition in filters: + if ( + isinstance(condition, dict) + and condition.get("column") in column_aliases + ): + condition["column"] = column_aliases[condition["column"]] + + if scope == "traces": + filters.append( + { + "type": "boolean", + "column": "isRootObservation", + "operator": "=", + "value": True, + } + ) + + return json.dumps(filters) + + @classmethod + def _observation_to_trace(cls, observation: ObservationV2) -> TraceWithFullDetails: + """Adapt a v2 root observation to the established trace mapper contract.""" + trace_id = observation.trace_id or observation.id + return TraceWithFullDetails.model_construct( + id=trace_id, + timestamp=observation.start_time, + name=observation.trace_name or observation.name, + input=cls._parse_io_value(observation.input), + output=cls._parse_io_value(observation.output), + session_id=observation.session_id, + release=observation.release, + version=observation.version, + user_id=observation.user_id, + metadata=observation.metadata, + tags=observation.tags or [], + public=observation.public or False, + environment=observation.environment or "default", + html_path="", + latency=observation.latency, + total_cost=observation.total_cost, + observations=[], + scores=[], + ) + + @classmethod + def _observation_to_legacy_view( + cls, observation: ObservationV2 + ) -> ObservationsView: + """Adapt a v2 observation to the established observation mapper contract.""" + usage_details = observation.usage_details or {} + return ObservationsView.model_construct( + id=observation.id, + trace_id=observation.trace_id, + type=observation.type, + name=observation.name, + start_time=observation.start_time, + end_time=observation.end_time, + completion_start_time=observation.completion_start_time, + model=observation.model, + model_parameters=observation.model_parameters or {}, + input=cls._parse_io_value(observation.input), + version=observation.version, + metadata=observation.metadata, + output=cls._parse_io_value(observation.output), + usage=Usage( + input=usage_details.get("input", 0), + output=usage_details.get("output", 0), + total=usage_details.get("total", 0), + ), + level=observation.level or ObservationLevel.DEFAULT, + status_message=observation.status_message, + parent_observation_id=observation.parent_observation_id, + prompt_id=observation.prompt_id, + usage_details=usage_details, + cost_details=observation.cost_details or {}, + environment=observation.environment or "default", + prompt_name=observation.prompt_name, + prompt_version=observation.prompt_version, + model_id=observation.model_id, + latency=observation.latency, + time_to_first_token=observation.time_to_first_token, + ) + + @staticmethod + def _parse_io_value(value: Any) -> Any: + """Restore the parsed JSON behavior of the legacy read endpoints.""" + if not isinstance(value, str): + return value + + try: + return json.loads(value) + except json.JSONDecodeError: + return value + async def _process_batch_evaluation_item( self, item: Union[TraceWithFullDetails, ObservationsView], @@ -1573,11 +1719,7 @@ def _get_timestamp_field_for_scope(scope: str) -> str: Returns: The field name to use in filters. """ - if scope == "traces": - return "timestamp" - elif scope == "observations": - return "start_time" - return "timestamp" # Default + return "startTime" @staticmethod def _dedupe_tags(tags: Optional[List[str]]) -> List[str]: diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py new file mode 100644 index 000000000..74a5bd13a --- /dev/null +++ b/tests/unit/test_batch_evaluation.py @@ -0,0 +1,167 @@ +import json +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from langfuse.api import ObservationsView, ObservationV2, TraceWithFullDetails +from langfuse.batch_evaluation import ( + BatchEvaluationRunner, + EvaluatorInputs, +) +from langfuse.experiment import Evaluation + + +def _observation( + *, + observation_id: str = "observation-id", + trace_id: str = "trace-id", + is_root: bool = False, +) -> ObservationV2: + return ObservationV2( + id=observation_id, + trace_id=trace_id, + start_time=datetime(2026, 1, 2, tzinfo=timezone.utc), + project_id="project-id", + parent_observation_id=None, + type="SPAN", + is_root_observation=is_root, + name="root-span", + trace_name="trace-name", + input='{"question": "hello"}', + output='"answer"', + metadata={"source": "test"}, + tags=["production"], + environment="production", + ) + + +@pytest.mark.asyncio +async def test_fetches_traces_as_root_observations_via_v2_api() -> None: + client = MagicMock() + client.api.observations.get_many.return_value = SimpleNamespace( + data=[_observation(is_root=True)], + meta=SimpleNamespace(cursor="next-cursor"), + ) + runner = BatchEvaluationRunner(client) + + items, cursor = await runner._fetch_batch_with_retry( + scope="traces", + filter='[{"type":"string","column":"user_id","operator":"=","value":"user"}]', + cursor=None, + limit=10, + max_retries=2, + fields="io", + ) + + assert cursor == "next-cursor" + assert len(items) == 1 + trace = items[0] + assert isinstance(trace, TraceWithFullDetails) + assert trace.id == "trace-id" + assert trace.timestamp == datetime(2026, 1, 2, tzinfo=timezone.utc) + assert trace.name == "trace-name" + assert trace.input == {"question": "hello"} + assert trace.output == "answer" + + kwargs = client.api.observations.get_many.call_args.kwargs + assert kwargs["cursor"] is None + assert kwargs["request_options"] == {"max_retries": 2} + assert set(kwargs["fields"].split(",")) == { + "basic", + "io", + "metadata", + "time", + "trace_context", + } + assert json.loads(kwargs["filter"]) == [ + { + "type": "string", + "column": "userId", + "operator": "=", + "value": "user", + }, + { + "type": "boolean", + "column": "isRootObservation", + "operator": "=", + "value": True, + }, + ] + + +@pytest.mark.asyncio +async def test_fetches_observations_via_v2_api() -> None: + client = MagicMock() + client.api.observations.get_many.return_value = SimpleNamespace( + data=[_observation()], + meta=SimpleNamespace(cursor=None), + ) + runner = BatchEvaluationRunner(client) + + items, cursor = await runner._fetch_batch_with_retry( + scope="observations", + filter=None, + cursor="current-cursor", + limit=25, + max_retries=3, + fields=None, + ) + + assert cursor is None + assert len(items) == 1 + observation = items[0] + assert isinstance(observation, ObservationsView) + assert observation.id == "observation-id" + assert observation.trace_id == "trace-id" + assert observation.input == {"question": "hello"} + assert observation.output == "answer" + + kwargs = client.api.observations.get_many.call_args.kwargs + assert kwargs["cursor"] == "current-cursor" + assert kwargs["filter"] == "[]" + assert "io" in kwargs["fields"].split(",") + + +def test_resume_filter_uses_v2_start_time_column() -> None: + assert BatchEvaluationRunner._get_timestamp_field_for_scope("traces") == "startTime" + assert ( + BatchEvaluationRunner._get_timestamp_field_for_scope("observations") + == "startTime" + ) + + +@pytest.mark.asyncio +async def test_run_uses_v2_cursor_for_next_batch() -> None: + client = MagicMock() + client.api.observations.get_many.side_effect = [ + SimpleNamespace( + data=[_observation(observation_id="first")], + meta=SimpleNamespace(cursor="next-cursor"), + ), + SimpleNamespace( + data=[_observation(observation_id="second")], + meta=SimpleNamespace(cursor=None), + ), + ] + runner = BatchEvaluationRunner(client) + + result = await runner.run_async( + scope="observations", + mapper=lambda *, item: EvaluatorInputs( + input=item.input, + output=item.output, + ), + evaluators=[ + lambda **kwargs: Evaluation(name="quality", value=1.0), + ], + fetch_batch_size=1, + ) + + assert result.total_items_processed == 2 + assert result.completed is True + assert [ + call.kwargs["cursor"] + for call in client.api.observations.get_many.call_args_list + ] == [None, "next-cursor"] From 65f638ab82ff52985263ca4141484d82bfb1432b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 7 Sep 2026 10:06:41 +0000 Subject: [PATCH 2/7] fix(evals): retain v3 batch evaluation fallback Co-authored-by: Hassieb Pakzad --- langfuse/batch_evaluation.py | 73 ++++++++++++++++++++++++++--- tests/unit/test_batch_evaluation.py | 47 ++++++++++++++++++- 2 files changed, 112 insertions(+), 8 deletions(-) diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index 56c86f569..55bacd068 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -24,6 +24,7 @@ ) from langfuse.api import ( + NotFoundError, ObservationLevel, ObservationsView, ObservationV2, @@ -831,6 +832,8 @@ class BatchEvaluationRunner: client: The Langfuse client instance used for API calls and score creation. """ + _LEGACY_PAGINATION_CURSOR = "__langfuse_batch_evaluation_legacy__" + def __init__(self, client: "Langfuse"): """Initialize the batch evaluation runner. @@ -951,6 +954,7 @@ async def run_async( items, next_cursor = await self._fetch_batch_with_retry( scope=scope, filter=effective_filter, + page=page, cursor=cursor, limit=fetch_batch_size, max_retries=max_retries, @@ -1152,6 +1156,7 @@ async def _fetch_batch_with_retry( *, scope: str, filter: Optional[str], + page: int, cursor: Optional[str], limit: int, max_retries: int, @@ -1162,6 +1167,7 @@ async def _fetch_batch_with_retry( Args: scope: The type of items ("traces", "observations"). filter: JSON filter string for querying. + page: Page number used by the v3 compatibility fallback. cursor: Cursor from the previous response. limit: Number of items per page. max_retries: Maximum number of retry attempts. @@ -1178,13 +1184,35 @@ async def _fetch_batch_with_retry( error_message = f"Invalid scope: {scope}" raise ValueError(error_message) - response = self.client.api.observations.get_many( - fields=self._get_v2_observation_fields(scope=scope, trace_fields=fields), - cursor=cursor, - limit=limit, - filter=self._build_v2_filter(filter=filter, scope=scope), - request_options={"max_retries": max_retries}, - ) + if cursor == self._LEGACY_PAGINATION_CURSOR: + return self._fetch_legacy_batch( + scope=scope, + filter=filter, + page=page, + limit=limit, + max_retries=max_retries, + fields=fields, + ) + + try: + response = self.client.api.observations.get_many( + fields=self._get_v2_observation_fields( + scope=scope, trace_fields=fields + ), + cursor=cursor, + limit=limit, + filter=self._build_v2_filter(filter=filter, scope=scope), + request_options={"max_retries": max_retries}, + ) + except NotFoundError: + return self._fetch_legacy_batch( + scope=scope, + filter=filter, + page=page, + limit=limit, + max_retries=max_retries, + fields=fields, + ) if scope == "traces": items: List[Union[TraceWithFullDetails, ObservationsView]] = [ @@ -1198,6 +1226,37 @@ async def _fetch_batch_with_retry( return items, response.meta.cursor + def _fetch_legacy_batch( + self, + *, + scope: str, + filter: Optional[str], + page: int, + limit: int, + max_retries: int, + fields: Optional[str], + ) -> Tuple[List[Union[TraceWithFullDetails, ObservationsView]], Optional[str]]: + """Fetch from v3 read APIs when the v2 endpoint is unavailable.""" + if scope == "traces": + response = self.client.api.trace.list( + page=page, + limit=limit, + filter=filter, + request_options={"max_retries": max_retries}, + fields=fields, + ) + else: + response = self.client.api.legacy.observations_v1.get_many( + page=page, + limit=limit, + filter=filter, + request_options={"max_retries": max_retries}, + ) + + items = list(response.data) + next_cursor = self._LEGACY_PAGINATION_CURSOR if len(items) == limit else None + return items, next_cursor + @staticmethod def _get_v2_observation_fields(*, scope: str, trace_fields: Optional[str]) -> str: """Map legacy trace field groups to v2 observation field groups.""" diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py index 74a5bd13a..b0890a78b 100644 --- a/tests/unit/test_batch_evaluation.py +++ b/tests/unit/test_batch_evaluation.py @@ -5,7 +5,12 @@ import pytest -from langfuse.api import ObservationsView, ObservationV2, TraceWithFullDetails +from langfuse.api import ( + NotFoundError, + ObservationsView, + ObservationV2, + TraceWithFullDetails, +) from langfuse.batch_evaluation import ( BatchEvaluationRunner, EvaluatorInputs, @@ -49,6 +54,7 @@ async def test_fetches_traces_as_root_observations_via_v2_api() -> None: items, cursor = await runner._fetch_batch_with_retry( scope="traces", filter='[{"type":"string","column":"user_id","operator":"=","value":"user"}]', + page=1, cursor=None, limit=10, max_retries=2, @@ -103,6 +109,7 @@ async def test_fetches_observations_via_v2_api() -> None: items, cursor = await runner._fetch_batch_with_retry( scope="observations", filter=None, + page=1, cursor="current-cursor", limit=25, max_retries=3, @@ -132,6 +139,44 @@ def test_resume_filter_uses_v2_start_time_column() -> None: ) +@pytest.mark.asyncio +async def test_falls_back_to_v3_read_api_when_v2_is_unavailable() -> None: + client = MagicMock() + client.api.observations.get_many.side_effect = NotFoundError(body="not found") + legacy_observation = MagicMock(spec=ObservationsView) + client.api.legacy.observations_v1.get_many.return_value = SimpleNamespace( + data=[legacy_observation] + ) + runner = BatchEvaluationRunner(client) + + items, cursor = await runner._fetch_batch_with_retry( + scope="observations", + filter=None, + page=1, + cursor=None, + limit=1, + max_retries=3, + fields=None, + ) + + assert items == [legacy_observation] + assert cursor == runner._LEGACY_PAGINATION_CURSOR + + client.api.observations.get_many.reset_mock() + await runner._fetch_batch_with_retry( + scope="observations", + filter=None, + page=2, + cursor=cursor, + limit=1, + max_retries=3, + fields=None, + ) + + client.api.observations.get_many.assert_not_called() + assert client.api.legacy.observations_v1.get_many.call_args.kwargs["page"] == 2 + + @pytest.mark.asyncio async def test_run_uses_v2_cursor_for_next_batch() -> None: client = MagicMock() From 7607236f57a02448b4a516552d4d6f4398bf33a6 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 7 Sep 2026 10:07:27 +0000 Subject: [PATCH 3/7] fix(evals): preserve v3 resume filters Co-authored-by: Hassieb Pakzad --- langfuse/batch_evaluation.py | 23 +++++++++++++++++++++-- tests/unit/test_batch_evaluation.py | 10 ++++++++-- 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index 55bacd068..3e94bad58 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -1237,11 +1237,12 @@ def _fetch_legacy_batch( fields: Optional[str], ) -> Tuple[List[Union[TraceWithFullDetails, ObservationsView]], Optional[str]]: """Fetch from v3 read APIs when the v2 endpoint is unavailable.""" + legacy_filter = self._build_legacy_filter(filter=filter, scope=scope) if scope == "traces": response = self.client.api.trace.list( page=page, limit=limit, - filter=filter, + filter=legacy_filter, request_options={"max_retries": max_retries}, fields=fields, ) @@ -1249,7 +1250,7 @@ def _fetch_legacy_batch( response = self.client.api.legacy.observations_v1.get_many( page=page, limit=limit, - filter=filter, + filter=legacy_filter, request_options={"max_retries": max_retries}, ) @@ -1321,6 +1322,24 @@ def _build_v2_filter(*, filter: Optional[str], scope: str) -> Optional[str]: return json.dumps(filters) + @staticmethod + def _build_legacy_filter(*, filter: Optional[str], scope: str) -> Optional[str]: + """Restore the v3 timestamp column when a resumed run falls back.""" + try: + filters = json.loads(filter) if filter else [] + except json.JSONDecodeError: + return filter + + if not isinstance(filters, list): + return filter + + timestamp_column = "timestamp" if scope == "traces" else "start_time" + for condition in filters: + if isinstance(condition, dict) and condition.get("column") == "startTime": + condition["column"] = timestamp_column + + return json.dumps(filters) + @classmethod def _observation_to_trace(cls, observation: ObservationV2) -> TraceWithFullDetails: """Adapt a v2 root observation to the established trace mapper contract.""" diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py index b0890a78b..9903bbe92 100644 --- a/tests/unit/test_batch_evaluation.py +++ b/tests/unit/test_batch_evaluation.py @@ -108,7 +108,7 @@ async def test_fetches_observations_via_v2_api() -> None: items, cursor = await runner._fetch_batch_with_retry( scope="observations", - filter=None, + filter='[{"type":"datetime","column":"startTime","operator":">","value":"2026-01-01"}]', page=1, cursor="current-cursor", limit=25, @@ -161,11 +161,17 @@ async def test_falls_back_to_v3_read_api_when_v2_is_unavailable() -> None: assert items == [legacy_observation] assert cursor == runner._LEGACY_PAGINATION_CURSOR + assert ( + json.loads( + client.api.legacy.observations_v1.get_many.call_args.kwargs["filter"] + )[0]["column"] + == "start_time" + ) client.api.observations.get_many.reset_mock() await runner._fetch_batch_with_retry( scope="observations", - filter=None, + filter='[{"type":"datetime","column":"startTime","operator":">","value":"2026-01-01"}]', page=2, cursor=cursor, limit=1, From 334ac151c03c78b94469be0511a178dc9f0740f7 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 7 Sep 2026 10:07:58 +0000 Subject: [PATCH 4/7] test(evals): correct fallback filter fixture Co-authored-by: Hassieb Pakzad --- tests/unit/test_batch_evaluation.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py index 9903bbe92..953e2e4ce 100644 --- a/tests/unit/test_batch_evaluation.py +++ b/tests/unit/test_batch_evaluation.py @@ -108,7 +108,7 @@ async def test_fetches_observations_via_v2_api() -> None: items, cursor = await runner._fetch_batch_with_retry( scope="observations", - filter='[{"type":"datetime","column":"startTime","operator":">","value":"2026-01-01"}]', + filter=None, page=1, cursor="current-cursor", limit=25, @@ -151,7 +151,7 @@ async def test_falls_back_to_v3_read_api_when_v2_is_unavailable() -> None: items, cursor = await runner._fetch_batch_with_retry( scope="observations", - filter=None, + filter='[{"type":"datetime","column":"startTime","operator":">","value":"2026-01-01"}]', page=1, cursor=None, limit=1, From 8d1993dda1301e4180146c1474522dff913012dd Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 7 Sep 2026 11:11:52 +0000 Subject: [PATCH 5/7] fix(evals): make v2 observation reads opt in Co-authored-by: Hassieb Pakzad --- langfuse/_client/client.py | 19 +++- langfuse/batch_evaluation.py | 164 ++++++++++++++++++---------- tests/unit/test_batch_evaluation.py | 104 +++++++++++++++--- 3 files changed, 213 insertions(+), 74 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index f8267ea45..3272f7f9f 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -3261,6 +3261,7 @@ def run_batched_evaluation( filter: Optional[str] = None, fetch_batch_size: int = 50, fetch_trace_fields: Optional[str] = None, + observation_read_api: Literal["legacy", "v2"] = "legacy", max_items: Optional[int] = None, max_retries: int = 3, evaluators: List[EvaluatorFunction], @@ -3305,7 +3306,22 @@ def run_batched_evaluation( Default: None (fetches all items). fetch_batch_size: Number of items to fetch per API call and hold in memory. Larger values may be faster but use more memory. Default: 50. - fetch_trace_fields: Comma-separated list of fields to include when fetching traces. Available field groups: 'core' (always included), 'io' (input, output, metadata), 'scores', 'observations', 'metrics'. If not specified, all fields are returned. Example: 'core,scores,metrics'. Note: Excluded 'observations' or 'scores' fields return empty arrays; excluded 'metrics' returns -1 for 'totalCost' and 'latency'. Only relevant if scope is 'traces'. + fetch_trace_fields: Comma-separated list of fields to include when + fetching traces. With `observation_read_api="legacy"`, available + groups are 'core' (always included), 'io', 'scores', + 'observations', and 'metrics'; if omitted, all groups are + returned. With `observation_read_api="v2"`, only 'core' and + 'io' are supported and omitting this option selects both. + observation_read_api: Observation Read API used to fetch items. + - "legacy" (default) calls `GET /api/public/traces` for + `scope="traces"` and the legacy + `GET /api/public/observations` endpoint for + `scope="observations"`. Use this with Langfuse platform v3. + - "v2" calls `GET /api/public/v2/observations` for both + scopes. Use this with Langfuse platform v4 `events_only` + deployments. Trace scope derives trace-shaped mapper items + from root observations and supports only the `core` and `io` + trace field groups. max_items: Maximum total number of items to process. If None, processes all items matching the filter. Useful for testing or limiting evaluation runs. Default: None (process all). @@ -3475,6 +3491,7 @@ def composite_evaluator(*, item, evaluations): filter=filter, fetch_batch_size=fetch_batch_size, fetch_trace_fields=fetch_trace_fields, + observation_read_api=observation_read_api, max_items=max_items, max_concurrency=max_concurrency, composite_evaluator=composite_evaluator, diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index 3e94bad58..b5ae5e3b5 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -15,6 +15,7 @@ Awaitable, Dict, List, + Literal, Optional, Protocol, Set, @@ -24,7 +25,6 @@ ) from langfuse.api import ( - NotFoundError, ObservationLevel, ObservationsView, ObservationV2, @@ -832,7 +832,7 @@ class BatchEvaluationRunner: client: The Langfuse client instance used for API calls and score creation. """ - _LEGACY_PAGINATION_CURSOR = "__langfuse_batch_evaluation_legacy__" + _LEGACY_NEXT_PAGE = "__langfuse_batch_evaluation_legacy__" def __init__(self, client: "Langfuse"): """Initialize the batch evaluation runner. @@ -851,6 +851,7 @@ async def run_async( filter: Optional[str] = None, fetch_batch_size: int = 50, fetch_trace_fields: Optional[str] = "io", + observation_read_api: Literal["legacy", "v2"] = "legacy", max_items: Optional[int] = None, max_concurrency: int = 5, composite_evaluator: Optional[CompositeEvaluatorFunction] = None, @@ -873,7 +874,18 @@ async def run_async( evaluators: List of evaluation functions to run on each item. filter: JSON filter string for querying items. fetch_batch_size: Number of items to fetch per API call. - fetch_trace_fields: Comma-separated list of fields to include when fetching traces. Available field groups: 'core' (always included), 'io' (input, output, metadata), 'scores', 'observations', 'metrics'. If not specified, all fields are returned. Example: 'core,scores,metrics'. Note: Excluded 'observations' or 'scores' fields return empty arrays; excluded 'metrics' returns -1 for 'totalCost' and 'latency'. Only relevant if scope is 'traces'. Default: 'io' + fetch_trace_fields: Comma-separated trace field groups. The legacy + API supports 'core', 'io', 'scores', 'observations', and + 'metrics'. The v2 API supports only 'core' and 'io'. + observation_read_api: Observation Read API used to fetch items. + - "legacy" (default) uses `GET /api/public/traces` for trace + scope and `GET /api/public/observations` for observation + scope. This is compatible with Langfuse platform v3. + - "v2" uses `GET /api/public/v2/observations` for both scopes. + This is compatible with Langfuse platform v4 `events_only` + deployments. For trace scope, root observations are adapted + to trace-shaped objects; legacy trace-specific fields and + filters are not all available. max_items: Maximum number of items to process (None = all). max_concurrency: Maximum number of concurrent evaluations. composite_evaluator: Optional function to create composite scores. @@ -889,6 +901,11 @@ async def run_async( Returns: BatchEvaluationResult with comprehensive statistics. """ + self._validate_read_options( + scope=scope, + observation_read_api=observation_read_api, + fetch_trace_fields=fetch_trace_fields, + ) start_time = time.time() # Initialize tracking variables @@ -911,13 +928,16 @@ async def run_async( } # Handle resume token by modifying filter - effective_filter = self._build_timestamp_filter(filter, resume_from) + effective_filter = self._build_timestamp_filter( + filter, resume_from, observation_read_api + ) normalized_additional_trace_tags = ( self._dedupe_tags(_additional_trace_tags) if _additional_trace_tags is not None else [] ) updated_trace_ids: Set[str] = set() + fetched_trace_ids: Set[str] = set() # Create semaphore for concurrency control semaphore = asyncio.Semaphore(max_concurrency) @@ -959,6 +979,7 @@ async def run_async( limit=fetch_batch_size, max_retries=max_retries, fields=fetch_trace_fields, + observation_read_api=observation_read_api, ) except Exception as e: # Failed after max_retries - create resume token and return @@ -990,12 +1011,25 @@ async def run_async( item_evaluations=item_evaluations, ) - # Check if we got any items + if scope == "traces" and observation_read_api == "v2": + unique_items = [] + for item in items: + if item.id not in fetched_trace_ids: + unique_items.append(item) + fetched_trace_ids.add(item.id) + items = unique_items + + # Check if we got any new items if not items: - has_more = False - if verbose: - logger.info("No more items to fetch") - break + if next_cursor is None: + has_more = False + if verbose: + logger.info("No more items to fetch") + break + + cursor = next_cursor + page += 1 + continue total_items_fetched += len(items) @@ -1161,6 +1195,7 @@ async def _fetch_batch_with_retry( limit: int, max_retries: int, fields: Optional[str], + observation_read_api: Literal["legacy", "v2"], ) -> Tuple[List[Union[TraceWithFullDetails, ObservationsView]], Optional[str]]: """Fetch a batch of items with retry logic. @@ -1173,6 +1208,7 @@ async def _fetch_batch_with_retry( max_retries: Maximum number of retry attempts. verbose: Whether to log retry attempts. fields: Trace fields to fetch + observation_read_api: Read API selected by the caller. Returns: A tuple containing the items and the next-page cursor. @@ -1184,7 +1220,7 @@ async def _fetch_batch_with_retry( error_message = f"Invalid scope: {scope}" raise ValueError(error_message) - if cursor == self._LEGACY_PAGINATION_CURSOR: + if observation_read_api == "legacy": return self._fetch_legacy_batch( scope=scope, filter=filter, @@ -1194,25 +1230,13 @@ async def _fetch_batch_with_retry( fields=fields, ) - try: - response = self.client.api.observations.get_many( - fields=self._get_v2_observation_fields( - scope=scope, trace_fields=fields - ), - cursor=cursor, - limit=limit, - filter=self._build_v2_filter(filter=filter, scope=scope), - request_options={"max_retries": max_retries}, - ) - except NotFoundError: - return self._fetch_legacy_batch( - scope=scope, - filter=filter, - page=page, - limit=limit, - max_retries=max_retries, - fields=fields, - ) + response = self.client.api.observations.get_many( + fields=self._get_v2_observation_fields(scope=scope, trace_fields=fields), + cursor=cursor, + limit=limit, + filter=self._build_v2_filter(filter=filter, scope=scope), + request_options={"max_retries": max_retries}, + ) if scope == "traces": items: List[Union[TraceWithFullDetails, ObservationsView]] = [ @@ -1236,13 +1260,12 @@ def _fetch_legacy_batch( max_retries: int, fields: Optional[str], ) -> Tuple[List[Union[TraceWithFullDetails, ObservationsView]], Optional[str]]: - """Fetch from v3 read APIs when the v2 endpoint is unavailable.""" - legacy_filter = self._build_legacy_filter(filter=filter, scope=scope) + """Fetch from the legacy trace or observation read API.""" if scope == "traces": response = self.client.api.trace.list( page=page, limit=limit, - filter=legacy_filter, + filter=filter, request_options={"max_retries": max_retries}, fields=fields, ) @@ -1250,14 +1273,48 @@ def _fetch_legacy_batch( response = self.client.api.legacy.observations_v1.get_many( page=page, limit=limit, - filter=legacy_filter, + filter=filter, request_options={"max_retries": max_retries}, ) items = list(response.data) - next_cursor = self._LEGACY_PAGINATION_CURSOR if len(items) == limit else None + next_cursor = self._LEGACY_NEXT_PAGE if len(items) == limit else None return items, next_cursor + @staticmethod + def _validate_read_options( + *, + scope: str, + observation_read_api: str, + fetch_trace_fields: Optional[str], + ) -> None: + """Validate options that differ between observation read APIs.""" + if observation_read_api not in {"legacy", "v2"}: + message = ( + "Invalid observation_read_api: " + f"{observation_read_api}. Expected 'legacy' or 'v2'." + ) + raise ValueError(message) + + if observation_read_api == "v2" and scope == "traces" and fetch_trace_fields: + requested_fields = { + field.strip() + for field in fetch_trace_fields.split(",") + if field.strip() + } + unsupported_fields = requested_fields & { + "metrics", + "observations", + "scores", + } + if unsupported_fields: + unsupported = ", ".join(sorted(unsupported_fields)) + message = ( + "observation_read_api='v2' does not support legacy trace " + f"field groups: {unsupported}. Use 'core' and/or 'io'." + ) + raise ValueError(message) + @staticmethod def _get_v2_observation_fields(*, scope: str, trace_fields: Optional[str]) -> str: """Map legacy trace field groups to v2 observation field groups.""" @@ -1272,11 +1329,13 @@ def _get_v2_observation_fields(*, scope: str, trace_fields: Optional[str]) -> st "metrics", "trace_context", } - if scope == "observations" or trace_fields is None: + if scope == "observations": selected_fields = all_fields else: requested_fields = { - field.strip() for field in trace_fields.split(",") if field.strip() + field.strip() + for field in (trace_fields or "core,io").split(",") + if field.strip() } selected_fields = {"basic", "time", "trace_context"} if "io" in requested_fields: @@ -1322,24 +1381,6 @@ def _build_v2_filter(*, filter: Optional[str], scope: str) -> Optional[str]: return json.dumps(filters) - @staticmethod - def _build_legacy_filter(*, filter: Optional[str], scope: str) -> Optional[str]: - """Restore the v3 timestamp column when a resumed run falls back.""" - try: - filters = json.loads(filter) if filter else [] - except json.JSONDecodeError: - return filter - - if not isinstance(filters, list): - return filter - - timestamp_column = "timestamp" if scope == "traces" else "start_time" - for condition in filters: - if isinstance(condition, dict) and condition.get("column") == "startTime": - condition["column"] = timestamp_column - - return json.dumps(filters) - @classmethod def _observation_to_trace(cls, observation: ObservationV2) -> TraceWithFullDetails: """Adapt a v2 root observation to the established trace mapper contract.""" @@ -1708,12 +1749,14 @@ def _build_timestamp_filter( self, original_filter: Optional[str], resume_from: Optional[BatchEvaluationResumeToken], + observation_read_api: Literal["legacy", "v2"], ) -> Optional[str]: """Build filter with timestamp constraint for resume capability. Args: original_filter: The original JSON filter string. resume_from: Optional resume token with timestamp information. + observation_read_api: Read API selected by the caller. Returns: Modified filter string with timestamp constraint, or original filter. @@ -1736,7 +1779,9 @@ def _build_timestamp_filter( filter_list = [] # Add timestamp constraint to filter array - timestamp_field = self._get_timestamp_field_for_scope(resume_from.scope) + timestamp_field = self._get_timestamp_field_for_scope( + resume_from.scope, observation_read_api + ) timestamp_filter = { "type": "datetime", "column": timestamp_field, @@ -1788,16 +1833,21 @@ def _get_item_timestamp( return "" @staticmethod - def _get_timestamp_field_for_scope(scope: str) -> str: + def _get_timestamp_field_for_scope( + scope: str, observation_read_api: Literal["legacy", "v2"] + ) -> str: """Get the timestamp field name for filtering based on scope. Args: scope: The type of items. + observation_read_api: Read API selected by the caller. Returns: The field name to use in filters. """ - return "startTime" + if observation_read_api == "v2": + return "startTime" + return "timestamp" if scope == "traces" else "start_time" @staticmethod def _dedupe_tags(tags: Optional[List[str]]) -> List[str]: diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py index 953e2e4ce..1875fb98a 100644 --- a/tests/unit/test_batch_evaluation.py +++ b/tests/unit/test_batch_evaluation.py @@ -6,7 +6,6 @@ import pytest from langfuse.api import ( - NotFoundError, ObservationsView, ObservationV2, TraceWithFullDetails, @@ -59,6 +58,7 @@ async def test_fetches_traces_as_root_observations_via_v2_api() -> None: limit=10, max_retries=2, fields="io", + observation_read_api="v2", ) assert cursor == "next-cursor" @@ -114,6 +114,7 @@ async def test_fetches_observations_via_v2_api() -> None: limit=25, max_retries=3, fields=None, + observation_read_api="v2", ) assert cursor is None @@ -131,18 +132,24 @@ async def test_fetches_observations_via_v2_api() -> None: assert "io" in kwargs["fields"].split(",") -def test_resume_filter_uses_v2_start_time_column() -> None: - assert BatchEvaluationRunner._get_timestamp_field_for_scope("traces") == "startTime" +def test_resume_filter_uses_read_api_timestamp_column() -> None: assert ( - BatchEvaluationRunner._get_timestamp_field_for_scope("observations") + BatchEvaluationRunner._get_timestamp_field_for_scope("traces", "v2") == "startTime" ) + assert ( + BatchEvaluationRunner._get_timestamp_field_for_scope("traces", "legacy") + == "timestamp" + ) + assert ( + BatchEvaluationRunner._get_timestamp_field_for_scope("observations", "legacy") + == "start_time" + ) @pytest.mark.asyncio -async def test_falls_back_to_v3_read_api_when_v2_is_unavailable() -> None: +async def test_legacy_read_api_uses_page_pagination() -> None: client = MagicMock() - client.api.observations.get_many.side_effect = NotFoundError(body="not found") legacy_observation = MagicMock(spec=ObservationsView) client.api.legacy.observations_v1.get_many.return_value = SimpleNamespace( data=[legacy_observation] @@ -151,38 +158,57 @@ async def test_falls_back_to_v3_read_api_when_v2_is_unavailable() -> None: items, cursor = await runner._fetch_batch_with_retry( scope="observations", - filter='[{"type":"datetime","column":"startTime","operator":">","value":"2026-01-01"}]', + filter='[{"type":"datetime","column":"start_time","operator":">","value":"2026-01-01"}]', page=1, cursor=None, limit=1, max_retries=3, fields=None, + observation_read_api="legacy", ) assert items == [legacy_observation] - assert cursor == runner._LEGACY_PAGINATION_CURSOR - assert ( - json.loads( - client.api.legacy.observations_v1.get_many.call_args.kwargs["filter"] - )[0]["column"] - == "start_time" - ) + assert cursor == runner._LEGACY_NEXT_PAGE + client.api.observations.get_many.assert_not_called() - client.api.observations.get_many.reset_mock() await runner._fetch_batch_with_retry( scope="observations", - filter='[{"type":"datetime","column":"startTime","operator":">","value":"2026-01-01"}]', + filter=None, page=2, cursor=cursor, limit=1, max_retries=3, fields=None, + observation_read_api="legacy", ) client.api.observations.get_many.assert_not_called() assert client.api.legacy.observations_v1.get_many.call_args.kwargs["page"] == 2 +@pytest.mark.asyncio +async def test_run_defaults_to_legacy_read_api() -> None: + client = MagicMock() + runner = BatchEvaluationRunner(client) + legacy_observation = runner._observation_to_legacy_view(_observation()) + client.api.legacy.observations_v1.get_many.return_value = SimpleNamespace( + data=[legacy_observation] + ) + + result = await runner.run_async( + scope="observations", + mapper=lambda *, item: EvaluatorInputs( + input=item.input, + output=item.output, + ), + evaluators=[], + ) + + assert result.total_items_processed == 1 + client.api.observations.get_many.assert_not_called() + client.api.legacy.observations_v1.get_many.assert_called_once() + + @pytest.mark.asyncio async def test_run_uses_v2_cursor_for_next_batch() -> None: client = MagicMock() @@ -208,6 +234,7 @@ async def test_run_uses_v2_cursor_for_next_batch() -> None: lambda **kwargs: Evaluation(name="quality", value=1.0), ], fetch_batch_size=1, + observation_read_api="v2", ) assert result.total_items_processed == 2 @@ -216,3 +243,48 @@ async def test_run_uses_v2_cursor_for_next_batch() -> None: call.kwargs["cursor"] for call in client.api.observations.get_many.call_args_list ] == [None, "next-cursor"] + + +@pytest.mark.asyncio +async def test_v2_trace_scope_deduplicates_logical_roots() -> None: + client = MagicMock() + client.api.observations.get_many.return_value = SimpleNamespace( + data=[ + _observation(observation_id="first-root", is_root=True), + _observation(observation_id="second-root", is_root=True), + ], + meta=SimpleNamespace(cursor=None), + ) + runner = BatchEvaluationRunner(client) + + result = await runner.run_async( + scope="traces", + mapper=lambda *, item: EvaluatorInputs( + input=item.input, + output=item.output, + ), + evaluators=[ + lambda **kwargs: Evaluation(name="quality", value=1.0), + ], + observation_read_api="v2", + ) + + assert result.total_items_processed == 1 + client.create_score.assert_called_once() + + +@pytest.mark.asyncio +async def test_v2_trace_scope_rejects_unsupported_field_groups() -> None: + runner = BatchEvaluationRunner(MagicMock()) + + with pytest.raises( + ValueError, + match="does not support legacy trace field groups: observations, scores", + ): + await runner.run_async( + scope="traces", + mapper=lambda *, item: EvaluatorInputs(input=None, output=None), + evaluators=[], + fetch_trace_fields="core,scores,observations", + observation_read_api="v2", + ) From 22aa20e0d603fdacf1aeb6227ea6b675b36b96fb Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 7 Sep 2026 12:23:18 +0000 Subject: [PATCH 6/7] fix(evals): restrict v2 reads to observations Co-authored-by: Hassieb Pakzad --- langfuse/_client/client.py | 16 ++-- langfuse/batch_evaluation.py | 126 ++++++---------------------- tests/unit/test_batch_evaluation.py | 95 +-------------------- 3 files changed, 33 insertions(+), 204 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 3272f7f9f..6634f20de 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -3307,21 +3307,17 @@ def run_batched_evaluation( fetch_batch_size: Number of items to fetch per API call and hold in memory. Larger values may be faster but use more memory. Default: 50. fetch_trace_fields: Comma-separated list of fields to include when - fetching traces. With `observation_read_api="legacy"`, available - groups are 'core' (always included), 'io', 'scores', - 'observations', and 'metrics'; if omitted, all groups are - returned. With `observation_read_api="v2"`, only 'core' and - 'io' are supported and omitting this option selects both. + fetching traces through the legacy API. Available groups are + 'core' (always included), 'io', 'scores', 'observations', and + 'metrics'; if omitted, all groups are returned. observation_read_api: Observation Read API used to fetch items. - "legacy" (default) calls `GET /api/public/traces` for `scope="traces"` and the legacy `GET /api/public/observations` endpoint for `scope="observations"`. Use this with Langfuse platform v3. - - "v2" calls `GET /api/public/v2/observations` for both - scopes. Use this with Langfuse platform v4 `events_only` - deployments. Trace scope derives trace-shaped mapper items - from root observations and supports only the `core` and `io` - trace field groups. + - "v2" calls `GET /api/public/v2/observations` and is only + supported with `scope="observations"`. Use this with Langfuse + platform v4 `events_only` deployments. max_items: Maximum total number of items to process. If None, processes all items matching the filter. Useful for testing or limiting evaluation runs. Default: None (process all). diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index b5ae5e3b5..58ba2a9ca 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -876,16 +876,14 @@ async def run_async( fetch_batch_size: Number of items to fetch per API call. fetch_trace_fields: Comma-separated trace field groups. The legacy API supports 'core', 'io', 'scores', 'observations', and - 'metrics'. The v2 API supports only 'core' and 'io'. + 'metrics'. Only relevant to the legacy trace scope. observation_read_api: Observation Read API used to fetch items. - "legacy" (default) uses `GET /api/public/traces` for trace scope and `GET /api/public/observations` for observation scope. This is compatible with Langfuse platform v3. - - "v2" uses `GET /api/public/v2/observations` for both scopes. - This is compatible with Langfuse platform v4 `events_only` - deployments. For trace scope, root observations are adapted - to trace-shaped objects; legacy trace-specific fields and - filters are not all available. + - "v2" uses `GET /api/public/v2/observations`. It is only + supported with observation scope and is compatible with + Langfuse platform v4 `events_only` deployments. max_items: Maximum number of items to process (None = all). max_concurrency: Maximum number of concurrent evaluations. composite_evaluator: Optional function to create composite scores. @@ -904,7 +902,6 @@ async def run_async( self._validate_read_options( scope=scope, observation_read_api=observation_read_api, - fetch_trace_fields=fetch_trace_fields, ) start_time = time.time() @@ -937,7 +934,6 @@ async def run_async( else [] ) updated_trace_ids: Set[str] = set() - fetched_trace_ids: Set[str] = set() # Create semaphore for concurrency control semaphore = asyncio.Semaphore(max_concurrency) @@ -1011,15 +1007,7 @@ async def run_async( item_evaluations=item_evaluations, ) - if scope == "traces" and observation_read_api == "v2": - unique_items = [] - for item in items: - if item.id not in fetched_trace_ids: - unique_items.append(item) - fetched_trace_ids.add(item.id) - items = unique_items - - # Check if we got any new items + # Check if we got any items if not items: if next_cursor is None: has_more = False @@ -1231,22 +1219,17 @@ async def _fetch_batch_with_retry( ) response = self.client.api.observations.get_many( - fields=self._get_v2_observation_fields(scope=scope, trace_fields=fields), + fields=self._get_v2_observation_fields(), cursor=cursor, limit=limit, - filter=self._build_v2_filter(filter=filter, scope=scope), + filter=self._build_v2_filter(filter=filter), request_options={"max_retries": max_retries}, ) - if scope == "traces": - items: List[Union[TraceWithFullDetails, ObservationsView]] = [ - self._observation_to_trace(observation) for observation in response.data - ] - else: - items = [ - self._observation_to_legacy_view(observation) - for observation in response.data - ] + items: List[Union[TraceWithFullDetails, ObservationsView]] = [ + self._observation_to_legacy_view(observation) + for observation in response.data + ] return items, response.meta.cursor @@ -1286,7 +1269,6 @@ def _validate_read_options( *, scope: str, observation_read_api: str, - fetch_trace_fields: Optional[str], ) -> None: """Validate options that differ between observation read APIs.""" if observation_read_api not in {"legacy", "v2"}: @@ -1296,29 +1278,18 @@ def _validate_read_options( ) raise ValueError(message) - if observation_read_api == "v2" and scope == "traces" and fetch_trace_fields: - requested_fields = { - field.strip() - for field in fetch_trace_fields.split(",") - if field.strip() - } - unsupported_fields = requested_fields & { - "metrics", - "observations", - "scores", - } - if unsupported_fields: - unsupported = ", ".join(sorted(unsupported_fields)) - message = ( - "observation_read_api='v2' does not support legacy trace " - f"field groups: {unsupported}. Use 'core' and/or 'io'." - ) - raise ValueError(message) + if observation_read_api == "v2" and scope != "observations": + message = ( + "observation_read_api='v2' is only supported with " + "scope='observations'. Use observation_read_api='legacy' " + "for scope='traces'." + ) + raise ValueError(message) @staticmethod - def _get_v2_observation_fields(*, scope: str, trace_fields: Optional[str]) -> str: - """Map legacy trace field groups to v2 observation field groups.""" - all_fields = { + def _get_v2_observation_fields() -> str: + """Return all v2 observation field groups needed by mappers.""" + fields = { "basic", "time", "io", @@ -1329,25 +1300,11 @@ def _get_v2_observation_fields(*, scope: str, trace_fields: Optional[str]) -> st "metrics", "trace_context", } - if scope == "observations": - selected_fields = all_fields - else: - requested_fields = { - field.strip() - for field in (trace_fields or "core,io").split(",") - if field.strip() - } - selected_fields = {"basic", "time", "trace_context"} - if "io" in requested_fields: - selected_fields.update({"io", "metadata"}) - if "metrics" in requested_fields: - selected_fields.update({"metrics", "usage"}) - - return ",".join(sorted(selected_fields)) + return ",".join(sorted(fields)) @staticmethod - def _build_v2_filter(*, filter: Optional[str], scope: str) -> Optional[str]: - """Adapt legacy filter columns and select root observations for traces.""" + def _build_v2_filter(*, filter: Optional[str]) -> Optional[str]: + """Adapt legacy observation filter columns to v2 names.""" try: filters = json.loads(filter) if filter else [] except json.JSONDecodeError: @@ -1369,43 +1326,8 @@ def _build_v2_filter(*, filter: Optional[str], scope: str) -> Optional[str]: ): condition["column"] = column_aliases[condition["column"]] - if scope == "traces": - filters.append( - { - "type": "boolean", - "column": "isRootObservation", - "operator": "=", - "value": True, - } - ) - return json.dumps(filters) - @classmethod - def _observation_to_trace(cls, observation: ObservationV2) -> TraceWithFullDetails: - """Adapt a v2 root observation to the established trace mapper contract.""" - trace_id = observation.trace_id or observation.id - return TraceWithFullDetails.model_construct( - id=trace_id, - timestamp=observation.start_time, - name=observation.trace_name or observation.name, - input=cls._parse_io_value(observation.input), - output=cls._parse_io_value(observation.output), - session_id=observation.session_id, - release=observation.release, - version=observation.version, - user_id=observation.user_id, - metadata=observation.metadata, - tags=observation.tags or [], - public=observation.public or False, - environment=observation.environment or "default", - html_path="", - latency=observation.latency, - total_cost=observation.total_cost, - observations=[], - scores=[], - ) - @classmethod def _observation_to_legacy_view( cls, observation: ObservationV2 diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py index 1875fb98a..1afabb0f7 100644 --- a/tests/unit/test_batch_evaluation.py +++ b/tests/unit/test_batch_evaluation.py @@ -1,4 +1,3 @@ -import json from datetime import datetime, timezone from types import SimpleNamespace from unittest.mock import MagicMock @@ -8,7 +7,6 @@ from langfuse.api import ( ObservationsView, ObservationV2, - TraceWithFullDetails, ) from langfuse.batch_evaluation import ( BatchEvaluationRunner, @@ -21,7 +19,6 @@ def _observation( *, observation_id: str = "observation-id", trace_id: str = "trace-id", - is_root: bool = False, ) -> ObservationV2: return ObservationV2( id=observation_id, @@ -30,7 +27,6 @@ def _observation( project_id="project-id", parent_observation_id=None, type="SPAN", - is_root_observation=is_root, name="root-span", trace_name="trace-name", input='{"question": "hello"}', @@ -41,62 +37,6 @@ def _observation( ) -@pytest.mark.asyncio -async def test_fetches_traces_as_root_observations_via_v2_api() -> None: - client = MagicMock() - client.api.observations.get_many.return_value = SimpleNamespace( - data=[_observation(is_root=True)], - meta=SimpleNamespace(cursor="next-cursor"), - ) - runner = BatchEvaluationRunner(client) - - items, cursor = await runner._fetch_batch_with_retry( - scope="traces", - filter='[{"type":"string","column":"user_id","operator":"=","value":"user"}]', - page=1, - cursor=None, - limit=10, - max_retries=2, - fields="io", - observation_read_api="v2", - ) - - assert cursor == "next-cursor" - assert len(items) == 1 - trace = items[0] - assert isinstance(trace, TraceWithFullDetails) - assert trace.id == "trace-id" - assert trace.timestamp == datetime(2026, 1, 2, tzinfo=timezone.utc) - assert trace.name == "trace-name" - assert trace.input == {"question": "hello"} - assert trace.output == "answer" - - kwargs = client.api.observations.get_many.call_args.kwargs - assert kwargs["cursor"] is None - assert kwargs["request_options"] == {"max_retries": 2} - assert set(kwargs["fields"].split(",")) == { - "basic", - "io", - "metadata", - "time", - "trace_context", - } - assert json.loads(kwargs["filter"]) == [ - { - "type": "string", - "column": "userId", - "operator": "=", - "value": "user", - }, - { - "type": "boolean", - "column": "isRootObservation", - "operator": "=", - "value": True, - }, - ] - - @pytest.mark.asyncio async def test_fetches_observations_via_v2_api() -> None: client = MagicMock() @@ -134,7 +74,7 @@ async def test_fetches_observations_via_v2_api() -> None: def test_resume_filter_uses_read_api_timestamp_column() -> None: assert ( - BatchEvaluationRunner._get_timestamp_field_for_scope("traces", "v2") + BatchEvaluationRunner._get_timestamp_field_for_scope("observations", "v2") == "startTime" ) assert ( @@ -246,45 +186,16 @@ async def test_run_uses_v2_cursor_for_next_batch() -> None: @pytest.mark.asyncio -async def test_v2_trace_scope_deduplicates_logical_roots() -> None: - client = MagicMock() - client.api.observations.get_many.return_value = SimpleNamespace( - data=[ - _observation(observation_id="first-root", is_root=True), - _observation(observation_id="second-root", is_root=True), - ], - meta=SimpleNamespace(cursor=None), - ) - runner = BatchEvaluationRunner(client) - - result = await runner.run_async( - scope="traces", - mapper=lambda *, item: EvaluatorInputs( - input=item.input, - output=item.output, - ), - evaluators=[ - lambda **kwargs: Evaluation(name="quality", value=1.0), - ], - observation_read_api="v2", - ) - - assert result.total_items_processed == 1 - client.create_score.assert_called_once() - - -@pytest.mark.asyncio -async def test_v2_trace_scope_rejects_unsupported_field_groups() -> None: +async def test_v2_read_api_rejects_trace_scope() -> None: runner = BatchEvaluationRunner(MagicMock()) with pytest.raises( ValueError, - match="does not support legacy trace field groups: observations, scores", + match="is only supported with scope='observations'", ): await runner.run_async( scope="traces", mapper=lambda *, item: EvaluatorInputs(input=None, output=None), evaluators=[], - fetch_trace_fields="core,scores,observations", observation_read_api="v2", ) From ff416d2c43188698839104e5f6404a844f52704f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 7 Sep 2026 12:52:44 +0000 Subject: [PATCH 7/7] docs(evals): mark batch evaluation as legacy Co-authored-by: Hassieb Pakzad --- langfuse/_client/client.py | 22 +-- langfuse/batch_evaluation.py | 252 ++++------------------------ tests/unit/test_batch_evaluation.py | 201 ---------------------- 3 files changed, 39 insertions(+), 436 deletions(-) delete mode 100644 tests/unit/test_batch_evaluation.py diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 6634f20de..42d861fd4 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -3261,7 +3261,6 @@ def run_batched_evaluation( filter: Optional[str] = None, fetch_batch_size: int = 50, fetch_trace_fields: Optional[str] = None, - observation_read_api: Literal["legacy", "v2"] = "legacy", max_items: Optional[int] = None, max_retries: int = 3, evaluators: List[EvaluatorFunction], @@ -3273,7 +3272,7 @@ def run_batched_evaluation( resume_from: Optional[BatchEvaluationResumeToken] = None, verbose: bool = False, ) -> BatchEvaluationResult: - """Fetch traces or observations and run evaluations on each item. + """Fetch traces or observations using legacy read APIs and evaluate each item. This method provides a powerful way to evaluate existing data in Langfuse at scale. It fetches items based on filters, transforms them using a mapper function, runs @@ -3289,6 +3288,11 @@ def run_batched_evaluation( it memory-efficient for large datasets. It includes comprehensive error handling, retry logic, and resume capability for long-running evaluations. + Legacy platform compatibility: + This method reads traces from `GET /api/public/traces` and observations + from the legacy `GET /api/public/observations` endpoint. It is supported + with Langfuse platform v3 and is not yet supported with platform v4. + Args: scope: The type of items to evaluate. Must be one of: - "traces": Evaluate complete traces with all their observations @@ -3306,18 +3310,7 @@ def run_batched_evaluation( Default: None (fetches all items). fetch_batch_size: Number of items to fetch per API call and hold in memory. Larger values may be faster but use more memory. Default: 50. - fetch_trace_fields: Comma-separated list of fields to include when - fetching traces through the legacy API. Available groups are - 'core' (always included), 'io', 'scores', 'observations', and - 'metrics'; if omitted, all groups are returned. - observation_read_api: Observation Read API used to fetch items. - - "legacy" (default) calls `GET /api/public/traces` for - `scope="traces"` and the legacy - `GET /api/public/observations` endpoint for - `scope="observations"`. Use this with Langfuse platform v3. - - "v2" calls `GET /api/public/v2/observations` and is only - supported with `scope="observations"`. Use this with Langfuse - platform v4 `events_only` deployments. + fetch_trace_fields: Comma-separated list of fields to include when fetching traces. Available field groups: 'core' (always included), 'io' (input, output, metadata), 'scores', 'observations', 'metrics'. If not specified, all fields are returned. Example: 'core,scores,metrics'. Note: Excluded 'observations' or 'scores' fields return empty arrays; excluded 'metrics' returns -1 for 'totalCost' and 'latency'. Only relevant if scope is 'traces'. max_items: Maximum total number of items to process. If None, processes all items matching the filter. Useful for testing or limiting evaluation runs. Default: None (process all). @@ -3487,7 +3480,6 @@ def composite_evaluator(*, item, evaluations): filter=filter, fetch_batch_size=fetch_batch_size, fetch_trace_fields=fetch_trace_fields, - observation_read_api=observation_read_api, max_items=max_items, max_concurrency=max_concurrency, composite_evaluator=composite_evaluator, diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index 58ba2a9ca..723b45757 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -15,7 +15,6 @@ Awaitable, Dict, List, - Literal, Optional, Protocol, Set, @@ -25,11 +24,8 @@ ) from langfuse.api import ( - ObservationLevel, ObservationsView, - ObservationV2, TraceWithFullDetails, - Usage, ) from langfuse.experiment import Evaluation, EvaluatorFunction from langfuse.logger import langfuse_logger as logger @@ -832,8 +828,6 @@ class BatchEvaluationRunner: client: The Langfuse client instance used for API calls and score creation. """ - _LEGACY_NEXT_PAGE = "__langfuse_batch_evaluation_legacy__" - def __init__(self, client: "Langfuse"): """Initialize the batch evaluation runner. @@ -851,7 +845,6 @@ async def run_async( filter: Optional[str] = None, fetch_batch_size: int = 50, fetch_trace_fields: Optional[str] = "io", - observation_read_api: Literal["legacy", "v2"] = "legacy", max_items: Optional[int] = None, max_concurrency: int = 5, composite_evaluator: Optional[CompositeEvaluatorFunction] = None, @@ -862,28 +855,23 @@ async def run_async( verbose: bool = False, resume_from: Optional[BatchEvaluationResumeToken] = None, ) -> BatchEvaluationResult: - """Run batch evaluation asynchronously. + """Run batch evaluation asynchronously using legacy read APIs. This is the main implementation method that orchestrates the entire batch evaluation process: fetching items, mapping, evaluating, creating scores, and tracking statistics. + This runner reads traces from `GET /api/public/traces` and observations + from the legacy `GET /api/public/observations` endpoint. It is supported + with Langfuse platform v3 and is not yet supported with platform v4. + Args: scope: The type of items to evaluate ("traces", "observations"). mapper: Function to transform API response items to evaluator inputs. evaluators: List of evaluation functions to run on each item. filter: JSON filter string for querying items. fetch_batch_size: Number of items to fetch per API call. - fetch_trace_fields: Comma-separated trace field groups. The legacy - API supports 'core', 'io', 'scores', 'observations', and - 'metrics'. Only relevant to the legacy trace scope. - observation_read_api: Observation Read API used to fetch items. - - "legacy" (default) uses `GET /api/public/traces` for trace - scope and `GET /api/public/observations` for observation - scope. This is compatible with Langfuse platform v3. - - "v2" uses `GET /api/public/v2/observations`. It is only - supported with observation scope and is compatible with - Langfuse platform v4 `events_only` deployments. + fetch_trace_fields: Comma-separated list of fields to include when fetching traces. Available field groups: 'core' (always included), 'io' (input, output, metadata), 'scores', 'observations', 'metrics'. If not specified, all fields are returned. Example: 'core,scores,metrics'. Note: Excluded 'observations' or 'scores' fields return empty arrays; excluded 'metrics' returns -1 for 'totalCost' and 'latency'. Only relevant if scope is 'traces'. Default: 'io' max_items: Maximum number of items to process (None = all). max_concurrency: Maximum number of concurrent evaluations. composite_evaluator: Optional function to create composite scores. @@ -899,10 +887,6 @@ async def run_async( Returns: BatchEvaluationResult with comprehensive statistics. """ - self._validate_read_options( - scope=scope, - observation_read_api=observation_read_api, - ) start_time = time.time() # Initialize tracking variables @@ -925,9 +909,7 @@ async def run_async( } # Handle resume token by modifying filter - effective_filter = self._build_timestamp_filter( - filter, resume_from, observation_read_api - ) + effective_filter = self._build_timestamp_filter(filter, resume_from) normalized_additional_trace_tags = ( self._dedupe_tags(_additional_trace_tags) if _additional_trace_tags is not None @@ -940,7 +922,6 @@ async def run_async( # Pagination state page = 1 - cursor: Optional[str] = None has_more = True last_item_timestamp: Optional[str] = None last_item_id: Optional[str] = None @@ -967,15 +948,13 @@ async def run_async( # Fetch next batch with retry logic try: - items, next_cursor = await self._fetch_batch_with_retry( + items = await self._fetch_batch_with_retry( scope=scope, filter=effective_filter, page=page, - cursor=cursor, limit=fetch_batch_size, max_retries=max_retries, fields=fetch_trace_fields, - observation_read_api=observation_read_api, ) except Exception as e: # Failed after max_retries - create resume token and return @@ -1009,15 +988,10 @@ async def run_async( # Check if we got any items if not items: - if next_cursor is None: - has_more = False - if verbose: - logger.info("No more items to fetch") - break - - cursor = next_cursor - page += 1 - continue + has_more = False + if verbose: + logger.info("No more items to fetch") + break total_items_fetched += len(items) @@ -1121,10 +1095,10 @@ async def process_item( ) # Check if we should continue to next page - if next_cursor is None: + if len(items) < fetch_batch_size: + # Last page - no more items available has_more = False else: - cursor = next_cursor page += 1 # Check max_items again before next fetch @@ -1179,71 +1153,27 @@ async def _fetch_batch_with_retry( scope: str, filter: Optional[str], page: int, - cursor: Optional[str], limit: int, max_retries: int, fields: Optional[str], - observation_read_api: Literal["legacy", "v2"], - ) -> Tuple[List[Union[TraceWithFullDetails, ObservationsView]], Optional[str]]: + ) -> List[Union[TraceWithFullDetails, ObservationsView]]: """Fetch a batch of items with retry logic. Args: scope: The type of items ("traces", "observations"). filter: JSON filter string for querying. - page: Page number used by the v3 compatibility fallback. - cursor: Cursor from the previous response. + page: Page number (1-indexed). limit: Number of items per page. max_retries: Maximum number of retry attempts. verbose: Whether to log retry attempts. fields: Trace fields to fetch - observation_read_api: Read API selected by the caller. Returns: - A tuple containing the items and the next-page cursor. + List of items from the API. Raises: Exception: If all retry attempts fail. """ - if scope not in {"traces", "observations"}: - error_message = f"Invalid scope: {scope}" - raise ValueError(error_message) - - if observation_read_api == "legacy": - return self._fetch_legacy_batch( - scope=scope, - filter=filter, - page=page, - limit=limit, - max_retries=max_retries, - fields=fields, - ) - - response = self.client.api.observations.get_many( - fields=self._get_v2_observation_fields(), - cursor=cursor, - limit=limit, - filter=self._build_v2_filter(filter=filter), - request_options={"max_retries": max_retries}, - ) - - items: List[Union[TraceWithFullDetails, ObservationsView]] = [ - self._observation_to_legacy_view(observation) - for observation in response.data - ] - - return items, response.meta.cursor - - def _fetch_legacy_batch( - self, - *, - scope: str, - filter: Optional[str], - page: int, - limit: int, - max_retries: int, - fields: Optional[str], - ) -> Tuple[List[Union[TraceWithFullDetails, ObservationsView]], Optional[str]]: - """Fetch from the legacy trace or observation read API.""" if scope == "traces": response = self.client.api.trace.list( page=page, @@ -1251,132 +1181,19 @@ def _fetch_legacy_batch( filter=filter, request_options={"max_retries": max_retries}, fields=fields, - ) - else: + ) # type: ignore + return list(response.data) # type: ignore + elif scope == "observations": response = self.client.api.legacy.observations_v1.get_many( page=page, limit=limit, filter=filter, request_options={"max_retries": max_retries}, - ) - - items = list(response.data) - next_cursor = self._LEGACY_NEXT_PAGE if len(items) == limit else None - return items, next_cursor - - @staticmethod - def _validate_read_options( - *, - scope: str, - observation_read_api: str, - ) -> None: - """Validate options that differ between observation read APIs.""" - if observation_read_api not in {"legacy", "v2"}: - message = ( - "Invalid observation_read_api: " - f"{observation_read_api}. Expected 'legacy' or 'v2'." - ) - raise ValueError(message) - - if observation_read_api == "v2" and scope != "observations": - message = ( - "observation_read_api='v2' is only supported with " - "scope='observations'. Use observation_read_api='legacy' " - "for scope='traces'." - ) - raise ValueError(message) - - @staticmethod - def _get_v2_observation_fields() -> str: - """Return all v2 observation field groups needed by mappers.""" - fields = { - "basic", - "time", - "io", - "metadata", - "model", - "usage", - "prompt", - "metrics", - "trace_context", - } - return ",".join(sorted(fields)) - - @staticmethod - def _build_v2_filter(*, filter: Optional[str]) -> Optional[str]: - """Adapt legacy observation filter columns to v2 names.""" - try: - filters = json.loads(filter) if filter else [] - except json.JSONDecodeError: - return filter - - if not isinstance(filters, list): - return filter - - column_aliases = { - "timestamp": "startTime", - "start_time": "startTime", - "user_id": "userId", - "session_id": "sessionId", - } - for condition in filters: - if ( - isinstance(condition, dict) - and condition.get("column") in column_aliases - ): - condition["column"] = column_aliases[condition["column"]] - - return json.dumps(filters) - - @classmethod - def _observation_to_legacy_view( - cls, observation: ObservationV2 - ) -> ObservationsView: - """Adapt a v2 observation to the established observation mapper contract.""" - usage_details = observation.usage_details or {} - return ObservationsView.model_construct( - id=observation.id, - trace_id=observation.trace_id, - type=observation.type, - name=observation.name, - start_time=observation.start_time, - end_time=observation.end_time, - completion_start_time=observation.completion_start_time, - model=observation.model, - model_parameters=observation.model_parameters or {}, - input=cls._parse_io_value(observation.input), - version=observation.version, - metadata=observation.metadata, - output=cls._parse_io_value(observation.output), - usage=Usage( - input=usage_details.get("input", 0), - output=usage_details.get("output", 0), - total=usage_details.get("total", 0), - ), - level=observation.level or ObservationLevel.DEFAULT, - status_message=observation.status_message, - parent_observation_id=observation.parent_observation_id, - prompt_id=observation.prompt_id, - usage_details=usage_details, - cost_details=observation.cost_details or {}, - environment=observation.environment or "default", - prompt_name=observation.prompt_name, - prompt_version=observation.prompt_version, - model_id=observation.model_id, - latency=observation.latency, - time_to_first_token=observation.time_to_first_token, - ) - - @staticmethod - def _parse_io_value(value: Any) -> Any: - """Restore the parsed JSON behavior of the legacy read endpoints.""" - if not isinstance(value, str): - return value - - try: - return json.loads(value) - except json.JSONDecodeError: - return value + ) # type: ignore + return list(response.data) # type: ignore + else: + error_message = f"Invalid scope: {scope}" + raise ValueError(error_message) async def _process_batch_evaluation_item( self, @@ -1671,14 +1488,12 @@ def _build_timestamp_filter( self, original_filter: Optional[str], resume_from: Optional[BatchEvaluationResumeToken], - observation_read_api: Literal["legacy", "v2"], ) -> Optional[str]: """Build filter with timestamp constraint for resume capability. Args: original_filter: The original JSON filter string. resume_from: Optional resume token with timestamp information. - observation_read_api: Read API selected by the caller. Returns: Modified filter string with timestamp constraint, or original filter. @@ -1701,9 +1516,7 @@ def _build_timestamp_filter( filter_list = [] # Add timestamp constraint to filter array - timestamp_field = self._get_timestamp_field_for_scope( - resume_from.scope, observation_read_api - ) + timestamp_field = self._get_timestamp_field_for_scope(resume_from.scope) timestamp_filter = { "type": "datetime", "column": timestamp_field, @@ -1755,21 +1568,20 @@ def _get_item_timestamp( return "" @staticmethod - def _get_timestamp_field_for_scope( - scope: str, observation_read_api: Literal["legacy", "v2"] - ) -> str: + def _get_timestamp_field_for_scope(scope: str) -> str: """Get the timestamp field name for filtering based on scope. Args: scope: The type of items. - observation_read_api: Read API selected by the caller. Returns: The field name to use in filters. """ - if observation_read_api == "v2": - return "startTime" - return "timestamp" if scope == "traces" else "start_time" + if scope == "traces": + return "timestamp" + elif scope == "observations": + return "start_time" + return "timestamp" # Default @staticmethod def _dedupe_tags(tags: Optional[List[str]]) -> List[str]: diff --git a/tests/unit/test_batch_evaluation.py b/tests/unit/test_batch_evaluation.py deleted file mode 100644 index 1afabb0f7..000000000 --- a/tests/unit/test_batch_evaluation.py +++ /dev/null @@ -1,201 +0,0 @@ -from datetime import datetime, timezone -from types import SimpleNamespace -from unittest.mock import MagicMock - -import pytest - -from langfuse.api import ( - ObservationsView, - ObservationV2, -) -from langfuse.batch_evaluation import ( - BatchEvaluationRunner, - EvaluatorInputs, -) -from langfuse.experiment import Evaluation - - -def _observation( - *, - observation_id: str = "observation-id", - trace_id: str = "trace-id", -) -> ObservationV2: - return ObservationV2( - id=observation_id, - trace_id=trace_id, - start_time=datetime(2026, 1, 2, tzinfo=timezone.utc), - project_id="project-id", - parent_observation_id=None, - type="SPAN", - name="root-span", - trace_name="trace-name", - input='{"question": "hello"}', - output='"answer"', - metadata={"source": "test"}, - tags=["production"], - environment="production", - ) - - -@pytest.mark.asyncio -async def test_fetches_observations_via_v2_api() -> None: - client = MagicMock() - client.api.observations.get_many.return_value = SimpleNamespace( - data=[_observation()], - meta=SimpleNamespace(cursor=None), - ) - runner = BatchEvaluationRunner(client) - - items, cursor = await runner._fetch_batch_with_retry( - scope="observations", - filter=None, - page=1, - cursor="current-cursor", - limit=25, - max_retries=3, - fields=None, - observation_read_api="v2", - ) - - assert cursor is None - assert len(items) == 1 - observation = items[0] - assert isinstance(observation, ObservationsView) - assert observation.id == "observation-id" - assert observation.trace_id == "trace-id" - assert observation.input == {"question": "hello"} - assert observation.output == "answer" - - kwargs = client.api.observations.get_many.call_args.kwargs - assert kwargs["cursor"] == "current-cursor" - assert kwargs["filter"] == "[]" - assert "io" in kwargs["fields"].split(",") - - -def test_resume_filter_uses_read_api_timestamp_column() -> None: - assert ( - BatchEvaluationRunner._get_timestamp_field_for_scope("observations", "v2") - == "startTime" - ) - assert ( - BatchEvaluationRunner._get_timestamp_field_for_scope("traces", "legacy") - == "timestamp" - ) - assert ( - BatchEvaluationRunner._get_timestamp_field_for_scope("observations", "legacy") - == "start_time" - ) - - -@pytest.mark.asyncio -async def test_legacy_read_api_uses_page_pagination() -> None: - client = MagicMock() - legacy_observation = MagicMock(spec=ObservationsView) - client.api.legacy.observations_v1.get_many.return_value = SimpleNamespace( - data=[legacy_observation] - ) - runner = BatchEvaluationRunner(client) - - items, cursor = await runner._fetch_batch_with_retry( - scope="observations", - filter='[{"type":"datetime","column":"start_time","operator":">","value":"2026-01-01"}]', - page=1, - cursor=None, - limit=1, - max_retries=3, - fields=None, - observation_read_api="legacy", - ) - - assert items == [legacy_observation] - assert cursor == runner._LEGACY_NEXT_PAGE - client.api.observations.get_many.assert_not_called() - - await runner._fetch_batch_with_retry( - scope="observations", - filter=None, - page=2, - cursor=cursor, - limit=1, - max_retries=3, - fields=None, - observation_read_api="legacy", - ) - - client.api.observations.get_many.assert_not_called() - assert client.api.legacy.observations_v1.get_many.call_args.kwargs["page"] == 2 - - -@pytest.mark.asyncio -async def test_run_defaults_to_legacy_read_api() -> None: - client = MagicMock() - runner = BatchEvaluationRunner(client) - legacy_observation = runner._observation_to_legacy_view(_observation()) - client.api.legacy.observations_v1.get_many.return_value = SimpleNamespace( - data=[legacy_observation] - ) - - result = await runner.run_async( - scope="observations", - mapper=lambda *, item: EvaluatorInputs( - input=item.input, - output=item.output, - ), - evaluators=[], - ) - - assert result.total_items_processed == 1 - client.api.observations.get_many.assert_not_called() - client.api.legacy.observations_v1.get_many.assert_called_once() - - -@pytest.mark.asyncio -async def test_run_uses_v2_cursor_for_next_batch() -> None: - client = MagicMock() - client.api.observations.get_many.side_effect = [ - SimpleNamespace( - data=[_observation(observation_id="first")], - meta=SimpleNamespace(cursor="next-cursor"), - ), - SimpleNamespace( - data=[_observation(observation_id="second")], - meta=SimpleNamespace(cursor=None), - ), - ] - runner = BatchEvaluationRunner(client) - - result = await runner.run_async( - scope="observations", - mapper=lambda *, item: EvaluatorInputs( - input=item.input, - output=item.output, - ), - evaluators=[ - lambda **kwargs: Evaluation(name="quality", value=1.0), - ], - fetch_batch_size=1, - observation_read_api="v2", - ) - - assert result.total_items_processed == 2 - assert result.completed is True - assert [ - call.kwargs["cursor"] - for call in client.api.observations.get_many.call_args_list - ] == [None, "next-cursor"] - - -@pytest.mark.asyncio -async def test_v2_read_api_rejects_trace_scope() -> None: - runner = BatchEvaluationRunner(MagicMock()) - - with pytest.raises( - ValueError, - match="is only supported with scope='observations'", - ): - await runner.run_async( - scope="traces", - mapper=lambda *, item: EvaluatorInputs(input=None, output=None), - evaluators=[], - observation_read_api="v2", - )