ag2-live

实时语音 / 实时音频 Agent

适用场景

• 用户需要一个可交谈的 Agent：对着麦克风说话，Agent 以语音回复，无需动手且延迟低（Gemini Live 或 OpenAI Realtime）。
• 用户需要在 Agent 说话时实时转录用户的语音。
• 用户需要基于现有文本 Agent 构建语音工作流：转录录音片段 → 运行 Agent →（可选）朗读回复。
• 用户需要为原本只支持文本流式输出的 Agent 添加语音输出（TTS）。

如果用户只需将录制的音频文件作为单次输入发送给 Agent（而非开启实时会话），请改用 ag2-multimodal-input（AudioInput）。

硬件 / 密钥注意事项。 实际会话需要（a）服务提供商 API 密钥，以及（b）可正常工作的麦克风和扬声器。无头环境不具备这两项条件。此技能中的所有内容都可以构建，但实际打开套接字或音频设备的部分会在下方标记为 [需要密钥 + 音频]。

安装

实时模块拆分为多个可选扩展包，请安装所需的扩展包：

# OpenAI Realtime + OpenAI TTS/STT
pip install "ag2[openai-realtime]"

# Gemini Live
pip install "ag2[gemini-realtime]"

# Local microphone / speaker I/O (SoundDeviceRecorder / SoundDevicePlayer)
pip install "sounddevice[numpy]"

使用本地声卡的典型 OpenAI 语音应用：pip install "ag2[openai-realtime]" "sounddevice[numpy]"。

必需。交付代码前，请先运行相关安装命令。如果未安装对应的扩展包，公共符号会解析为占位项：使用时会立即引发 ImportError（openai/gemini），或额外依赖错误（sounddevice[numpy]）。

公共 API

所有符号均从 ag2.live 导出：

符号	作用
LiveAgent	封装提示词和实时配置；agent.run() 会开启双向会话
GeminiRealTimeConfig	Gemini Live 实时服务提供商配置
OpenAIRealTimeConfig	OpenAI Realtime 服务提供商配置
OpenAITTSConfig	文本 → 语音（PCM 字节）
OpenAITranscriber	一次性语音 → 文本（转录）
OpenAITranslationTranscriber	一次性语音 → 英文文本（翻译）
SoundDeviceRecorder	麦克风采集 → 在流中生成 RecordedAudioEvent
SoundDevicePlayer	通过扬声器播放 SynthesizedAudioEvent PCM 音频
TTSObserver	通过 TTS 配置朗读文本 Agent 流式输出的令牌的观察器

实时会话的工作方式

LiveAgent 围绕一个事件流（ConversationContext）构建。录音器、服务提供商会话和播放器共享同一个上下文：

mic ──SoundDeviceRecorder──▶ RecordedAudioEvent ──▶ provider session (Gemini/OpenAI)
                                                       │
                                       SynthesizedAudioEvent (assistant audio)
                                                       ▼
                                              SoundDevicePlayer ──▶ speaker

在此过程中，服务提供商还会发出 TranscriptionChunkEvent / TranscriptionCompletedEvent（用户语音）、ModelMessageChunk（助手文本）、ToolCallEvent/ToolResultEvent 和 UsageEvent。agent.run() 是一个异步上下文管理器，会返回共享的 ConversationContext，以便你将录音器和播放器附加到其中。

最简示例 — 全双工语音循环 [需要密钥 + 音频]

import asyncio

from ag2.live import (
    LiveAgent,
    OpenAIRealTimeConfig,
    SoundDevicePlayer,
    SoundDeviceRecorder,
)
from ag2.live.openai import AudioOutput, InputConfig

agent = LiveAgent(
    "voice_bot",
    "You are a friendly realtime voice assistant. Keep replies short and conversational.",
    config=OpenAIRealTimeConfig(
        "gpt-realtime",
        output=AudioOutput(voice="marin"),
        # Transcribe the user's speech so we can see it too.
        input=InputConfig(transcription={"model": "gpt-4o-mini-transcribe"}),
    ),
)

async def main() -> None:
    # run() yields the shared ConversationContext.
    async with agent.run() as ctx:
        # Recorder and player bind to the SAME context so audio flows on one bus.
        async with (
            SoundDeviceRecorder(context=ctx),   # mic  -> RecordedAudioEvent
            SoundDevicePlayer(context=ctx),     # SynthesizedAudioEvent -> speaker
        ):
            print("Talk to the agent. Ctrl-C to stop.")
            await asyncio.Event().wait()  # keep the session open

if __name__ == "__main__":
    asyncio.run(main())

设置 OPENAI_API_KEY（或传入 client=AsyncOpenAI(...)）。使用 Gemini 时，设置 GEMINI_API_KEY / GOOGLE_API_KEY（或传入 client=）。

只需修改配置即可切换到 Gemini：

from ag2.live import GeminiRealTimeConfig
from ag2.live.gemini import AudioOutput, InputConfig

agent = LiveAgent(
    "voice_bot",
    "You are a friendly realtime voice assistant.",
    config=GeminiRealTimeConfig(
        "gemini-2.5-flash-native-audio-preview-12-2025",
        output=AudioOutput(voice="Kore"),
        input=InputConfig(transcribe=True),   # user-speech transcription
    ),
)

服务商矩阵 — Gemini Live 与 OpenAI Realtime

	Gemini（GeminiRealTimeConfig）	OpenAI（OpenAIRealTimeConfig）
导入旋钮类型的位置	ag2.live.gemini	ag2.live.openai
输出模式	AudioOutput（默认）/ TextOutput	AudioOutput（默认）/ TextOutput
语音（AudioOutput(voice=...)）	Aoede、Charon、Fenrir、Kore（默认）、Leda、Orus、Puck、Zephyr	alloy（默认）、ash、ballad、coral、echo、sage、shimmer、verse、marin、cedar
示例模型（第一个位置参数）	gemini-2.5-flash-native-audio-preview-12-2025、gemini-3.1-flash-live-preview、gemini-live-2.5-flash-preview、gemini-2.0-flash-live-001	gpt-realtime、gpt-realtime-mini、gpt-audio-1.5、gpt-4o-realtime-preview-2024-10-01
用户语音转录	InputConfig(transcribe=True, transcription_languages=[...])	InputConfig(transcription={"model": "gpt-4o-mini-transcribe"})
轮次检测 / VAD	InputConfig(automatic_activity_detection=..., activity_handling=..., turn_coverage=...)	InputConfig(turn_detection={"type": "semantic_vad", ...})（默认开启）
音频 I/O 采样率	输入 16 kHz / 输出 24 kHz（由 API 固定）	可通过 AudioOutput.format / InputConfig.format 配置（默认 PCM 24 kHz）
生成参数	temperature=、max_output_tokens=	max_output_tokens=（整数或 "inf"）、tool_choice=、tracing=
原始服务商配置的备用入口	config=<LiveConnectConfigDict>	session=<RealtimeSessionCreateRequestParam>
使用自有客户端	client=google.genai.Client(...)	client=openai.AsyncOpenAI(...)

模型名称是位置参数；其他参数均为仅限关键字参数。上方列出的 Literal 模型 / 语音列表均可使用，但也允许任意字符串（向前兼容新模型）。

AudioOutput 还支持 language_code=（Gemini）以及 speed= + format=（OpenAI）。TextOutput() 不接受参数——仅返回文本（ModelMessageChunk），不播放音频。

LiveAgent 上的工具、HITL 和观察器

LiveAgent(...) 接受与普通 Agent 相同的一组参数：tools=、hitl_hook=、middleware=、observers=、dependencies=、variables=、plugins=。模型发起的工具调用会以 ToolCallEvent 的形式传入，结果会自动转发回会话。每次会话都可以通过 run() 覆盖 config=、prompt=、tools=、observers=、hitl_hook=。

注意： LiveAgent 在实时模式下仅支持函数工具——两种后端中，服务商的服务端工具类型都会触发 NotImplementedError。

音频 I/O — SoundDeviceRecorder / SoundDevicePlayer

两者都是异步上下文管理器（async with），并通过 context= 绑定到 ConversationContext。设备会在 __aenter__ 时打开 [需要音频]；仅构造对象不会触及任何硬件。

from ag2.live import SoundDeviceRecorder, SoundDevicePlayer

# Recorder: mic -> RecordedAudioEvent on the stream.
recorder = SoundDeviceRecorder(context=ctx, sample_rate=16000, channels=1)  # block_size optional

# Player: subscribes to SynthesizedAudioEvent, writes PCM to the speaker (24 kHz mono).
player = SoundDevicePlayer(context=ctx)

SoundDeviceRecorder 还提供一次性阻塞式 record(duration)，返回一个 VoiceInput（16 位 PCM）——无需运行实时会话，即可将其传入下方的 STT 流水线：

voice = recorder.record(duration=4.0)   # blocks 4s, returns VoiceInput  [needs audio]

一次性语音转文本 + 语音流水线 [需要密钥；仅实时录制时需要音频]

OpenAITranscriber（转录）和 OpenAITranslationTranscriber（翻译成英语）接收 VoiceInput 并返回文本。.pipe(agent) 会构建一个 VoicePipeline，先转录，再运行普通的文本 Agent：

from ag2 import Agent
from ag2.config import OpenAIConfig
from ag2.live import OpenAITranscriber, SoundDeviceRecorder
from ag2.live.stt import VoiceInput

agent = Agent("assistant", "You answer questions.", config=OpenAIConfig(model="gpt-4o-mini"))
pipeline = OpenAITranscriber("gpt-4o-transcribe").pipe(agent)

# Record a clip, transcribe it, and ask the agent — in one call.  [needs keys + audio]
voice: VoiceInput = SoundDeviceRecorder().record(4.0)
reply = await pipeline.ask(voice)
print(await reply.content())

# Continue the conversation with another clip:
followup = await reply.ask(SoundDeviceRecorder().record(4.0))

VoiceInput(content: bytes, frame_rate: int, channels: int) 封装 16 位 PCM；如果不使用录音器，也可以直接从任意 PCM 源构建。

OpenAITranslationTranscriber 的用法相同，但始终输出英语（适用于输入非英语语音、输出英语文本）。

文本转语音 — OpenAITTSConfig 和 TTSObserver

OpenAITTSConfig 可将文本合成为 PCM 字节：

from ag2.live import OpenAITTSConfig

tts = OpenAITTSConfig("gpt-4o-mini-tts", voice="alloy", speed=1.0)
pcm: bytes = await tts.synthesize("Hello there!")   # [needs keys]

TTSObserver 可将任意流式传输 文本 的 Agent 转变为会说话的 Agent：将它附加为观察器后，它会累积 ModelMessageChunk 令牌，合成完整句子，并发出 SynthesizedAudioEvent — 同一流上的 SoundDevicePlayer 会将其播放出来。这是普通文本 Agent 与实时音频输出之间的桥梁（无需实时提供商）。

from ag2 import Agent, MemoryStream
from ag2.config import OpenAIConfig
from ag2.context import ConversationContext
from ag2.live import OpenAITTSConfig, SoundDevicePlayer, TTSObserver

agent = Agent(
    "narrator",
    "You are a helpful assistant.",
    config=OpenAIConfig(model="gpt-4o-mini"),
    observers=[TTSObserver(OpenAITTSConfig("gpt-4o-mini-tts", voice="nova"))],
)

# Agent and Player must share ONE stream: agent.ask(stream=...) and
# SoundDevicePlayer(context=...) whose .stream is that same stream.
stream = MemoryStream()
ctx = ConversationContext(stream=stream)

async with SoundDevicePlayer(context=ctx):     # plays the synthesized speech  [needs audio]
    reply = await agent.ask("Tell me a one-line joke.", stream=stream)
    print(await reply.content())

TTSObserver(config) 会返回一个 CompositeObserver。它会在最终的 ModelMessage 上刷新所有剩余的缓冲文本，因此未完成的句子也会被朗读出来。

用量报告

await LiveAgent.usage_report(ctx) 会将实时会话事件日志中的令牌用量汇总为 UsageReport。

构造测试与需要实时密钥和硬件的测试

由此技能的 references/test_samples.py 运行（全部通过）— 离线构造／运行测试：

• 所有 9 个公共符号都能以真实类的形式导入（不是缺少依赖时的占位符）。
• GeminiRealTimeConfig / OpenAIRealTimeConfig 的构造，包括 AudioOutput / TextOutput / InputConfig、语音、temperature/max-tokens，以及 _build_session(instructions=...) 合并。
• OpenAITTSConfig、OpenAITranscriber、OpenAITranslationTranscriber 的构造。
• LiveAgent 的构造（字符串提示词、提示词列表，以及使用 tools=/observers= 的情况）。
• SoundDeviceRecorder / SoundDevicePlayer 的构造（不会打开设备）。
• TTSObserver 返回 CompositeObserver。
• OpenAITranscriber(...).pipe(agent) → VoicePipeline；VoiceInput 数据类。

需要真实 API 密钥和麦克风／扬声器（无法在无头环境中运行）：

• agent.run() 打开与 Gemini/OpenAI 的实时 websocket。
• async with SoundDeviceRecorder(...) / SoundDevicePlayer(...)（在 __aenter__ 时打开声卡）。
• SoundDeviceRecorder.record(duration)（会阻塞等待麦克风输入）。
• OpenAITTSConfig.synthesize(...)、OpenAITranscriber.transcribe(...) 和 VoicePipeline.ask(...)（会调用 OpenAI API）。

常见问题

• 安装了错误的扩展包。 OpenAI 使用 pip install "ag2[openai-realtime]"，Gemini 使用 "ag2[gemini-realtime]"，本地音频使用 "sounddevice[numpy]"。符号可以正常导入，但否则会在首次使用时引发错误。
• 缺少 numpy。 SoundDeviceRecorder / SoundDevicePlayer 需要 numpy — sounddevice[numpy] 会将其一并安装。缺少它时会出现额外依赖相关的 ImportError。
• 从错误的模块导入旋钮类型。 AudioOutput / TextOutput / InputConfig 是特定于提供商的：ag2.live.gemini 与 ag2.live.openai 中的定义不同。它们不会从 ag2.live 重新导出。
• 录音器、播放器和 run() 未共用同一上下文。 将 agent.run() 生成的 ctx 传入 SoundDeviceRecorder(context=ctx) / SoundDevicePlayer(context=ctx)，否则音频事件无法传递到会话／扬声器。
• 语音不在提供商的列表中。 Gemini 和 OpenAI 的语音名称不同（请参见对照表）。可以接受普通字符串，但如果语音未知，提供商会在会话打开时拒绝它。
• 期望实时模式支持服务端／提供商工具。 仅支持函数工具；其他工具类型会引发 NotImplementedError。
• 忘记保持循环运行。 agent.run() 是一个上下文管理器 — 退出时会关闭会话。要保持持续对话，请让它保持打开状态（使用 await asyncio.Event().wait() 或你自己的循环）。

深入了解

• 源码：ag2/live/{realtime.py,gemini.py,openai.py,protocols.py,observer.py,stt.py,sound_device.py}。
• realtime.py — LiveAgent 和 RealtimeConfig 协议。
• 若要将录制的音频 文件 作为一次性输入发送给文本 Agent，请参见 ag2-multimodal-input（AudioInput）。
• 有关观察器的一般信息（令牌监视器、循环检测），请参见 ag2-observers-and-alerts。