agentscope.realtimeRealtimeModelBase
class RealtimeModelBase(ABC)
agentscope/realtime/_base.py:20–237实时模型基类,处理低延迟的实时语音与多模态交互。位于 agentscope.realtime 包,继承自 ABC。核心成员包括 list_models、connect、close、events、push_audio、push_text。
Facts live in three places: class attributes for what the adapter implements, the model card for what the model is, and properties for what is a constant on one provider and a parameter on another.
嵌套类摘要Nested Class Summary
| 类Class | 类型Kind | 说明Description |
|---|---|---|
Parameters | 类class | 参数,处理低延迟的实时语音与多模态交互。 Provider-specific tuneables, surfaced to the UI through the card's ``parameter_schema``. |
构造器摘要Constructor Summary
| 构造器Constructor | 说明Description |
|---|---|
def __init__(model: str, credential: CredentialBase, parameters: 'RealtimeModelBase.Parameters | None' = None, model_ca… | 初始化 `RealtimeModelBase` 实例。 Initialize the realtime model. |
属性摘要Property Summary
| 属性Property | 说明Description |
|---|---|
input_sample_rate | 读取 `input_sample_rate` 属性。 The PCM rate :meth:`push_audio` expects, in Hz. |
output_sample_rate | 音频 in:类:`AudioDelta`,in Hz的PCM 速率。 The PCM rate of the audio in :class:`AudioDelta`, in Hz. |
input_transcription_enabled | 读取 `input_transcription_enabled` 属性。 Whether a settled user transcript is expected after speech. |
属性摘要Attribute Summary
| 属性Attribute | 类型Type | 说明Description | 值Value |
|---|---|---|---|
type | str | 类型字段。 | '' |
supports_text_input | bool | 支持文本输入字段。 | False |
supports_history_replay | bool | `supports_history_replay` 字段。 | False |
方法摘要Method Summary
| 方法Method | 说明Description |
|---|---|
def list_models(custom_yaml_dir: str | None = None) -> list[RealtimeModelCard] | 列出模型。 List the candidate models of this adapter. |
async def connect(instructions: str, tools: list[dict] | None = None, **kwargs: Any) -> Noneasyncabstractabstractmethod | 建立连接并完成初始化。 Open the session. |
async def close() -> Noneasyncabstractabstractmethod | 关闭连接并释放相关资源。 Close the session and release the connection. |
async def events() -> AsyncIterator[ModelEvent]asyncabstractabstractmethod | 返回事件。 Yield model events until the session ends. |
async def push_audio(pcm: bytes) -> Noneasyncabstractabstractmethod | 设置推送音频。 Feed user audio: PCM16 mono at :attr:`input_sample_rate`. |
async def push_text(text: str) -> Noneasyncabstractabstractmethod | 设置推送文本。 Feed a text user turn. |
async def replay_history(messages: Sequence[Msg]) -> Noneasync | 设置该参数。 Seed a newly opened session without requesting a response. |
async def push_tool_result(block: ToolResultBlock) -> Noneasyncabstractabstractmethod | 设置推送工具结果。 Feed the result of a tool the caller executed. |
async def commit_turn() -> Noneasyncabstractabstractmethod | 返回提交轮次。 Declare the user turn finished. Only called when the caller owns endpointing, i.e. the session was opened with turn detection off. |
async def request_response() -> Noneasyncabstractabstractmethod | 返回请求响应。 Ask the model to start generating a reply. |
async def cancel_response() -> Noneasyncabstractabstractmethod | 返回取消响应。 Stop the active reply without closing the session. |
构造器详情Constructor Detail
RealtimeModelBase
def __init__(model: str, credential: CredentialBase, parameters: 'RealtimeModelBase.Parameters | None' = None, model_card: RealtimeModelCard | None = None) -> None
初始化 RealtimeModelBase 实例。
Initialize the realtime model.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
model | str | 模型实例(类型 `str`)。The model name, e.g. `"qwen3-omni-flash-realtime"`. |
credential | CredentialBase | 凭证对象(类型 `CredentialBase`)。The credential used to authenticate against the provider. |
parameters默认default None | 'RealtimeModelBase.Parameters | None' | 参数(类型 `'RealtimeModelBase.Parameters | None'`)。Provider-specific parameters; defaults are used if omitted. |
model_card默认default None | RealtimeModelCard | None | 模型卡片(类型 `RealtimeModelCard | None`)。The model card. When `None` it is looked up by *model* among the cards shipped with this class, so that constructing a model by name alone still yields the right sample rates and limits. |
- 可能抛出 `Exception`。`ValueError`: If no card is given and none matches *model*.
agentscope/realtime/_base.py:46–87属性详情Attribute Detail
type
str = ''类型字段。声明为 str。
The type field.
agentscope/realtime/_base.py:36supports_text_input
bool = False支持文本输入字段。声明为 bool。
The supports text input field.
agentscope/realtime/_base.py:40supports_history_replay
bool = Falsesupports_history_replay 字段。声明为 bool。
The supports history replay field.
agentscope/realtime/_base.py:43属性详情Property Detail
input_sample_rate
def input_sample_rate() -> int
读取 input_sample_rate 属性。
The PCM rate :meth:push_audio expects, in Hz.
agentscope/realtime/_base.py:132–134output_sample_rate
def output_sample_rate() -> int
音频 in:类:AudioDelta,in Hz的PCM 速率。
The PCM rate of the audio in :class:AudioDelta, in Hz.
agentscope/realtime/_base.py:137–139input_transcription_enabled
def input_transcription_enabled() -> bool
读取 input_transcription_enabled 属性。
Whether a settled user transcript is expected after speech.
agentscope/realtime/_base.py:142–144方法详情Method Detail
list_models
def list_models(custom_yaml_dir: str | None = None) -> list[RealtimeModelCard]
列出模型。
List the candidate models of this adapter.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
custom_yaml_dir默认default None | str | None | 自定义 yaml 目录(类型 `str | None`)。Directory to scan. Defaults to the `_models` directory next to the concrete subclass's source file. |
agentscope/realtime/_base.py:102–125connect
async def connect(instructions: str, tools: list[dict] | None = None, **kwargs: Any) -> None
建立连接并完成初始化。
Open the session.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
instructions | str | 指令(类型 `str`)。System instructions. |
tools默认default None | list[dict] | None | 工具列表(类型 `list[dict] | None`)。Tool JSON schemas. Ignored unless the card declares `supports_tools`. |
kwargs | Any | 关键字参数(类型 `Any`)。The kwargs. |
agentscope/realtime/_base.py:151–169close
async def close() -> None
关闭连接并释放相关资源。
Close the session and release the connection.
agentscope/realtime/_base.py:172–173events
async def events() -> AsyncIterator[ModelEvent]
返回事件。
Yield model events until the session ends.
agentscope/realtime/_base.py:176–179push_audio
async def push_audio(pcm: bytes) -> None
设置推送音频。
Feed user audio: PCM16 mono at :attr:input_sample_rate.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
pcm | bytes | pcm(类型 `bytes`)。The pcm. |
agentscope/realtime/_base.py:186–187push_text
async def push_text(text: str) -> None
设置推送文本。
Feed a text user turn.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
text | str | 文本(类型 `str`)。The text. |
- 可能抛出 `Exception`。`NotImplementedError`: If :attr:`supports_text_input` is
- 可能抛出 `Exception`。`False`.
agentscope/realtime/_base.py:190–196replay_history
async def replay_history(messages: Sequence[Msg]) -> None
设置该参数。
Providers that implement this method must preserve message roles and must not trigger model inference.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
messages | Sequence[Msg] | 消息列表(类型 `Sequence[Msg]`)。Completed conversation messages in chronological order. |
- 可能抛出 `Exception`。`NotImplementedError`: If the adapter does not support structured
- 可能抛出 `Exception`。history replay.
agentscope/realtime/_base.py:198–216push_tool_result
async def push_tool_result(block: ToolResultBlock) -> None
设置推送工具结果。
Feed the result of a tool the caller executed.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
block | ToolResultBlock | 内容块(类型 `ToolResultBlock`)。The block. |
agentscope/realtime/_base.py:219–220commit_turn
async def commit_turn() -> None
返回提交轮次。
Declare the user turn finished.
agentscope/realtime/_base.py:223–225request_response
async def request_response() -> None
返回请求响应。
Ask the model to start generating a reply.
agentscope/realtime/_base.py:232–233cancel_response
async def cancel_response() -> None
返回取消响应。
Stop the active reply without closing the session.
agentscope/realtime/_base.py:236–237嵌套类详情Nested Class Detail
Parameters
class Parameters参数,处理低延迟的实时语音与多模态交互。位于 agentscope.realtime 包。
Provider-specific tuneables, surfaced to the UI through the card's parameter_schema. Declared in agentscope.realtime.