From f519f58e4d5a9ad2b43b02962e1845f25873e983 Mon Sep 17 00:00:00 2001 From: maxind Date: Thu, 30 Jul 2026 13:40:34 +0000 Subject: [PATCH] feat(util-genai): add per-signal context-scoped attributes Adds `set_context_scoped_attributes` so a caller can attach attributes to an OTel context and have GenAI telemetry emitted within it carry them, with each attribute declaring whether it applies to spans or to events. This bridges the layering of the GenAI ecosystem: an agentic framework knows which agent, workflow, or conversation is running, while the model-client instrumentation that emits the inference telemetry sits a layer below it and has no way to learn any of it. Span-targeted attributes are applied when the span starts, so they are visible to samplers. Metrics are deliberately not supported. Assisted-by: Claude Opus 5 --- .../.changelog/337.added | 1 + util/opentelemetry-util-genai/README.rst | 42 ++++ .../util/genai/_inference_invocation.py | 2 + .../opentelemetry/util/genai/_invocation.py | 11 ++ .../util/genai/context_attributes.py | 97 ++++++++++ .../tests/test_context_attributes.py | 182 ++++++++++++++++++ 6 files changed, 335 insertions(+) create mode 100644 util/opentelemetry-util-genai/.changelog/337.added create mode 100644 util/opentelemetry-util-genai/src/opentelemetry/util/genai/context_attributes.py create mode 100644 util/opentelemetry-util-genai/tests/test_context_attributes.py diff --git a/util/opentelemetry-util-genai/.changelog/337.added b/util/opentelemetry-util-genai/.changelog/337.added new file mode 100644 index 000000000..31fe97e7b --- /dev/null +++ b/util/opentelemetry-util-genai/.changelog/337.added @@ -0,0 +1 @@ +Add ``set_context_scoped_attributes`` for attaching attributes to an OpenTelemetry context so that GenAI telemetry emitted within it carries them, targeted per signal at spans or events. diff --git a/util/opentelemetry-util-genai/README.rst b/util/opentelemetry-util-genai/README.rst index 0a9f8f971..788b86c7f 100644 --- a/util/opentelemetry-util-genai/README.rst +++ b/util/opentelemetry-util-genai/README.rst @@ -12,6 +12,7 @@ Key Components - ``TelemetryHandler`` -- manages LLM invocation lifecycles (spans, metrics, events) - ``InferenceInvocation`` and message types (``Text``, ``Reasoning``, ``Blob``, etc.) -- structured data model for GenAI interactions - ``CompletionHook`` -- protocol for uploading content to external storage (built-in ``fsspec`` support) +- ``set_context_scoped_attributes`` -- attach attributes to a context so GenAI telemetry emitted within it carries them - Metrics -- ``gen_ai.client.operation.duration`` and ``gen_ai.client.token.usage`` histograms, plus the streaming timing histograms ``gen_ai.client.operation.time_to_first_chunk`` and ``gen_ai.client.operation.time_per_output_chunk`` @@ -24,6 +25,47 @@ See the module docstring in ``opentelemetry.util.genai.handler`` for usage examp including context manager and manual lifecycle patterns. +Context-scoped Attributes +------------------------- + +An agentic framework knows which agent, workflow, or conversation is running. +The model-client instrumentation that emits the inference telemetry sits a layer +below it and has no way to learn any of it -- there is no shared call path, and +attributes cannot be read back off a parent span. +``set_context_scoped_attributes`` bridges the two through the OpenTelemetry +context. + +Each attribute declares which signal it applies to, so content that is unsafe on +a sampled, widely-read span can still be recorded on the event:: + + from opentelemetry import context + from opentelemetry.util.genai.context_attributes import ( + set_context_scoped_attributes, + ) + + token = context.attach( + set_context_scoped_attributes( + span_attributes={"gen_ai.agent.name": "trip-planner"}, + log_attributes={"user.id": user_id}, + ) + ) + try: + client.chat.completions.create(...) # instrumented elsewhere + finally: + context.detach(token) + +- Attributes apply to GenAI telemetry emitted by this package only. Other + instrumentation, and telemetry the application emits directly, are unaffected. +- Attributes are never propagated out of the process. +- ``span_attributes`` are applied when the span starts, so they are visible to + samplers. Attributes an invocation sets itself take precedence. +- ``log_attributes`` apply to the ``gen_ai.client.inference.operation.details`` + event, which is the only event this package emits. +- Nested calls merge, with the inner call taking precedence for keys it sets. +- An invocation reads the context once, when it starts. +- Metrics are deliberately not supported, to avoid unbounded cardinality. + + Environment Variables --------------------- diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py index a77a1fe47..3eb5add8f 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py @@ -246,6 +246,8 @@ def _maybe_create_event(self) -> LogRecord | None: return None attributes = self._get_start_attributes() + if self._context_scoped_attributes.log: + attributes = {**self._context_scoped_attributes.log, **attributes} attributes.update(self._get_attributes()) attributes.update(self._get_message_attributes(for_span=False)) attributes.update(self.attributes) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py index 3be633294..1168e25fe 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py @@ -21,6 +21,10 @@ from opentelemetry.trace import Span, SpanKind, Tracer, set_span_in_context from opentelemetry.trace.status import Status, StatusCode from opentelemetry.util.genai.completion_hook import CompletionHook +from opentelemetry.util.genai.context_attributes import ( + _ContextScopedAttributes, + _get_context_scoped_attributes, +) from opentelemetry.util.genai.types import ( Error, ErrorTypeResolver, @@ -94,6 +98,7 @@ def __init__( self._request_stream: bool | None = None self._ttfc_seconds: float | None = None self._stream_last_chunk_at: float | None = None + self._context_scoped_attributes = _ContextScopedAttributes() def _start( self, attributes: dict[str, AttributeValue] | None = None @@ -103,6 +108,12 @@ def _start( Args: attributes: Initial span attributes available for sampling decisions. """ + self._context_scoped_attributes = _get_context_scoped_attributes() + if self._context_scoped_attributes.span: + attributes = { + **self._context_scoped_attributes.span, + **(attributes or {}), + } self.span = self._tracer.start_span( name=self._span_name, kind=self._span_kind, diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/context_attributes.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/context_attributes.py new file mode 100644 index 000000000..0db4b3f0d --- /dev/null +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/context_attributes.py @@ -0,0 +1,97 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Attach attributes to a context so GenAI telemetry emitted within it carries them. + +This lets a caller that knows something the instrumentation cannot -- which +agent is running, which user made the request -- add it to telemetry emitted +further down the stack. Each attribute targets a signal, so data that is unsafe +on a span can still go on the event. + +.. code-block:: python + + token = context.attach( + set_context_scoped_attributes( + span_attributes={"gen_ai.agent.name": "trip-planner"}, + log_attributes={"user.id": user_id}, + ) + ) + try: + client.chat.completions.create(...) # instrumented elsewhere + finally: + context.detach(token) + +Only telemetry from this package is affected, and attributes never leave the +process. Metrics are not supported, to avoid unbounded cardinality. + +The naming follows `OTEP 4931 +`_, +but this is narrower: that OTEP has the SDK stamp all telemetry and gates it +per signal on the provider. +""" + +from __future__ import annotations + +from types import MappingProxyType +from typing import Mapping, NamedTuple + +from opentelemetry.context import ( + Context, + create_key, + get_current, + get_value, + set_value, +) +from opentelemetry.util.types import AttributeValue + +__all__ = ["set_context_scoped_attributes"] + +_CONTEXT_SCOPED_ATTRIBUTES_KEY = create_key( + "opentelemetry.util.genai.context_scoped_attributes" +) + +_EMPTY: Mapping[str, AttributeValue] = MappingProxyType({}) + + +class _ContextScopedAttributes(NamedTuple): + """The per-signal attribute bags carried by a single context.""" + + span: Mapping[str, AttributeValue] = _EMPTY + log: Mapping[str, AttributeValue] = _EMPTY + + +_NO_ATTRIBUTES = _ContextScopedAttributes() + + +def set_context_scoped_attributes( + *, + span_attributes: Mapping[str, AttributeValue] | None = None, + log_attributes: Mapping[str, AttributeValue] | None = None, + context: Context | None = None, +) -> Context: + """Return a new context carrying the given attributes. Does not attach it. + + Args: + span_attributes: Added to GenAI spans on span start. + log_attributes: Added to the GenAI logs emitted. + context: Base context. Defaults to the current context. + """ + if not span_attributes and not log_attributes: + return context if context is not None else get_current() + + existing = _get_context_scoped_attributes(context) + merged = _ContextScopedAttributes( + span=MappingProxyType({**existing.span, **(span_attributes or {})}), + log=MappingProxyType({**existing.log, **(log_attributes or {})}), + ) + return set_value(_CONTEXT_SCOPED_ATTRIBUTES_KEY, merged, context) + + +def _get_context_scoped_attributes( + context: Context | None = None, +) -> _ContextScopedAttributes: + """Return the attribute bags on the given context, or empty ones.""" + value = get_value(_CONTEXT_SCOPED_ATTRIBUTES_KEY, context) + if isinstance(value, _ContextScopedAttributes): + return value + return _NO_ATTRIBUTES diff --git a/util/opentelemetry-util-genai/tests/test_context_attributes.py b/util/opentelemetry-util-genai/tests/test_context_attributes.py new file mode 100644 index 000000000..2f2cb59dd --- /dev/null +++ b/util/opentelemetry-util-genai/tests/test_context_attributes.py @@ -0,0 +1,182 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +import os +import unittest +from contextlib import contextmanager +from unittest.mock import patch + +from opentelemetry import context as otel_context +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import ( + InMemoryLogRecordExporter, + SimpleLogRecordProcessor, +) +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, +) +from opentelemetry.sdk.trace.sampling import Decision, Sampler, SamplingResult +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAI, +) +from opentelemetry.semconv._incubating.attributes import user_attributes +from opentelemetry.util.genai.context_attributes import ( + set_context_scoped_attributes, +) +from opentelemetry.util.genai.handler import get_telemetry_handler + + +class _RecordingSampler(Sampler): + """Samples everything, remembering the attributes it was given.""" + + def __init__(self): + self.seen = [] + + def should_sample( + self, + parent_context, + trace_id, + name, + kind=None, + attributes=None, + *_, + **__, + ): + self.seen.append(attributes) + return SamplingResult(Decision.RECORD_AND_SAMPLE, attributes) + + def get_description(self): + return "RecordingSampler" + + +@patch.dict( + os.environ, + { + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "EVENT_ONLY", + "OTEL_INSTRUMENTATION_GENAI_EMIT_EVENT": "true", + }, +) +class TestContextScopedAttributes(unittest.TestCase): + def setUp(self): + self.sampler = _RecordingSampler() + self.span_exporter = InMemorySpanExporter() + tracer_provider = TracerProvider(sampler=self.sampler) + tracer_provider.add_span_processor( + SimpleSpanProcessor(self.span_exporter) + ) + self.log_exporter = InMemoryLogRecordExporter() + logger_provider = LoggerProvider() + logger_provider.add_log_record_processor( + SimpleLogRecordProcessor(self.log_exporter) + ) + self.handler = get_telemetry_handler( + tracer_provider=tracer_provider, logger_provider=logger_provider + ) + + def tearDown(self): + if hasattr(get_telemetry_handler, "_default_handler"): + delattr(get_telemetry_handler, "_default_handler") + + @contextmanager + def _inference(self, context): + """Run one inference invocation with `context` attached.""" + token = otel_context.attach(context) + try: + with self.handler.inference( + "test-provider", request_model="test-model" + ) as invocation: + yield invocation + finally: + otel_context.detach(token) + + @property + def span_attributes(self): + (span,) = self.span_exporter.get_finished_spans() + return span.attributes + + @property + def event_attributes(self): + (log,) = self.log_exporter.get_finished_logs() + return log.log_record.attributes + + def test_attributes_go_only_to_the_signal_they_target(self): + with self._inference( + set_context_scoped_attributes( + span_attributes={GenAI.GEN_AI_AGENT_NAME: "trip-planner"}, + log_attributes={user_attributes.USER_ID: "u-1"}, + ) + ): + pass + + self.assertEqual( + self.span_attributes[GenAI.GEN_AI_AGENT_NAME], "trip-planner" + ) + self.assertNotIn(user_attributes.USER_ID, self.span_attributes) + self.assertEqual(self.event_attributes[user_attributes.USER_ID], "u-1") + self.assertNotIn(GenAI.GEN_AI_AGENT_NAME, self.event_attributes) + + def test_invocation_attributes_win(self): + context = set_context_scoped_attributes( + span_attributes={ + GenAI.GEN_AI_PROVIDER_NAME: "from-context", + "custom": "from-context", + }, + log_attributes={GenAI.GEN_AI_PROVIDER_NAME: "from-context"}, + ) + with self._inference(context) as invocation: + invocation.attributes["custom"] = "from-invocation" + + self.assertEqual( + self.span_attributes[GenAI.GEN_AI_PROVIDER_NAME], "test-provider" + ) + self.assertEqual(self.span_attributes["custom"], "from-invocation") + self.assertEqual( + self.event_attributes[GenAI.GEN_AI_PROVIDER_NAME], "test-provider" + ) + + def test_nested_contexts_merge_with_the_inner_one_winning(self): + outer = set_context_scoped_attributes( + span_attributes={"outer": "outer", "shared": "outer"} + ) + with self._inference( + set_context_scoped_attributes( + span_attributes={"inner": "inner", "shared": "inner"}, + context=outer, + ) + ): + pass + + self.assertEqual(self.span_attributes["outer"], "outer") + self.assertEqual(self.span_attributes["inner"], "inner") + self.assertEqual(self.span_attributes["shared"], "inner") + + def test_span_attributes_are_visible_to_the_sampler(self): + with self._inference( + set_context_scoped_attributes( + span_attributes={GenAI.GEN_AI_AGENT_NAME: "trip-planner"} + ) + ): + pass + + (seen,) = self.sampler.seen + self.assertEqual(seen[GenAI.GEN_AI_AGENT_NAME], "trip-planner") + + def test_attributes_do_not_apply_outside_the_attached_context(self): + with self._inference( + set_context_scoped_attributes( + span_attributes={GenAI.GEN_AI_AGENT_NAME: "trip-planner"} + ) + ): + pass + self.span_exporter.clear() + + with self.handler.inference("test-provider"): + pass + + self.assertNotIn(GenAI.GEN_AI_AGENT_NAME, self.span_attributes) + + +if __name__ == "__main__": + unittest.main()