From e946c7185a5a35a46f5adeddf49a4f2de2d0b9ec Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Thu, 30 Jul 2026 11:45:25 +0200 Subject: [PATCH 1/3] fix(ci): make test suite compatible with Langfuse v4 --- langfuse/_client/attributes.py | 1 + langfuse/_client/client.py | 28 +- langfuse/_client/propagation.py | 8 + langfuse/_client/span_processor.py | 1 + langfuse/batch_evaluation.py | 133 ++++- tests/e2e/test_batch_evaluation.py | 8 +- tests/e2e/test_core_sdk.py | 37 +- tests/e2e/test_decorators.py | 2 +- tests/support/api_wrapper.py | 146 ++++-- tests/support/utils.py | 4 +- tests/support/v4_api.py | 494 +++++++++++++++++++ tests/unit/test_additional_headers_simple.py | 2 + tests/unit/test_e2e_support.py | 42 +- tests/unit/test_propagate_attributes.py | 48 +- 14 files changed, 780 insertions(+), 174 deletions(-) create mode 100644 tests/support/v4_api.py diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py index 4660b50f0..43a85c2fd 100644 --- a/langfuse/_client/attributes.py +++ b/langfuse/_client/attributes.py @@ -68,6 +68,7 @@ class LangfuseOtelSpanAttributes: EXPERIMENT_METADATA = "langfuse.experiment.metadata" EXPERIMENT_DATASET_ID = "langfuse.experiment.dataset.id" EXPERIMENT_ITEM_ID = "langfuse.experiment.item.id" + EXPERIMENT_ITEM_VERSION = "langfuse.experiment.item.version" EXPERIMENT_ITEM_EXPECTED_OUTPUT = "langfuse.experiment.item.expected_output" EXPERIMENT_ITEM_METADATA = "langfuse.experiment.item.metadata" EXPERIMENT_ITEM_ROOT_OBSERVATION_ID = "langfuse.experiment.item.root_observation_id" diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index d97708efd..1f0ec926b 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -2997,37 +2997,23 @@ async def _process_experiment_item( # Link dataset runs to the canonical task observation so their # latency excludes the subsequent evaluator subtree. if hasattr(item, "id") and hasattr(item, "dataset_id"): - try: - # Use sync API to avoid event loop issues when - # run_async_safely creates multiple event loops across - # different threads. - dataset_run_item = await asyncio.to_thread( - self.api.dataset_run_items.create, - run_name=experiment_run_name, - run_description=experiment_description, - metadata=experiment_metadata, - dataset_item_id=item.id, # type: ignore - trace_id=trace_id, - observation_id=task_span.id, - dataset_version=dataset_version, - ) - - dataset_run_id = dataset_run_item.dataset_run_id - - except Exception as e: - langfuse_logger.error( - f"Failed to create dataset run item: {e}" - ) + # Langfuse v4 synthesizes the experiment and item links from + # the OpenTelemetry attributes propagated below. + dataset_run_id = fallback_experiment_id experiment_id = dataset_run_id or fallback_experiment_id propagated_experiment_attributes = PropagatedExperimentAttributes( experiment_id=experiment_id, experiment_name=experiment_run_name, + experiment_description=experiment_description, experiment_metadata=_flatten_and_serialize_metadata_values( experiment_metadata ), experiment_dataset_id=dataset_id, experiment_item_id=experiment_item_id, + experiment_item_version=( + dataset_version.isoformat() if dataset_version else None + ), experiment_item_metadata=_flatten_and_serialize_metadata_values( item_metadata if isinstance(item_metadata, dict) else None ), diff --git a/langfuse/_client/propagation.py b/langfuse/_client/propagation.py index d6e26163a..6afbec2ec 100644 --- a/langfuse/_client/propagation.py +++ b/langfuse/_client/propagation.py @@ -58,9 +58,11 @@ InternalPropagatedKeys = Literal[ "experiment_id", "experiment_name", + "experiment_description", "experiment_metadata", "experiment_dataset_id", "experiment_item_id", + "experiment_item_version", "experiment_item_metadata", "experiment_item_root_observation_id", ] @@ -77,9 +79,11 @@ "prompt_version", "experiment_id", "experiment_name", + "experiment_description", "experiment_metadata", "experiment_dataset_id", "experiment_item_id", + "experiment_item_version", "experiment_item_metadata", "experiment_item_root_observation_id", ] @@ -88,9 +92,11 @@ class PropagatedExperimentAttributes(TypedDict): experiment_id: str experiment_name: str + experiment_description: Optional[str] experiment_metadata: Optional[Dict[str, str]] experiment_dataset_id: Optional[str] experiment_item_id: str + experiment_item_version: Optional[str] experiment_item_metadata: Optional[Dict[str, str]] experiment_item_root_observation_id: str @@ -758,9 +764,11 @@ def _get_propagated_span_key(key: str) -> str: "prompt_version": LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_VERSION, "experiment_id": LangfuseOtelSpanAttributes.EXPERIMENT_ID, "experiment_name": LangfuseOtelSpanAttributes.EXPERIMENT_NAME, + "experiment_description": LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION, "experiment_metadata": LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, "experiment_dataset_id": LangfuseOtelSpanAttributes.EXPERIMENT_DATASET_ID, "experiment_item_id": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ID, + "experiment_item_version": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_VERSION, "experiment_item_metadata": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA, "experiment_item_root_observation_id": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ROOT_OBSERVATION_ID, }.get(key) or f"{LangfuseOtelSpanAttributes.TRACE_METADATA}.{key}" diff --git a/langfuse/_client/span_processor.py b/langfuse/_client/span_processor.py index f01b081c2..de2a961c8 100644 --- a/langfuse/_client/span_processor.py +++ b/langfuse/_client/span_processor.py @@ -109,6 +109,7 @@ def __init__( "x-langfuse-sdk-name": "python", "x-langfuse-sdk-version": langfuse_version, "x-langfuse-public-key": public_key, + "x-langfuse-ingestion-version": "4", } # Merge additional headers if provided diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index d28fd085d..ad3080d30 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -9,6 +9,7 @@ import asyncio import json import time +from dataclasses import dataclass from typing import ( TYPE_CHECKING, Any, @@ -34,6 +35,41 @@ from langfuse._client.client import Langfuse +def _parse_observation_value(value: Any) -> Any: + if not isinstance(value, str): + return value + + try: + return json.loads(value) + except (json.JSONDecodeError, TypeError): + return value + + +@dataclass +class _BatchEvaluationObservation: + id: str + trace_id: Optional[str] + start_time: Any + input: Any + output: Any + metadata: Any + name: Optional[str] + user_id: Optional[str] + tags: List[str] + + +@dataclass +class _BatchEvaluationTrace: + id: str + timestamp: Any + input: Any + output: Any + metadata: Any + name: Optional[str] + user_id: Optional[str] + tags: List[str] + + class EvaluatorInputs: """Input data structure for evaluators, returned by mapper functions. @@ -918,6 +954,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 @@ -943,10 +980,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, @@ -1085,10 +1122,11 @@ async def process_item( ) # Check if we should continue to next page - if len(items) < fetch_batch_size: + if next_cursor is None: # Last page - no more items available has_more = False else: + cursor = next_cursor page += 1 # Check max_items again before next fetch @@ -1141,49 +1179,94 @@ 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[Any], 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 returned by the previous Observations API page. 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. + The items and cursor for the next page. 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) + effective_filter = filter + if scope == "traces": + filters = json.loads(filter) if filter else [] + filters.append( + { + "type": "boolean", + "column": "isRootObservation", + "operator": "=", + "value": True, + } + ) + effective_filter = json.dumps(filters) + + response = self.client.api.observations.get_many( + fields="core,basic,io,metadata,trace_context", + limit=limit, + cursor=cursor, + filter=effective_filter, + request_options={"max_retries": max_retries}, + ) + + observations = [ + _BatchEvaluationObservation( + id=observation.id, + trace_id=observation.trace_id, + start_time=observation.start_time, + input=_parse_observation_value(observation.input), + output=_parse_observation_value(observation.output), + metadata=_parse_observation_value(observation.metadata), + name=observation.name, + user_id=observation.user_id, + tags=observation.tags or [], + ) + for observation in response.data + ] + if scope == "observations": + return observations, response.meta.cursor # type: ignore[return-value] + + return ( + [ + _BatchEvaluationTrace( + id=observation.trace_id or observation.id, + timestamp=observation.start_time, + input=observation.input, + output=observation.output, + metadata=observation.metadata, + name=next( + ( + item.trace_name + for item in response.data + if item.id == observation.id and item.trace_name + ), + observation.name, + ), + user_id=observation.user_id, + tags=observation.tags, + ) + for observation in observations + ], + response.meta.cursor, + ) # type: ignore[return-value] + async def _process_batch_evaluation_item( self, item: Union[TraceWithFullDetails, ObservationsView], diff --git a/tests/e2e/test_batch_evaluation.py b/tests/e2e/test_batch_evaluation.py index 0632b21b8..299e82480 100644 --- a/tests/e2e/test_batch_evaluation.py +++ b/tests/e2e/test_batch_evaluation.py @@ -52,7 +52,7 @@ def _seed_trace_corpus( trace_names.append(trace_name) with langfuse_client.start_as_current_observation(name=trace_name) as span: with propagate_attributes(tags=[corpus_tag]): - span.set_trace_io( + span.update( input=f"Seed input {index}", output=f"Seed output {index}", ) @@ -161,7 +161,7 @@ def test_batch_evaluation_with_filter(langfuse_client): name=f"filtered-trace-{create_uuid()}" ) as span: with propagate_attributes(tags=[unique_tag]): - span.set_trace_io( + span.update( input="Filtered test", output="Filtered output", ) @@ -789,7 +789,7 @@ def test_pagination_with_max_items(langfuse_client): name=f"pagination-test-{create_uuid()}" ) as span: with propagate_attributes(tags=["pagination_test"]): - span.set_trace_io( + span.update( input=f"Input {i}", output=f"Output {i}", ) @@ -821,7 +821,7 @@ def test_has_more_items_flag(langfuse_client): name=f"more-items-test-{i}" ) as span: with propagate_attributes(tags=[batch_tag]): - span.set_trace_io( + span.update( input=f"Input {i}", output=f"Output {i}", ) diff --git a/tests/e2e/test_core_sdk.py b/tests/e2e/test_core_sdk.py index 614d6da41..5bcfea0c0 100644 --- a/tests/e2e/test_core_sdk.py +++ b/tests/e2e/test_core_sdk.py @@ -270,8 +270,7 @@ def test_create_boolean_score(): assert created_score is not None, "Score not found in trace" assert created_score["id"] == score_id assert created_score["dataType"] == "BOOLEAN" - assert created_score["value"] == 1 - assert created_score["stringValue"] == "True" + assert created_score["value"] is True def test_create_categorical_score(): @@ -328,8 +327,7 @@ def test_create_categorical_score(): assert created_score is not None, "Score not found in trace" assert created_score["id"] == score_id assert created_score["dataType"] == "CATEGORICAL" - assert created_score["value"] == 0 - assert created_score["stringValue"] == "high score" + assert created_score["value"] == "high score" def test_create_text_score(): @@ -386,9 +384,8 @@ def test_create_text_score(): assert created_score is not None, "Score not found in trace" assert created_score["id"] == score_id assert created_score["dataType"] == "TEXT" - assert ( - created_score["stringValue"] + created_score["value"] == "This is a detailed text evaluation of the output quality." ) @@ -494,7 +491,6 @@ def test_create_trace(): assert trace["metadata"]["key"] == "value" assert trace["tags"] == ["tag1", "tag2"] assert trace["public"] is True - assert True if not trace["externalId"] else False def test_create_update_trace(): @@ -556,7 +552,7 @@ def test_create_update_current_trace(): user_id="test", metadata={"key": "value"}, ): - langfuse.set_current_trace_io(input="test_input") + langfuse.update_current_span(input="test_input") langfuse.set_current_trace_as_public() # Get trace ID for later reference trace_id = span.trace_id @@ -660,7 +656,7 @@ def test_create_generation(): ( { "input": 51, - "output": 0, + "output": 49, "total": 100, }, "TOKENS", @@ -671,7 +667,7 @@ def test_create_generation(): ( { "input": 51, - "output": 0, + "output": 49, "total": 100, }, "CHARACTERS", @@ -701,6 +697,7 @@ def test_create_generation_complex( }, ], output=[{"foo": "bar"}], + model="gpt-3.5-turbo", usage_details=usage, metadata={"tags": ["yo"]}, ).end() @@ -736,7 +733,7 @@ def test_create_generation_complex( assert generation_api.metadata["tags"] == ["yo"] assert generation_api.start_time is not None - assert generation_api.usage_details == {"input": 51, "output": 0, "total": 100} + assert generation_api.usage_details == {"input": 51, "output": 49, "total": 100} def test_create_span(): @@ -989,7 +986,7 @@ def test_create_trace_and_generation(): # Create parent span and set trace properties with langfuse.start_as_current_observation(name=trace_name) as parent_span: with propagate_attributes(trace_name=trace_name, session_id="test-session-id"): - parent_span.set_trace_io(input={"key": "value"}) + parent_span.update(input={"key": "value"}) # Create a generation as child generation = parent_span.start_observation( @@ -1394,6 +1391,7 @@ def test_end_generation_with_openai_token_format(): generation = langfuse.start_observation( as_type="generation", name="query-generation", + model="gpt-3.5-turbo", ) # Get trace ID for verification @@ -1434,7 +1432,6 @@ def test_end_generation_with_openai_token_format(): assert generation_api.usage.input == 100 # prompt_tokens mapped to input assert generation_api.usage.output == 200 # completion_tokens mapped to output assert generation_api.usage.total == 500 - assert generation_api.usage.unit == "TOKENS" # Default unit for OpenAI format assert generation_api.calculated_input_cost == 111 assert generation_api.calculated_output_cost == 222 assert generation_api.calculated_total_cost == 444 @@ -1812,7 +1809,7 @@ def test_fetch_traces(): # First trace with langfuse.start_as_current_observation(name="test1") as span: with propagate_attributes(trace_name=name, session_id="session-1"): - span.set_trace_io(input={"key": "value"}, output="output-value") + span.update(input={"key": "value"}, output="output-value") trace_ids.append(span.trace_id) sleep(1) # Ensure traces have different timestamps @@ -1820,7 +1817,7 @@ def test_fetch_traces(): # Second trace with langfuse.start_as_current_observation(name="test2") as span: with propagate_attributes(trace_name=name, session_id="session-1"): - span.set_trace_io(input={"key": "value"}, output="output-value") + span.update(input={"key": "value"}, output="output-value") trace_ids.append(span.trace_id) sleep(1) # Ensure traces have different timestamps @@ -1828,7 +1825,7 @@ def test_fetch_traces(): # Third trace with langfuse.start_as_current_observation(name="test3") as span: with propagate_attributes(trace_name=name, session_id="session-1"): - span.set_trace_io(input={"key": "value"}, output="output-value") + span.update(input={"key": "value"}, output="output-value") trace_ids.append(span.trace_id) # Ensure data is sent @@ -2088,11 +2085,11 @@ def mask_func(data): # Create a root span with trace properties with langfuse.start_as_current_observation(name="test-span") as root_span: with propagate_attributes(trace_name="test_trace"): - root_span.set_trace_io(input={"sensitive": "data"}) + root_span.update(input={"sensitive": "data"}) # Get trace ID for later use trace_id = root_span.trace_id # Add output to the trace - root_span.set_trace_io(output={"more": "sensitive"}) + root_span.update(output={"more": "sensitive"}) # Create a generation as child gen = root_span.start_observation( @@ -2136,11 +2133,11 @@ def mask_func(data): # Create a root span with trace properties with langfuse.start_as_current_observation(name="test-span") as root_span: with propagate_attributes(trace_name="test_trace"): - root_span.set_trace_io(input={"should_raise": "data"}) + root_span.update(input={"should_raise": "data"}) # Get trace ID for later use trace_id = root_span.trace_id # Add output to the trace - root_span.set_trace_io(output={"should_raise": "sensitive"}) + root_span.update(output={"should_raise": "sensitive"}) # Ensure data is sent langfuse.flush() diff --git a/tests/e2e/test_decorators.py b/tests/e2e/test_decorators.py index 9e302f799..da8048d2d 100644 --- a/tests/e2e/test_decorators.py +++ b/tests/e2e/test_decorators.py @@ -1075,7 +1075,7 @@ def test_media(): @observe() def main(): sleep(1) - langfuse.set_current_trace_io( + langfuse.update_current_span( input={ "context": { "nested": media, diff --git a/tests/support/api_wrapper.py b/tests/support/api_wrapper.py index c4519252f..53549a999 100644 --- a/tests/support/api_wrapper.py +++ b/tests/support/api_wrapper.py @@ -1,47 +1,87 @@ -import os - -import httpx - -from langfuse.api.commons.errors.not_found_error import NotFoundError from tests.support.retry import ( DEFAULT_RETRY_INTERVAL_SECONDS, DEFAULT_RETRY_TIMEOUT_SECONDS, - is_not_found_payload, - retry_until_ready, ) +from tests.support.utils import get_api, wait_for_result +from tests.support.v4_api import _score_view class LangfuseAPI: def __init__(self, username=None, password=None, base_url=None): - username = username if username else os.environ["LANGFUSE_PUBLIC_KEY"] - password = password if password else os.environ["LANGFUSE_SECRET_KEY"] - self.auth = (username, password) - self.BASE_URL = base_url if base_url else os.environ["LANGFUSE_BASE_URL"] + self._api = get_api(retry=False) - def _get_json( - self, - url, - params=None, - *, - retry=True, - is_result_ready=None, - timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, - interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, - ): - def _request(): - response = httpx.get(url, params=params, auth=self.auth) - payload = response.json() + @staticmethod + def _score_json(score): + return { + "id": score.id, + "name": score.name, + "value": score.value, + "timestamp": score.timestamp.isoformat(), + "dataType": score.data_type, + "stringValue": score.string_value, + "traceId": score.trace_id, + "observationId": score.observation_id, + "sessionId": score.session_id, + "comment": getattr(score, "comment", None), + "metadata": getattr(score, "metadata", None), + } - if response.status_code == 404 and is_not_found_payload(payload): - raise NotFoundError(body=payload, headers=dict(response.headers)) + @classmethod + def _observation_json(cls, observation): + return { + "id": observation.id, + "traceId": observation.trace_id, + "type": observation.type, + "name": observation.name, + "startTime": observation.start_time.isoformat(), + "endTime": ( + observation.end_time.isoformat() if observation.end_time else None + ), + "input": observation.input, + "output": observation.output, + "metadata": observation.metadata, + "model": observation.model, + "usageDetails": observation.usage_details, + "costDetails": observation.cost_details, + "promptId": observation.prompt_id, + } - return payload + @classmethod + def _trace_json(cls, trace): + return { + "id": trace.id, + "timestamp": trace.timestamp.isoformat(), + "name": trace.name, + "input": trace.input, + "output": trace.output, + "sessionId": trace.session_id, + "release": trace.release, + "version": trace.version, + "userId": trace.user_id, + "metadata": trace.metadata, + "tags": trace.tags, + "public": trace.public, + "environment": trace.environment, + "observations": [ + cls._observation_json(observation) for observation in trace.observations + ], + "scores": [cls._score_json(score) for score in trace.scores], + } + @staticmethod + def _read( + operation, + *, + retry, + is_result_ready, + timeout_seconds, + interval_seconds, + ): if not retry: - return _request() + return operation() - return retry_until_ready( - _request, + return wait_for_result( + operation, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, interval_seconds=interval_seconds, @@ -56,9 +96,8 @@ def get_observation( timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, ): - url = f"{self.BASE_URL}/api/public/observations/{observation_id}" - return self._get_json( - url, + return self._read( + lambda: self._observation_json(self._api.observations.get(observation_id)), retry=retry, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, @@ -77,11 +116,21 @@ def get_scores( timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, ): - params = {"page": page, "limit": limit, "userId": user_id, "name": name} - url = f"{self.BASE_URL}/api/public/scores" - return self._get_json( - url, - params=params, + def operation(): + response = self._api._client.scores_v3.get_many_v3( + name=name, + user_id=user_id, + limit=limit, + fields="details,subject,annotation", + ) + data = [self._score_json(_score_view(score)) for score in response.data] + return { + "data": data, + "meta": {"page": page or 1, "limit": limit or 50}, + } + + return self._read( + operation, retry=retry, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, @@ -100,11 +149,17 @@ def get_traces( timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, ): - params = {"page": page, "limit": limit, "userId": user_id, "name": name} - url = f"{self.BASE_URL}/api/public/traces" - return self._get_json( - url, - params=params, + def operation(): + response = self._api.trace.list( + page=page, limit=limit, user_id=user_id, name=name + ) + return { + "data": [self._trace_json(trace) for trace in response.data], + "meta": vars(response.meta), + } + + return self._read( + operation, retry=retry, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, @@ -120,9 +175,8 @@ def get_trace( timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, ): - url = f"{self.BASE_URL}/api/public/traces/{trace_id}" - return self._get_json( - url, + return self._read( + lambda: self._trace_json(self._api.trace.get(trace_id)), retry=retry, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, diff --git a/tests/support/utils.py b/tests/support/utils.py index a29274d3a..e22654d13 100644 --- a/tests/support/utils.py +++ b/tests/support/utils.py @@ -9,6 +9,7 @@ DEFAULT_RETRY_TIMEOUT_SECONDS, retry_until_ready, ) +from tests.support.v4_api import V4TestAPI READ_METHOD_NAMES = {"get", "get_by_id", "get_many", "get_run", "list"} PAGINATION_ARGUMENTS = {"limit", "page"} @@ -71,7 +72,8 @@ def get_api(*, retry: bool = True): password=os.environ.get("LANGFUSE_SECRET_KEY"), base_url=os.environ.get("LANGFUSE_BASE_URL"), ) - return _RetryingApiProxy(client) if retry else client + test_api = V4TestAPI(client) + return _RetryingApiProxy(test_api) if retry else test_api def wait_for_result( diff --git a/tests/support/v4_api.py b/tests/support/v4_api.py new file mode 100644 index 000000000..2b6a667e8 --- /dev/null +++ b/tests/support/v4_api.py @@ -0,0 +1,494 @@ +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from typing import Any, Iterable + +from langfuse.api.commons.errors.not_found_error import NotFoundError + +OBSERVATION_FIELDS = ( + "core,basic,time,io,metadata,model,usage,prompt,metrics,trace_context" +) +DEFAULT_LOOKBACK = timedelta(days=1) +DEFAULT_LOOKAHEAD = timedelta(minutes=5) + + +def _time_bounds() -> tuple[datetime, datetime]: + now = datetime.now(timezone.utc) + return now - DEFAULT_LOOKBACK, now + DEFAULT_LOOKAHEAD + + +def _parse_json(value: Any) -> Any: + if not isinstance(value, str): + return value + + try: + return json.loads(value) + except (json.JSONDecodeError, TypeError): + return value + + +def _as_float(value: Any) -> float | None: + if value is None: + return None + + return float(value) + + +def _none_if_empty(value: Any) -> Any: + return None if value == "" else value + + +def _score_view(score: Any) -> SimpleNamespace: + subject = getattr(score, "subject", None) + kind = getattr(subject, "kind", None) + value = score.value + score_type = type(score).__name__.split("_")[-1].removesuffix("ScoreV3").upper() + string_value = value if score_type in {"CATEGORICAL", "TEXT"} else None + trace_id = None + if subject is not None: + trace_id = getattr(subject, "trace_id", None) + if trace_id is None and kind == "trace": + trace_id = subject.id + + data = score.model_dump() + data.update( + data_type=score_type, + string_value=string_value, + trace_id=trace_id, + observation_id=( + subject.id if subject is not None and kind == "observation" else None + ), + session_id=subject.id if subject is not None and kind == "session" else None, + ) + return SimpleNamespace(**data) + + +class V4ObservationView: + """Test-friendly view of one row returned by Observations API v2.""" + + def __init__(self, observation: Any): + self._observation = observation + self.id = observation.id + self.trace_id = observation.trace_id + self.type = observation.type + self.name = observation.name + self.start_time = observation.start_time + self.end_time = observation.end_time + self.completion_start_time = observation.completion_start_time + self.model = ( + getattr(observation, "model", None) or observation.provided_model_name + ) + self.model_parameters = _parse_json(observation.model_parameters) + self.input = _parse_json(observation.input) + self.output = _parse_json(observation.output) + self.version = observation.version + self.metadata = _parse_json(observation.metadata) or {} + self.level = observation.level + self.status_message = observation.status_message + self.parent_observation_id = observation.parent_observation_id + self.is_root_observation = observation.is_root_observation + self.prompt_id = _none_if_empty(observation.prompt_id) + self.prompt_name = _none_if_empty(observation.prompt_name) + self.prompt_version = observation.prompt_version + self.usage_details = observation.usage_details or {} + self.cost_details = observation.cost_details or {} + self.environment = observation.environment + self.latency = observation.latency + self.time_to_first_token = observation.time_to_first_token + self.model_id = observation.model_id + self.input_price = _as_float(observation.input_price) + self.output_price = _as_float(observation.output_price) + self.total_price = _as_float(observation.total_price) + self.calculated_input_cost = self.cost_details.get("input") + self.calculated_output_cost = self.cost_details.get("output") + self.calculated_total_cost = observation.total_cost or self.cost_details.get( + "total" + ) + self.usage = SimpleNamespace( + input=self.usage_details.get("input", 0), + output=self.usage_details.get("output", 0), + total=self.usage_details.get("total", 0), + unit=self.usage_details.get("unit"), + input_cost=self.calculated_input_cost, + output_cost=self.calculated_output_cost, + total_cost=self.calculated_total_cost, + ) + + +class V4TraceView: + """Observation-backed trace view used by SDK behavior tests.""" + + def __init__( + self, trace_id: str, observations: list[V4ObservationView], scores: list + ): + if not observations: + raise NotFoundError(body={"error": "LangfuseNotFoundError"}) + + observations = list( + {observation.id: observation for observation in observations}.values() + ) + observations.sort(key=lambda observation: observation.start_time) + root = next( + ( + observation + for observation in observations + if observation.is_root_observation is True + ), + None, + ) + if root is None: + root = next( + ( + observation + for observation in observations + if observation.parent_observation_id is None + ), + observations[0], + ) + + source = root._observation + + def first_populated(attribute: str, fallback: Any = None) -> Any: + return next( + ( + value + for observation in observations + if (value := getattr(observation._observation, attribute, None)) + ), + fallback, + ) + + self.id = trace_id + self.timestamp = root.start_time + self.name = first_populated("trace_name", root.name) + self.input = root.input + self.output = root.output + self.session_id = first_populated("session_id") + self.release = first_populated("release") + self.version = root.version + self.user_id = first_populated("user_id") + self.metadata = max( + (observation.metadata for observation in observations), + key=len, + default={}, + ) + self.tags = list( + dict.fromkeys( + tag + for observation in observations + for tag in observation._observation.tags + ) + ) + self.public = bool(source.public) + self.environment = source.environment + self.observations = observations + self.scores = scores + + +class _V4ObservationsClient: + def __init__(self, client: Any): + self._client = client + + def get(self, observation_id: str) -> V4ObservationView: + response = self.get_many( + filter=json.dumps( + [ + { + "type": "string", + "column": "id", + "operator": "=", + "value": observation_id, + } + ] + ), + limit=1, + ) + if not response.data: + raise NotFoundError(body={"error": "LangfuseNotFoundError"}) + + return response.data[0] + + def get_many( + self, + *, + fields: str | None = None, + limit: int | None = None, + cursor: str | None = None, + page: int | None = None, + name: str | None = None, + user_id: str | None = None, + type: str | None = None, + trace_id: str | None = None, + parent_observation_id: str | None = None, + environment: str | Iterable[str] | None = None, + version: str | None = None, + filter: str | None = None, + **_: Any, + ) -> SimpleNamespace: + from_start_time, to_start_time = _time_bounds() + requested_page = page or 1 + response = None + current_cursor = cursor + fetched_page = 0 + for fetched_page in range(1, requested_page + 1): + response = self._client.observations.get_many( + fields=fields or OBSERVATION_FIELDS, + limit=limit, + cursor=current_cursor, + name=name, + user_id=user_id, + type=type, + trace_id=trace_id, + parent_observation_id=parent_observation_id, + environment=environment, + version=version, + filter=filter, + from_start_time=from_start_time, + to_start_time=to_start_time, + ) + current_cursor = response.meta.cursor + if current_cursor is None and fetched_page < requested_page: + break + + if response is None or fetched_page < requested_page: + return SimpleNamespace( + data=[], + meta=SimpleNamespace( + cursor=None, + total_items=0, + total_pages=requested_page - 1, + page=requested_page, + limit=limit or 50, + ), + ) + + data = [V4ObservationView(observation) for observation in response.data] + return SimpleNamespace( + data=data, + meta=SimpleNamespace( + cursor=response.meta.cursor, + total_items=len(data), + total_pages=1, + page=requested_page, + limit=limit or 50, + ), + ) + + +class _V4TraceClient: + def __init__(self, client: Any, observations: _V4ObservationsClient): + self._client = client + self._observations = observations + + def _scores(self, trace_id: str) -> list[Any]: + response = self._client.scores_v3.get_many_v3( + trace_id=trace_id, + fields="details,subject,annotation", + limit=100, + ) + return [_score_view(score) for score in response.data] + + def get(self, trace_id: str | None = None, **kwargs: Any) -> V4TraceView: + resolved_trace_id = trace_id or kwargs["trace_id"] + response = self._observations.get_many(trace_id=resolved_trace_id, limit=1000) + return V4TraceView( + resolved_trace_id, + response.data, + self._scores(resolved_trace_id), + ) + + def list( + self, + *, + name: str | None = None, + session_id: str | None = None, + user_id: str | None = None, + filter: str | None = None, + limit: int | None = None, + page: int | None = None, + **_: Any, + ) -> SimpleNamespace: + filters: list[dict[str, Any]] = [] + if filter: + filters.extend(json.loads(filter)) + if name is not None: + filters.append( + { + "type": "string", + "column": "traceName", + "operator": "=", + "value": name, + } + ) + if session_id is not None: + filters.append( + { + "type": "string", + "column": "sessionId", + "operator": "=", + "value": session_id, + } + ) + + response = self._observations.get_many( + user_id=user_id, + filter=json.dumps(filters) if filters else None, + limit=1000, + ) + observations_by_trace: dict[str, list[V4ObservationView]] = {} + for observation in response.data: + if observation.trace_id is not None: + observations_by_trace.setdefault(observation.trace_id, []).append( + observation + ) + + traces = [ + V4TraceView(trace_id, observations, self._scores(trace_id)) + for trace_id, observations in observations_by_trace.items() + ] + traces.sort(key=lambda trace: trace.timestamp, reverse=True) + + requested_limit = limit or 50 + requested_page = page or 1 + start = (requested_page - 1) * requested_limit + data = traces[start : start + requested_limit] + total_items = len(traces) + total_pages = (total_items + requested_limit - 1) // requested_limit + return SimpleNamespace( + data=data, + meta=SimpleNamespace( + page=requested_page, + limit=requested_limit, + total_items=total_items, + total_pages=total_pages, + ), + ) + + +class _V4SessionsClient: + def __init__(self, traces: _V4TraceClient): + self._traces = traces + + def list( + self, *, limit: int | None = None, page: int | None = None + ) -> SimpleNamespace: + traces = self._traces.list(limit=1000).data + sessions: dict[str, list[V4TraceView]] = {} + for trace in traces: + if trace.session_id is not None: + sessions.setdefault(trace.session_id, []).append(trace) + + data = [ + SimpleNamespace(id=session_id, traces=session_traces) + for session_id, session_traces in sessions.items() + ] + requested_limit = limit or 50 + requested_page = page or 1 + start = (requested_page - 1) * requested_limit + page_data = data[start : start + requested_limit] + total_items = len(data) + return SimpleNamespace( + data=page_data, + meta=SimpleNamespace( + page=requested_page, + limit=requested_limit, + total_items=total_items, + total_pages=(total_items + requested_limit - 1) // requested_limit, + ), + ) + + +class _V4ScoresClient: + def __init__(self, client: Any): + self._client = client + + def get_by_id(self, score_id: str) -> Any: + response = self._client.scores_v3.get_many_v3( + id=score_id, + fields="details,subject,annotation", + limit=1, + ) + if not response.data: + raise NotFoundError(body={"error": "LangfuseNotFoundError"}) + + return _score_view(response.data[0]) + + +class _V4DatasetsClient: + def __init__(self, client: Any): + self._client = client + self._datasets = client.datasets + + def __getattr__(self, name: str) -> Any: + return getattr(self._datasets, name) + + def get_run(self, dataset_name: str, run_name: str) -> Any: + dataset = self._datasets.get(dataset_name) + from_start_time, to_start_time = _time_bounds() + response = self._client.experiments.list( + from_start_time=from_start_time, + to_start_time=to_start_time, + dataset_id=dataset.id, + name=run_name, + fields="core,metadata,scores", + limit=1, + ) + if not response.data: + raise NotFoundError(body={"error": "LangfuseNotFoundError"}) + + return response.data[0] + + def get_runs(self, dataset_name: str) -> SimpleNamespace: + dataset = self._datasets.get(dataset_name) + from_start_time, to_start_time = _time_bounds() + return self._client.experiments.list( + from_start_time=from_start_time, + to_start_time=to_start_time, + dataset_id=dataset.id, + fields="core,metadata,scores", + limit=100, + ) + + +class _V4ExperimentItemsClient: + def __init__(self, client: Any): + self._client = client + + def list(self, *, dataset_id: str, run_name: str) -> Any: + from_start_time, to_start_time = _time_bounds() + experiments = self._client.experiments.list( + from_start_time=from_start_time, + to_start_time=to_start_time, + dataset_id=dataset_id, + name=run_name, + limit=1, + ) + if not experiments.data: + raise NotFoundError(body={"error": "LangfuseNotFoundError"}) + + return self._client.experiments.list_items( + from_start_time=from_start_time, + to_start_time=to_start_time, + experiment_id=experiments.data[0].id, + fields=("core,dataset,io,metadata,itemMetadata,experimentMetadata,scores"), + limit=100, + ) + + +class V4TestAPI: + """Expose v4-native read paths behind the test suite's existing API helper.""" + + def __init__(self, client: Any): + self._client = client + self.observations = _V4ObservationsClient(client) + self.trace = _V4TraceClient(client, self.observations) + self.legacy = SimpleNamespace(observations_v1=self.observations) + self.sessions = _V4SessionsClient(self.trace) + self.scores = _V4ScoresClient(client) + self.datasets = _V4DatasetsClient(client) + self.dataset_run_items = _V4ExperimentItemsClient(client) + + def __getattr__(self, name: str) -> Any: + return getattr(self._client, name) diff --git a/tests/unit/test_additional_headers_simple.py b/tests/unit/test_additional_headers_simple.py index dd843b35a..32664810a 100644 --- a/tests/unit/test_additional_headers_simple.py +++ b/tests/unit/test_additional_headers_simple.py @@ -187,6 +187,7 @@ def test_span_processor_has_additional_headers_in_otel_exporter(self): assert "Authorization" in exporter._headers assert "x-langfuse-sdk-name" in exporter._headers assert "x-langfuse-public-key" in exporter._headers + assert exporter._headers["x-langfuse-ingestion-version"] == "4" # Check that our override worked assert exporter._headers["X-Override-Default"] == "override-value" @@ -210,6 +211,7 @@ def test_span_processor_none_additional_headers_works(self): assert "Authorization" in exporter._headers assert "x-langfuse-sdk-name" in exporter._headers assert "x-langfuse-public-key" in exporter._headers + assert exporter._headers["x-langfuse-ingestion-version"] == "4" def test_span_processor_uses_custom_span_exporter_when_provided(self): """Test that a custom exporter bypasses the default OTLP exporter construction.""" diff --git a/tests/unit/test_e2e_support.py b/tests/unit/test_e2e_support.py index 8320bd2fe..46376b1dc 100644 --- a/tests/unit/test_e2e_support.py +++ b/tests/unit/test_e2e_support.py @@ -29,6 +29,7 @@ class FakeClient: trace = FakeTraceService() monkeypatch.setattr("tests.support.utils.LangfuseAPI", lambda **_: FakeClient()) + monkeypatch.setattr("tests.support.utils.V4TestAPI", lambda client: client) trace = get_api().trace.get("trace-123") @@ -54,6 +55,7 @@ class FakeClient: trace = FakeTraceService() monkeypatch.setattr("tests.support.utils.LangfuseAPI", lambda **_: FakeClient()) + monkeypatch.setattr("tests.support.utils.V4TestAPI", lambda client: client) response = get_api().trace.list(name="ready-trace") @@ -73,6 +75,7 @@ class FakeClient: trace = FakeTraceService() monkeypatch.setattr("tests.support.utils.LangfuseAPI", lambda **_: FakeClient()) + monkeypatch.setattr("tests.support.utils.V4TestAPI", lambda client: client) response = get_api(retry=False).trace.list(name="missing-trace") @@ -85,30 +88,26 @@ def test_raw_api_wrapper_retries_not_found_payload(monkeypatch): attempts = {"count": 0} - class FakeResponse: - def __init__(self, status_code, payload): - self.status_code = status_code - self._payload = payload - self.headers = {} - - def json(self): - return self._payload - - def fake_get(*args, **kwargs): - attempts["count"] += 1 + class FakeTraceService: + def get(self, trace_id): + attempts["count"] += 1 + if attempts["count"] < 3: + raise NotFoundError( + body={ + "error": "LangfuseNotFoundError", + "message": "Trace trace-123 not found within authorized project", + } + ) - if attempts["count"] < 3: - return FakeResponse( - 404, - { - "error": "LangfuseNotFoundError", - "message": "Trace trace-123 not found within authorized project", - }, - ) + return {"id": trace_id, "observations": []} - return FakeResponse(200, {"id": "trace-123", "observations": []}) + class FakeClient: + trace = FakeTraceService() - monkeypatch.setattr("tests.support.api_wrapper.httpx.get", fake_get) + monkeypatch.setattr("tests.support.api_wrapper.get_api", lambda **_: FakeClient()) + monkeypatch.setattr( + SupportLangfuseAPI, "_trace_json", staticmethod(lambda trace: trace) + ) api = SupportLangfuseAPI(username="user", password="pass", base_url="http://test") trace = api.get_trace("trace-123") @@ -131,6 +130,7 @@ class FakeClient: trace = FakeTraceService() monkeypatch.setattr("tests.support.utils.LangfuseAPI", lambda **_: FakeClient()) + monkeypatch.setattr("tests.support.utils.V4TestAPI", lambda client: client) trace = wait_for_trace( "trace-123", is_result_ready=lambda trace: len(trace["observations"]) == 3 diff --git a/tests/unit/test_propagate_attributes.py b/tests/unit/test_propagate_attributes.py index cdf4f9351..212f1b663 100644 --- a/tests/unit/test_propagate_attributes.py +++ b/tests/unit/test_propagate_attributes.py @@ -2706,11 +2706,13 @@ def task_with_child_spans(*, item, **kwargs): LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT, ) - # Root-only attributes should NOT be present on children - self.verify_missing_attribute( + # Experiment context is propagated to children. + self.verify_span_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION, + "Test experiment description", ) + # Item payload attributes remain root-only. self.verify_missing_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_EXPECTED_OUTPUT, @@ -2783,33 +2785,9 @@ def test_experiment_run_metadata_overrides_item_metadata( ) def test_experiment_attributes_propagate_with_dataset( - self, langfuse_client, memory_exporter, monkeypatch + self, langfuse_client, memory_exporter ): """Test experiment attribute propagation with Langfuse dataset.""" - created_run_items = [] - - # Mock the sync API used by run_experiment to create dataset run items - def mock_create_dataset_run_item(*args, **kwargs): - from langfuse.api import DatasetRunItem - - created_run_items.append(kwargs) - return DatasetRunItem( - id="mock-run-item-id", - dataset_run_id="mock-dataset-run-id-123", - dataset_run_name=kwargs.get("run_name", "Dataset Test"), - dataset_item_id=kwargs.get("dataset_item_id", "mock-item-id"), - trace_id="mock-trace-id", - observation_id=kwargs.get("observation_id"), - created_at=datetime.now(), - updated_at=datetime.now(), - ) - - monkeypatch.setattr( - langfuse_client.api.dataset_run_items, - "create", - mock_create_dataset_run_item, - ) - # Create a mock dataset with items dataset_id = "test-dataset-id-456" dataset_item_id = "test-dataset-item-id-789" @@ -2871,10 +2849,7 @@ def task_with_children(*, item, **kwargs): root_spans = self.get_spans_by_name(memory_exporter, "experiment-item-run") assert len(root_spans) >= 1, "Should have at least 1 root span" first_root = root_spans[0] - task_span = self.get_span_by_name(memory_exporter, "experiment-item-task") - assert result.experiment_id == "mock-dataset-run-id-123" - assert len(created_run_items) == 1 - assert created_run_items[0]["observation_id"] == task_span["span_id"] + assert result.experiment_id == result.dataset_run_id # Root-only attributes should be on root self.verify_span_attribute( @@ -2971,11 +2946,13 @@ def task_with_children(*, item, **kwargs): LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT, ) - # Root-only attributes should NOT be present on children - self.verify_missing_attribute( + # Experiment context is propagated to children. + self.verify_span_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION, + "Dataset experiment description", ) + # Item payload attributes remain root-only. self.verify_missing_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_EXPECTED_OUTPUT, @@ -3054,10 +3031,11 @@ def task_with_nested_spans(*, item, **kwargs): LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT, ) - # Root-only attributes should NOT be present - self.verify_missing_attribute( + # Experiment context is propagated to nested spans. + self.verify_span_attribute( span_data, LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION, + "Nested test", ) self.verify_missing_attribute( span_data, From e63e5cbb9fc9b53db0d6c3075414db972cc080ac Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:02:11 +0200 Subject: [PATCH 2/3] fix(tests): merge v4 trace metadata --- tests/support/v4_api.py | 8 ++--- tests/unit/test_e2e_support.py | 62 ++++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 5 deletions(-) diff --git a/tests/support/v4_api.py b/tests/support/v4_api.py index 2b6a667e8..c14518800 100644 --- a/tests/support/v4_api.py +++ b/tests/support/v4_api.py @@ -169,11 +169,9 @@ def first_populated(attribute: str, fallback: Any = None) -> Any: self.release = first_populated("release") self.version = root.version self.user_id = first_populated("user_id") - self.metadata = max( - (observation.metadata for observation in observations), - key=len, - default={}, - ) + self.metadata = {} + for observation in observations: + self.metadata.update(observation.metadata) self.tags = list( dict.fromkeys( tag diff --git a/tests/unit/test_e2e_support.py b/tests/unit/test_e2e_support.py index 46376b1dc..32da136ed 100644 --- a/tests/unit/test_e2e_support.py +++ b/tests/unit/test_e2e_support.py @@ -1,9 +1,71 @@ +from datetime import datetime, timedelta, timezone from types import SimpleNamespace from langfuse.api.commons.errors.not_found_error import NotFoundError from tests.support.api_wrapper import LangfuseAPI as SupportLangfuseAPI from tests.support.retry import retry_until_ready from tests.support.utils import get_api, wait_for_trace +from tests.support.v4_api import V4TraceView + + +def test_v4_trace_view_merges_metadata_from_all_observations(): + started_at = datetime.now(timezone.utc) + + def observation( + *, + observation_id, + metadata, + start_time, + is_root_observation, + parent_observation_id, + ): + return SimpleNamespace( + id=observation_id, + name=observation_id, + start_time=start_time, + is_root_observation=is_root_observation, + parent_observation_id=parent_observation_id, + input=None, + output=None, + version=None, + metadata=metadata, + _observation=SimpleNamespace( + trace_name="experiment-trace", + session_id=None, + release=None, + user_id=None, + tags=[], + public=False, + environment="default", + ), + ) + + trace = V4TraceView( + "trace-123", + [ + observation( + observation_id="root", + metadata={"experiment_name": "metadata-test"}, + start_time=started_at, + is_root_observation=True, + parent_observation_id=None, + ), + observation( + observation_id="external-child", + metadata={"mode": "offline", "job_name": "agent-eval/PR-4"}, + start_time=started_at + timedelta(milliseconds=1), + is_root_observation=False, + parent_observation_id="root", + ), + ], + [], + ) + + assert trace.metadata == { + "experiment_name": "metadata-test", + "mode": "offline", + "job_name": "agent-eval/PR-4", + } def test_get_api_retries_not_found(monkeypatch): From 8676ddab732c5fa7c9aeb5d52523c32c1802a863 Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:20:33 +0200 Subject: [PATCH 3/3] fix(ci): run Langfuse v4 in dual-write mode --- .github/langfuse-v4-dual-write.override.yml | 7 + .github/workflows/ci.yml | 2 + langfuse/_client/attributes.py | 1 - langfuse/_client/client.py | 28 +- langfuse/_client/propagation.py | 8 - langfuse/_client/span_processor.py | 1 - langfuse/batch_evaluation.py | 133 +---- tests/e2e/test_batch_evaluation.py | 8 +- tests/e2e/test_core_sdk.py | 37 +- tests/e2e/test_decorators.py | 2 +- tests/support/api_wrapper.py | 146 ++---- tests/support/utils.py | 4 +- tests/support/v4_api.py | 492 ------------------- tests/unit/test_additional_headers_simple.py | 2 - tests/unit/test_e2e_support.py | 104 +--- tests/unit/test_propagate_attributes.py | 48 +- 16 files changed, 183 insertions(+), 840 deletions(-) create mode 100644 .github/langfuse-v4-dual-write.override.yml delete mode 100644 tests/support/v4_api.py diff --git a/.github/langfuse-v4-dual-write.override.yml b/.github/langfuse-v4-dual-write.override.yml new file mode 100644 index 000000000..48ab6ff48 --- /dev/null +++ b/.github/langfuse-v4-dual-write.override.yml @@ -0,0 +1,7 @@ +services: + langfuse-worker: + environment: + LANGFUSE_MIGRATION_V4_WRITE_MODE: dual + langfuse-web: + environment: + LANGFUSE_MIGRATION_V4_WRITE_MODE: dual diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2d01df52b..4b44daed6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -160,6 +160,8 @@ jobs: LANGFUSE_SERVER_SHA="$(git ls-remote https://github.com/langfuse/langfuse.git HEAD | cut -f1)" curl -fsSL "https://raw.githubusercontent.com/langfuse/langfuse/${LANGFUSE_SERVER_SHA}/docker-compose.yml" \ -o ./langfuse-server/docker-compose.yml + cp ./.github/langfuse-v4-dual-write.override.yml \ + ./langfuse-server/docker-compose.override.yml echo "${LANGFUSE_SERVER_SHA}" - name: Run langfuse server diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py index 43a85c2fd..4660b50f0 100644 --- a/langfuse/_client/attributes.py +++ b/langfuse/_client/attributes.py @@ -68,7 +68,6 @@ class LangfuseOtelSpanAttributes: EXPERIMENT_METADATA = "langfuse.experiment.metadata" EXPERIMENT_DATASET_ID = "langfuse.experiment.dataset.id" EXPERIMENT_ITEM_ID = "langfuse.experiment.item.id" - EXPERIMENT_ITEM_VERSION = "langfuse.experiment.item.version" EXPERIMENT_ITEM_EXPECTED_OUTPUT = "langfuse.experiment.item.expected_output" EXPERIMENT_ITEM_METADATA = "langfuse.experiment.item.metadata" EXPERIMENT_ITEM_ROOT_OBSERVATION_ID = "langfuse.experiment.item.root_observation_id" diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 1f0ec926b..d97708efd 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -2997,23 +2997,37 @@ async def _process_experiment_item( # Link dataset runs to the canonical task observation so their # latency excludes the subsequent evaluator subtree. if hasattr(item, "id") and hasattr(item, "dataset_id"): - # Langfuse v4 synthesizes the experiment and item links from - # the OpenTelemetry attributes propagated below. - dataset_run_id = fallback_experiment_id + try: + # Use sync API to avoid event loop issues when + # run_async_safely creates multiple event loops across + # different threads. + dataset_run_item = await asyncio.to_thread( + self.api.dataset_run_items.create, + run_name=experiment_run_name, + run_description=experiment_description, + metadata=experiment_metadata, + dataset_item_id=item.id, # type: ignore + trace_id=trace_id, + observation_id=task_span.id, + dataset_version=dataset_version, + ) + + dataset_run_id = dataset_run_item.dataset_run_id + + except Exception as e: + langfuse_logger.error( + f"Failed to create dataset run item: {e}" + ) experiment_id = dataset_run_id or fallback_experiment_id propagated_experiment_attributes = PropagatedExperimentAttributes( experiment_id=experiment_id, experiment_name=experiment_run_name, - experiment_description=experiment_description, experiment_metadata=_flatten_and_serialize_metadata_values( experiment_metadata ), experiment_dataset_id=dataset_id, experiment_item_id=experiment_item_id, - experiment_item_version=( - dataset_version.isoformat() if dataset_version else None - ), experiment_item_metadata=_flatten_and_serialize_metadata_values( item_metadata if isinstance(item_metadata, dict) else None ), diff --git a/langfuse/_client/propagation.py b/langfuse/_client/propagation.py index 6afbec2ec..d6e26163a 100644 --- a/langfuse/_client/propagation.py +++ b/langfuse/_client/propagation.py @@ -58,11 +58,9 @@ InternalPropagatedKeys = Literal[ "experiment_id", "experiment_name", - "experiment_description", "experiment_metadata", "experiment_dataset_id", "experiment_item_id", - "experiment_item_version", "experiment_item_metadata", "experiment_item_root_observation_id", ] @@ -79,11 +77,9 @@ "prompt_version", "experiment_id", "experiment_name", - "experiment_description", "experiment_metadata", "experiment_dataset_id", "experiment_item_id", - "experiment_item_version", "experiment_item_metadata", "experiment_item_root_observation_id", ] @@ -92,11 +88,9 @@ class PropagatedExperimentAttributes(TypedDict): experiment_id: str experiment_name: str - experiment_description: Optional[str] experiment_metadata: Optional[Dict[str, str]] experiment_dataset_id: Optional[str] experiment_item_id: str - experiment_item_version: Optional[str] experiment_item_metadata: Optional[Dict[str, str]] experiment_item_root_observation_id: str @@ -764,11 +758,9 @@ def _get_propagated_span_key(key: str) -> str: "prompt_version": LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_VERSION, "experiment_id": LangfuseOtelSpanAttributes.EXPERIMENT_ID, "experiment_name": LangfuseOtelSpanAttributes.EXPERIMENT_NAME, - "experiment_description": LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION, "experiment_metadata": LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, "experiment_dataset_id": LangfuseOtelSpanAttributes.EXPERIMENT_DATASET_ID, "experiment_item_id": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ID, - "experiment_item_version": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_VERSION, "experiment_item_metadata": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA, "experiment_item_root_observation_id": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ROOT_OBSERVATION_ID, }.get(key) or f"{LangfuseOtelSpanAttributes.TRACE_METADATA}.{key}" diff --git a/langfuse/_client/span_processor.py b/langfuse/_client/span_processor.py index de2a961c8..f01b081c2 100644 --- a/langfuse/_client/span_processor.py +++ b/langfuse/_client/span_processor.py @@ -109,7 +109,6 @@ def __init__( "x-langfuse-sdk-name": "python", "x-langfuse-sdk-version": langfuse_version, "x-langfuse-public-key": public_key, - "x-langfuse-ingestion-version": "4", } # Merge additional headers if provided diff --git a/langfuse/batch_evaluation.py b/langfuse/batch_evaluation.py index ad3080d30..d28fd085d 100644 --- a/langfuse/batch_evaluation.py +++ b/langfuse/batch_evaluation.py @@ -9,7 +9,6 @@ import asyncio import json import time -from dataclasses import dataclass from typing import ( TYPE_CHECKING, Any, @@ -35,41 +34,6 @@ from langfuse._client.client import Langfuse -def _parse_observation_value(value: Any) -> Any: - if not isinstance(value, str): - return value - - try: - return json.loads(value) - except (json.JSONDecodeError, TypeError): - return value - - -@dataclass -class _BatchEvaluationObservation: - id: str - trace_id: Optional[str] - start_time: Any - input: Any - output: Any - metadata: Any - name: Optional[str] - user_id: Optional[str] - tags: List[str] - - -@dataclass -class _BatchEvaluationTrace: - id: str - timestamp: Any - input: Any - output: Any - metadata: Any - name: Optional[str] - user_id: Optional[str] - tags: List[str] - - class EvaluatorInputs: """Input data structure for evaluators, returned by mapper functions. @@ -954,7 +918,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 @@ -980,10 +943,10 @@ 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, - cursor=cursor, + page=page, limit=fetch_batch_size, max_retries=max_retries, fields=fetch_trace_fields, @@ -1122,11 +1085,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,94 +1141,49 @@ async def _fetch_batch_with_retry( *, scope: str, filter: Optional[str], - cursor: Optional[str], + page: int, limit: int, max_retries: int, fields: Optional[str], - ) -> Tuple[List[Any], 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. - cursor: Cursor returned by the previous Observations API page. + 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 Returns: - The items and cursor for the next page. + List of items from the API. Raises: Exception: If all retry attempts fail. """ - if scope not in {"traces", "observations"}: + 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: error_message = f"Invalid scope: {scope}" raise ValueError(error_message) - effective_filter = filter - if scope == "traces": - filters = json.loads(filter) if filter else [] - filters.append( - { - "type": "boolean", - "column": "isRootObservation", - "operator": "=", - "value": True, - } - ) - effective_filter = json.dumps(filters) - - response = self.client.api.observations.get_many( - fields="core,basic,io,metadata,trace_context", - limit=limit, - cursor=cursor, - filter=effective_filter, - request_options={"max_retries": max_retries}, - ) - - observations = [ - _BatchEvaluationObservation( - id=observation.id, - trace_id=observation.trace_id, - start_time=observation.start_time, - input=_parse_observation_value(observation.input), - output=_parse_observation_value(observation.output), - metadata=_parse_observation_value(observation.metadata), - name=observation.name, - user_id=observation.user_id, - tags=observation.tags or [], - ) - for observation in response.data - ] - if scope == "observations": - return observations, response.meta.cursor # type: ignore[return-value] - - return ( - [ - _BatchEvaluationTrace( - id=observation.trace_id or observation.id, - timestamp=observation.start_time, - input=observation.input, - output=observation.output, - metadata=observation.metadata, - name=next( - ( - item.trace_name - for item in response.data - if item.id == observation.id and item.trace_name - ), - observation.name, - ), - user_id=observation.user_id, - tags=observation.tags, - ) - for observation in observations - ], - response.meta.cursor, - ) # type: ignore[return-value] - async def _process_batch_evaluation_item( self, item: Union[TraceWithFullDetails, ObservationsView], diff --git a/tests/e2e/test_batch_evaluation.py b/tests/e2e/test_batch_evaluation.py index 299e82480..0632b21b8 100644 --- a/tests/e2e/test_batch_evaluation.py +++ b/tests/e2e/test_batch_evaluation.py @@ -52,7 +52,7 @@ def _seed_trace_corpus( trace_names.append(trace_name) with langfuse_client.start_as_current_observation(name=trace_name) as span: with propagate_attributes(tags=[corpus_tag]): - span.update( + span.set_trace_io( input=f"Seed input {index}", output=f"Seed output {index}", ) @@ -161,7 +161,7 @@ def test_batch_evaluation_with_filter(langfuse_client): name=f"filtered-trace-{create_uuid()}" ) as span: with propagate_attributes(tags=[unique_tag]): - span.update( + span.set_trace_io( input="Filtered test", output="Filtered output", ) @@ -789,7 +789,7 @@ def test_pagination_with_max_items(langfuse_client): name=f"pagination-test-{create_uuid()}" ) as span: with propagate_attributes(tags=["pagination_test"]): - span.update( + span.set_trace_io( input=f"Input {i}", output=f"Output {i}", ) @@ -821,7 +821,7 @@ def test_has_more_items_flag(langfuse_client): name=f"more-items-test-{i}" ) as span: with propagate_attributes(tags=[batch_tag]): - span.update( + span.set_trace_io( input=f"Input {i}", output=f"Output {i}", ) diff --git a/tests/e2e/test_core_sdk.py b/tests/e2e/test_core_sdk.py index 5bcfea0c0..614d6da41 100644 --- a/tests/e2e/test_core_sdk.py +++ b/tests/e2e/test_core_sdk.py @@ -270,7 +270,8 @@ def test_create_boolean_score(): assert created_score is not None, "Score not found in trace" assert created_score["id"] == score_id assert created_score["dataType"] == "BOOLEAN" - assert created_score["value"] is True + assert created_score["value"] == 1 + assert created_score["stringValue"] == "True" def test_create_categorical_score(): @@ -327,7 +328,8 @@ def test_create_categorical_score(): assert created_score is not None, "Score not found in trace" assert created_score["id"] == score_id assert created_score["dataType"] == "CATEGORICAL" - assert created_score["value"] == "high score" + assert created_score["value"] == 0 + assert created_score["stringValue"] == "high score" def test_create_text_score(): @@ -384,8 +386,9 @@ def test_create_text_score(): assert created_score is not None, "Score not found in trace" assert created_score["id"] == score_id assert created_score["dataType"] == "TEXT" + assert ( - created_score["value"] + created_score["stringValue"] == "This is a detailed text evaluation of the output quality." ) @@ -491,6 +494,7 @@ def test_create_trace(): assert trace["metadata"]["key"] == "value" assert trace["tags"] == ["tag1", "tag2"] assert trace["public"] is True + assert True if not trace["externalId"] else False def test_create_update_trace(): @@ -552,7 +556,7 @@ def test_create_update_current_trace(): user_id="test", metadata={"key": "value"}, ): - langfuse.update_current_span(input="test_input") + langfuse.set_current_trace_io(input="test_input") langfuse.set_current_trace_as_public() # Get trace ID for later reference trace_id = span.trace_id @@ -656,7 +660,7 @@ def test_create_generation(): ( { "input": 51, - "output": 49, + "output": 0, "total": 100, }, "TOKENS", @@ -667,7 +671,7 @@ def test_create_generation(): ( { "input": 51, - "output": 49, + "output": 0, "total": 100, }, "CHARACTERS", @@ -697,7 +701,6 @@ def test_create_generation_complex( }, ], output=[{"foo": "bar"}], - model="gpt-3.5-turbo", usage_details=usage, metadata={"tags": ["yo"]}, ).end() @@ -733,7 +736,7 @@ def test_create_generation_complex( assert generation_api.metadata["tags"] == ["yo"] assert generation_api.start_time is not None - assert generation_api.usage_details == {"input": 51, "output": 49, "total": 100} + assert generation_api.usage_details == {"input": 51, "output": 0, "total": 100} def test_create_span(): @@ -986,7 +989,7 @@ def test_create_trace_and_generation(): # Create parent span and set trace properties with langfuse.start_as_current_observation(name=trace_name) as parent_span: with propagate_attributes(trace_name=trace_name, session_id="test-session-id"): - parent_span.update(input={"key": "value"}) + parent_span.set_trace_io(input={"key": "value"}) # Create a generation as child generation = parent_span.start_observation( @@ -1391,7 +1394,6 @@ def test_end_generation_with_openai_token_format(): generation = langfuse.start_observation( as_type="generation", name="query-generation", - model="gpt-3.5-turbo", ) # Get trace ID for verification @@ -1432,6 +1434,7 @@ def test_end_generation_with_openai_token_format(): assert generation_api.usage.input == 100 # prompt_tokens mapped to input assert generation_api.usage.output == 200 # completion_tokens mapped to output assert generation_api.usage.total == 500 + assert generation_api.usage.unit == "TOKENS" # Default unit for OpenAI format assert generation_api.calculated_input_cost == 111 assert generation_api.calculated_output_cost == 222 assert generation_api.calculated_total_cost == 444 @@ -1809,7 +1812,7 @@ def test_fetch_traces(): # First trace with langfuse.start_as_current_observation(name="test1") as span: with propagate_attributes(trace_name=name, session_id="session-1"): - span.update(input={"key": "value"}, output="output-value") + span.set_trace_io(input={"key": "value"}, output="output-value") trace_ids.append(span.trace_id) sleep(1) # Ensure traces have different timestamps @@ -1817,7 +1820,7 @@ def test_fetch_traces(): # Second trace with langfuse.start_as_current_observation(name="test2") as span: with propagate_attributes(trace_name=name, session_id="session-1"): - span.update(input={"key": "value"}, output="output-value") + span.set_trace_io(input={"key": "value"}, output="output-value") trace_ids.append(span.trace_id) sleep(1) # Ensure traces have different timestamps @@ -1825,7 +1828,7 @@ def test_fetch_traces(): # Third trace with langfuse.start_as_current_observation(name="test3") as span: with propagate_attributes(trace_name=name, session_id="session-1"): - span.update(input={"key": "value"}, output="output-value") + span.set_trace_io(input={"key": "value"}, output="output-value") trace_ids.append(span.trace_id) # Ensure data is sent @@ -2085,11 +2088,11 @@ def mask_func(data): # Create a root span with trace properties with langfuse.start_as_current_observation(name="test-span") as root_span: with propagate_attributes(trace_name="test_trace"): - root_span.update(input={"sensitive": "data"}) + root_span.set_trace_io(input={"sensitive": "data"}) # Get trace ID for later use trace_id = root_span.trace_id # Add output to the trace - root_span.update(output={"more": "sensitive"}) + root_span.set_trace_io(output={"more": "sensitive"}) # Create a generation as child gen = root_span.start_observation( @@ -2133,11 +2136,11 @@ def mask_func(data): # Create a root span with trace properties with langfuse.start_as_current_observation(name="test-span") as root_span: with propagate_attributes(trace_name="test_trace"): - root_span.update(input={"should_raise": "data"}) + root_span.set_trace_io(input={"should_raise": "data"}) # Get trace ID for later use trace_id = root_span.trace_id # Add output to the trace - root_span.update(output={"should_raise": "sensitive"}) + root_span.set_trace_io(output={"should_raise": "sensitive"}) # Ensure data is sent langfuse.flush() diff --git a/tests/e2e/test_decorators.py b/tests/e2e/test_decorators.py index da8048d2d..9e302f799 100644 --- a/tests/e2e/test_decorators.py +++ b/tests/e2e/test_decorators.py @@ -1075,7 +1075,7 @@ def test_media(): @observe() def main(): sleep(1) - langfuse.update_current_span( + langfuse.set_current_trace_io( input={ "context": { "nested": media, diff --git a/tests/support/api_wrapper.py b/tests/support/api_wrapper.py index 53549a999..c4519252f 100644 --- a/tests/support/api_wrapper.py +++ b/tests/support/api_wrapper.py @@ -1,87 +1,47 @@ +import os + +import httpx + +from langfuse.api.commons.errors.not_found_error import NotFoundError from tests.support.retry import ( DEFAULT_RETRY_INTERVAL_SECONDS, DEFAULT_RETRY_TIMEOUT_SECONDS, + is_not_found_payload, + retry_until_ready, ) -from tests.support.utils import get_api, wait_for_result -from tests.support.v4_api import _score_view class LangfuseAPI: def __init__(self, username=None, password=None, base_url=None): - self._api = get_api(retry=False) + username = username if username else os.environ["LANGFUSE_PUBLIC_KEY"] + password = password if password else os.environ["LANGFUSE_SECRET_KEY"] + self.auth = (username, password) + self.BASE_URL = base_url if base_url else os.environ["LANGFUSE_BASE_URL"] - @staticmethod - def _score_json(score): - return { - "id": score.id, - "name": score.name, - "value": score.value, - "timestamp": score.timestamp.isoformat(), - "dataType": score.data_type, - "stringValue": score.string_value, - "traceId": score.trace_id, - "observationId": score.observation_id, - "sessionId": score.session_id, - "comment": getattr(score, "comment", None), - "metadata": getattr(score, "metadata", None), - } + def _get_json( + self, + url, + params=None, + *, + retry=True, + is_result_ready=None, + timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, + interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, + ): + def _request(): + response = httpx.get(url, params=params, auth=self.auth) + payload = response.json() - @classmethod - def _observation_json(cls, observation): - return { - "id": observation.id, - "traceId": observation.trace_id, - "type": observation.type, - "name": observation.name, - "startTime": observation.start_time.isoformat(), - "endTime": ( - observation.end_time.isoformat() if observation.end_time else None - ), - "input": observation.input, - "output": observation.output, - "metadata": observation.metadata, - "model": observation.model, - "usageDetails": observation.usage_details, - "costDetails": observation.cost_details, - "promptId": observation.prompt_id, - } + if response.status_code == 404 and is_not_found_payload(payload): + raise NotFoundError(body=payload, headers=dict(response.headers)) - @classmethod - def _trace_json(cls, trace): - return { - "id": trace.id, - "timestamp": trace.timestamp.isoformat(), - "name": trace.name, - "input": trace.input, - "output": trace.output, - "sessionId": trace.session_id, - "release": trace.release, - "version": trace.version, - "userId": trace.user_id, - "metadata": trace.metadata, - "tags": trace.tags, - "public": trace.public, - "environment": trace.environment, - "observations": [ - cls._observation_json(observation) for observation in trace.observations - ], - "scores": [cls._score_json(score) for score in trace.scores], - } + return payload - @staticmethod - def _read( - operation, - *, - retry, - is_result_ready, - timeout_seconds, - interval_seconds, - ): if not retry: - return operation() + return _request() - return wait_for_result( - operation, + return retry_until_ready( + _request, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, interval_seconds=interval_seconds, @@ -96,8 +56,9 @@ def get_observation( timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, ): - return self._read( - lambda: self._observation_json(self._api.observations.get(observation_id)), + url = f"{self.BASE_URL}/api/public/observations/{observation_id}" + return self._get_json( + url, retry=retry, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, @@ -116,21 +77,11 @@ def get_scores( timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, ): - def operation(): - response = self._api._client.scores_v3.get_many_v3( - name=name, - user_id=user_id, - limit=limit, - fields="details,subject,annotation", - ) - data = [self._score_json(_score_view(score)) for score in response.data] - return { - "data": data, - "meta": {"page": page or 1, "limit": limit or 50}, - } - - return self._read( - operation, + params = {"page": page, "limit": limit, "userId": user_id, "name": name} + url = f"{self.BASE_URL}/api/public/scores" + return self._get_json( + url, + params=params, retry=retry, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, @@ -149,17 +100,11 @@ def get_traces( timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, ): - def operation(): - response = self._api.trace.list( - page=page, limit=limit, user_id=user_id, name=name - ) - return { - "data": [self._trace_json(trace) for trace in response.data], - "meta": vars(response.meta), - } - - return self._read( - operation, + params = {"page": page, "limit": limit, "userId": user_id, "name": name} + url = f"{self.BASE_URL}/api/public/traces" + return self._get_json( + url, + params=params, retry=retry, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, @@ -175,8 +120,9 @@ def get_trace( timeout_seconds=DEFAULT_RETRY_TIMEOUT_SECONDS, interval_seconds=DEFAULT_RETRY_INTERVAL_SECONDS, ): - return self._read( - lambda: self._trace_json(self._api.trace.get(trace_id)), + url = f"{self.BASE_URL}/api/public/traces/{trace_id}" + return self._get_json( + url, retry=retry, is_result_ready=is_result_ready, timeout_seconds=timeout_seconds, diff --git a/tests/support/utils.py b/tests/support/utils.py index e22654d13..a29274d3a 100644 --- a/tests/support/utils.py +++ b/tests/support/utils.py @@ -9,7 +9,6 @@ DEFAULT_RETRY_TIMEOUT_SECONDS, retry_until_ready, ) -from tests.support.v4_api import V4TestAPI READ_METHOD_NAMES = {"get", "get_by_id", "get_many", "get_run", "list"} PAGINATION_ARGUMENTS = {"limit", "page"} @@ -72,8 +71,7 @@ def get_api(*, retry: bool = True): password=os.environ.get("LANGFUSE_SECRET_KEY"), base_url=os.environ.get("LANGFUSE_BASE_URL"), ) - test_api = V4TestAPI(client) - return _RetryingApiProxy(test_api) if retry else test_api + return _RetryingApiProxy(client) if retry else client def wait_for_result( diff --git a/tests/support/v4_api.py b/tests/support/v4_api.py deleted file mode 100644 index c14518800..000000000 --- a/tests/support/v4_api.py +++ /dev/null @@ -1,492 +0,0 @@ -from __future__ import annotations - -import json -from datetime import datetime, timedelta, timezone -from types import SimpleNamespace -from typing import Any, Iterable - -from langfuse.api.commons.errors.not_found_error import NotFoundError - -OBSERVATION_FIELDS = ( - "core,basic,time,io,metadata,model,usage,prompt,metrics,trace_context" -) -DEFAULT_LOOKBACK = timedelta(days=1) -DEFAULT_LOOKAHEAD = timedelta(minutes=5) - - -def _time_bounds() -> tuple[datetime, datetime]: - now = datetime.now(timezone.utc) - return now - DEFAULT_LOOKBACK, now + DEFAULT_LOOKAHEAD - - -def _parse_json(value: Any) -> Any: - if not isinstance(value, str): - return value - - try: - return json.loads(value) - except (json.JSONDecodeError, TypeError): - return value - - -def _as_float(value: Any) -> float | None: - if value is None: - return None - - return float(value) - - -def _none_if_empty(value: Any) -> Any: - return None if value == "" else value - - -def _score_view(score: Any) -> SimpleNamespace: - subject = getattr(score, "subject", None) - kind = getattr(subject, "kind", None) - value = score.value - score_type = type(score).__name__.split("_")[-1].removesuffix("ScoreV3").upper() - string_value = value if score_type in {"CATEGORICAL", "TEXT"} else None - trace_id = None - if subject is not None: - trace_id = getattr(subject, "trace_id", None) - if trace_id is None and kind == "trace": - trace_id = subject.id - - data = score.model_dump() - data.update( - data_type=score_type, - string_value=string_value, - trace_id=trace_id, - observation_id=( - subject.id if subject is not None and kind == "observation" else None - ), - session_id=subject.id if subject is not None and kind == "session" else None, - ) - return SimpleNamespace(**data) - - -class V4ObservationView: - """Test-friendly view of one row returned by Observations API v2.""" - - def __init__(self, observation: Any): - self._observation = observation - self.id = observation.id - self.trace_id = observation.trace_id - self.type = observation.type - self.name = observation.name - self.start_time = observation.start_time - self.end_time = observation.end_time - self.completion_start_time = observation.completion_start_time - self.model = ( - getattr(observation, "model", None) or observation.provided_model_name - ) - self.model_parameters = _parse_json(observation.model_parameters) - self.input = _parse_json(observation.input) - self.output = _parse_json(observation.output) - self.version = observation.version - self.metadata = _parse_json(observation.metadata) or {} - self.level = observation.level - self.status_message = observation.status_message - self.parent_observation_id = observation.parent_observation_id - self.is_root_observation = observation.is_root_observation - self.prompt_id = _none_if_empty(observation.prompt_id) - self.prompt_name = _none_if_empty(observation.prompt_name) - self.prompt_version = observation.prompt_version - self.usage_details = observation.usage_details or {} - self.cost_details = observation.cost_details or {} - self.environment = observation.environment - self.latency = observation.latency - self.time_to_first_token = observation.time_to_first_token - self.model_id = observation.model_id - self.input_price = _as_float(observation.input_price) - self.output_price = _as_float(observation.output_price) - self.total_price = _as_float(observation.total_price) - self.calculated_input_cost = self.cost_details.get("input") - self.calculated_output_cost = self.cost_details.get("output") - self.calculated_total_cost = observation.total_cost or self.cost_details.get( - "total" - ) - self.usage = SimpleNamespace( - input=self.usage_details.get("input", 0), - output=self.usage_details.get("output", 0), - total=self.usage_details.get("total", 0), - unit=self.usage_details.get("unit"), - input_cost=self.calculated_input_cost, - output_cost=self.calculated_output_cost, - total_cost=self.calculated_total_cost, - ) - - -class V4TraceView: - """Observation-backed trace view used by SDK behavior tests.""" - - def __init__( - self, trace_id: str, observations: list[V4ObservationView], scores: list - ): - if not observations: - raise NotFoundError(body={"error": "LangfuseNotFoundError"}) - - observations = list( - {observation.id: observation for observation in observations}.values() - ) - observations.sort(key=lambda observation: observation.start_time) - root = next( - ( - observation - for observation in observations - if observation.is_root_observation is True - ), - None, - ) - if root is None: - root = next( - ( - observation - for observation in observations - if observation.parent_observation_id is None - ), - observations[0], - ) - - source = root._observation - - def first_populated(attribute: str, fallback: Any = None) -> Any: - return next( - ( - value - for observation in observations - if (value := getattr(observation._observation, attribute, None)) - ), - fallback, - ) - - self.id = trace_id - self.timestamp = root.start_time - self.name = first_populated("trace_name", root.name) - self.input = root.input - self.output = root.output - self.session_id = first_populated("session_id") - self.release = first_populated("release") - self.version = root.version - self.user_id = first_populated("user_id") - self.metadata = {} - for observation in observations: - self.metadata.update(observation.metadata) - self.tags = list( - dict.fromkeys( - tag - for observation in observations - for tag in observation._observation.tags - ) - ) - self.public = bool(source.public) - self.environment = source.environment - self.observations = observations - self.scores = scores - - -class _V4ObservationsClient: - def __init__(self, client: Any): - self._client = client - - def get(self, observation_id: str) -> V4ObservationView: - response = self.get_many( - filter=json.dumps( - [ - { - "type": "string", - "column": "id", - "operator": "=", - "value": observation_id, - } - ] - ), - limit=1, - ) - if not response.data: - raise NotFoundError(body={"error": "LangfuseNotFoundError"}) - - return response.data[0] - - def get_many( - self, - *, - fields: str | None = None, - limit: int | None = None, - cursor: str | None = None, - page: int | None = None, - name: str | None = None, - user_id: str | None = None, - type: str | None = None, - trace_id: str | None = None, - parent_observation_id: str | None = None, - environment: str | Iterable[str] | None = None, - version: str | None = None, - filter: str | None = None, - **_: Any, - ) -> SimpleNamespace: - from_start_time, to_start_time = _time_bounds() - requested_page = page or 1 - response = None - current_cursor = cursor - fetched_page = 0 - for fetched_page in range(1, requested_page + 1): - response = self._client.observations.get_many( - fields=fields or OBSERVATION_FIELDS, - limit=limit, - cursor=current_cursor, - name=name, - user_id=user_id, - type=type, - trace_id=trace_id, - parent_observation_id=parent_observation_id, - environment=environment, - version=version, - filter=filter, - from_start_time=from_start_time, - to_start_time=to_start_time, - ) - current_cursor = response.meta.cursor - if current_cursor is None and fetched_page < requested_page: - break - - if response is None or fetched_page < requested_page: - return SimpleNamespace( - data=[], - meta=SimpleNamespace( - cursor=None, - total_items=0, - total_pages=requested_page - 1, - page=requested_page, - limit=limit or 50, - ), - ) - - data = [V4ObservationView(observation) for observation in response.data] - return SimpleNamespace( - data=data, - meta=SimpleNamespace( - cursor=response.meta.cursor, - total_items=len(data), - total_pages=1, - page=requested_page, - limit=limit or 50, - ), - ) - - -class _V4TraceClient: - def __init__(self, client: Any, observations: _V4ObservationsClient): - self._client = client - self._observations = observations - - def _scores(self, trace_id: str) -> list[Any]: - response = self._client.scores_v3.get_many_v3( - trace_id=trace_id, - fields="details,subject,annotation", - limit=100, - ) - return [_score_view(score) for score in response.data] - - def get(self, trace_id: str | None = None, **kwargs: Any) -> V4TraceView: - resolved_trace_id = trace_id or kwargs["trace_id"] - response = self._observations.get_many(trace_id=resolved_trace_id, limit=1000) - return V4TraceView( - resolved_trace_id, - response.data, - self._scores(resolved_trace_id), - ) - - def list( - self, - *, - name: str | None = None, - session_id: str | None = None, - user_id: str | None = None, - filter: str | None = None, - limit: int | None = None, - page: int | None = None, - **_: Any, - ) -> SimpleNamespace: - filters: list[dict[str, Any]] = [] - if filter: - filters.extend(json.loads(filter)) - if name is not None: - filters.append( - { - "type": "string", - "column": "traceName", - "operator": "=", - "value": name, - } - ) - if session_id is not None: - filters.append( - { - "type": "string", - "column": "sessionId", - "operator": "=", - "value": session_id, - } - ) - - response = self._observations.get_many( - user_id=user_id, - filter=json.dumps(filters) if filters else None, - limit=1000, - ) - observations_by_trace: dict[str, list[V4ObservationView]] = {} - for observation in response.data: - if observation.trace_id is not None: - observations_by_trace.setdefault(observation.trace_id, []).append( - observation - ) - - traces = [ - V4TraceView(trace_id, observations, self._scores(trace_id)) - for trace_id, observations in observations_by_trace.items() - ] - traces.sort(key=lambda trace: trace.timestamp, reverse=True) - - requested_limit = limit or 50 - requested_page = page or 1 - start = (requested_page - 1) * requested_limit - data = traces[start : start + requested_limit] - total_items = len(traces) - total_pages = (total_items + requested_limit - 1) // requested_limit - return SimpleNamespace( - data=data, - meta=SimpleNamespace( - page=requested_page, - limit=requested_limit, - total_items=total_items, - total_pages=total_pages, - ), - ) - - -class _V4SessionsClient: - def __init__(self, traces: _V4TraceClient): - self._traces = traces - - def list( - self, *, limit: int | None = None, page: int | None = None - ) -> SimpleNamespace: - traces = self._traces.list(limit=1000).data - sessions: dict[str, list[V4TraceView]] = {} - for trace in traces: - if trace.session_id is not None: - sessions.setdefault(trace.session_id, []).append(trace) - - data = [ - SimpleNamespace(id=session_id, traces=session_traces) - for session_id, session_traces in sessions.items() - ] - requested_limit = limit or 50 - requested_page = page or 1 - start = (requested_page - 1) * requested_limit - page_data = data[start : start + requested_limit] - total_items = len(data) - return SimpleNamespace( - data=page_data, - meta=SimpleNamespace( - page=requested_page, - limit=requested_limit, - total_items=total_items, - total_pages=(total_items + requested_limit - 1) // requested_limit, - ), - ) - - -class _V4ScoresClient: - def __init__(self, client: Any): - self._client = client - - def get_by_id(self, score_id: str) -> Any: - response = self._client.scores_v3.get_many_v3( - id=score_id, - fields="details,subject,annotation", - limit=1, - ) - if not response.data: - raise NotFoundError(body={"error": "LangfuseNotFoundError"}) - - return _score_view(response.data[0]) - - -class _V4DatasetsClient: - def __init__(self, client: Any): - self._client = client - self._datasets = client.datasets - - def __getattr__(self, name: str) -> Any: - return getattr(self._datasets, name) - - def get_run(self, dataset_name: str, run_name: str) -> Any: - dataset = self._datasets.get(dataset_name) - from_start_time, to_start_time = _time_bounds() - response = self._client.experiments.list( - from_start_time=from_start_time, - to_start_time=to_start_time, - dataset_id=dataset.id, - name=run_name, - fields="core,metadata,scores", - limit=1, - ) - if not response.data: - raise NotFoundError(body={"error": "LangfuseNotFoundError"}) - - return response.data[0] - - def get_runs(self, dataset_name: str) -> SimpleNamespace: - dataset = self._datasets.get(dataset_name) - from_start_time, to_start_time = _time_bounds() - return self._client.experiments.list( - from_start_time=from_start_time, - to_start_time=to_start_time, - dataset_id=dataset.id, - fields="core,metadata,scores", - limit=100, - ) - - -class _V4ExperimentItemsClient: - def __init__(self, client: Any): - self._client = client - - def list(self, *, dataset_id: str, run_name: str) -> Any: - from_start_time, to_start_time = _time_bounds() - experiments = self._client.experiments.list( - from_start_time=from_start_time, - to_start_time=to_start_time, - dataset_id=dataset_id, - name=run_name, - limit=1, - ) - if not experiments.data: - raise NotFoundError(body={"error": "LangfuseNotFoundError"}) - - return self._client.experiments.list_items( - from_start_time=from_start_time, - to_start_time=to_start_time, - experiment_id=experiments.data[0].id, - fields=("core,dataset,io,metadata,itemMetadata,experimentMetadata,scores"), - limit=100, - ) - - -class V4TestAPI: - """Expose v4-native read paths behind the test suite's existing API helper.""" - - def __init__(self, client: Any): - self._client = client - self.observations = _V4ObservationsClient(client) - self.trace = _V4TraceClient(client, self.observations) - self.legacy = SimpleNamespace(observations_v1=self.observations) - self.sessions = _V4SessionsClient(self.trace) - self.scores = _V4ScoresClient(client) - self.datasets = _V4DatasetsClient(client) - self.dataset_run_items = _V4ExperimentItemsClient(client) - - def __getattr__(self, name: str) -> Any: - return getattr(self._client, name) diff --git a/tests/unit/test_additional_headers_simple.py b/tests/unit/test_additional_headers_simple.py index 32664810a..dd843b35a 100644 --- a/tests/unit/test_additional_headers_simple.py +++ b/tests/unit/test_additional_headers_simple.py @@ -187,7 +187,6 @@ def test_span_processor_has_additional_headers_in_otel_exporter(self): assert "Authorization" in exporter._headers assert "x-langfuse-sdk-name" in exporter._headers assert "x-langfuse-public-key" in exporter._headers - assert exporter._headers["x-langfuse-ingestion-version"] == "4" # Check that our override worked assert exporter._headers["X-Override-Default"] == "override-value" @@ -211,7 +210,6 @@ def test_span_processor_none_additional_headers_works(self): assert "Authorization" in exporter._headers assert "x-langfuse-sdk-name" in exporter._headers assert "x-langfuse-public-key" in exporter._headers - assert exporter._headers["x-langfuse-ingestion-version"] == "4" def test_span_processor_uses_custom_span_exporter_when_provided(self): """Test that a custom exporter bypasses the default OTLP exporter construction.""" diff --git a/tests/unit/test_e2e_support.py b/tests/unit/test_e2e_support.py index 32da136ed..8320bd2fe 100644 --- a/tests/unit/test_e2e_support.py +++ b/tests/unit/test_e2e_support.py @@ -1,71 +1,9 @@ -from datetime import datetime, timedelta, timezone from types import SimpleNamespace from langfuse.api.commons.errors.not_found_error import NotFoundError from tests.support.api_wrapper import LangfuseAPI as SupportLangfuseAPI from tests.support.retry import retry_until_ready from tests.support.utils import get_api, wait_for_trace -from tests.support.v4_api import V4TraceView - - -def test_v4_trace_view_merges_metadata_from_all_observations(): - started_at = datetime.now(timezone.utc) - - def observation( - *, - observation_id, - metadata, - start_time, - is_root_observation, - parent_observation_id, - ): - return SimpleNamespace( - id=observation_id, - name=observation_id, - start_time=start_time, - is_root_observation=is_root_observation, - parent_observation_id=parent_observation_id, - input=None, - output=None, - version=None, - metadata=metadata, - _observation=SimpleNamespace( - trace_name="experiment-trace", - session_id=None, - release=None, - user_id=None, - tags=[], - public=False, - environment="default", - ), - ) - - trace = V4TraceView( - "trace-123", - [ - observation( - observation_id="root", - metadata={"experiment_name": "metadata-test"}, - start_time=started_at, - is_root_observation=True, - parent_observation_id=None, - ), - observation( - observation_id="external-child", - metadata={"mode": "offline", "job_name": "agent-eval/PR-4"}, - start_time=started_at + timedelta(milliseconds=1), - is_root_observation=False, - parent_observation_id="root", - ), - ], - [], - ) - - assert trace.metadata == { - "experiment_name": "metadata-test", - "mode": "offline", - "job_name": "agent-eval/PR-4", - } def test_get_api_retries_not_found(monkeypatch): @@ -91,7 +29,6 @@ class FakeClient: trace = FakeTraceService() monkeypatch.setattr("tests.support.utils.LangfuseAPI", lambda **_: FakeClient()) - monkeypatch.setattr("tests.support.utils.V4TestAPI", lambda client: client) trace = get_api().trace.get("trace-123") @@ -117,7 +54,6 @@ class FakeClient: trace = FakeTraceService() monkeypatch.setattr("tests.support.utils.LangfuseAPI", lambda **_: FakeClient()) - monkeypatch.setattr("tests.support.utils.V4TestAPI", lambda client: client) response = get_api().trace.list(name="ready-trace") @@ -137,7 +73,6 @@ class FakeClient: trace = FakeTraceService() monkeypatch.setattr("tests.support.utils.LangfuseAPI", lambda **_: FakeClient()) - monkeypatch.setattr("tests.support.utils.V4TestAPI", lambda client: client) response = get_api(retry=False).trace.list(name="missing-trace") @@ -150,26 +85,30 @@ def test_raw_api_wrapper_retries_not_found_payload(monkeypatch): attempts = {"count": 0} - class FakeTraceService: - def get(self, trace_id): - attempts["count"] += 1 - if attempts["count"] < 3: - raise NotFoundError( - body={ - "error": "LangfuseNotFoundError", - "message": "Trace trace-123 not found within authorized project", - } - ) + class FakeResponse: + def __init__(self, status_code, payload): + self.status_code = status_code + self._payload = payload + self.headers = {} - return {"id": trace_id, "observations": []} + def json(self): + return self._payload - class FakeClient: - trace = FakeTraceService() + def fake_get(*args, **kwargs): + attempts["count"] += 1 - monkeypatch.setattr("tests.support.api_wrapper.get_api", lambda **_: FakeClient()) - monkeypatch.setattr( - SupportLangfuseAPI, "_trace_json", staticmethod(lambda trace: trace) - ) + if attempts["count"] < 3: + return FakeResponse( + 404, + { + "error": "LangfuseNotFoundError", + "message": "Trace trace-123 not found within authorized project", + }, + ) + + return FakeResponse(200, {"id": "trace-123", "observations": []}) + + monkeypatch.setattr("tests.support.api_wrapper.httpx.get", fake_get) api = SupportLangfuseAPI(username="user", password="pass", base_url="http://test") trace = api.get_trace("trace-123") @@ -192,7 +131,6 @@ class FakeClient: trace = FakeTraceService() monkeypatch.setattr("tests.support.utils.LangfuseAPI", lambda **_: FakeClient()) - monkeypatch.setattr("tests.support.utils.V4TestAPI", lambda client: client) trace = wait_for_trace( "trace-123", is_result_ready=lambda trace: len(trace["observations"]) == 3 diff --git a/tests/unit/test_propagate_attributes.py b/tests/unit/test_propagate_attributes.py index 212f1b663..cdf4f9351 100644 --- a/tests/unit/test_propagate_attributes.py +++ b/tests/unit/test_propagate_attributes.py @@ -2706,13 +2706,11 @@ def task_with_child_spans(*, item, **kwargs): LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT, ) - # Experiment context is propagated to children. - self.verify_span_attribute( + # Root-only attributes should NOT be present on children + self.verify_missing_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION, - "Test experiment description", ) - # Item payload attributes remain root-only. self.verify_missing_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_EXPECTED_OUTPUT, @@ -2785,9 +2783,33 @@ def test_experiment_run_metadata_overrides_item_metadata( ) def test_experiment_attributes_propagate_with_dataset( - self, langfuse_client, memory_exporter + self, langfuse_client, memory_exporter, monkeypatch ): """Test experiment attribute propagation with Langfuse dataset.""" + created_run_items = [] + + # Mock the sync API used by run_experiment to create dataset run items + def mock_create_dataset_run_item(*args, **kwargs): + from langfuse.api import DatasetRunItem + + created_run_items.append(kwargs) + return DatasetRunItem( + id="mock-run-item-id", + dataset_run_id="mock-dataset-run-id-123", + dataset_run_name=kwargs.get("run_name", "Dataset Test"), + dataset_item_id=kwargs.get("dataset_item_id", "mock-item-id"), + trace_id="mock-trace-id", + observation_id=kwargs.get("observation_id"), + created_at=datetime.now(), + updated_at=datetime.now(), + ) + + monkeypatch.setattr( + langfuse_client.api.dataset_run_items, + "create", + mock_create_dataset_run_item, + ) + # Create a mock dataset with items dataset_id = "test-dataset-id-456" dataset_item_id = "test-dataset-item-id-789" @@ -2849,7 +2871,10 @@ def task_with_children(*, item, **kwargs): root_spans = self.get_spans_by_name(memory_exporter, "experiment-item-run") assert len(root_spans) >= 1, "Should have at least 1 root span" first_root = root_spans[0] - assert result.experiment_id == result.dataset_run_id + task_span = self.get_span_by_name(memory_exporter, "experiment-item-task") + assert result.experiment_id == "mock-dataset-run-id-123" + assert len(created_run_items) == 1 + assert created_run_items[0]["observation_id"] == task_span["span_id"] # Root-only attributes should be on root self.verify_span_attribute( @@ -2946,13 +2971,11 @@ def task_with_children(*, item, **kwargs): LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT, ) - # Experiment context is propagated to children. - self.verify_span_attribute( + # Root-only attributes should NOT be present on children + self.verify_missing_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION, - "Dataset experiment description", ) - # Item payload attributes remain root-only. self.verify_missing_attribute( child_span, LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_EXPECTED_OUTPUT, @@ -3031,11 +3054,10 @@ def task_with_nested_spans(*, item, **kwargs): LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT, ) - # Experiment context is propagated to nested spans. - self.verify_span_attribute( + # Root-only attributes should NOT be present + self.verify_missing_attribute( span_data, LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION, - "Nested test", ) self.verify_missing_attribute( span_data,