ag2-live

リアルタイム音声／ライブオーディオエージェント

使用する場面

• 会話するエージェントが必要な場合：マイクに向かって話すと、エージェントが音声で応答するハンズフリーで低遅延なやり取り（Gemini Live または OpenAI Realtime）。
• エージェントが話している間に、ユーザーの発話をリアルタイムで文字起こししたい場合。
• 既存のテキスト Agent に音声パイプラインを追加したい場合：録音した音声クリップを文字起こしし → エージェントを実行し →（任意で）応答を音声で話す。
• それ以外はテキストのみのストリーミング Agent に**音声出力（TTS）**を追加したい場合。

ライブセッションではなく、録音した音声ファイルをエージェントへの入力として一度だけ送信したい場合は、代わりに 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=（int または "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 には、VoiceInput（16 ビット PCM）を返す一度だけのブロッキング record(duration) もあります。ライブセッションを実行せずに、以下の STT パイプラインへ入力するのに便利です：

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

一度だけの音声テキスト変換＋音声パイプライン [キーが必要。ライブ録音の場合のみ音声が必要]

OpenAITranscriber（文字起こし）と OpenAITranslationTranscriber（英語への翻訳）は VoiceInput を受け取り、テキストを返します。.pipe(agent) は、文字起こしを行った後に通常のテキスト Agent を実行する VoicePipeline を構築します：

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 を音声で応答するエージェントにします。オブザーバーとして追加すると、ModelMessageChunk トークンを蓄積し、文全体を合成して SynthesizedAudioEvent を出力します。同じストリーム上の SoundDevicePlayer がその音声を再生します。これは、通常のテキストエージェントとリアルタイム音声出力をつなぐ橋渡しです（リアルタイムプロバイダーは不要です）。

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] をインストールすると 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 プロトコル。
• 録音した音声ファイルを単発入力としてテキストエージェントに送信する方法については、ag2-multimodal-input（AudioInput）を参照してください。
• オブザーバー全般（トークンモニター、ループ検出など）については、ag2-observers-and-alerts を参照してください。