Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,7 @@ typing = [
"huggingface-hub>=1.26.1",
"anthropic>=1.2.0",
"google-genai>=2.21.0",
"openai>=3.7.0",
]
test = [
"dataclasses ; python_full_version < '3.7'",
Expand Down
18 changes: 11 additions & 7 deletions sentry_sdk/ai/_openai_completions_api.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
from collections.abc import Iterable
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, cast

if TYPE_CHECKING:
from typing import Union
from typing import TypeGuard, Union

from openai.types.chat import (
ChatCompletionContentPartParam,
ChatCompletionContentPartRefusalParam,
ChatCompletionContentPartTextParam,
ChatCompletionMessageParam,
ChatCompletionSystemMessageParam,
ChatCompletionToolUnionParam,
Expand All @@ -14,32 +16,34 @@
from sentry_sdk._types import TextPart, ToolDefinition


def _is_system_instruction(message: "ChatCompletionMessageParam") -> bool:
def _is_system_instruction(
message: "ChatCompletionMessageParam",
) -> "TypeGuard[ChatCompletionSystemMessageParam]":
return isinstance(message, dict) and message.get("role") == "system"


def _get_system_instructions(
messages: "Iterable[ChatCompletionMessageParam]",
) -> "list[ChatCompletionMessageParam]":
) -> "list[ChatCompletionSystemMessageParam]":
if not isinstance(messages, Iterable):
return []

return [message for message in messages if _is_system_instruction(message)]


def _get_text_items(
content: "Union[str, Iterable[ChatCompletionContentPartParam]]",
content: "Union[str, Iterable[Union[ChatCompletionContentPartParam, ChatCompletionContentPartTextParam, ChatCompletionContentPartRefusalParam]]]",
) -> "list[str]":
if isinstance(content, str):
return [content]

if not isinstance(content, Iterable):
return []

text_items = []
text_items: "list[str]" = []
for part in content:
if isinstance(part, dict) and part.get("type") == "text":
text = part.get("text", None)
text = cast("ChatCompletionContentPartTextParam", part).get("text", None)
if text is not None:
text_items.append(text)

Expand Down
8 changes: 5 additions & 3 deletions sentry_sdk/ai/_openai_responses_api.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from typing import TYPE_CHECKING, Iterable, cast

if TYPE_CHECKING:
from typing import Iterable, Union
from typing import Iterable, TypeGuard, Union

from openai.types.responses import (
ResponseInputItemParam,
Expand All @@ -15,7 +15,9 @@
from sentry_sdk._types import TextPart, ToolDefinition


def _is_system_instruction(message: "ResponseInputItemParam") -> bool:
def _is_system_instruction(
message: "ResponseInputItemParam",
) -> "TypeGuard[Union[EasyInputMessageParam, Message]]":
if not isinstance(message, dict) or not message.get("role") == "system":
return False

Expand All @@ -24,7 +26,7 @@ def _is_system_instruction(message: "ResponseInputItemParam") -> bool:

def _get_system_instructions(
messages: "Union[str, ResponseInputParam]",
) -> "list[ResponseInputItemParam]":
) -> "list[Union[EasyInputMessageParam, Message]]":
if not isinstance(messages, list):
return []

Expand Down
Loading
Loading