feat(types): add is_given TypeGuard helpers for NotGiven sentinels

Pyright can't narrow identity checks against module-level NotGiven
sentinels (they aren't typed as singletons), which leaves many
NotGiven-bearing unions stuck as unnarrowed types throughout the
codebase. Introduce is_given TypeGuard helpers so narrowing works via
isinstance under the hood.

Each helper is co-located with the NotGiven flavor it guards:

- services/settings.py: upgrade the existing is_given to a TypeGuard.
- processors/aggregators/llm_context.py: add an is_given for
  LLMContext's NotGiven. Treat LLMContext's re-exported types
  (LLMStandardMessage, LLMContextToolChoice, NOT_GIVEN, NotGiven) as
  LLMContext's own — independent definitions that happen to coincide
  with OpenAI's as an implementation detail.
- adapters/services/anthropic_adapter.py: add is_given for anthropic's
  NotGiven.
- adapters/services/open_ai_adapter.py: add is_given for openai's
  NotGiven.
This commit is contained in:
Paul Kompfner
2026-04-23 15:19:32 -04:00
parent 092b1dcb0f
commit 1624d7a474
4 changed files with 93 additions and 9 deletions

View File

@@ -9,7 +9,7 @@
import copy import copy
import json import json
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, TypedDict from typing import Any, TypedDict, TypeGuard, TypeVar
from anthropic import NOT_GIVEN, NotGiven from anthropic import NOT_GIVEN, NotGiven
from anthropic.types.message_param import MessageParam from anthropic.types.message_param import MessageParam
@@ -26,6 +26,29 @@ from pipecat.processors.aggregators.llm_context import (
LLMStandardMessage, LLMStandardMessage,
) )
_T = TypeVar("_T")
def is_given(value: _T | NotGiven) -> TypeGuard[_T]:
"""Check whether a value was explicitly provided.
Typically used when checking whether a parameter or field typed with
Anthropic's ``NotGiven`` was set::
if is_given(system):
...
Also acts as a type guard: inside a true branch, the value is narrowed
to exclude ``NotGiven`` (e.g. ``str | NotGiven`` becomes ``str``).
Args:
value: The value to check.
Returns:
``True`` if *value* is anything other than ``NOT_GIVEN``.
"""
return not isinstance(value, NotGiven)
class AnthropicLLMInvocationParams(TypedDict): class AnthropicLLMInvocationParams(TypedDict):
"""Context-based parameters for invoking Anthropic's LLM API.""" """Context-based parameters for invoking Anthropic's LLM API."""

View File

@@ -6,7 +6,7 @@
"""OpenAI LLM adapter for Pipecat.""" """OpenAI LLM adapter for Pipecat."""
from typing import Any, TypedDict from typing import Any, TypedDict, TypeGuard, TypeVar
from openai._types import NotGiven as OpenAINotGiven from openai._types import NotGiven as OpenAINotGiven
from openai.types.chat import ( from openai.types.chat import (
@@ -25,6 +25,31 @@ from pipecat.processors.aggregators.llm_context import (
NotGiven, NotGiven,
) )
_T = TypeVar("_T")
def is_given(value: _T | OpenAINotGiven) -> TypeGuard[_T]:
"""Check whether a value was explicitly provided.
Typically used when checking whether a parameter or field typed with
OpenAI's ``NotGiven`` was set::
if is_given(tool_choice):
...
Also acts as a type guard: inside a true branch, the value is narrowed
to exclude ``OpenAINotGiven`` (e.g.
``ChatCompletionToolChoiceOptionParam | OpenAINotGiven`` becomes
``ChatCompletionToolChoiceOptionParam``).
Args:
value: The value to check.
Returns:
``True`` if *value* is anything other than ``NOT_GIVEN``.
"""
return not isinstance(value, OpenAINotGiven)
class OpenAILLMInvocationParams(TypedDict): class OpenAILLMInvocationParams(TypedDict):
"""Context-based parameters for invoking OpenAI ChatCompletion API.""" """Context-based parameters for invoking OpenAI ChatCompletion API."""

View File

@@ -21,7 +21,7 @@ import io
import wave import wave
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, TypeAlias from typing import Any, TypeAlias, TypeGuard, TypeVar
from loguru import logger from loguru import logger
from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
@@ -36,16 +36,45 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import AudioRawFrame from pipecat.frames.frames import AudioRawFrame
# "Re-export" types from OpenAI that we're using as universal context types. # "Re-export" types from OpenAI that we're using as universal context types.
# NOTE: if universal message types need to someday diverge from OpenAI's, we # NOTE: these are aliased to OpenAI's today, but callers should treat them as
# should consider managing our own definitions. But we should do so carefully, # LLMContext's own types — independent definitions that happen to coincide
# as the OpenAI messages are somewhat of a standard and we want to continue # with OpenAI's as an implementation detail. If universal context types need
# supporting them. # to someday diverge from OpenAI's, we should consider managing our own
# definitions (but with care, since OpenAI's types are somewhat of a standard
# and we want to continue supporting them). In the meantime, code at the
# LLMContext/OpenAI boundary should use explicit casts rather than rely on
# the aliasing.
LLMStandardMessage = ChatCompletionMessageParam LLMStandardMessage = ChatCompletionMessageParam
LLMContextToolChoice = ChatCompletionToolChoiceOptionParam LLMContextToolChoice = ChatCompletionToolChoiceOptionParam
NOT_GIVEN = OPEN_AI_NOT_GIVEN NOT_GIVEN = OPEN_AI_NOT_GIVEN
NotGiven = OpenAINotGiven NotGiven = OpenAINotGiven
_T = TypeVar("_T")
def is_given(value: _T | NotGiven) -> TypeGuard[_T]:
"""Check whether a value was explicitly provided.
Typically used when checking whether a ``NotGiven``-valued field or
parameter was set::
if is_given(context.tools):
...
Also acts as a type guard: inside a true branch, the value is narrowed
to exclude ``NotGiven`` (e.g. ``ToolsSchema | NotGiven`` becomes
``ToolsSchema``).
Args:
value: The value to check.
Returns:
``True`` if *value* is anything other than ``NOT_GIVEN``.
"""
return not isinstance(value, NotGiven)
@dataclass @dataclass
class LLMSpecificMessage: class LLMSpecificMessage:
"""A container for a context message that is specific to a particular LLM service. """A container for a context message that is specific to a particular LLM service.

View File

@@ -39,7 +39,7 @@ from __future__ import annotations
import copy import copy
from collections.abc import Mapping from collections.abc import Mapping
from dataclasses import dataclass, field, fields from dataclasses import dataclass, field, fields
from typing import TYPE_CHECKING, Any, ClassVar, TypeVar from typing import TYPE_CHECKING, Any, ClassVar, TypeGuard, TypeVar
from loguru import logger from loguru import logger
@@ -88,7 +88,10 @@ Valid only in delta-mode settings objects. Must never appear in a service's
""" """
def is_given(value: Any) -> bool: _T = TypeVar("_T")
def is_given(value: _T | _NotGiven) -> TypeGuard[_T]:
"""Check whether a delta field was explicitly provided. """Check whether a delta field was explicitly provided.
Typically used when processing a delta to decide whether a field Typically used when processing a delta to decide whether a field
@@ -98,6 +101,10 @@ def is_given(value: Any) -> bool:
# caller wants to change the voice # caller wants to change the voice
... ...
Also acts as a type guard: inside a true branch, the value is narrowed
to exclude ``_NotGiven`` (e.g. ``str | None | _NotGiven`` becomes
``str | None``).
For store-mode objects this always returns ``True`` (since For store-mode objects this always returns ``True`` (since
``validate_complete`` ensures no ``NOT_GIVEN`` fields remain). ``validate_complete`` ensures no ``NOT_GIVEN`` fields remain).