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()