模型

实时播放状态

基础: TypedDict

源代码在 src/agents/realtime/model.py

class RealtimePlaybackState(TypedDict):
    current_item_id: str | None
    """The item ID of the current item being played."""

    current_item_content_index: int | None
    """The index of the current item content being played."""

    elapsed_ms: float | None
    """The number of milliseconds of audio that have been played."""

current_item_id 实例属性

current_item_id: str | None

当前正在播放的项目的 ID。

current_item_content_index 实例属性

current_item_content_index: int | None

当前正在播放的项目内容的索引。

elapsed_ms 实例属性

elapsed_ms: float | None

已播放的音频的毫秒数。

实时播放跟踪器

如果您有自定义播放逻辑或预计音频播放存在延迟或以不同的速度进行，请创建一个 RealtimePlaybackTracker 实例并将其传递给会话。您负责跟踪音频播放进度，并在用户播放了一些音频时调用 on_play_bytes 或 on_play_ms。

源代码在 src/agents/realtime/model.py

class RealtimePlaybackTracker:
    """If you have custom playback logic or expect that audio is played with delays or at different
    speeds, create an instance of RealtimePlaybackTracker and pass it to the session. You are
    responsible for tracking the audio playback progress and calling `on_play_bytes` or
    `on_play_ms` when the user has played some audio."""

    def __init__(self) -> None:
        self._format: RealtimeAudioFormat | None = None
        # (item_id, item_content_index)
        self._current_item: tuple[str, int] | None = None
        self._elapsed_ms: float | None = None

    def on_play_bytes(self, item_id: str, item_content_index: int, bytes: bytes) -> None:
        """Called by you when you have played some audio.

        Args:
            item_id: The item ID of the audio being played.
            item_content_index: The index of the audio content in `item.content`
            bytes: The audio bytes that have been fully played.
        """
        ms = calculate_audio_length_ms(self._format, bytes)
        self.on_play_ms(item_id, item_content_index, ms)

    def on_play_ms(self, item_id: str, item_content_index: int, ms: float) -> None:
        """Called by you when you have played some audio.

        Args:
            item_id: The item ID of the audio being played.
            item_content_index: The index of the audio content in `item.content`
            ms: The number of milliseconds of audio that have been played.
        """
        if self._current_item != (item_id, item_content_index):
            self._current_item = (item_id, item_content_index)
            self._elapsed_ms = ms
        else:
            assert self._elapsed_ms is not None
            self._elapsed_ms += ms

    def on_interrupted(self) -> None:
        """Called by the model when the audio playback has been interrupted."""
        self._current_item = None
        self._elapsed_ms = None

    def set_audio_format(self, format: RealtimeAudioFormat) -> None:
        """Will be called by the model to set the audio format.

        Args:
            format: The audio format to use.
        """
        self._format = format

    def get_state(self) -> RealtimePlaybackState:
        """Will be called by the model to get the current playback state."""
        if self._current_item is None:
            return {
                "current_item_id": None,
                "current_item_content_index": None,
                "elapsed_ms": None,
            }
        assert self._elapsed_ms is not None

        item_id, item_content_index = self._current_item
        return {
            "current_item_id": item_id,
            "current_item_content_index": item_content_index,
            "elapsed_ms": self._elapsed_ms,
        }

on_play_bytes

on_play_bytes(
    item_id: str, item_content_index: int, bytes: bytes
) -> None

当您播放了一些音频时调用。

参数

名称	类型	描述	默认
item_id	str	正在播放的音频的项目 ID。	required
item_content_index	int	item.content 中的音频内容的索引	required
bytes	bytes	已完全播放的音频字节数。	required

源代码在 src/agents/realtime/model.py

def on_play_bytes(self, item_id: str, item_content_index: int, bytes: bytes) -> None:
    """Called by you when you have played some audio.

    Args:
        item_id: The item ID of the audio being played.
        item_content_index: The index of the audio content in `item.content`
        bytes: The audio bytes that have been fully played.
    """
    ms = calculate_audio_length_ms(self._format, bytes)
    self.on_play_ms(item_id, item_content_index, ms)

on_play_ms

on_play_ms(
    item_id: str, item_content_index: int, ms: float
) -> None

当您播放了一些音频时调用。

参数

名称	类型	描述	默认
item_id	str	正在播放的音频的项目 ID。	required
item_content_index	int	item.content 中的音频内容的索引	required
ms	float	已播放的音频的毫秒数。	required

源代码在 src/agents/realtime/model.py

def on_play_ms(self, item_id: str, item_content_index: int, ms: float) -> None:
    """Called by you when you have played some audio.

    Args:
        item_id: The item ID of the audio being played.
        item_content_index: The index of the audio content in `item.content`
        ms: The number of milliseconds of audio that have been played.
    """
    if self._current_item != (item_id, item_content_index):
        self._current_item = (item_id, item_content_index)
        self._elapsed_ms = ms
    else:
        assert self._elapsed_ms is not None
        self._elapsed_ms += ms

on_interrupted

on_interrupted() -> None

当音频播放被中断时，由模型调用。

源代码在 src/agents/realtime/model.py

def on_interrupted(self) -> None:
    """Called by the model when the audio playback has been interrupted."""
    self._current_item = None
    self._elapsed_ms = None

set_audio_format

set_audio_format(format: RealtimeAudioFormat) -> None

模型将调用此方法来设置音频格式。

参数

名称	类型	描述	默认
format	RealtimeAudioFormat	要使用的音频格式。	required

源代码在 src/agents/realtime/model.py

def set_audio_format(self, format: RealtimeAudioFormat) -> None:
    """Will be called by the model to set the audio format.

    Args:
        format: The audio format to use.
    """
    self._format = format

get_state

get_state() -> RealtimePlaybackState

模型将调用此方法来获取当前的播放状态。

源代码在 src/agents/realtime/model.py

def get_state(self) -> RealtimePlaybackState:
    """Will be called by the model to get the current playback state."""
    if self._current_item is None:
        return {
            "current_item_id": None,
            "current_item_content_index": None,
            "elapsed_ms": None,
        }
    assert self._elapsed_ms is not None

    item_id, item_content_index = self._current_item
    return {
        "current_item_id": item_id,
        "current_item_content_index": item_content_index,
        "elapsed_ms": self._elapsed_ms,
    }

实时模型监听器

基础: ABC

实时传输事件的监听器。

源代码在 src/agents/realtime/model.py

class RealtimeModelListener(abc.ABC):
    """A listener for realtime transport events."""

    @abc.abstractmethod
    async def on_event(self, event: RealtimeModelEvent) -> None:
        """Called when an event is emitted by the realtime transport."""
        pass

on_event 抽象方法 异步

on_event(event: RealtimeModelEvent) -> None

当实时传输发出事件时调用。

源代码在 src/agents/realtime/model.py

@abc.abstractmethod
async def on_event(self, event: RealtimeModelEvent) -> None:
    """Called when an event is emitted by the realtime transport."""
    pass

实时模型配置

基础: TypedDict

连接到实时模型的选项。

源代码在 src/agents/realtime/model.py

class RealtimeModelConfig(TypedDict):
    """Options for connecting to a realtime model."""

    api_key: NotRequired[str | Callable[[], MaybeAwaitable[str]]]
    """The API key (or function that returns a key) to use when connecting. If unset, the model will
    try to use a sane default. For example, the OpenAI Realtime model will try to use the
    `OPENAI_API_KEY`  environment variable.
    """

    url: NotRequired[str]
    """The URL to use when connecting. If unset, the model will use a sane default. For example,
    the OpenAI Realtime model will use the default OpenAI WebSocket URL.
    """

    headers: NotRequired[dict[str, str]]
    """The headers to use when connecting. If unset, the model will use a sane default.
    Note that, when you set this, authorization header won't be set under the hood.
    e.g., {"api-key": "your api key here"} for Azure OpenAI Realtime WebSocket connections.
    """

    initial_model_settings: NotRequired[RealtimeSessionModelSettings]
    """The initial model settings to use when connecting."""

    playback_tracker: NotRequired[RealtimePlaybackTracker]
    """The playback tracker to use when tracking audio playback progress. If not set, the model will
    use a default implementation that assumes audio is played immediately, at realtime speed.

    A playback tracker is useful for interruptions. The model generates audio much faster than
    realtime playback speed. So if there's an interruption, its useful for the model to know how
    much of the audio has been played by the user. In low-latency scenarios, it's fine to assume
    that audio is played back immediately at realtime speed. But in scenarios like phone calls or
    other remote interactions, you can set a playback tracker that lets the model know when audio
    is played to the user.
    """

    call_id: NotRequired[str]
    """Attach to an existing realtime call instead of creating a new session.

    When provided, the transport connects using the `call_id` query string parameter rather than a
    model name. This is used for SIP-originated calls that are accepted via the Realtime Calls API.
    """

api_key 实例属性

api_key: NotRequired[
    str | Callable[[], MaybeAwaitable[str]]
]

连接时要使用的 API 密钥（或返回密钥的函数）。如果未设置，模型将尝试使用合理的默认值。例如，OpenAI 实时模型将尝试使用 OPENAI_API_KEY 环境变量。

url 实例属性

url: NotRequired[str]

连接时要使用的 URL。如果未设置，模型将使用合理的默认值。例如，OpenAI 实时模型将使用默认的 OpenAI WebSocket URL。

headers 实例属性

headers: NotRequired[dict[str, str]]

连接时要使用的标头。如果未设置，模型将使用合理的默认值。请注意，当您设置此项时，授权标头不会在底层设置。例如，对于 Azure OpenAI 实时 WebSocket 连接，可以使用 {"api-key": "您的 api 密钥"}。

initial_model_settings 实例属性

initial_model_settings: NotRequired[
    RealtimeSessionModelSettings
]

连接时要使用的初始模型设置。

playback_tracker 实例属性

playback_tracker: NotRequired[RealtimePlaybackTracker]

用于跟踪音频播放进度的播放跟踪器。如果未设置，模型将使用默认实现，该实现假定音频立即以实时速度播放。

播放跟踪器对于中断很有用。模型以比实时播放速度快得多的速度生成音频。因此，如果发生中断，模型知道用户播放了多少音频是有用的。在低延迟场景中，假设音频立即以实时速度播放是可以的。但在电话或其他远程交互等场景中，您可以设置一个播放跟踪器，让模型知道何时将音频播放给用户。

call_id 实例属性

call_id: NotRequired[str]

附加到现有的实时通话，而不是创建新的会话。

当提供时，传输将使用 call_id 查询字符串参数而不是模型名称进行连接。这用于通过实时通话 API 接受的 SIP 发起的呼叫。

实时模型

基础: ABC

连接到实时模型并发送/接收事件的接口。

源代码在 src/agents/realtime/model.py

class RealtimeModel(abc.ABC):
    """Interface for connecting to a realtime model and sending/receiving events."""

    @abc.abstractmethod
    async def connect(self, options: RealtimeModelConfig) -> None:
        """Establish a connection to the model and keep it alive."""
        pass

    @abc.abstractmethod
    def add_listener(self, listener: RealtimeModelListener) -> None:
        """Add a listener to the model."""
        pass

    @abc.abstractmethod
    def remove_listener(self, listener: RealtimeModelListener) -> None:
        """Remove a listener from the model."""
        pass

    @abc.abstractmethod
    async def send_event(self, event: RealtimeModelSendEvent) -> None:
        """Send an event to the model."""
        pass

    @abc.abstractmethod
    async def close(self) -> None:
        """Close the session."""
        pass

connect 抽象方法 异步

connect(options: RealtimeModelConfig) -> None

建立与模型的连接并保持其活动状态。

源代码在 src/agents/realtime/model.py

@abc.abstractmethod
async def connect(self, options: RealtimeModelConfig) -> None:
    """Establish a connection to the model and keep it alive."""
    pass

add_listener 抽象方法

add_listener(listener: RealtimeModelListener) -> None

将监听器添加到模型。

源代码在 src/agents/realtime/model.py

@abc.abstractmethod
def add_listener(self, listener: RealtimeModelListener) -> None:
    """Add a listener to the model."""
    pass

remove_listener 抽象方法

remove_listener(listener: RealtimeModelListener) -> None

从模型中删除监听器。

源代码在 src/agents/realtime/model.py

@abc.abstractmethod
def remove_listener(self, listener: RealtimeModelListener) -> None:
    """Remove a listener from the model."""
    pass

send_event 抽象方法 异步

send_event(event: RealtimeModelSendEvent) -> None

将事件发送到模型。

源代码在 src/agents/realtime/model.py

@abc.abstractmethod
async def send_event(self, event: RealtimeModelSendEvent) -> None:
    """Send an event to the model."""
    pass

close 抽象方法 异步

close() -> None

关闭会话。

源代码在 src/agents/realtime/model.py

@abc.abstractmethod
async def close(self) -> None:
    """Close the session."""
    pass