RealtimeRunner

RealtimeRunner 是针对实时代理的 Runner 的等效物。它通过与底层模型层保持持久连接，自动处理多个回合。

会话管理本地历史记录副本，执行工具，运行护栏，并促进代理之间的交接。

由于此代码在您的服务器上运行，因此默认使用 WebSockets。您可以选择通过实现 RealtimeModel 接口来创建自己的自定义模型层。

源代码位于 src/agents/realtime/runner.py

class RealtimeRunner:
    """A `RealtimeRunner` is the equivalent of `Runner` for realtime agents. It automatically
    handles multiple turns by maintaining a persistent connection with the underlying model
    layer.

    The session manages the local history copy, executes tools, runs guardrails and facilitates
    handoffs between agents.

    Since this code runs on your server, it uses WebSockets by default. You can optionally create
    your own custom model layer by implementing the `RealtimeModel` interface.
    """

    def __init__(
        self,
        starting_agent: RealtimeAgent,
        *,
        model: RealtimeModel | None = None,
        config: RealtimeRunConfig | None = None,
    ) -> None:
        """Initialize the realtime runner.

        Args:
            starting_agent: The agent to start the session with.
            context: The context to use for the session.
            model: The model to use. If not provided, will use a default OpenAI realtime model.
            config: Override parameters to use for the entire run.
        """
        self._starting_agent = starting_agent
        self._config = config
        self._model = model or OpenAIRealtimeWebSocketModel()

    async def run(
        self, *, context: TContext | None = None, model_config: RealtimeModelConfig | None = None
    ) -> RealtimeSession:
        """Start and returns a realtime session.

        Returns:
            RealtimeSession: A session object that allows bidirectional communication with the
            realtime model.

        Example:
            ```python
            runner = RealtimeRunner(agent)
            async with await runner.run() as session:
                await session.send_message("Hello")
                async for event in session:
                    print(event)
            ```
        """
        # Create and return the connection
        session = RealtimeSession(
            model=self._model,
            agent=self._starting_agent,
            context=context,
            model_config=model_config,
            run_config=self._config,
        )

        return session

__init__

__init__(
    starting_agent: RealtimeAgent,
    *,
    model: RealtimeModel | None = None,
    config: RealtimeRunConfig | None = None,
) -> None

初始化实时运行器。

参数

名称	类型	描述	默认
starting_agent	RealtimeAgent	用于启动会话的代理。	required
context		会话要使用的上下文。	required
model	RealtimeModel | None	要使用的模型。如果未提供，将使用默认的 OpenAI 实时模型。	None
config	RealtimeRunConfig | None	用于整个运行的参数覆盖。	None

源代码位于 src/agents/realtime/runner.py

def __init__(
    self,
    starting_agent: RealtimeAgent,
    *,
    model: RealtimeModel | None = None,
    config: RealtimeRunConfig | None = None,
) -> None:
    """Initialize the realtime runner.

    Args:
        starting_agent: The agent to start the session with.
        context: The context to use for the session.
        model: The model to use. If not provided, will use a default OpenAI realtime model.
        config: Override parameters to use for the entire run.
    """
    self._starting_agent = starting_agent
    self._config = config
    self._model = model or OpenAIRealtimeWebSocketModel()

run async

run(
    *,
    context: TContext | None = None,
    model_config: RealtimeModelConfig | None = None,
) -> RealtimeSession

启动并返回一个实时会话。

返回值

名称	类型	描述
RealtimeSession	RealtimeSession	一个会话对象，允许与实时模型进行双向通信。
	RealtimeSession	实时模型。

示例

runner = RealtimeRunner(agent)
async with await runner.run() as session:
    await session.send_message("Hello")
    async for event in session:
        print(event)

源代码位于 src/agents/realtime/runner.py

async def run(
    self, *, context: TContext | None = None, model_config: RealtimeModelConfig | None = None
) -> RealtimeSession:
    """Start and returns a realtime session.

    Returns:
        RealtimeSession: A session object that allows bidirectional communication with the
        realtime model.

    Example:
        ```python
        runner = RealtimeRunner(agent)
        async with await runner.run() as session:
            await session.send_message("Hello")
            async for event in session:
                print(event)
        ```
    """
    # Create and return the connection
    session = RealtimeSession(
        model=self._model,
        agent=self._starting_agent,
        context=context,
        model_config=model_config,
        run_config=self._config,
    )

    return session