Input

AudioInput dataclass

用于 VoicePipeline 的静态音频输入。

Source code in src/agents/voice/input.py

@dataclass
class AudioInput:
    """用于 VoicePipeline 的静态音频输入。"""

    buffer: npt.NDArray[np.int16 | np.float32]
    """
    包含代理音频数据的缓冲区。必须是 int16 或 float32 类型的 numpy 数组。
    """

    frame_rate: int = DEFAULT_SAMPLE_RATE
    """音频数据的采样率。默认为 24000。"""

    sample_width: int = 2
    """音频数据的采样宽度。默认为 2。"""

    channels: int = 1
    """音频数据的通道数。默认为 1。"""

    def to_audio_file(self) -> tuple[str, io.BytesIO, str]:
        """返回一个 (文件名, 字节流, 内容类型) 的元组"""
        return _buffer_to_audio_file(self.buffer, self.frame_rate, self.sample_width, self.channels)

    def to_base64(self) -> str:
        """以 base64 编码字符串的形式返回音频数据。"""
        if self.buffer.dtype == np.float32:
            # 转换为 int16
            self.buffer = np.clip(self.buffer, -1.0, 1.0)
            self.buffer = (self.buffer * 32767).astype(np.int16)
        elif self.buffer.dtype != np.int16:
            raise UserError("Buffer must be a numpy array of int16 or float32")

        return base64.b64encode(self.buffer.tobytes()).decode("utf-8")

buffer instance-attribute

buffer: NDArray[int16 | float32]

包含代理音频数据的缓冲区。必须是 int16 或 float32 类型的 numpy 数组。

frame_rate class-attribute instance-attribute

frame_rate: int = DEFAULT_SAMPLE_RATE

音频数据的采样率。默认为 24000。

sample_width class-attribute instance-attribute

sample_width: int = 2

音频数据的采样宽度。默认为 2。

channels class-attribute instance-attribute

channels: int = 1

音频数据的通道数。默认为 1。

to_audio_file

to_audio_file() -> tuple[str, BytesIO, str]

返回一个 (文件名, 字节流, 内容类型) 的元组

Source code in src/agents/voice/input.py

def to_audio_file(self) -> tuple[str, io.BytesIO, str]:
    """返回一个 (文件名, 字节流, 内容类型) 的元组"""
    return _buffer_to_audio_file(self.buffer, self.frame_rate, self.sample_width, self.channels)

to_base64

to_base64() -> str

以 base64 编码字符串的形式返回音频数据。

Source code in src/agents/voice/input.py

def to_base64(self) -> str:
    """以 base64 编码字符串的形式返回音频数据。"""
    if self.buffer.dtype == np.float32:
        # 转换为 int16
        self.buffer = np.clip(self.buffer, -1.0, 1.0)
        self.buffer = (self.buffer * 32767).astype(np.int16)
    elif self.buffer.dtype != np.int16:
        raise UserError("Buffer must be a numpy array of int16 or float32")

    return base64.b64encode(self.buffer.tobytes()).decode("utf-8")

StreamedAudioInput

以音频数据流形式表示的音频输入。你可以将其传递给 VoicePipeline， 然后使用 add_audio 方法将音频数据推入队列。

Source code in src/agents/voice/input.py

class StreamedAudioInput:
    """以音频数据流形式表示的音频输入。你可以将其传递给 `VoicePipeline`，
    然后使用 `add_audio` 方法将音频数据推入队列。
    """

    def __init__(self):
        self.queue: asyncio.Queue[npt.NDArray[np.int16 | np.float32]] = asyncio.Queue()

    async def add_audio(self, audio: npt.NDArray[np.int16 | np.float32]):
        """向音频流中添加更多音频数据。

        参数:
            audio: 要添加的音频数据。必须是 int16 或 float32 类型的 numpy 数组。
        """
        await self.queue.put(audio)

add_audio async

add_audio(audio: NDArray[int16 | float32])

向音频流中添加更多音频数据。

参数

audio: 要添加的音频数据。必须是 int16 或 float32 类型的 numpy 数组。

Source code in src/agents/voice/input.py

async def add_audio(self, audio: npt.NDArray[np.int16 | np.float32]):
    """向音频流中添加更多音频数据。

    参数:
        audio: 要添加的音频数据。必须是 int16 或 float32 类型的 numpy 数组。
    """
    await self.queue.put(audio)