feat(openai): Set system instruction attribute

sentry_sdk/ai/_openai_completions_api.py

@@ -0,0 +1,20 @@
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from openai.types.chat import (
+        ChatCompletionMessageParam,
+        ChatCompletionSystemMessageParam,
+    )
+    from typing import Iterable, Union
+
+
+def _get_system_instructions(
+    messages: "Iterable[Union[ChatCompletionMessageParam, str]]",
+) -> "list[ChatCompletionSystemMessageParam]":
+    system_messages = []
+
+    for message in messages:
+        if isinstance(message, dict) and message.get("role") == "system":
+            system_messages.append(message)
+
+    return system_messages

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,9 @@
     normalize_message_roles,
     truncate_and_annotate_messages,
 )
+from sentry_sdk.ai._openai_completions_api import (
+    _get_system_instructions as _get_system_instructions_completions,
+)
 from sentry_sdk.consts import SPANDATA
 from sentry_sdk.integrations import DidNotEnable, Integration
 from sentry_sdk.scope import should_send_default_pii
@@ -35,7 +38,7 @@
     )
     from sentry_sdk.tracing import Span
 
-    from openai.types.responses import ResponseInputParam
+    from openai.types.responses import ResponseInputParam, ResponseInputItemParam
 
 try:
     try:
@@ -193,6 +196,25 @@ def _calculate_token_usage(
     )
 
 
+def _get_system_instructions_responses(
+    input_items: "Union[ResponseInputParam, list[str]]",
+) -> "list[ResponseInputItemParam]":
+    if isinstance(input_items, str):
+        return []
+
+    system_messages = []
+
+    for item in input_items:
+        if (
+            isinstance(item, dict)
+            and item.get("type") == "message"
+            and item.get("role") == "system"
+        ):
+            system_messages.append(item)
+
+    return system_messages
+
+
 def _get_input_messages(
     kwargs: "dict[str, Any]",
 ) -> "Optional[Union[Iterable[Any], list[str]]]":
@@ -247,6 +269,15 @@ def _set_responses_api_input_data(
         kwargs
     )
 
+    if messages is not None:
+        system_instructions = _get_system_instructions_responses(messages)
+        set_data_normalized(
+            span,
+            SPANDATA.GEN_AI_SYSTEM_INSTRUCTIONS,
+            system_instructions,
+            unpack=False,
+        )
+
     if (
         messages is not None
         and len(messages) > 0
@@ -274,6 +305,15 @@ def _set_completions_api_input_data(
         _get_input_messages(kwargs)
     )
 
+    if messages is not None:
+        system_instructions = _get_system_instructions_completions(messages)
+        set_data_normalized(
+            span,
+            SPANDATA.GEN_AI_SYSTEM_INSTRUCTIONS,
+            system_instructions,
+            unpack=False,
+        )
+
     if (
         messages is not None
         and len(messages) > 0  # type: ignore