Pydantic AI、音声対話機能を追加し複数プロバイダー対応へ

Pydantic AI エージェントは、特別なインターフェースを内蔵していない通常の Python オブジェクトです。同じエージェントは、ターミナルで run() メソッドを呼び出して実行したり、ビルトインの Web チャットの背後で動作したり、独自のフロントエンドへストリーミングされたりしています。これにさらに、ライブ音声会話という機能が加わりました。音声インターフェースは、すでに持っているエージェントに対する別の接続手段に過ぎません。

内部では、これはリアルタイムの音声対音声処理です。OpenAI Realtime、Azure OpenAI、Gemini Live、xAI Grok Voice など複数のプロバイダーが、プロバイダー非依存な API の背後で統合されています。モデルは直接聞き取り、直接発話するため、中間に「文字起こし→生成→音声合成」というパイプラインが存在せず、レイテンシが低く、割り込みも自然に感じられます。

バックエンド側では常にエージェントが動作するため、ツールや履歴、API キーはサーバー側に保持されます。オーディオの経路はアプリに合わせて柔軟に選べます。サーバーでキャプチャする場合もあれば、ブラウザの場合は WebRTC を介してブラウザとプロバイダー間で直接通信し、バックエンドはその通話にサイドバンドとして接続される構成です。これにより、クライアント側へ移動するのはオーディオデータのみとなります。OpenAI および Azure OpenAI ではブラウザの WebRTC がサポートされており、すぐに始められる FastAPI の実行可能例も用意されています。

1 つのファイルで実現する音声エージェント

リアルタイム機能は追加の依存関係として提供されます。Logfire を併せてインストールすれば、セッションが自動的にトレーシングされます：

uv add "pydantic-ai[realtime]" logfire

サーバー側でオーディオをキャプチャする場合、音声エージェントは 1 つのセッションと 3 つの小さなループ（マイク入力、スピーカー出力、ライブ文字起こし）で構成されます。

import asyncio

import contextlib

from collections.abc import AsyncIterator

Pydantic AI エージェントに音声機能が追加されました

以下は、Pydantic AI を使用してリアルタイム音声アシスタントを構築するサンプルコードです。

まず、logfire と pydantic_ai の主要コンポーネントを読み込みます。RealtimeSession は音声入出力のセッション管理に使用されます。

import logfire
from pydantic_ai import Agent
from pydantic_ai.realtime import RealtimeSession

logfire.configure()
logfire.instrument_pydantic_ai()

次に、ヘルプ機能を持つ音声アシスタントとして動作するエージェントを定義します。

agent = Agent(instructions='You are a helpful voice assistant.')

このエージェントには、天気情報を取得するためのツールが追加されます。ここでは簡易的な実装として、指定された都市の天気を「晴れ」として返す関数を登録しています。

@agent.tool_plain
async def get_weather(city: str) -> str:
    return f'Sunny in {city}'

音声入力を処理する非同期関数 stream_microphone では、マイクから取得したオーディオデータをセッションに送信します。具体的には、16 ビットモノラル PCM データをキャプチャまたはリサンプリングし、session.audio_input_sample_rate に合わせて await session.send_audio(chunk) を呼び出します。

async def stream_microphone(session: RealtimeSession) -> None:
    ...  # capture or resample 16-bit mono PCM at `session.audio_input_sample_rate` and `await session.send_audio(chunk)`

一方、エージェントからの音声出力を再生する関数 play_audio は、非同期イテレータから得られたオーディオチャンクをスピーカーに書き込みます。

async def play_audio(chunks: AsyncIterator[bytes]) -> None:
    async for chunk in chunks:
        ...  # write the PCM chunk to your speaker

メイン処理では、OpenAI のリアルタイムモデル（openai:gpt-realtime-2.1）を使用してセッションを開始します。マイクからの入力を非同期タスクとして実行し、同時にスピーカーへの出力も並列で処理します。

async def main():
    async with agent.realtime('openai:gpt-realtime-2.1').session() as session:
        mic = asyncio.create_task(stream_microphone(session))
        speaker = asyncio.create_task(play_audio(session.stream_audio()))

セッションからは、リアルタイムで生成される文字起こし（キャプション）をストリーム形式で取得できます。これにより、ユーザーとアシスタントの対話内容を即座に確認できます。

async for part in session.stream_transcripts():  # live captions
    print(f'{part.speaker}: {part.transcript}')
    #> user: What's the weather like in Paris?
    #> assistant: It's sunny in Paris right now.
    if part.speaker == 'assistant':
        break  # one exchange; a real call keeps listening

対話が完了したら、マイク入力を停止し、エラーを適切に処理した上でスピーカータスクを終了します。

mic.cancel()
with contextlib.suppress(asyncio.CancelledError):
    await mic
await speaker

最後に、この非同期メイン関数を実行してアプリケーションを開始します。

asyncio.run(main())

send_audio() は通話者のマイクからの音声ストリームを受け取り、stream_audio() は応答の音声をストリームとして返します。また stream_transcripts() を使えば、両者のリアルタイム文字起こしを取得できます。

この音声アシスタントの例では、2 つの音声プレースホルダーに sounddevice を使用して音声を埋め込んでいます。ここで登場するエージェントは新しい VoiceAgent 型ではなく、通常の Agent です。get_weather は通話中に呼び出される一般的なサーバーサイドツールです。logfire の 2 行が可観測性設定全体を担っています。

引き継がれる機能

プロバイダーとのリアルタイムソケットを開くには、どの SDK でも数行で済みます。実際に取り組むべきはモデル周辺のことです。実際に動作するツール、信頼できる履歴、コストの可視化、そして API キーをブラウザから守るセキュリティモデルなどです。Pydantic AI が担うのはまさにこの部分であり、テキストエージェントの場合と同じ仕組みで動作します。

• 型安全なツールはサーバーサイドで実行されます。@agent.tool で登録したツールとその依存関係、バリデーション、リトライ機構もそのまま使えます。各呼び出しはバックグラウンドで実行されるためセッションをブロックしません。モデルが待機中も話し続けるかどうかはプロバイダー次第です。
• 機能の引き継ぎも同様です。テキストエージェントに組み込むことができるオプトイン機能（サードパーティ製のものも含む）は、接続時に一度解決されます。（出力バリデーターなどの一部ラングラフ機能は対象外です。）

セッションでは実際のメッセージ履歴が構築されます。音声によるやり取りは、ツール呼び出しやプロバイダーから提供される文字起こしを含む、テキスト実行で生成される ModelRequest/ModelResponse メッセージと同じ形式になります。

利用状況と制限機能は正常に動作します。session.usage は音声トークンとキャッシュされた内訳を累積し、usage_limits は実行中のセッションが暴走するのを防ぐため、通常のランと同様に上限を設定します。

計測も実装済みです。リアルタイムセッションはエージェントの実行時と同じように OpenTelemetry スパンを生成するため、Logfire やその他の OTel バックエンドでも特別な処理なしで読み込むことができます。

1 つのエージェントが複数のモダリティに対応

音声セッションは正規のメッセージ履歴を記録するため、音声とテキストを組み合わせて利用できます。通話終了後にテキスト用エージェントに渡せば、構造化されたデータ抽出が可能です。

from typing import Literal

from pydantic import BaseModel

from pydantic_ai import Agent
from pydantic_ai.realtime import RealtimeSession

logfire.configure()
logfire.instrument_pydantic_ai()

class SupportTicket(BaseModel):
    summary: str
    severity: Literal['low', 'medium', 'high']
    next_action: str

support_agent = Agent(instructions='You are a friendly support line. Keep replies short.')
notetaker = Agent('openai:gpt-5.6-sol', output_type=SupportTicket)

async def take_call(session: RealtimeSession) -> None:
    ...  # stream the caller's mic in and play replies back until they hang up

async def main():
    async with support_agent.realtime('openai:gpt-realtime-2.1').session() as session:
        await take_call(session)

ticket = await notetaker.run(
        'Summarize this call as a support ticket.',
        message_history=session.all_messages(),
    )
    print(ticket.output)

音声通話とテキスト要約は、同じメッセージ履歴を共有する 2 つのエージェントです。逆も同様で、以前のテキスト会話から message_history をシードして音声セッションを開始すれば、通話者は中断した場所からすぐに再開できます。電話ボットとアプリ内のアシスタントは、同じ監査証跡を持つ同一の Agent で実現可能です。

プロバイダーやゲートウェイをまたぐポータビリティ

すべてのプロバイダーが同じ RealtimeModel インターフェースを実装し、型付きイベント語彙へ正規化するため、プロバイダーを変更してもイベントループのコア部分は変わりません。各モデルは、そのプロフィールで自らの機能を宣言します。具体的には、手動のターン制、バジーインによるトランケーション（通話中の割り込み処理）、ブロッキングしないツール呼び出し、そして Gemini の検索グラウンディングのようなプロバイダー固有のサポート機能です。これにより、通話中にモデルの能力を発見するのではなく、事前に機能を分岐させることが可能になります。

ポータビリティには Pydantic AI Gateway も含まれます。アップストリームプロバイダーの名前を指定してセッションをルーティングするだけで、コードの他の部分は一切変更する必要がありません：

from pydantic_ai import Agent

agent = Agent(instructions='You are a helpful voice assistant.')

agent.realtime('gateway/openai:gpt-realtime-2.1')
agent.realtime('gateway/google:gemini-3.1-flash-live-preview')

音声トラフィックも、テキストトラフィックと同じ単一のキー、支出制限、ルーティングルールを適用されます。明らかな理由以外に、これはオーディオにおいて特に重要です。リアルタイム利用は各プロバイダーのオーディオトークンまたは通話時間あたりの課金体系であるため、無人の通話ではコストが瞬く間に膨れ上がってしまう可能性があるからです。

実際に試す

本日、pydantic-ai にリアルタイム音声サポートが実装されました。

uv add "pydantic-ai[realtime]"

これにより、OpenAI、Azure OpenAI、Gemini Live が利用可能になります。これは pydantic-ai パッケージにこれらの SDK がバンドルされているためです。xAI Grok Voice を利用するには [xai-realtime] を追加してください。軽量版の pydantic-ai-slim パッケージを使用する場合は、プロバイダーを手動で指定する必要があります：pydantic-ai-slim[openai-realtime]（Azure も対応）、[google-realtime]、または [xai-realtime] です。

まずはリアルタイム音声のドキュメントを確認し、以下の例から選んでみてください。ターミナル用音声アシスタント、ブラウザ上の WebRTC アプリ、監視しながら解説するカメラエージェント、あるいは通話をテキスト処理用のエージェントに引き継ぐ機能などです。

通話結果の確認

音声データは追跡（trace）上でテキストとして記録されます。発話ターンは pydantic_ai.all_messages に格納され、これは通常テキストストリームを扱う属性で、オーディオの代わりに文字起こしテキストが置かれます。

1 つのセッションスパンが通話全体を保持します。通話者が話した各区間はユーザー音声スパンとして、応答はチャットスパンとして記録されます。会話中にツール呼び出しが発生した場合、その呼び出しには独自のスパンが割り当てられ、モデルが選択した引数と関数の戻り値が含まれます。pydantic_ai.audio_chunks_dropped と transcript_items_dropped は、セッション処理が追いつかなかったデータ数をカウントします。

以下は、この記事の実通話におけるエージェントの追跡データです。4 つの質問に対し、3 回の execute_tool get_weather スパンが発生し、処理時間は 61.87 秒、コストは 0.08 ドルでした。このレートで計算すると、1 時間の通話にかかる費用は約 4.50 ドルとなります。

Pydantic Logfire は、各セッションを他のエージェント実行と並行してファイルに保存します。これは、セッションスパンが Runs 表示でグループ化される基準となる agent_name を保持しているためです。

もしこれを使って何かを構築したり、違和感のある点に遭遇したりした場合は、GitHub や Slack でぜひご報告ください。