Better follow otel semconv

pydantic_ai_slim/pydantic_ai/_otel_messages.py

@@ -1,6 +1,8 @@
 """Type definitions of OpenTelemetry GenAI spec message parts.
 
-Based on https://github.com/lmolkova/semantic-conventions/blob/eccd1f806e426a32c98271c3ce77585492d26de2/docs/gen-ai/non-normative/models.ipynb
+Based on the OpenTelemetry semantic conventions for GenAI:
+https://github.com/open-telemetry/semantic-conventions/blob/main/docs/gen-ai/gen-ai-input-messages.json
+https://github.com/open-telemetry/semantic-conventions/blob/main/docs/gen-ai/gen-ai-output-messages.json
 """
 
 from __future__ import annotations
@@ -12,11 +14,15 @@
 
 
 class TextPart(TypedDict):
+    """A text part in a GenAI message."""
+
     type: Literal['text']
     content: NotRequired[str]
 
 
 class ToolCallPart(TypedDict):
+    """A tool call part in a GenAI message."""
+
     type: Literal['tool_call']
     id: str
     name: str
@@ -25,36 +31,57 @@ class ToolCallPart(TypedDict):
 
 
 class ToolCallResponsePart(TypedDict):
+    """A tool call response part in a GenAI message."""
+
     type: Literal['tool_call_response']
     id: str
     name: str
+    # TODO: This should be `response` not `result`
     result: NotRequired[JsonValue]
     builtin: NotRequired[bool]  # Not (currently?) part of the spec, used by Logfire
 
 
-class MediaUrlPart(TypedDict):
-    type: Literal['image-url', 'audio-url', 'video-url', 'document-url']
-    url: NotRequired[str]
+class UriPart(TypedDict):
+    """A URI part in a GenAI message (for images, audio, video, documents).
 
+    Per the semantic conventions, uses 'uri' type with modality field.
+    """
 
-class BinaryDataPart(TypedDict):
-    type: Literal['binary']
-    media_type: str
-    content: NotRequired[str]
+    type: Literal['uri']
+    uri: NotRequired[str]
+    modality: NotRequired[str]
+
+
+class BlobPart(TypedDict):
+    """A blob (binary data) part in a GenAI message.
 
+    Per the semantic conventions, uses 'blob' type with modality field.
+    """
 
-class ThinkingPart(TypedDict):
-    type: Literal['thinking']
+    type: Literal['blob']
+    blob: NotRequired[str]
+    modality: NotRequired[str]
+
+
+class ReasoningPart(TypedDict):
+    """A reasoning/thinking part in a GenAI message.
+
+    Per the semantic conventions, uses 'reasoning' type.
+    """
+
+    type: Literal['reasoning']
     content: NotRequired[str]
 
 
-MessagePart: TypeAlias = 'TextPart | ToolCallPart | ToolCallResponsePart | MediaUrlPart | BinaryDataPart | ThinkingPart'
+MessagePart: TypeAlias = 'TextPart | ToolCallPart | ToolCallResponsePart | UriPart | BlobPart | ReasoningPart'
 
 
 Role = Literal['system', 'user', 'assistant']
 
 
 class ChatMessage(TypedDict):
+    """A chat message in the GenAI format."""
+
     role: Role
     parts: list[MessagePart]
 
@@ -63,6 +90,8 @@ class ChatMessage(TypedDict):
 
 
 class OutputMessage(ChatMessage):
+    """An output message with optional finish reason."""
+
     finish_reason: NotRequired[str]
 
 

pydantic_ai_slim/pydantic_ai/_output.py

@@ -102,8 +102,10 @@ async def execute_traced_output_function(
     instrumentation_names = InstrumentationNames.for_version(run_context.instrumentation_version)
     # Set up span attributes
     tool_name = run_context.tool_name or getattr(function_schema.function, '__name__', 'output_function')
-    attributes = {
+    attributes: dict[str, Any] = {
         'gen_ai.tool.name': tool_name,
+        'gen_ai.operation.name': 'execute_tool',
+        'gen_ai.tool.type': 'function',
         'logfire.msg': f'running output function: {tool_name}',
     }
     if run_context.tool_call_id:

pydantic_ai_slim/pydantic_ai/_tool_manager.py

@@ -219,10 +219,12 @@ async def _call_function_tool(
         """See <https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-spans/#execute-tool-span>."""
         instrumentation_names = InstrumentationNames.for_version(instrumentation_version)
 
-        span_attributes = {
+        span_attributes: dict[str, Any] = {
             'gen_ai.tool.name': call.tool_name,
             # NOTE: this means `gen_ai.tool.call.id` will be included even if it was generated by pydantic-ai
             'gen_ai.tool.call.id': call.tool_call_id,
+            'gen_ai.operation.name': 'execute_tool',
+            'gen_ai.tool.type': 'function',
             **({instrumentation_names.tool_arguments_attr: call.args_as_json_str()} if include_content else {}),
             'logfire.msg': f'running tool: {call.tool_name}',
             # add the JSON schema so these attributes are formatted nicely in Logfire

pydantic_ai_slim/pydantic_ai/agent/__init__.py

@@ -665,6 +665,7 @@ async def get_instructions(run_context: RunContext[AgentDepsT]) -> str | None:
                 'model_name': model_used.model_name if model_used else 'no-model',
                 'agent_name': agent_name,
                 'gen_ai.agent.name': agent_name,
+                'gen_ai.operation.name': 'invoke_agent',
                 'logfire.msg': f'{agent_name} run',
             },
         )

pydantic_ai_slim/pydantic_ai/messages.py

@@ -767,17 +767,33 @@ def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_me
                     _otel_messages.TextPart(type='text', **({'content': part} if settings.include_content else {}))
                 )
             elif isinstance(part, ImageUrl | AudioUrl | DocumentUrl | VideoUrl):
-                parts.append(
-                    _otel_messages.MediaUrlPart(
-                        type=part.kind,
-                        **{'url': part.url} if settings.include_content else {},
-                    )
-                )
+                # Map file URL kinds to modality values
+                modality_map = {
+                    'image-url': 'image',
+                    'audio-url': 'audio',
+                    'document-url': 'document',
+                    'video-url': 'video',
+                }
+                uri_part = _otel_messages.UriPart(type='uri', modality=modality_map.get(part.kind, 'unknown'))
+                if settings.include_content:
+                    uri_part['uri'] = part.url
+                parts.append(uri_part)
             elif isinstance(part, BinaryContent):
-                converted_part = _otel_messages.BinaryDataPart(type='binary', media_type=part.media_type)
+                # Map media type prefix to modality
+                if part.is_image:
+                    modality = 'image'
+                elif part.is_audio:
+                    modality = 'audio'
+                elif part.is_video:
+                    modality = 'video'
+                elif part.is_document:
+                    modality = 'document'
+                else:
+                    modality = part.media_type
+                blob_part = _otel_messages.BlobPart(type='blob', modality=modality)
                 if settings.include_content and settings.include_binary_content:
-                    converted_part['content'] = base64.b64encode(part.data).decode()
-                parts.append(converted_part)
+                    blob_part['blob'] = base64.b64encode(part.data).decode()
+                parts.append(blob_part)
             elif isinstance(part, CachePoint):
                 # CachePoint is a marker, not actual content - skip it for otel
                 pass
@@ -1421,43 +1437,19 @@ def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_me
                     )
                 )
             elif isinstance(part, ThinkingPart):
+                # Per semantic conventions, thinking/reasoning uses 'reasoning' type
                 parts.append(
-                    _otel_messages.ThinkingPart(
-                        type='thinking',
+                    _otel_messages.ReasoningPart(
+                        type='reasoning',
                         **({'content': part.content} if settings.include_content else {}),
                     )
                 )
             elif isinstance(part, FilePart):
-                converted_part = _otel_messages.BinaryDataPart(type='binary', media_type=part.content.media_type)
-                if settings.include_content and settings.include_binary_content:
-                    converted_part['content'] = base64.b64encode(part.content.data).decode()
-                parts.append(converted_part)
+                parts.append(_file_part_to_otel(part.content, settings))
             elif isinstance(part, BaseToolCallPart):
-                call_part = _otel_messages.ToolCallPart(type='tool_call', id=part.tool_call_id, name=part.tool_name)
-                if isinstance(part, BuiltinToolCallPart):
-                    call_part['builtin'] = True
-                if settings.include_content and part.args is not None:
-                    from .models.instrumented import InstrumentedModel
-
-                    if isinstance(part.args, str):
-                        call_part['arguments'] = part.args
-                    else:
-                        call_part['arguments'] = {k: InstrumentedModel.serialize_any(v) for k, v in part.args.items()}
-
-                parts.append(call_part)
+                parts.append(_tool_call_to_otel(part, settings))
             elif isinstance(part, BuiltinToolReturnPart):
-                return_part = _otel_messages.ToolCallResponsePart(
-                    type='tool_call_response',
-                    id=part.tool_call_id,
-                    name=part.tool_name,
-                    builtin=True,
-                )
-                if settings.include_content and part.content is not None:  # pragma: no branch
-                    from .models.instrumented import InstrumentedModel
-
-                    return_part['result'] = InstrumentedModel.serialize_any(part.content)
-
-                parts.append(return_part)
+                parts.append(_builtin_tool_return_to_otel(part, settings))
         return parts
 
     @property
@@ -1478,6 +1470,57 @@ def provider_request_id(self) -> str | None:
     __repr__ = _utils.dataclasses_no_defaults_repr
 
 
+def _file_part_to_otel(bc: BinaryContent, settings: InstrumentationSettings) -> _otel_messages.BlobPart:
+    """Convert a FilePart's BinaryContent to an otel BlobPart."""
+    if bc.is_image:
+        modality = 'image'
+    elif bc.is_audio:
+        modality = 'audio'
+    elif bc.is_video:
+        modality = 'video'
+    elif bc.is_document:
+        modality = 'document'
+    else:
+        modality = bc.media_type
+    blob_part = _otel_messages.BlobPart(type='blob', modality=modality)
+    if settings.include_content and settings.include_binary_content:
+        blob_part['blob'] = base64.b64encode(bc.data).decode()
+    return blob_part
+
+
+def _tool_call_to_otel(part: BaseToolCallPart, settings: InstrumentationSettings) -> _otel_messages.ToolCallPart:
+    """Convert a BaseToolCallPart to an otel ToolCallPart."""
+    call_part = _otel_messages.ToolCallPart(type='tool_call', id=part.tool_call_id, name=part.tool_name)
+    if settings.include_content and part.args is not None:
+        from .models.instrumented import InstrumentedModel
+
+        if isinstance(part.args, str):
+            call_part['arguments'] = part.args
+        else:
+            call_part['arguments'] = {k: InstrumentedModel.serialize_any(v) for k, v in part.args.items()}
+
+    if isinstance(part, BuiltinToolCallPart):
+        call_part['builtin'] = True
+    return call_part
+
+
+def _builtin_tool_return_to_otel(
+    part: BuiltinToolReturnPart, settings: InstrumentationSettings
+) -> _otel_messages.ToolCallResponsePart:
+    """Convert a BuiltinToolReturnPart to an otel ToolCallResponsePart."""
+    return_part = _otel_messages.ToolCallResponsePart(
+        type='tool_call_response',
+        id=part.tool_call_id,
+        name=part.tool_name,
+        builtin=True,
+    )
+    if settings.include_content and part.content is not None:  # pragma: no branch
+        from .models.instrumented import InstrumentedModel
+
+        return_part['result'] = InstrumentedModel.serialize_any(part.content)
+    return return_part
+
+
 ModelMessage = Annotated[ModelRequest | ModelResponse, pydantic.Discriminator('kind')]
 """Any message sent to or returned by a model."""