Merge pull request #3449 from kingster/telemetry-fix-system-message

fix: Record correct system_instruction in LLM spans for LLM services

changelog/3449.changed.md

@@ -0,0 +1 @@
+- Renamed tracing span attributes to align with OpenTelemetry GenAI semantic conventions: `gen_ai.system` to `gen_ai.provider.name`, `system` to `gen_ai.system_instructions`, `gen_ai.usage.cache_read_input_tokens` to `gen_ai.usage.cache_read.input_tokens`, and `gen_ai.usage.cache_creation_input_tokens` to `gen_ai.usage.cache_creation.input_tokens`.

changelog/3449.fixed.md

@@ -0,0 +1 @@
+- Fixed stale `system_instruction` in LLM tracing spans by reading from `_settings.system_instruction` instead of the removed `_system_instruction` attribute.

src/pipecat/utils/tracing/service_attributes.py

@@ -25,20 +25,20 @@ if is_tracing_available():
     from opentelemetry.trace import Span
 
 
-def _get_gen_ai_system_from_service_name(service_name: str) -> str:
-    """Extract the standardized gen_ai.system value from a service class name.
+def _get_provider_name_from_service_name(service_name: str) -> str:
+    """Extract the standardized gen_ai.provider.name value from a service class name.
 
     Source:
-    https://opentelemetry.io/docs/specs/semconv/attributes-registry/gen-ai/#gen-ai-system
+    https://opentelemetry.io/docs/specs/semconv/attributes-registry/gen-ai/
 
     Uses standard OTel names where possible, with special case mappings for
     service names that don't follow the pattern.
 
     Args:
-        service_name: The service class name to extract system name from.
+        service_name: The service class name to extract provider name from.
 
     Returns:
-        The standardized gen_ai.system value.
+        The standardized gen_ai.provider.name value.
     """
     SPECIAL_CASE_MAPPINGS = {
         # AWS
@@ -91,7 +91,7 @@ def add_tts_span_attributes(
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", service_name.replace("TTSService", "").lower())
+    span.set_attribute("gen_ai.provider.name", service_name.replace("TTSService", "").lower())
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", operation_name)
     span.set_attribute("gen_ai.output.type", "speech")
@@ -150,7 +150,7 @@ def add_stt_span_attributes(
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", service_name.replace("STTService", "").lower())
+    span.set_attribute("gen_ai.provider.name", service_name.replace("STTService", "").lower())
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", operation_name)
     span.set_attribute("vad_enabled", vad_enabled)
@@ -193,7 +193,7 @@ def add_llm_span_attributes(
     tools: Optional[str] = None,
     tool_count: Optional[int] = None,
     tool_choice: Optional[str] = None,
-    system: Optional[str] = None,
+    system_instructions: Optional[str] = None,
     parameters: Optional[Dict[str, Any]] = None,
     extra_parameters: Optional[Dict[str, Any]] = None,
     ttfb: Optional[float] = None,
@@ -211,14 +211,14 @@ def add_llm_span_attributes(
         tools: JSON-serialized tools configuration.
         tool_count: Number of tools available.
         tool_choice: Tool selection configuration.
-        system: System message.
+        system_instructions: System instructions.
         parameters: Service parameters.
         extra_parameters: Additional parameters.
         ttfb: Time to first byte in seconds.
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", _get_gen_ai_system_from_service_name(service_name))
+    span.set_attribute("gen_ai.provider.name", _get_provider_name_from_service_name(service_name))
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", "chat")
     span.set_attribute("gen_ai.output.type", "text")
@@ -240,8 +240,8 @@ def add_llm_span_attributes(
     if tool_choice:
         span.set_attribute("tool_choice", tool_choice)
 
-    if system:
-        span.set_attribute("system", system)
+    if system_instructions:
+        span.set_attribute("gen_ai.system_instructions", system_instructions)
 
     if ttfb is not None:
         span.set_attribute("metrics.ttfb", ttfb)
@@ -313,7 +313,7 @@ def add_gemini_live_span_attributes(
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", "gcp.gemini")
+    span.set_attribute("gen_ai.provider.name", "gcp.gemini")
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", operation_name)
     span.set_attribute("service.operation", operation_name)
@@ -414,7 +414,7 @@ def add_openai_realtime_span_attributes(
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", "openai")
+    span.set_attribute("gen_ai.provider.name", "openai")
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", operation_name)
     span.set_attribute("service.operation", operation_name)

src/pipecat/utils/tracing/service_decorators.py

@@ -137,14 +137,14 @@ def _add_token_usage_to_span(span, token_usage):
             and token_usage["cache_read_input_tokens"] is not None
         ):
             span.set_attribute(
-                "gen_ai.usage.cache_read_input_tokens", token_usage["cache_read_input_tokens"]
+                "gen_ai.usage.cache_read.input_tokens", token_usage["cache_read_input_tokens"]
             )
         if (
             "cache_creation_input_tokens" in token_usage
             and token_usage["cache_creation_input_tokens"] is not None
         ):
             span.set_attribute(
-                "gen_ai.usage.cache_creation_input_tokens",
+                "gen_ai.usage.cache_creation.input_tokens",
                 token_usage["cache_creation_input_tokens"],
             )
         if "reasoning_tokens" in token_usage and token_usage["reasoning_tokens"] is not None:
@@ -159,11 +159,11 @@ def _add_token_usage_to_span(span, token_usage):
         # Add cached token metrics for LLMTokenUsage object
         cache_read_tokens = getattr(token_usage, "cache_read_input_tokens", None)
         if cache_read_tokens is not None:
-            span.set_attribute("gen_ai.usage.cache_read_input_tokens", cache_read_tokens)
+            span.set_attribute("gen_ai.usage.cache_read.input_tokens", cache_read_tokens)
 
         cache_creation_tokens = getattr(token_usage, "cache_creation_input_tokens", None)
         if cache_creation_tokens is not None:
-            span.set_attribute("gen_ai.usage.cache_creation_input_tokens", cache_creation_tokens)
+            span.set_attribute("gen_ai.usage.cache_creation.input_tokens", cache_creation_tokens)
 
         reasoning_tokens = getattr(token_usage, "reasoning_tokens", None)
         if reasoning_tokens is not None:
@@ -502,18 +502,45 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
 
                             # Handle system message for different services
                             system_message = None
-                            if hasattr(context, "system"):
+                            if isinstance(context, LLMContext):
+                                # settings.system_instruction takes priority (matches service behavior)
+                                if hasattr(self, "_settings") and getattr(
+                                    self._settings, "system_instruction", None
+                                ):
+                                    system_message = self._settings.system_instruction
+                                else:
+                                    # Fall back to extracting from context messages
+                                    ctx_messages = context.get_messages()
+                                    if ctx_messages:
+                                        first = ctx_messages[0]
+                                        if (
+                                            isinstance(first, dict)
+                                            and first.get("role") == "system"
+                                        ):
+                                            content = first.get("content")
+                                            if isinstance(content, str):
+                                                system_message = content
+                                            elif isinstance(content, list):
+                                                system_message = " ".join(
+                                                    part.get("text", "")
+                                                    for part in content
+                                                    if isinstance(part, dict)
+                                                    and part.get("type") == "text"
+                                                )
+                            elif hasattr(context, "system"):
                                 system_message = context.system
                             elif hasattr(context, "system_message"):
                                 system_message = context.system_message
-                            elif hasattr(self, "_system_instruction"):
-                                system_message = self._system_instruction
 
                             # Use given_fields() defensively in case a service doesn't
                             # initialize all settings.
                             params = {}
                             if hasattr(self, "_settings"):
                                 for key, value in self._settings.given_fields().items():
+                                    # system_instruction is already captured as the
+                                    # "system_instructions" span attribute above.
+                                    if key == "system_instruction":
+                                        continue
                                     if isinstance(value, (int, float, bool, str)):
                                         params[key] = value
                                     elif value is None:
@@ -534,7 +561,7 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
                                 attribute_kwargs["tools"] = serialized_tools
                                 attribute_kwargs["tool_count"] = tool_count
                             if system_message:
-                                attribute_kwargs["system"] = system_message
+                                attribute_kwargs["system_instructions"] = system_message
 
                             # Add all gathered attributes to the span
                             add_llm_span_attributes(span=current_span, **attribute_kwargs)