From 87798513ab435383c2d7ea2327ea70b0c0511968 Mon Sep 17 00:00:00 2001 From: yucheng Date: Thu, 15 Oct 2026 07:28:25 -0700 Subject: [PATCH] fix(langfuse_otel): split string caller tags the way Langfuse reads them A caller tags value given as a comma list or a JSON array string (body metadata.tags or the langfuse_tags header) used to reach Langfuse raw, where ingestion split it into separate tags. Merging it with the derived tags kept it as one tag, so parse it the same way before the merge --- .../integrations/langfuse/langfuse_otel.py | 20 ++++++++++------ .../test_langfuse_otel_trace_fields.py | 24 ++++++++++++++++++- tests/unit/integrations/test_langfuse_otel.py | 17 ++++++++++--- 3 files changed, 50 insertions(+), 11 deletions(-) diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 7421225501f..9f5b19d8d02 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -54,6 +54,18 @@ def _preset_cache_key(kwargs: Mapping[str, object]) -> object: return litellm.cache.get_preset_cache_key_from_kwargs(**kwargs) +def _caller_tag_list(value: object) -> tuple[str, ...]: + """Caller tags as Langfuse ingestion reads a raw tags value: a list, a JSON array string, a comma list or one tag.""" + parsed: Final = ( + safe_json_loads(value, default=value) if isinstance(value, str) and value.strip().startswith("[") else value + ) + if isinstance(parsed, list): + return tuple(tag if isinstance(tag, str) else json.dumps(tag, default=str) for tag in parsed) + if isinstance(parsed, str) and parsed: + return tuple(tag.strip() for tag in parsed.split(",")) if "," in parsed else (parsed,) + return () + + class LangfuseOtelLogger(OpenTelemetry): def __init__(self, config=None, *args, **kwargs): # Prevent LangfuseOtelLogger from modifying global environment variables by constructing config manually @@ -183,13 +195,7 @@ class LangfuseOtelLogger(OpenTelemetry): expanded: Final = tuple(_default_tag(key) for key in default_tags) if isinstance(default_tags, list) else () candidates: Final = ( - ( - (caller_tags,) - if isinstance(caller_tags, str) - else tuple(tag for tag in caller_tags if isinstance(tag, str)) - if isinstance(caller_tags, list) - else () - ) + _caller_tag_list(caller_tags) + (tuple(tag for tag in request_tags if isinstance(tag, str)) if isinstance(request_tags, list) else ()) + tuple(tag for tag in expanded if tag is not None) ) diff --git a/tests/integration/observability/test_langfuse_otel_trace_fields.py b/tests/integration/observability/test_langfuse_otel_trace_fields.py index 54e41d5ad51..c4dcc55e503 100644 --- a/tests/integration/observability/test_langfuse_otel_trace_fields.py +++ b/tests/integration/observability/test_langfuse_otel_trace_fields.py @@ -34,10 +34,15 @@ class SpanRecord: attributes: Mapping[str, object] def tags(self) -> list[object]: + """The trace tags Langfuse ingestion reads from the raw attribute.""" raw: Final = self.attributes.get("langfuse.trace.tags") if isinstance(raw, tuple): return list(raw) - return json.loads(str(raw)) if raw is not None else [] + if not isinstance(raw, str) or not raw: + return [] + if raw.strip().startswith("["): + return json.loads(raw) + return [tag.strip() for tag in raw.split(",")] if "," in raw else [raw] def _attribute_value(value: AnyValue) -> object: @@ -276,6 +281,23 @@ def test_caller_trace_name_and_tags_win_over_derived_values(v1: Rig) -> None: assert span.attributes.get("langfuse.trace.input") == span.attributes.get("langfuse.observation.input") +@pytest.mark.parametrize( + ("metadata", "headers", "expected"), + [ + pytest.param({"tags": "c1, c2"}, None, ["c1", "c2"], id="body_comma_string"), + pytest.param({"tags": '["j1", "j2"]'}, None, ["j1", "j2"], id="body_json_array_string"), + pytest.param(None, {"langfuse_tags": "h1,h2"}, ["h1", "h2"], id="header_comma_string"), + ], +) +def test_string_caller_tags_keep_the_tags_langfuse_split_them_into( + v1: Rig, metadata: Mapping[str, object] | None, headers: Mapping[str, str] | None, expected: list[str] +) -> None: + marker: Final = "lit8281-strtags-" + uuid.uuid4().hex + _chat(v1, marker, metadata=metadata, headers=headers) + span: Final = v1.sink.one(marker) + assert span.tags()[: len(expected)] == expected, span.attributes + + def test_trace_name_header_wins_over_the_derived_name(v1: Rig) -> None: marker: Final = "lit8281-header-" + uuid.uuid4().hex _chat(v1, marker, headers={"langfuse_trace_name": "header-trace"}) diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index 294a5319db1..a202573a7c8 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -1240,14 +1240,25 @@ class TestDerivedTraceFields: ) assert json.loads(attributes["langfuse.trace.tags"]) == ["cache_key:explicit-key"] - def test_caller_tags_string_is_a_single_tag(self): + @pytest.mark.parametrize( + ("caller_tags", "expected"), + [ + ("single", ["single", "req"]), + ("h1, h2", ["h1", "h2", "req"]), + ('["j1", "j2"]', ["j1", "j2", "req"]), + (["x", 1], ["x", "1", "req"]), + ("", ["req"]), + ], + ) + def test_caller_tags_are_read_the_way_langfuse_reads_a_raw_tags_value(self, caller_tags, expected): attributes = _emitted( { "call_type": "acompletion", - "litellm_params": {"metadata": {"tags": "single"}}, + "litellm_params": {"metadata": {"tags": caller_tags}}, + "standard_logging_object": {"request_tags": ["req"]}, } ) - assert json.loads(attributes["langfuse.trace.tags"]) == ["single"] + assert json.loads(attributes["langfuse.trace.tags"]) == expected def test_no_tags_anywhere_leaves_trace_tags_unset(self): attributes = _emitted(