类class agentscope.agent

RealtimeAgent

class RealtimeAgent

源码Source agentscope/agent/_realtime/_agent.py:102–1345

实时智能体，驱动智能体的推理、工具调用与会话执行。位于 agentscope.agent 包。核心成员包括 connect、close、reply_stream、send、interrupt、checkpoint_snapshot。

Unlike :class:~agentscope.agent.Agent it is bidirectional and has no request/reply boundary — audio flows in continuously while events flow out of :meth:reply_stream.

Three lifetimes are kept apart on purpose. The agent owns the model session and the state; a transport is owned by whoever created it; a :meth:reply_stream borrows both for as long as both are alive. So a client can drop and reconnect without losing the model session, and the model session can time out during a long silence and be re-established on the next word without touching the transport.

The public methods are the only entry point for discrete input. A transport carrying a browser's ControlFrame calls exactly those methods rather than reaching into the agent, so one code path handles the semantics whether the caller is Python or a browser. Continuous audio is separate and always arrives through the transport.

示例Example

agent = RealtimeAgent("Friday", "Be brief.", model)
async with agent:                              # model session
    async with LocalAudioTransport() as t:     # sound card
        async for event in agent.reply_stream(t):       # this call
            print(event)

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(name: str, system_prompt: str, model:` `RealtimeModelBase, toolkit: Toolkit \| None = None, state: AgentState…`	初始化 `RealtimeAgent` 实例。 Initialize the realtime agent.

属性摘要Property Summary

属性Property	说明Description
`last_turn_metrics`	获取最后轮次指标。 Latency breakdown of the most recent turn.

方法摘要Method Summary

方法Method	说明Description
`async def __aenter__() -> 'RealtimeAgent'`async	进入异步上下文并返回自身。 Connect on entry.
`async def __aexit__(*exc: Any) -> None`async	退出异步上下文并清理资源。 Disconnect on exit.
`async def connect() -> None`async	建立连接并完成初始化。 Open the model session. Safe to call again after the provider closed it; a no-op while connected.
`async def close() -> None`async	关闭连接并释放相关资源。 Cancel everything in flight and close the model session.
`async def reply_stream(transport:` `TransportBase) -> AsyncIterator[AgentEvent]`async	流式回复输入消息，并逐个产出事件或消息。 Pump transport and yield agent events until it ends.
`async def send(inputs: str \|` `Msg \| UserConfirmResultEvent \| UserInterruptEvent) -> None`async	设置发送。 Feed the agent anything that is not audio.
`async def interrupt() -> None`async	请求中断当前执行。 Stop the active reply, as when the user presses stop.
`def checkpoint_snapshot(event: AgentEvent) ->` `AgentState \| None`	设置检查点快照。 Return the immutable state paired with the yielded event.

构造器详情Constructor Detail

RealtimeAgent

def __init__(name: str, system_prompt: str, model: RealtimeModelBase, toolkit: Toolkit | None = None, state: AgentState | None = None, vad: VADBase | None = None, aggregator: TurnAggregator | None = None) -> None

初始化 RealtimeAgent 实例。

Initialize the realtime agent.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	名称（类型 `str`）。Display name stamped on assistant messages and events.
`system_prompt`	str	系统提示词（类型 `str`）。System instructions sent to the model on connect, with the toolkit's skill instructions appended.
`model`	RealtimeModelBase	模型实例（类型 `RealtimeModelBase`）。The realtime model. Its session is opened by :meth:`connect` and lives as long as this agent — not as long as any one transport — and is re-established on the next user input if the provider closes it.
`toolkit`默认default `None`	Toolkit \| None	工具包（类型 `Toolkit \| None`）。Tools the model may call. Executed here, with permission checks against `state.permission_context`.
`state`默认default `None`	AgentState \| None	状态对象（类型 `AgentState \| None`）。Conversation history, permission rules and tool context. A new one is created if omitted.
`vad`默认default `None`	VADBase \| None	语音活动检测（类型 `VADBase \| None`）。Server-side voice activity detection over the audio arriving from the transport. `STARTED` cuts off a reply the user speaks over; `ENDED` closes the user's turn and hands it to the model. When given, the provider's own turn detection is switched off so that exactly one source decides turn boundaries; when `None`, the provider decides and this agent only reacts to what it reports.
`aggregator`默认default `None`	TurnAggregator \| None	聚合器（类型 `TurnAggregator \| None`）。Collapses the provider's transcripts into clean user turns: merging one that endpointing split, dropping bare acknowledgements. Subclass it to change what counts as a turn. A default with no backchannel list is used if omitted.

源码Source agentscope/agent/_realtime/_agent.py:133–212

属性详情Property Detail

last_turn_metrics

def last_turn_metrics() -> TurnMetrics

获取最后轮次指标。

Latency breakdown of the most recent turn.

返回Returns TurnMetrics

返回TurnMetrics。Returns turn metrics.

源码Source agentscope/agent/_realtime/_agent.py:546–548

方法详情Method Detail

aenter

async def __aenter__() -> 'RealtimeAgent'

async

进入异步上下文并返回自身。

Connect on entry.

返回Returns 'RealtimeAgent'

返回'RealtimeAgent'。Returns realtime agent.

源码Source agentscope/agent/_realtime/_agent.py:218–221

aexit

async def __aexit__(*exc: Any) -> None

async

退出异步上下文并清理资源。

Disconnect on exit.

参数Parameters

名称Name	类型Type	说明Description
`exc`	Any	exc（类型 `Any`）。The exc.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/agent/_realtime/_agent.py:223–225

connect

async def connect() -> None

async

建立连接并完成初始化。

Open the model session.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/agent/_realtime/_agent.py:227–327

close

async def close() -> None

async

关闭连接并释放相关资源。

Cancel everything in flight and close the model session.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/agent/_realtime/_agent.py:329–344

reply_stream

async def reply_stream(transport: TransportBase) -> AsyncIterator[AgentEvent]

async

流式回复输入消息，并逐个产出事件或消息。

The transport is borrowed, not owned: it must already be started and is not closed here, so the caller can keep it, reuse it or hand it on.

参数Parameters

名称Name	类型Type	说明Description
`transport`	TransportBase	传输层（类型 `TransportBase`）。The transport.

返回Returns AsyncIterator[AgentEvent]

以异步流的形式返回AgentEvent。Returns a stream of agent event.

源码Source agentscope/agent/_realtime/_agent.py:394–484

send

async def send(inputs: str | Msg | UserConfirmResultEvent | UserInterruptEvent) -> None

async

设置发送。

Mirrors :meth:Agent.reply's inputs: a text turn (which cuts off any reply in progress first), the outcome of a permission prompt, or an interrupt.

参数Parameters

名称Name	类型Type	说明Description
`inputs`	str \| Msg \| UserConfirmResultEvent \| UserInterruptEvent	输入消息（类型 `str \| Msg \| UserConfirmResultEvent \| UserInterruptEvent`）。The inputs.

返回Returns None

无返回值。No value is returned.

异常Raises

可能抛出 `Exception`。`NotImplementedError`: For a text turn when the provider takes
可能抛出 `Exception`。no text input.
当用于文本轮次当提供者的`ModelDisconnectedError`时抛出 `Exception`。`ModelDisconnectedError`: For a text turn when the provider
可能抛出 `Exception`。cannot be reached, or disconnects before accepting it.

源码Source agentscope/agent/_realtime/_agent.py:490–539

interrupt

async def interrupt() -> None

async

请求中断当前执行。

Stop the active reply, as when the user presses stop.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/agent/_realtime/_agent.py:541–543

checkpoint_snapshot

def checkpoint_snapshot(event: AgentEvent) -> AgentState | None

设置检查点快照。

Return the immutable state paired with the yielded event.

参数Parameters

名称Name	类型Type	说明Description
`event`	AgentEvent	事件（类型 `AgentEvent`）。The event.

返回Returns AgentState | None

返回AgentState|None。Returns agent state none.

源码Source agentscope/agent/_realtime/_agent.py:1118–1123