GenAI Utils | Add _BaseAgent base class and agent creation lifecycle

docs/nitpick-exceptions.ini

@@ -47,6 +47,7 @@ py-class=
     fastapi.applications.FastAPI
     starlette.applications.Starlette
     _contextvars.Token
+    opentelemetry.util.genai._agent_creation.AgentCreation
 
 any=
     ; API

util/opentelemetry-util-genai/CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Add `AgentCreation` type with `create_agent` span lifecycle via `start_create_agent` factory method and `create_agent` context manager
+  ([#4217](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4217))
 - Add metrics support for EmbeddingInvocation
   ([#4377](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4377))
 - Add support for workflow in genAI utils handler.

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

@@ -0,0 +1,153 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Agent creation invocation type.
+
+Represents a ``create_agent`` operation as defined by the OpenTelemetry
+GenAI semantic conventions:
+https://github.com/open-telemetry/semantic-conventions/blob/main/docs/gen-ai/gen-ai-agent-spans.md#create-agent
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+from typing import Any
+
+from opentelemetry._logs import Logger
+from opentelemetry.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAI,
+)
+from opentelemetry.semconv.attributes import server_attributes
+from opentelemetry.trace import SpanKind, Tracer
+from opentelemetry.util.genai._invocation import Error, GenAIInvocation
+from opentelemetry.util.genai.metrics import InvocationMetricsRecorder
+from opentelemetry.util.genai.types import MessagePart
+from opentelemetry.util.genai.utils import (
+    ContentCapturingMode,
+    gen_ai_json_dumps,
+    get_content_capturing_mode,
+    is_experimental_mode,
+)
+
+
+class AgentCreation(GenAIInvocation):
+    """Represents an agent creation/initialization.
+
+    Use ``handler.start_create_agent()`` or ``handler.create_agent()``
+    context manager rather than constructing this directly.
+
+    Spec:
+    https://github.com/open-telemetry/semantic-conventions/blob/main/docs/gen-ai/gen-ai-agent-spans.md#create-agent
+    """
+
+    def __init__(
+        self,
+        tracer: Tracer,
+        metrics_recorder: InvocationMetricsRecorder,
+        logger: Logger,
+        provider: str,
+        *,
+        request_model: str | None = None,
+        server_address: str | None = None,
+        server_port: int | None = None,
+        attributes: dict[str, Any] | None = None,
+        metric_attributes: dict[str, Any] | None = None,
+    ) -> None:
+        """Use handler.start_create_agent() or handler.create_agent() instead of calling this directly."""
+        _operation_name = GenAI.GenAiOperationNameValues.CREATE_AGENT.value
+        super().__init__(
+            tracer,
+            metrics_recorder,
+            logger,
+            operation_name=_operation_name,
+            span_name=_operation_name,
+            span_kind=SpanKind.CLIENT,
+            attributes=attributes,
+            metric_attributes=metric_attributes,
+        )
+        self.provider = provider
+        self.request_model = request_model
+        self.server_address = server_address
+        self.server_port = server_port
+
+        self.agent_name: str | None = None
+        self.agent_id: str | None = None
+        self.agent_description: str | None = None
+        self.agent_version: str | None = None
+
+        self.system_instruction: list[MessagePart] = []
+
+        self._start()
+
+    def _get_common_attributes(self) -> dict[str, Any]:
+        optional_attrs = (
+            (GenAI.GEN_AI_REQUEST_MODEL, self.request_model),
+            (server_attributes.SERVER_ADDRESS, self.server_address),
+            (server_attributes.SERVER_PORT, self.server_port),
+            (GenAI.GEN_AI_AGENT_NAME, self.agent_name),
+            (GenAI.GEN_AI_AGENT_ID, self.agent_id),
+            (GenAI.GEN_AI_AGENT_DESCRIPTION, self.agent_description),
+            (GenAI.GEN_AI_AGENT_VERSION, self.agent_version),
+        )
+        return {
+            GenAI.GEN_AI_OPERATION_NAME: self._operation_name,
+            GenAI.GEN_AI_PROVIDER_NAME: self.provider,
+            **{k: v for k, v in optional_attrs if v is not None},
+        }
+
+    def _get_system_instructions_for_span(self) -> dict[str, Any]:
+        if (
+            not is_experimental_mode()
+            or get_content_capturing_mode()
+            not in (
+                ContentCapturingMode.SPAN_ONLY,
+                ContentCapturingMode.SPAN_AND_EVENT,
+            )
+            or not self.system_instruction
+        ):
+            return {}
+        return {
+            GenAI.GEN_AI_SYSTEM_INSTRUCTIONS: gen_ai_json_dumps(
+                [asdict(p) for p in self.system_instruction]
+            ),
+        }
+
+    def _get_metric_attributes(self) -> dict[str, Any]:
+        optional_attrs = (
+            (GenAI.GEN_AI_PROVIDER_NAME, self.provider),
+            (GenAI.GEN_AI_REQUEST_MODEL, self.request_model),
+            (server_attributes.SERVER_ADDRESS, self.server_address),
+            (server_attributes.SERVER_PORT, self.server_port),
+        )
+        attrs: dict[str, Any] = {
+            GenAI.GEN_AI_OPERATION_NAME: self._operation_name,
+            **{k: v for k, v in optional_attrs if v is not None},
+        }
+        attrs.update(self.metric_attributes)
+        return attrs
+
+    def _apply_finish(self, error: Error | None = None) -> None:
+        if error is not None:
+            self._apply_error_attributes(error)
+
+        # Update span name if agent_name was set after construction
+        if self.agent_name:
+            self.span.update_name(f"{self._operation_name} {self.agent_name}")
+
+        attributes: dict[str, Any] = {}
+        attributes.update(self._get_common_attributes())
+        attributes.update(self._get_system_instructions_for_span())
+        attributes.update(self.attributes)
+        self.span.set_attributes(attributes)
+        self._metrics_recorder.record(self)

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

@@ -60,6 +60,7 @@
     TracerProvider,
     get_tracer,
 )
+from opentelemetry.util.genai._agent_creation import AgentCreation
 from opentelemetry.util.genai._inference_invocation import (
     LLMInvocation,
 )
@@ -299,6 +300,52 @@ def tool(
             tool_description=tool_description,
         )._managed()
 
+    def start_create_agent(
+        self,
+        provider: str,
+        *,
+        request_model: str | None = None,
+        server_address: str | None = None,
+        server_port: int | None = None,
+    ) -> AgentCreation:
+        """Create and start an agent creation invocation.
+
+        Set remaining attributes (agent_name, etc.) on the returned
+        invocation, then call invocation.stop() or invocation.fail().
+        """
+        return AgentCreation(
+            self._tracer,
+            self._metrics_recorder,
+            self._logger,
+            provider,
+            request_model=request_model,
+            server_address=server_address,
+            server_port=server_port,
+        )
+
+    def create_agent(
+        self,
+        provider: str,
+        *,
+        request_model: str | None = None,
+        server_address: str | None = None,
+        server_port: int | None = None,
+    ) -> AbstractContextManager[AgentCreation]:
+        """Context manager for agent creation.
+
+        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.
+        """
+        return self.start_create_agent(
+            provider=provider,
+            request_model=request_model,
+            server_address=server_address,
+            server_port=server_port,
+        )._managed()
+
     def workflow(
         self,
         name: str | None = None,

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

@@ -26,6 +26,7 @@
     )
 """
 
+from opentelemetry.util.genai._agent_creation import AgentCreation
 from opentelemetry.util.genai._embedding_invocation import EmbeddingInvocation
 from opentelemetry.util.genai._inference_invocation import InferenceInvocation
 from opentelemetry.util.genai._invocation import (
@@ -37,6 +38,7 @@
 from opentelemetry.util.genai._workflow_invocation import WorkflowInvocation
 
 __all__ = [
+    "AgentCreation",
     "ContextToken",
     "Error",
     "GenAIInvocation",

util/opentelemetry-util-genai/tests/test_handler_agent.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import unittest
+
+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.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAI,
+)
+from opentelemetry.semconv.attributes import server_attributes
+from opentelemetry.trace import SpanKind
+from opentelemetry.util.genai.handler import TelemetryHandler
+
+
+class TestAgentCreation(unittest.TestCase):
+    def setUp(self):
+        self.span_exporter = InMemorySpanExporter()
+        tracer_provider = TracerProvider()
+        tracer_provider.add_span_processor(
+            SimpleSpanProcessor(self.span_exporter)
+        )
+        self.handler = TelemetryHandler(tracer_provider=tracer_provider)
+
+    def test_start_stop_creates_span(self):
+        creation = self.handler.start_create_agent(
+            "openai",
+            request_model="gpt-4",
+        )
+        creation.agent_name = "New Agent"
+        creation.agent_id = "agent-new-1"
+        creation.stop()
+
+        spans = self.span_exporter.get_finished_spans()
+        assert len(spans) == 1
+        span = spans[0]
+        assert span.name == "create_agent New Agent"
+        assert span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "create_agent"
+        assert span.attributes[GenAI.GEN_AI_AGENT_NAME] == "New Agent"
+        assert span.attributes[GenAI.GEN_AI_AGENT_ID] == "agent-new-1"
+        assert span.attributes[GenAI.GEN_AI_PROVIDER_NAME] == "openai"
+        assert span.attributes[GenAI.GEN_AI_REQUEST_MODEL] == "gpt-4"
+
+    def test_span_kind_is_client(self):
+        creation = self.handler.start_create_agent("openai")
+        creation.stop()
+
+        assert (
+            self.span_exporter.get_finished_spans()[0].kind == SpanKind.CLIENT
+        )
+
+    def test_all_attributes(self):
+        creation = self.handler.start_create_agent(
+            "openai",
+            request_model="gpt-4",
+            server_address="api.openai.com",
+            server_port=443,
+        )
+        creation.agent_name = "Full Agent"
+        creation.agent_id = "agent-123"
+        creation.agent_description = "A test agent"
+        creation.agent_version = "1.0.0"
+        creation.stop()
+
+        spans = self.span_exporter.get_finished_spans()
+        assert len(spans) == 1
+        attrs = spans[0].attributes
+        assert attrs[GenAI.GEN_AI_OPERATION_NAME] == "create_agent"
+        assert attrs[GenAI.GEN_AI_AGENT_NAME] == "Full Agent"
+        assert attrs[GenAI.GEN_AI_AGENT_ID] == "agent-123"
+        assert attrs[GenAI.GEN_AI_AGENT_DESCRIPTION] == "A test agent"
+        assert attrs[GenAI.GEN_AI_AGENT_VERSION] == "1.0.0"
+        assert attrs[GenAI.GEN_AI_PROVIDER_NAME] == "openai"
+        assert attrs[GenAI.GEN_AI_REQUEST_MODEL] == "gpt-4"
+        assert attrs[server_attributes.SERVER_ADDRESS] == "api.openai.com"
+        assert attrs[server_attributes.SERVER_PORT] == 443
+
+    def test_no_server_attributes_when_not_provided(self):
+        creation = self.handler.start_create_agent("openai")
+        creation.stop()
+
+        attrs = self.span_exporter.get_finished_spans()[0].attributes
+        assert server_attributes.SERVER_ADDRESS not in attrs
+        assert server_attributes.SERVER_PORT not in attrs
+
+    def test_fail_create_agent(self):
+        creation = self.handler.start_create_agent("openai")
+        creation.agent_name = "Bad Agent"
+        creation.fail(RuntimeError("creation failed"))
+
+        spans = self.span_exporter.get_finished_spans()
+        assert len(spans) == 1
+        assert spans[0].status.description == "creation failed"
+        assert spans[0].attributes.get("error.type") == "RuntimeError"
+
+    def test_context_manager(self):
+        with self.handler.create_agent(
+            "openai", request_model="gpt-4"
+        ) as creation:
+            creation.agent_name = "CM Agent"
+            creation.agent_id = "assigned-id"
+
+        spans = self.span_exporter.get_finished_spans()
+        assert len(spans) == 1
+        assert spans[0].name == "create_agent CM Agent"
+        assert spans[0].attributes[GenAI.GEN_AI_AGENT_ID] == "assigned-id"
+
+    def test_context_manager_error(self):
+        with self.assertRaises(TypeError):
+            with self.handler.create_agent("openai") as creation:
+                creation.agent_name = "Err"
+                raise TypeError("bad type")
+
+        spans = self.span_exporter.get_finished_spans()
+        assert len(spans) == 1
+        assert spans[0].attributes.get("error.type") == "TypeError"
+
+    def test_custom_attributes(self):
+        creation = self.handler.start_create_agent(
+            "openai", request_model="gpt-4"
+        )
+        creation.attributes["custom.key"] = "custom_value"
+        creation.stop()
+
+        attrs = self.span_exporter.get_finished_spans()[0].attributes
+        assert attrs["custom.key"] == "custom_value"
+
+    def test_span_name_without_agent_name(self):
+        creation = self.handler.start_create_agent("openai")
+        creation.stop()
+
+        assert (
+            self.span_exporter.get_finished_spans()[0].name == "create_agent"
+        )