ガイド
このガイドでは、OpenAI Agents SDK のリアルタイム機能を用いて音声対応の AI エージェントを構築する方法を詳しく説明します。
ベータ機能
リアルタイム エージェントはベータ版です。実装の改善に伴い、互換性のない変更が発生する可能性があります。
概要
リアルタイム エージェントは、音声とテキストの入力をリアルタイムに処理し、リアルタイム音声で応答する会話フローを可能にします。OpenAI の Realtime API との永続的な接続を維持し、低遅延で自然な音声対話と、割り込みへの柔軟な対応を実現します。
アーキテクチャ
中核コンポーネント
リアルタイム システムは、次の主要コンポーネントで構成されます。
- RealtimeAgent: instructions、tools、handoffs で構成されたエージェント。
- RealtimeRunner: 設定を管理します。
runner.run()を呼び出してセッションを取得できます。 - RealtimeSession: 単一の対話セッション。通常、ユーザーが会話を開始するたびに作成し、会話が完了するまで維持します。
- RealtimeModel: 基盤となるモデル インターフェース(通常は OpenAI の WebSocket 実装)
セッションの流れ
一般的なリアルタイム セッションは次の流れに従います。
- RealtimeAgent を作成 し、instructions、tools、handoffs を設定します。
- RealtimeRunner をセットアップ し、エージェントと設定オプションを渡します。
- セッションを開始 します。
await runner.run()を使用すると RealtimeSession が返されます。 - 音声またはテキスト メッセージを送信 します。
send_audio()またはsend_message()を使用します。 - イベントをリッスン します。セッションをイテレートしてイベント(音声出力、文字起こし、ツール呼び出し、ハンドオフ、エラー)を受け取ります。
- 割り込みに対応 します。ユーザーが発話で割り込むと、現在の音声生成は自動的に停止します。
セッションは会話履歴を保持し、リアルタイム モデルとの永続接続を管理します。
エージェントの設定
RealtimeAgent は、通常の Agent クラスと同様に動作しますが、いくつか重要な違いがあります。完全な API 詳細は、RealtimeAgent の API リファレンスをご覧ください。
通常のエージェントとの主な違い:
- モデルの選択はエージェント レベルではなくセッション レベルで設定します。
- 構造化出力のサポートはありません(
outputTypeはサポートされません)。 - 音声はエージェントごとに設定できますが、最初のエージェントが発話した後は変更できません。
- tools、handoffs、instructions など、その他の機能は同様に動作します。
セッションの設定
モデル設定
セッション設定では、基盤となるリアルタイム モデルの動作を制御できます。モデル名(gpt-realtime など)、音声の選択(alloy、echo、fable、onyx、nova、shimmer)、サポートするモダリティ(テキストや音声)を構成できます。音声フォーマットは入力と出力の両方で設定でき、デフォルトは PCM16 です。
音声設定
音声設定では、セッションが音声入力と出力をどのように処理するかを制御します。Whisper などのモデルを使った入力音声の文字起こし、言語設定、ドメイン固有用語の精度向上のための文字起こしプロンプトを設定できます。応答開始/停止の検出設定では、音声活動検知のしきい値、無音時間、検出された音声の前後パディングなどを指定できます。
ツールと関数
ツールの追加
通常のエージェントと同様に、リアルタイム エージェントは会話中に実行される 関数ツール をサポートします。
from agents import function_tool
@function_tool
def get_weather(city: str) -> str:
"""Get current weather for a city."""
# Your weather API logic here
return f"The weather in {city} is sunny, 72°F"
@function_tool
def book_appointment(date: str, time: str, service: str) -> str:
"""Book an appointment."""
# Your booking logic here
return f"Appointment booked for {service} on {date} at {time}"
agent = RealtimeAgent(
name="Assistant",
instructions="You can help with weather and appointments.",
tools=[get_weather, book_appointment],
)
ハンドオフ
ハンドオフの作成
ハンドオフにより、特化したエージェント間で会話を引き継ぐことができます。
from agents.realtime import realtime_handoff
# Specialized agents
billing_agent = RealtimeAgent(
name="Billing Support",
instructions="You specialize in billing and payment issues.",
)
technical_agent = RealtimeAgent(
name="Technical Support",
instructions="You handle technical troubleshooting.",
)
# Main agent with handoffs
main_agent = RealtimeAgent(
name="Customer Service",
instructions="You are the main customer service agent. Hand off to specialists when needed.",
handoffs=[
realtime_handoff(billing_agent, tool_description="Transfer to billing support"),
realtime_handoff(technical_agent, tool_description="Transfer to technical support"),
]
)
イベント処理
セッションはイベントをストリーミングし、セッション オブジェクトをイテレートしてリッスンできます。イベントには、音声出力チャンク、文字起こし結果、ツール実行の開始と終了、エージェントのハンドオフ、エラーなどが含まれます。特に次のイベントを処理してください。
- audio: エージェントの応答からの raw 音声データ
- audio_end: エージェントが話し終えた
- audio_interrupted: ユーザーがエージェントを割り込んだ
- tool_start/tool_end: ツール実行のライフサイクル
- handoff: エージェントのハンドオフが発生
- error: 処理中にエラーが発生
完全なイベントの詳細は、RealtimeSessionEvent を参照してください。
ガードレール
リアルタイム エージェントでサポートされるのは出力 ガードレール のみです。パフォーマンス問題を避けるため、これらのガードレールは(単語ごとではなく)定期的にデバウンスして実行されます。デフォルトのデバウンス長は 100 文字ですが、設定可能です。
ガードレールは RealtimeAgent に直接アタッチするか、セッションの run_config で提供できます。両方のソースからのガードレールは併せて実行されます。
from agents.guardrail import GuardrailFunctionOutput, OutputGuardrail
def sensitive_data_check(context, agent, output):
return GuardrailFunctionOutput(
tripwire_triggered="password" in output,
output_info=None,
)
agent = RealtimeAgent(
name="Assistant",
instructions="...",
output_guardrails=[OutputGuardrail(guardrail_function=sensitive_data_check)],
)
ガードレールが発火すると、guardrail_tripped イベントが生成され、エージェントの現在の応答を中断できます。デバウンス動作は、安全性とリアルタイム性能要件のバランスを取るのに役立ちます。テキスト エージェントと異なり、リアルタイム エージェントはガードレールが発火しても Exception を送出しません。
音声処理
session.send_audio(audio_bytes) を使って音声を送信するか、session.send_message() を使ってテキストを送信します。
音声出力については、audio イベントをリッスンし、任意の音声ライブラリで再生してください。ユーザーがエージェントを割り込んだ場合にすぐに再生を停止し、キュー済みの音声をクリアするため、audio_interrupted イベントを必ず監視してください。
SIP 連携
Realtime Calls API 経由で着信する電話にリアルタイム エージェントを接続できます。SDK は OpenAIRealtimeSIPModel を提供しており、SIP 上でメディアをネゴシエーションしながら同じエージェント フローを再利用します。
使用するには、モデル インスタンスを runner に渡し、セッション開始時に SIP の call_id を指定します。コール ID は、着信を通知する webhook によって渡されます。
from agents.realtime import RealtimeAgent, RealtimeRunner
from agents.realtime.openai_realtime import OpenAIRealtimeSIPModel
runner = RealtimeRunner(
starting_agent=agent,
model=OpenAIRealtimeSIPModel(),
)
async with await runner.run(
model_config={
"call_id": call_id_from_webhook,
"initial_model_settings": {
"turn_detection": {"type": "semantic_vad", "interrupt_response": True},
},
},
) as session:
async for event in session:
...
発信者が電話を切ると、SIP セッションは終了し、リアルタイム接続は自動的にクローズされます。完全なテレフォニーの例は、examples/realtime/twilio_sip を参照してください。
モデルへの直接アクセス
基盤となるモデルにアクセスして、カスタム リスナーを追加したり高度な操作を実行したりできます。
これにより、接続を低レベルに制御する必要がある高度なユースケース向けに、RealtimeModel インターフェースへ直接アクセスできます。
コード例
完全に動作するコード例は、examples/realtime ディレクトリ を参照してください。UI コンポーネントの有無それぞれのデモが含まれています。