feat(openai): Set system instruction attribute

sentry_sdk/_types.py

@@ -359,3 +359,7 @@ class SDKInfo(TypedDict):
     )
 
     HttpStatusCodeRange = Union[int, Container[int]]
+
+    class TextPart(TypedDict):
+        type: Literal["text"]
+        content: str

sentry_sdk/ai/_openai_completions_api.py

@@ -0,0 +1,50 @@
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from openai.types.chat import (
+        ChatCompletionMessageParam,
+        ChatCompletionSystemMessageParam,
+    )
+    from typing import Iterable
+
+    from sentry_sdk._types import TextPart
+
+
+def _is_system_instruction(message: "ChatCompletionMessageParam") -> bool:
+    return isinstance(message, dict) and message.get("role") == "system"
+
+
+def _get_system_instructions(
+    messages: "Iterable[ChatCompletionMessageParam]",
+) -> "list[ChatCompletionSystemMessageParam]":
+    system_instructions = []
+
+    for message in messages:
+        if _is_system_instruction(message):
+            system_instructions.append(message)
+
+    return system_instructions
+
+
+def _transform_system_instructions(
+    system_instructions: "list[ChatCompletionSystemMessageParam]",
+) -> "list[TextPart]":
+    instruction_text_parts: "list[TextPart]" = []
+
+    for instruction in system_instructions:
+        if not isinstance(instruction, dict):
+            continue
+
+        content = instruction.get("content")
+
+        if isinstance(content, str):
+            instruction_text_parts.append({"type": "text", "content": content})
+
+        elif isinstance(content, list):
+            for part in content:
+                if isinstance(part, dict) and part.get("type") == "text":
+                    text = part.get("text", "")
+                    if text:
+                        instruction_text_parts.append({"type": "text", "content": text})
+
+    return instruction_text_parts

sentry_sdk/consts.py

@@ -542,6 +542,12 @@ class SPANDATA:
     Example: 2048
     """
 
+    GEN_AI_SYSTEM_INSTRUCTIONS = "gen_ai.system_instructions"
+    """
+    The system instructions passed to the model.
+    Example: [{"type": "text", "text": "You are a helpful assistant."},{"type": "text", "text": "Be concise and clear."}]
+    """
+
     GEN_AI_REQUEST_MESSAGES = "gen_ai.request.messages"
     """
     The messages passed to the model. The "content" can be a string or an array of objects.

sentry_sdk/integrations/openai.py

@@ -9,6 +9,11 @@
     normalize_message_roles,
     truncate_and_annotate_messages,
 )
+from sentry_sdk.ai._openai_completions_api import (
+    _get_system_instructions as _get_system_instructions_completions,
+    _is_system_instruction as _is_system_instruction_completions,
+    _transform_system_instructions,
+)
 from sentry_sdk.consts import SPANDATA
 from sentry_sdk.integrations import DidNotEnable, Integration
 from sentry_sdk.scope import should_send_default_pii
@@ -35,7 +40,8 @@
     )
     from sentry_sdk.tracing import Span
 
-    from openai.types.responses import ResponseInputParam
+    from openai.types.responses import ResponseInputParam, ResponseInputItemParam
+    from openai import Omit
 
 try:
     try:
@@ -193,6 +199,29 @@ def _calculate_token_usage(
     )
 
 
+def _is_system_instruction_responses(message: "ResponseInputItemParam") -> bool:
+    return (
+        isinstance(message, dict)
+        and message.get("type") == "message"
+        and message.get("role") == "system"
+    )
+
+
+def _get_system_instructions_responses(
+    messages: "Union[str, ResponseInputParam]",
+) -> "list[ResponseInputItemParam]":
+    if isinstance(messages, str):
+        return []
+
+    system_instructions = []
+
+    for message in messages:
+        if _is_system_instruction_responses(message):
+            system_instructions.append(message)
+
+    return system_instructions
+
+
 def _get_input_messages(
     kwargs: "dict[str, Any]",
 ) -> "Optional[Union[Iterable[Any], list[str]]]":
@@ -243,24 +272,68 @@ def _set_responses_api_input_data(
     kwargs: "dict[str, Any]",
     integration: "OpenAIIntegration",
 ) -> None:
-    messages: "Optional[Union[ResponseInputParam, list[str]]]" = _get_input_messages(
-        kwargs
-    )
+    messages: "Optional[Union[str, ResponseInputParam]]" = kwargs.get("input")
 
+    if messages is None:
+        set_data_normalized(span, SPANDATA.GEN_AI_OPERATION_NAME, "responses")
+        _commmon_set_input_data(span, kwargs)
+        return
+
+    explicit_instructions: "Union[Optional[str], Omit]" = kwargs.get("instructions")
+    system_instructions = _get_system_instructions_responses(messages)
     if (
-        messages is not None
-        and len(messages) > 0
+        (_is_given(explicit_instructions) or len(system_instructions) > 0)
         and should_send_default_pii()
         and integration.include_prompts
     ):
-        normalized_messages = normalize_message_roles(messages)  # type: ignore
+        # Deliberate use of function accepting completions API type because
+        # of shared structure FOR THIS PURPOSE ONLY.
+        instructions_text_parts = _transform_system_instructions(system_instructions)
+        if _is_given(explicit_instructions):
+            instructions_text_parts.append(
+                {
+                    "type": "text",
+                    "content": explicit_instructions,  # type: ignore
+                }
+            )
+
+        set_data_normalized(
+            span,
+            SPANDATA.GEN_AI_SYSTEM_INSTRUCTIONS,
+            instructions_text_parts,
+            unpack=False,
+        )
+
+    if (
+        isinstance(messages, str)
+        and should_send_default_pii()
+        and integration.include_prompts
+    ):
+        normalized_messages = normalize_message_roles([messages])  # type: ignore
         scope = sentry_sdk.get_current_scope()
         messages_data = truncate_and_annotate_messages(normalized_messages, span, scope)
         if messages_data is not None:
             set_data_normalized(
                 span, SPANDATA.GEN_AI_REQUEST_MESSAGES, messages_data, unpack=False
             )
 
+    elif should_send_default_pii() and integration.include_prompts:
+        non_system_messages = [
+            message
+            for message in messages
+            if not _is_system_instruction_responses(message)
+        ]
+        if len(non_system_messages) > 0:
+            normalized_messages = normalize_message_roles(non_system_messages)  # type: ignore
+            scope = sentry_sdk.get_current_scope()
+            messages_data = truncate_and_annotate_messages(
+                normalized_messages, span, scope
+            )
+            if messages_data is not None:
+                set_data_normalized(
+                    span, SPANDATA.GEN_AI_REQUEST_MESSAGES, messages_data, unpack=False
+                )
+
     set_data_normalized(span, SPANDATA.GEN_AI_OPERATION_NAME, "responses")
     _commmon_set_input_data(span, kwargs)
 
@@ -270,26 +343,56 @@ def _set_completions_api_input_data(
     kwargs: "dict[str, Any]",
     integration: "OpenAIIntegration",
 ) -> None:
-    messages: "Optional[Union[Iterable[ChatCompletionMessageParam], list[str]]]" = (
-        _get_input_messages(kwargs)
+    messages: "Optional[Union[str, Iterable[ChatCompletionMessageParam]]]" = kwargs.get(
+        "messages"
     )
 
+    if messages is None:
+        set_data_normalized(span, SPANDATA.GEN_AI_OPERATION_NAME, "chat")
+        _commmon_set_input_data(span, kwargs)
+        return
+
+    system_instructions = _get_system_instructions_completions(messages)
     if (
-        messages is not None
-        and len(messages) > 0  # type: ignore
+        len(system_instructions) > 0
         and should_send_default_pii()
         and integration.include_prompts
     ):
-        normalized_messages = normalize_message_roles(messages)  # type: ignore
+        set_data_normalized(
+            span,
+            SPANDATA.GEN_AI_SYSTEM_INSTRUCTIONS,
+            _transform_system_instructions(system_instructions),
+            unpack=False,
+        )
+
+    if (
+        isinstance(messages, str)
+        and should_send_default_pii()
+        and integration.include_prompts
+    ):
+        normalized_messages = normalize_message_roles([messages])  # type: ignore
         scope = sentry_sdk.get_current_scope()
         messages_data = truncate_and_annotate_messages(normalized_messages, span, scope)
         if messages_data is not None:
             set_data_normalized(
                 span, SPANDATA.GEN_AI_REQUEST_MESSAGES, messages_data, unpack=False
             )
-
-    set_data_normalized(span, SPANDATA.GEN_AI_OPERATION_NAME, "chat")
-    _commmon_set_input_data(span, kwargs)
+    elif should_send_default_pii() and integration.include_prompts:
+        non_system_messages = [
+            message
+            for message in messages
+            if not _is_system_instruction_completions(message)
+        ]
+        if len(non_system_messages) > 0:
+            normalized_messages = normalize_message_roles(non_system_messages)  # type: ignore
+            scope = sentry_sdk.get_current_scope()
+            messages_data = truncate_and_annotate_messages(
+                normalized_messages, span, scope
+            )
+            if messages_data is not None:
+                set_data_normalized(
+                    span, SPANDATA.GEN_AI_REQUEST_MESSAGES, messages_data, unpack=False
+                )
 
 
 def _set_embeddings_input_data(