From e007c1480d91fcd5b2d0f9e4cec6e61f794bfba2 Mon Sep 17 00:00:00 2001 From: Leroyyyyyyyyy <150530443+Leroyyyyyyyyy@users.noreply.github.com> Date: Wed, 19 Aug 2026 22:01:58 +0800 Subject: [PATCH] fix(otel): share one AnyValue decoder across all OTLP paths Three AnyValue decoders had drifted apart. extraction's flatten_otlp_attributes silently dropped array/kvlist/bytes attributes, and the OTLP JSON loader json.dumps()'d the raw proto wrapper instead of decoding it. gen_ai.response.finish_reasons therefore surfaced as '{"values": [{"stringValue": "stop"}]}' on one path and vanished on another, where both should yield ["stop"]. Move the already-correct recursive decoder out of api/otlp_processing.py into a dependency-free module and route all three call sites through it. The new module imports only the standard library, so extraction, loader.otlp and api.otlp_processing can all share it without creating an import cycle. bytesValue is returned unchanged: MessageToDict base64-encodes protobuf bytes fields, so callers already receive a str today. Decoding it here would change existing behaviour, which is out of scope for this fix. Adds coverage for array/kvlist/bytes attributes, which previously had none on either path. Fixes #173 --- src/agentevals/api/otlp_processing.py | 30 +--------- src/agentevals/extraction.py | 21 +++---- src/agentevals/loader/otlp.py | 21 +------ src/agentevals/otlp_anyvalue.py | 80 +++++++++++++++++++++++++++ tests/test_extraction.py | 52 +++++++++++++++++ tests/test_otlp_loader.py | 60 ++++++++++++++++++++ 6 files changed, 204 insertions(+), 60 deletions(-) create mode 100644 src/agentevals/otlp_anyvalue.py diff --git a/src/agentevals/api/otlp_processing.py b/src/agentevals/api/otlp_processing.py index 948e31b..613510e 100644 --- a/src/agentevals/api/otlp_processing.py +++ b/src/agentevals/api/otlp_processing.py @@ -15,6 +15,7 @@ ) from ..extraction import flatten_otlp_attributes +from ..otlp_anyvalue import decode_any_value from ..trace_attrs import ( OTEL_GENAI_CONVERSATION_ID, OTEL_GENAI_INPUT_MESSAGES, @@ -310,37 +311,12 @@ def _convert_otlp_log_record(log_record: dict) -> dict | None: return result -def _parse_otlp_any_value(value_obj: dict): - """Recursively parse an OTLP AnyValue to native Python types. - - Handles the full AnyValue union: stringValue, intValue, doubleValue, - boolValue, kvlistValue (→ dict), arrayValue (→ list), bytesValue. - """ - if "stringValue" in value_obj: - return value_obj["stringValue"] - if "intValue" in value_obj: - return int(value_obj["intValue"]) - if "doubleValue" in value_obj: - return float(value_obj["doubleValue"]) - if "boolValue" in value_obj: - return value_obj["boolValue"] - if "kvlistValue" in value_obj: - kv = value_obj["kvlistValue"] - return {item.get("key", ""): _parse_otlp_any_value(item.get("value", {})) for item in kv.get("values", [])} - if "arrayValue" in value_obj: - arr = value_obj["arrayValue"] - return [_parse_otlp_any_value(v) for v in arr.get("values", [])] - if "bytesValue" in value_obj: - return value_obj["bytesValue"] - return value_obj - - def _parse_otlp_body(body_raw: dict) -> dict | str: """Parse OTLP log record body value. Top-level stringValue bodies are JSON-decoded (Strands-style logs store message content as JSON strings). All other AnyValue types are parsed - recursively via ``_parse_otlp_any_value`` (handles the nested kvlistValue / + recursively via ``decode_any_value`` (handles the nested kvlistValue / arrayValue structures used by the OpenAI instrumentor). """ if "stringValue" in body_raw: @@ -351,4 +327,4 @@ def _parse_otlp_body(body_raw: dict) -> dict | str: return json.loads(raw) except (json.JSONDecodeError, TypeError): return raw - return _parse_otlp_any_value(body_raw) + return decode_any_value(body_raw) diff --git a/src/agentevals/extraction.py b/src/agentevals/extraction.py index 141f230..efcc9d6 100644 --- a/src/agentevals/extraction.py +++ b/src/agentevals/extraction.py @@ -17,6 +17,7 @@ from typing import Any, Protocol, TypedDict, TypeVar from .loader.base import Span, Trace +from .otlp_anyvalue import decode_attributes from .trace_attrs import ( ADK_LLM_REQUEST, ADK_LLM_RESPONSE, @@ -523,20 +524,12 @@ def is_invocation_span(span: Span) -> bool: def flatten_otlp_attributes(attrs_list: list[dict]) -> dict[str, Any]: - """Convert OTLP attributes array [{key, value: {stringValue|...}}] to flat dict.""" - result: dict[str, Any] = {} - for attr in attrs_list: - key = attr.get("key", "") - value_obj = attr.get("value", {}) - if "stringValue" in value_obj: - result[key] = value_obj["stringValue"] - elif "intValue" in value_obj: - result[key] = int(value_obj["intValue"]) - elif "doubleValue" in value_obj: - result[key] = float(value_obj["doubleValue"]) - elif "boolValue" in value_obj: - result[key] = value_obj["boolValue"] - return result + """Convert OTLP attributes array [{key, value: {stringValue|...}}] to flat dict. + + Delegates to the shared ``AnyValue`` decoder so array/kvlist/bytes + attributes survive instead of being dropped. + """ + return decode_attributes(attrs_list) # --------------------------------------------------------------------------- diff --git a/src/agentevals/loader/otlp.py b/src/agentevals/loader/otlp.py index ef26cb2..26a11ea 100644 --- a/src/agentevals/loader/otlp.py +++ b/src/agentevals/loader/otlp.py @@ -5,6 +5,7 @@ import json import logging +from ..otlp_anyvalue import decode_attributes from ..trace_attrs import ( OTEL_GENAI_INPUT_MESSAGES, OTEL_GENAI_OUTPUT_MESSAGES, @@ -192,25 +193,7 @@ def _extract_attributes(self, attrs) -> dict: if isinstance(attrs, dict): return self._flatten_nested_dict(attrs) - result = {} - for attr in attrs: - key = attr.get("key", "") - value_obj = attr.get("value", {}) - - if "stringValue" in value_obj: - result[key] = value_obj["stringValue"] - elif "intValue" in value_obj: - result[key] = int(value_obj["intValue"]) - elif "doubleValue" in value_obj: - result[key] = float(value_obj["doubleValue"]) - elif "boolValue" in value_obj: - result[key] = value_obj["boolValue"] - elif "arrayValue" in value_obj: - result[key] = json.dumps(value_obj["arrayValue"]) - elif "kvlistValue" in value_obj: - result[key] = json.dumps(value_obj["kvlistValue"]) - - return result + return decode_attributes(attrs) @staticmethod def _flatten_nested_dict(d: dict, prefix: str = "") -> dict: diff --git a/src/agentevals/otlp_anyvalue.py b/src/agentevals/otlp_anyvalue.py new file mode 100644 index 0000000..9f498d3 --- /dev/null +++ b/src/agentevals/otlp_anyvalue.py @@ -0,0 +1,80 @@ +"""Shared decoder for the OTLP ``AnyValue`` union. + +OTLP encodes every attribute value, log body and nested element as an +``AnyValue``: a one-of wrapper such as ``{"stringValue": "chat"}`` or +``{"arrayValue": {"values": [...]}}``. The protobuf receiver (via +``MessageToDict``) and OTLP/JSON payloads deliver that same dict shape, so +every consumer needs identical decoding rules. + +This module is deliberately dependency-free: importing only the standard +library lets ``extraction``, ``loader.otlp`` and ``api.otlp_processing`` all +use it without creating an import cycle. +""" + +from __future__ import annotations + +from typing import Any + +ANY_VALUE_FIELDS = ( + "stringValue", + "intValue", + "doubleValue", + "boolValue", + "kvlistValue", + "arrayValue", + "bytesValue", +) + + +def decode_any_value(value_obj: dict) -> Any: + """Recursively decode an OTLP ``AnyValue`` to a native Python value. + + Handles the full union: stringValue, intValue (OTLP sends it as a + string), doubleValue, boolValue, kvlistValue (→ dict), arrayValue + (→ list), bytesValue. + + ``bytesValue`` is returned unchanged. ``MessageToDict`` base64-encodes + protobuf bytes fields and OTLP/JSON does the same, so callers already + receive a str; decoding it here would change the value they see today. + + A value carrying none of the union fields is returned as-is. + """ + if "stringValue" in value_obj: + return value_obj["stringValue"] + if "intValue" in value_obj: + return int(value_obj["intValue"]) + if "doubleValue" in value_obj: + return float(value_obj["doubleValue"]) + if "boolValue" in value_obj: + return value_obj["boolValue"] + if "kvlistValue" in value_obj: + kv = value_obj["kvlistValue"] + return {item.get("key", ""): decode_any_value(item.get("value", {})) for item in kv.get("values", [])} + if "arrayValue" in value_obj: + arr = value_obj["arrayValue"] + return [decode_any_value(v) for v in arr.get("values", [])] + if "bytesValue" in value_obj: + return value_obj["bytesValue"] + return value_obj + + +def is_any_value(value_obj: dict) -> bool: + """Return True when *value_obj* carries one of the ``AnyValue`` fields.""" + for field in ANY_VALUE_FIELDS: + if field in value_obj: + return True + return False + + +def decode_attributes(attrs_list: list[dict]) -> dict[str, Any]: + """Decode an OTLP attributes array to a flat ``{key: value}`` dict. + + Entries whose value carries no ``AnyValue`` field are skipped, matching + the behaviour every call site had before they shared this decoder. + """ + result: dict[str, Any] = {} + for attr in attrs_list: + value_obj = attr.get("value", {}) + if is_any_value(value_obj): + result[attr.get("key", "")] = decode_any_value(value_obj) + return result diff --git a/tests/test_extraction.py b/tests/test_extraction.py index 96544f7..ec3591f 100644 --- a/tests/test_extraction.py +++ b/tests/test_extraction.py @@ -400,6 +400,58 @@ def test_mixed_types(self): ) assert result == {"str": "hello", "num": 3.14, "flag": True} + def test_array_value(self): + result = flatten_otlp_attributes( + [ + { + "key": "gen_ai.response.finish_reasons", + "value": {"arrayValue": {"values": [{"stringValue": "stop"}]}}, + }, + ] + ) + assert result == {"gen_ai.response.finish_reasons": ["stop"]} + + def test_kvlist_value(self): + result = flatten_otlp_attributes( + [ + { + "key": "gen_ai.request.params", + "value": { + "kvlistValue": { + "values": [ + {"key": "temperature", "value": {"doubleValue": 0.7}}, + {"key": "stream", "value": {"boolValue": False}}, + ] + } + }, + }, + ] + ) + assert result == {"gen_ai.request.params": {"temperature": 0.7, "stream": False}} + + def test_array_of_kvlist(self): + """Tool calls arrive as an arrayValue of kvlistValue.""" + result = flatten_otlp_attributes( + [ + { + "key": "gen_ai.tool.calls", + "value": { + "arrayValue": { + "values": [ + {"kvlistValue": {"values": [{"key": "name", "value": {"stringValue": "get_weather"}}]}}, + ] + } + }, + }, + ] + ) + assert result == {"gen_ai.tool.calls": [{"name": "get_weather"}]} + + def test_bytes_value(self): + """MessageToDict base64-encodes bytes fields, so the decoder sees a str.""" + result = flatten_otlp_attributes([{"key": "payload", "value": {"bytesValue": "AP9oaQ=="}}]) + assert result == {"payload": "AP9oaQ=="} + def test_empty(self): assert flatten_otlp_attributes([]) == {} diff --git a/tests/test_otlp_loader.py b/tests/test_otlp_loader.py index c3a3428..402ff25 100644 --- a/tests/test_otlp_loader.py +++ b/tests/test_otlp_loader.py @@ -258,6 +258,66 @@ def test_load_from_dict_empty_resource_spans(self): assert traces == [] +class TestAnyValueAttributes: + """Attributes carrying the full OTLP AnyValue union (array / kvlist / bytes).""" + + @staticmethod + def _load_span_with(attribute): + loader = OtlpJsonLoader() + data = { + "resourceSpans": [ + { + "resource": {"attributes": []}, + "scopeSpans": [ + { + "scope": {"name": "test-scope"}, + "spans": [ + { + "traceId": "t1", + "spanId": "s1", + "name": "test", + "startTimeUnixNano": "1000000000", + "endTimeUnixNano": "2000000000", + "attributes": [attribute], + } + ], + } + ], + } + ], + } + return loader.load_from_dict(data)[0].all_spans[0] + + def test_array_value(self): + span = self._load_span_with( + { + "key": "gen_ai.response.finish_reasons", + "value": {"arrayValue": {"values": [{"stringValue": "stop"}]}}, + } + ) + assert span.tags["gen_ai.response.finish_reasons"] == ["stop"] + + def test_kvlist_value(self): + span = self._load_span_with( + { + "key": "gen_ai.request.params", + "value": { + "kvlistValue": { + "values": [ + {"key": "temperature", "value": {"doubleValue": 0.7}}, + {"key": "stream", "value": {"boolValue": False}}, + ] + } + }, + } + ) + assert span.tags["gen_ai.request.params"] == {"temperature": 0.7, "stream": False} + + def test_bytes_value(self): + span = self._load_span_with({"key": "payload", "value": {"bytesValue": "AP9oaQ=="}}) + assert span.tags["payload"] == "AP9oaQ==" + + class TestFlatDictAttributes: """Tests for flat dict attribute format (e.g. from simplified producers)."""