From ecbd5891d3b9ad0e924efcb0e98e462b1e98286e Mon Sep 17 00:00:00 2001 From: Moritz Althaus Date: Fri, 18 Sep 2026 07:56:59 +0200 Subject: [PATCH] fix(scores): apply trace sampling to scores again create_score passed camelCase kwargs to the v4 ScoreBody, which takes snake_case field names and keeps unknown kwargs as extras. body.trace_id was always None, so add_score_task never sampled a score out. --- langfuse/_client/client.py | 12 ++++----- .../_task_manager/score_ingestion_consumer.py | 7 ++++- tests/unit/test_otel.py | 20 ++++++++++++++ tests/unit/test_resource_manager.py | 27 +++++++++++++++++++ 4 files changed, 59 insertions(+), 7 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 42d861fd4..fe0605a6e 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -2022,15 +2022,15 @@ def create_score( try: new_body = ScoreBody( id=score_id, - sessionId=session_id, - datasetRunId=dataset_run_id, - traceId=trace_id, - observationId=observation_id, + session_id=session_id, + dataset_run_id=dataset_run_id, + trace_id=trace_id, + observation_id=observation_id, name=name, value=value, - dataType=data_type, # type: ignore + data_type=data_type, # type: ignore comment=comment, - configId=config_id, + config_id=config_id, environment=environment or self._environment, metadata=metadata, ) diff --git a/langfuse/_task_manager/score_ingestion_consumer.py b/langfuse/_task_manager/score_ingestion_consumer.py index 1dce00f80..27eb9588b 100644 --- a/langfuse/_task_manager/score_ingestion_consumer.py +++ b/langfuse/_task_manager/score_ingestion_consumer.py @@ -11,6 +11,7 @@ from langfuse._utils.parse_error import handle_exception from langfuse._utils.request import APIError, LangfuseClient from langfuse._utils.serializer import EventSerializer +from langfuse.api.core.serialization import convert_and_respect_annotation_metadata from langfuse.logger import langfuse_logger as logger from .._version import __version__ as langfuse_version @@ -78,7 +79,11 @@ def _next(self) -> list: # convert pydantic models to dicts if "body" in event and isinstance(event["body"], BaseModel): - event["body"] = event["body"].model_dump(exclude_none=True) + event["body"] = convert_and_respect_annotation_metadata( + object_=event["body"].model_dump(exclude_none=True), + annotation=type(event["body"]), + direction="write", + ) item_size = self._get_item_size(event) diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index 46a085a71..4343c40e4 100644 --- a/tests/unit/test_otel.py +++ b/tests/unit/test_otel.py @@ -1,6 +1,7 @@ import json from datetime import datetime from hashlib import sha256 +from queue import Queue from typing import List, Sequence import pytest @@ -1468,6 +1469,25 @@ def test_sampling(self, monkeypatch, tracer_provider, mock_processor_init): # Restore the original provider trace_api.set_tracer_provider(original_provider) + def test_score_sampling_follows_trace_sampling(self): + client = Langfuse( + public_key="test-public-key", + secret_key="test-secret-key", + base_url="http://test-host", + sample_rate=0.5, + ) + # Detach from the consumer thread so it cannot drain the queue mid-test. + queue = Queue() + client._resources._score_ingestion_queue = queue + + # The sampler keeps a trace when its low 64 bits fall below rate * 2^64. + client.create_score(name="sampled", value=1.0, trace_id="0" * 32) + client.create_score(name="dropped", value=1.0, trace_id="f" * 32) + client.create_score(name="session", value=1.0, session_id="session-1") + + names = [queue.get_nowait()["body"].name for _ in range(queue.qsize())] + assert names == ["sampled", "session"] + @pytest.mark.skip("Calling shutdown will pollute the global context") def test_shutdown_and_flush(self, langfuse_client, memory_exporter): """Test shutdown and flush operations.""" diff --git a/tests/unit/test_resource_manager.py b/tests/unit/test_resource_manager.py index f66a1e052..5bf1cbd39 100644 --- a/tests/unit/test_resource_manager.py +++ b/tests/unit/test_resource_manager.py @@ -14,6 +14,7 @@ from langfuse._task_manager.media_manager import MediaManager from langfuse._task_manager.media_upload_consumer import MediaUploadConsumer from langfuse._task_manager.score_ingestion_consumer import ScoreIngestionConsumer +from langfuse.api.ingestion.types.score_body import ScoreBody from langfuse.types import MaskOtelSpansResult @@ -156,6 +157,32 @@ def test_score_ingestion_consumer_pause_wakes_blocked_thread(): assert not consumer.is_alive() +def test_score_ingestion_consumer_serializes_body_by_alias(): + queue = Queue() + queue.put( + { + "type": "score-create", + "body": ScoreBody(id="s1", trace_id="t" * 32, name="quality", value=1.0), + } + ) + consumer = ScoreIngestionConsumer( + ingestion_queue=queue, + identifier=0, + client=Mock(), + public_key="pk-test", + flush_interval=0.01, + ) + + batch = consumer._next() + + assert batch[0]["body"] == { + "id": "s1", + "traceId": "t" * 32, + "name": "quality", + "value": 1.0, + } + + def test_media_upload_consumer_signal_shutdown_wakes_blocked_thread(): media_manager = MediaManager( api_client=Mock(),