GenAI Utils | Agent Invocation Span

Author: etserend

util/opentelemetry-util-genai/src/opentelemetry/util/genai/handler.py

@@ -80,13 +80,15 @@
 )
 from opentelemetry.util.genai.metrics import InvocationMetricsRecorder
 from opentelemetry.util.genai.span_utils import (
+    _apply_agent_finish_attributes,
     _apply_creation_finish_attributes,
     _apply_error_attributes,
     _apply_llm_finish_attributes,
     _maybe_emit_llm_event,
 )
 from opentelemetry.util.genai.types import (
     AgentCreation,
+    AgentInvocation,
     Error,
     LLMInvocation,
 )
@@ -213,6 +215,75 @@ def llm(
             raise
         self.stop_llm(invocation)
 
+    # ---- Agent invocation lifecycle ----
+
+    def start_agent(
+        self,
+        invocation: AgentInvocation,
+    ) -> AgentInvocation:
+        """Start an agent invocation and create a pending span entry."""
+        span_name = f"{invocation.operation_name} {invocation.agent_name}".strip()
+        kind = SpanKind.CLIENT if invocation.is_remote else SpanKind.INTERNAL
+        span = self._tracer.start_span(
+            name=span_name,
+            kind=kind,
+        )
+        invocation.monotonic_start_s = timeit.default_timer()
+        invocation.span = span
+        invocation.context_token = otel_context.attach(
+            set_span_in_context(span)
+        )
+        return invocation
+
+    def stop_agent(self, invocation: AgentInvocation) -> AgentInvocation:  # pylint: disable=no-self-use
+        """Finalize an agent invocation successfully and end its span."""
+        if invocation.context_token is None or invocation.span is None:
+            return invocation
+
+        span = invocation.span
+        _apply_agent_finish_attributes(span, invocation)
+        otel_context.detach(invocation.context_token)
+        span.end()
+        return invocation
+
+    def fail_agent(  # pylint: disable=no-self-use
+        self, invocation: AgentInvocation, error: Error
+    ) -> AgentInvocation:
+        """Fail an agent invocation and end its span with error status."""
+        if invocation.context_token is None or invocation.span is None:
+            return invocation
+
+        span = invocation.span
+        _apply_agent_finish_attributes(span, invocation)
+        _apply_error_attributes(span, error)
+        otel_context.detach(invocation.context_token)
+        span.end()
+        return invocation
+
+    @contextmanager
+    def agent(
+        self, invocation: AgentInvocation | None = None
+    ) -> Iterator[AgentInvocation]:
+        """Context manager for agent invocations.
+
+        Only set data attributes on the invocation object, do not modify the span or context.
+
+        Starts the span on entry. On normal exit, finalizes the invocation and ends the span.
+        If an exception occurs inside the context, marks the span as error, ends it, and
+        re-raises the original exception.
+        """
+        if invocation is None:
+            invocation = AgentInvocation()
+        self.start_agent(invocation)
+        try:
+            yield invocation
+        except Exception as exc:
+            self.fail_agent(
+                invocation, Error(message=str(exc), type=type(exc))
+            )
+            raise
+        self.stop_agent(invocation)
+
     # ---- Agent creation lifecycle ----
 
     def start_create_agent(

util/opentelemetry-util-genai/src/opentelemetry/util/genai/span_utils.py

@@ -33,6 +33,7 @@
 from opentelemetry.trace.status import Status, StatusCode
 from opentelemetry.util.genai.types import (
     AgentCreation,
+    AgentInvocation,
     Error,
     InputMessage,
     LLMInvocation,
@@ -62,6 +63,16 @@ def _agent_attr(name: str, fallback: str) -> str:
 _GEN_AI_AGENT_VERSION = _agent_attr(
     "GEN_AI_AGENT_VERSION", "gen_ai.agent.version"
 )
+_GEN_AI_CONVERSATION_ID = _agent_attr(
+    "GEN_AI_CONVERSATION_ID", "gen_ai.conversation.id"
+)
+_GEN_AI_DATA_SOURCE_ID = _agent_attr(
+    "GEN_AI_DATA_SOURCE_ID", "gen_ai.data_source.id"
+)
+_GEN_AI_OUTPUT_TYPE = _agent_attr("GEN_AI_OUTPUT_TYPE", "gen_ai.output.type")
+_GEN_AI_TOOL_DEFINITIONS = _agent_attr(
+    "GEN_AI_TOOL_DEFINITIONS", "gen_ai.tool.definitions"
+)
 
 
 def _get_llm_common_attributes(
@@ -324,6 +335,149 @@ def _get_base_agent_span_name(agent: _BaseAgent) -> str:
     return agent.operation_name
 
 
+def _get_agent_common_attributes(
+    invocation: AgentInvocation,
+) -> dict[str, Any]:
+    """Get common agent invocation attributes shared by finish() and error() paths."""
+    attrs = _get_base_agent_common_attributes(invocation)
+
+    # Invoke-specific conditionally required attributes
+    invoke_attrs = (
+        (_GEN_AI_CONVERSATION_ID, invocation.conversation_id),
+        (_GEN_AI_DATA_SOURCE_ID, invocation.data_source_id),
+        (_GEN_AI_OUTPUT_TYPE, invocation.output_type),
+    )
+    attrs.update(
+        {key: value for key, value in invoke_attrs if value is not None}
+    )
+
+    return attrs
+
+
+def _get_agent_span_name(invocation: AgentInvocation) -> str:
+    """Get the span name for an agent invocation."""
+    return _get_base_agent_span_name(invocation)
+
+
+def _get_agent_request_attributes(
+    invocation: AgentInvocation,
+) -> dict[str, Any]:
+    """Get GenAI request semantic convention attributes for agent invocation."""
+    optional_attrs = (
+        (GenAI.GEN_AI_REQUEST_TEMPERATURE, invocation.temperature),
+        (GenAI.GEN_AI_REQUEST_TOP_P, invocation.top_p),
+        (GenAI.GEN_AI_REQUEST_FREQUENCY_PENALTY, invocation.frequency_penalty),
+        (GenAI.GEN_AI_REQUEST_PRESENCE_PENALTY, invocation.presence_penalty),
+        (GenAI.GEN_AI_REQUEST_MAX_TOKENS, invocation.max_tokens),
+        (GenAI.GEN_AI_REQUEST_STOP_SEQUENCES, invocation.stop_sequences),
+        (GenAI.GEN_AI_REQUEST_SEED, invocation.seed),
+        (GenAI.GEN_AI_REQUEST_CHOICE_COUNT, invocation.choice_count),
+    )
+
+    return {key: value for key, value in optional_attrs if value is not None}
+
+
+def _get_agent_response_attributes(
+    invocation: AgentInvocation,
+) -> dict[str, Any]:
+    """Get GenAI response semantic convention attributes for agent invocation."""
+    finish_reasons: list[str] | None
+    if invocation.finish_reasons is not None:
+        finish_reasons = invocation.finish_reasons
+    elif invocation.output_messages:
+        finish_reasons = [
+            message.finish_reason
+            for message in invocation.output_messages
+            if message.finish_reason
+        ]
+    else:
+        finish_reasons = None
+
+    unique_finish_reasons = (
+        sorted(set(finish_reasons)) if finish_reasons else None
+    )
+
+    optional_attrs = (
+        (
+            GenAI.GEN_AI_RESPONSE_FINISH_REASONS,
+            unique_finish_reasons if unique_finish_reasons else None,
+        ),
+        (GenAI.GEN_AI_RESPONSE_MODEL, invocation.response_model_name),
+        (GenAI.GEN_AI_RESPONSE_ID, invocation.response_id),
+        (GenAI.GEN_AI_USAGE_INPUT_TOKENS, invocation.input_tokens),
+        (GenAI.GEN_AI_USAGE_OUTPUT_TOKENS, invocation.output_tokens),
+    )
+
+    return {key: value for key, value in optional_attrs if value is not None}
+
+
+def _get_agent_messages_attributes_for_span(
+    input_messages: list[InputMessage],
+    output_messages: list[OutputMessage],
+    system_instruction: list[MessagePart] | None = None,
+    tool_definitions: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Get message attributes formatted for span (JSON string format) for agent invocation."""
+    if not is_experimental_mode() or get_content_capturing_mode() not in (
+        ContentCapturingMode.SPAN_ONLY,
+        ContentCapturingMode.SPAN_AND_EVENT,
+    ):
+        return {}
+
+    optional_attrs = (
+        (
+            GenAI.GEN_AI_INPUT_MESSAGES,
+            gen_ai_json_dumps([asdict(m) for m in input_messages])
+            if input_messages
+            else None,
+        ),
+        (
+            GenAI.GEN_AI_OUTPUT_MESSAGES,
+            gen_ai_json_dumps([asdict(m) for m in output_messages])
+            if output_messages
+            else None,
+        ),
+        (
+            GenAI.GEN_AI_SYSTEM_INSTRUCTIONS,
+            gen_ai_json_dumps([asdict(p) for p in system_instruction])
+            if system_instruction
+            else None,
+        ),
+        (
+            _GEN_AI_TOOL_DEFINITIONS,
+            gen_ai_json_dumps(tool_definitions)
+            if tool_definitions
+            else None,
+        ),
+    )
+
+    return {key: value for key, value in optional_attrs if value is not None}
+
+
+def _apply_agent_finish_attributes(
+    span: Span, invocation: AgentInvocation
+) -> None:
+    """Apply attributes/messages common to agent finish() paths."""
+    span.update_name(_get_agent_span_name(invocation))
+
+    attributes: dict[str, Any] = {}
+    attributes.update(_get_agent_common_attributes(invocation))
+    attributes.update(_get_agent_request_attributes(invocation))
+    attributes.update(_get_agent_response_attributes(invocation))
+    attributes.update(
+        _get_agent_messages_attributes_for_span(
+            invocation.input_messages,
+            invocation.output_messages,
+            invocation.system_instruction,
+            invocation.tool_definitions,
+        )
+    )
+    attributes.update(invocation.attributes)
+
+    if attributes:
+        span.set_attributes(attributes)
+
+
 def _get_creation_common_attributes(
     creation: AgentCreation,
 ) -> dict[str, Any]:
@@ -375,7 +529,12 @@ def _apply_creation_finish_attributes(
     "_maybe_emit_llm_event",
     "_get_base_agent_common_attributes",
     "_get_base_agent_span_name",
+    "_apply_agent_finish_attributes",
     "_apply_creation_finish_attributes",
+    "_get_agent_common_attributes",
+    "_get_agent_request_attributes",
+    "_get_agent_response_attributes",
+    "_get_agent_span_name",
     "_get_creation_common_attributes",
     "_get_creation_span_name",
 ]

util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py

@@ -298,6 +298,64 @@ class AgentCreation(_BaseAgent):
     operation_name: str = "create_agent"
 
 
+@dataclass
+class AgentInvocation(_BaseAgent):
+    """
+    Represents an agent invocation (invoke_agent operation).
+
+    Follows semconv for GenAI agent spans:
+    https://github.com/open-telemetry/semantic-conventions/blob/main/docs/gen-ai/gen-ai-agent-spans.md#invoke-agent-span
+
+    When creating an AgentInvocation object, only update the data attributes.
+    The span and context_token attributes are set by the TelemetryHandler.
+    """
+
+    # Override default operation name
+    operation_name: str = "invoke_agent"
+
+    # Invoke-specific request attributes (Cond. Required)
+    conversation_id: str | None = None
+    data_source_id: str | None = None
+    output_type: str | None = None
+
+    # Request parameters (Recommended)
+    temperature: float | None = None
+    top_p: float | None = None
+    frequency_penalty: float | None = None
+    presence_penalty: float | None = None
+    max_tokens: int | None = None
+    stop_sequences: list[str] | None = None
+    seed: int | None = None
+    choice_count: int | None = None
+
+    # Response (Recommended)
+    response_model_name: str | None = None
+    response_id: str | None = None
+    finish_reasons: list[str] | None = None
+    input_tokens: int | None = None
+    output_tokens: int | None = None
+
+    # Content (Opt-In) — input/output messages and tool definitions
+    input_messages: list[InputMessage] = field(
+        default_factory=_new_input_messages
+    )
+    output_messages: list[OutputMessage] = field(
+        default_factory=_new_output_messages
+    )
+    tool_definitions: list[dict[str, Any]] | None = None
+
+    # Span kind: CLIENT for remote agents, INTERNAL for in-process agents
+    is_remote: bool = True
+
+    metric_attributes: dict[str, Any] = field(
+        default_factory=_new_str_any_dict
+    )
+    """
+    Additional attributes to set on metrics. Must be of a low cardinality.
+    These attributes will not be set on spans or events.
+    """
+
+
 @dataclass
 class Error:
     message: str