MENU
CyberAgent
サーバーサイドエンジニアはこちら フロントエンドエンジニアはこちら コンサルティングセールス・PMはこちら カジュアル面談をご希望の方はこちら
  1. TOP
  2. TECH BLOG
  3. Realtime APIとTwilioを用いた電話予約デモシステムの構築

Realtime APIとTwilioを用いた電話予約デモシステムの構築

2024.10.29

Development

はじめに

こんにちは、AI チームの長澤 (@sp_1999N) です。 今回の記事では 2024/10 に OpenAI から発表された Realtime API と Twilio を用いて、音声予約デモシステムを構築してみます。

Twilio が公開しているデモコードを参考に、function calling などを用いて簡単な対話管理・予約可能日の取得（空き枠検索）を組み合わせたシステムの構築を行い、現時点でのプロダクト適用可能性を探ります。

RealtimeAPI とは

Realtime API は 2024/10 に OpenAI より提供が開始された、音声対話を低レイテンシーで実現できる API になっています。 入出力としてはテキストと音声の両方に対応し、function calling も組み合わせて利用することができます。

従来の音声対話アプリケーションといえば、Speech-to-Text の変換を行い、テキストの世界で対話管理・応答内容を決定、そして Text-to-Speech の変換を行うといったパイプライン処理が主流でした。一方で Realtime API は end-to-end な処理が可能になっているため、レイテンシーの抑制に加え、トーンや感情等の情報を活用したより自然な対話処理が可能になっています。

従って Realtime API の利用により、これまで以上に気軽に音声モダリティを使用したアプリケーションを構築することができるようになりました。

また、Realtime API は WebSocket を用いたステートフルな通信を行い、上述のような処理を実現しています。 ここでは詳細な仕様の記述は控えますが、Session State 等気になる方は是非 公式ドキュメント をご参照ください。

準備

この記事ではTwilio が公開しているデモコードを参考に、音声予約デモシステムの構築を目指します。

上記のデモコードをそのまま実行するだけでも、Twilio を介した電話応答システムを動かすことができます。 雑談等は可能ですが、来店予約など、一定程度の対話管理を要するユースケースにおいては、そのままではうまくいきません。

本記事ではデモコードに少し修正を加え、function calling を用いたナイーブな対話管理を実現し、レストランの来店予約受付システムを構築してみます。 以下は今回実装するシステムの大まかな構成図です。

RealtimeAPI と Twilio を繋ぐデモシステム概念図

各種画像については以下から引用しました

基本構成

ここでは、Twilio のデモコードに従って、大きく3つの関数を使いながらシステムを構築します。

  • handle_incoming_call(...)

    Twilio からの着信を受け取り、TwiML を利用した初期メッセージの送信・WebSocket を介したストリーミング通信の開始処理を行う関数です。本記事では特にデモコードからの大きな修正は行いません。

  • send_session_update(...)

    Realtime API のセッション情報を送信するための関数です。ターン検出や SYSTEM_PROMPT、function calling などの種々の設定を行います。 本記事では主にセッション開始時にのみ使用します。

  • handle_media_stream(...)

    Twilio の Media Stream と OpenAI それぞれとの WebSocket 通信の仲介を行います。この関数内で両者への入出力情報を管理します。Function calling に関わる処理をこの関数に追記します。またこの関数では receive_from_twilio() と send_to_twilio() の2つの関数が内部で定義されます。


    • receive_from_twilio()

      Twilio から送信される音声データ（電話を介した入力音声）などを受け取り、Realtime API の WebSocket に送信する処理を行います。

    • send_to_twilio()

      Realtime API から送信される諸情報を処理し、応答音声データを Twilio に送り返します。Function calling に関する処理はここに追記することになります。

またこれらの他に、function calling で使用する関数（対話状態のスロット管理や日付検索）を定義します。 Realtime API における function calling の使用イメージとしては、まず呼び出し可能な関数をsend_session_update(...)で設定し、セッション更新イベントを作成します。 そしてhandle_media_stream(...)内で、ユーザーの発話内容をもとに、Realtime API から呼び出す関数とその引数などの情報が送られてくるので、それらを実行します。 最後に、関数の実行結果を再度 Realtime API に送り返すことで、その内容を踏まえたモデルの応答を生成します。

実際のコード

それでは上述の内容について、具体的なコードをご紹介します。ここでは、Twilio のデモコードに対する修正点を中心にご紹介します。 （handle_incoming_call(...) については、デモコードからほとんど修正を加えていないため、紹介を省略します）

Function calling で使用する関数例

まずは今回の実装で組み込みたい関数の一部をご紹介します。

本記事ではレストランへの来店予約システムの構築を考えます。 従ってシステムは希望日付や人数など、予約に必要な情報をユーザーから聞き出す必要があります。

これらの情報をスロットとして管理し、対話進行につれて何がヒアリングできているのか/いないのかを厳密に管理できるようにします。 ここでは、まずスロット管理を行うグローバル変数 SLOT を定義します。 そして、現状のスロット状態を取得してモデルに渡すための get_slot_status() 関数、ユーザーの発話内容に基づいてスロットを更新するための update_slot_status(...) 関数を定義します。

SLOT = {
    "date": None,
    "time": None,
    "number_of_people": None,
    "name": None,
    "phone_number": None,
    "note": None
}

def get_slot_status():
    # 現状のスロットを取得します
    return SLOT

def update_slot_status(date=None, time=None, name=None,
                       number_of_people=None, phone_number=None, note=None):
    # 引数として設定された情報のスロットを更新します
    if date:
        SLOT["date"] = date
    if time:
        SLOT["time"] = time
    if name:
        SLOT["name"] = name
    if number_of_people:
        SLOT["number_of_people"] = number_of_people
    if phone_number:
        SLOT["phone_number"] = phone_number
    if note:
        SLOT["note"] = note

    return SLOT

また、上記のような関数を function calling として使用するための設定辞書も定義します。以下はupdate_slot_status(...)関数に対応するものです。description に説明文を書くことにより、関数の結果を待つ際の、適切なフィラー文言の生成につながります。

dct_update_slot = {
    "type": "function",
    "name": "update_slot_status",
    "description": "予約に必要なスロットを埋めます。ユーザー発話を元に、スロットの更新も行います。,
    "parameters": {
        "type": "object",
        "properties": {
            "date": {
                "type": "string",
                "description": "reservation date"
            },
            "time": {
                "type": "string",
                "description": "reservation time"
            },
            "name": {
                "type": "string",
                "description": "予約代表者名"
            },
            "number_of_people": {
                "type": "integer",
                "description": "予約人数"
            },
            "phone_number": {
                "type": "string",
                "description": "電話番号"
            },
            "note": {
                "type": "string",
                "description": "その他予約に関するメモ"
            },
        },
    },
}

この関数の他にも、予約可能日やFAQ（よくある質問とその回答集）を取得する関数、および呼び出しのための設定辞書も定義して使用します。（特に設定辞書については、後述のセッション情報の設定で使用することになります。）

上記で具体的なコードを示したものも含め、本システムで利用する関数および設定辞書をご紹介します。

    使用する関数および辞書
  • get_slot_status()
    現状の対話スロットを取得する関数です。対話進行の確認等で使用する想定となります。設定辞書は dct_check_slot_status です。
  • update_slot_status(...)
    対話スロット項目の更新を行う関数です。スロットに関してユーザーから情報を聞き出すことができた際に使用する想定となります。設定辞書は dct_update_slot_status です。
  • get_available_dates(...)
    予約可能な日を取得し、リストで返却する関数です。予約の空き枠検索を行い、予約可能日をユーザーに共有するために使用する想定となります。設定辞書は dct_get_available_date です。
  • get_faq_list()
    お店に関するFAQリストを取得する関数です。ユーザーからお店について質問があった場合に、この取得結果を元に質問回答を行うためのものになります。設定辞書は dct_get_faq_list です。

send_session_update(...)

async def send_session_update(openai_ws):
    """Send session update to OpenAI WebSocket."""
    session_update = {
        "type": "session.update",
        "session": {
            "turn_detection": {
                "type": "server_vad",
                "threshold": 0.5,
                "prefix_padding_ms": 300,
                "silence_duration_ms": 200
                },
            "input_audio_format": "g711_ulaw",
            "output_audio_format": "g711_ulaw",
            "voice": VOICE,
            "instructions": SYSTEM_PROMPT,
            "modalities": ["text", "audio"],
            "temperature": 0.8,
            "tools": [dct_check_slot_status, dct_update_slot_status,
                         dct_get_available_date, dct_get_faq_list],
            "tool_choice": "auto"
        }
    }
    print('Sending session update:', json.dumps(session_update))
    await openai_ws.send(json.dumps(session_update))

先述した通り、この関数ではターン検出や SYSTEM_PROMPT など、セッション中の基幹となる情報を設定します。 ここでの主な追記事項は tools と tool_choice です。

まず tools では function calling に必要な関数呼び出しの設定辞書を設定します。具体的には、今回は4つの関数を function calling の対象とするため、それぞれに対応する4つの辞書を設定します。

また他の設定項目として tool_choice があります。none, auto, requiredの3つの値を設定することが可能です。 今回は呼び出す関数をモデルによしなに選択させるautoを設定しますが、requiredを設定すると function calling の発動を強制できたりします。（noneは反対に function calling の発動を禁止できます。） 場面に応じてセッション情報を更新することで、function calling に対する発動要望も変更できるようになっています。

handle_media_stream(...) / send_to_twilio()

続いて、実際のストリーミング処理の中で function calling を使用するための実装をご紹介します。対象となる関数は handle_media_stream(...) のうち、send_to_twilio()になります。

async def send_to_twilio():
    """Receive events from the OpenAI Realtime API, send audio back to Twilio."""
    nonlocal stream_sid
    try:
        async for openai_message in openai_ws:
            response = json.loads(openai_message)
            # ログ出力
            if response['type'] in LOG_EVENT_TYPES:
                logger.info(f"Received event: {response['type']}", extra={"event": response})
            if response['type'] == 'session.updated':
                logger.info("Session updated successfully:", extra={"event": response})
            if response['type'] == 'response.function_call_arguments.done':
                logger.info("Function call requests:", extra={"event": response})

                # function calling
                func_to_call = response['name']
                try:
                    result = None
                    arguments = json.loads(response['arguments'])
                    if func_to_call == 'update_slot_status':
                        result = update_slot_status(**arguments)
                    elif func_to_call == 'get_available_dates':
                        result = get_available_dates(**arguments)
                    elif func_to_call == 'get_slot_status':
                        result = get_slot_status()
                    elif func_to_call == "get_faq_list":
                        result = get_faq_list()

                    await asyncio.sleep(0.3)

                    # 関数実行結果の整理・送信
                    result_to_send_openai = {
                    "type": "conversation.item.create",
                    "item": {
                        "id": response['call_id'],
                        "call_id": response['call_id'],
                        "type": "function_call_output",
                        "output": json.dumps(result)
                    }
                }
                    await openai_ws.send(json.dumps(result_to_send_openai))
                    await openai_ws.send(json.dumps({"type": "response.create", "response": {}}))

                except Exception as e:
                    print(f"Error calling function: {e}")

            if response['type'] == 'response.audio.delta' and response.get('delta'):
                # Audio from OpenAI
                try:
                    audio_payload = base64.b64encode(base64.b64decode(response['delta'])).decode('utf-8')
                    audio_delta = {
                        "event": "media",
                        "streamSid": stream_sid,
                        "media": {
                            "payload": audio_payload
                        }
                    }
                    await websocket.send_json(audio_delta)
                except Exception as e:
                    logger.warning(f"Error processing audio data: {e}")

    except Exception as e:
        logger.warning(f"Error in send_to_twilio: {e}")

send_to_twilio() の関数では主に Realtime API から送られてくる情報の処理を行います。 ここで function calling に関連する重要なイベントとして response.function_call_arguments.done があります。これはモデル側で関数呼び出しに必要な引数が準備されたことを通知するものになっています。

従ってこのイベントを受け取った際に、呼び出す関数名とその引数を取り出して、実際に関数を呼び出す処理を追記します。

そして関数の実行結果を Realtime API に送り返す処理が必要になります。ここでこちらから送信するべきイベントは2つあります。

1つは関数の実行結果を格納した conversation.item.create です。 type フィールドには function_call_output を設定し、サーバー側のコンテクスト管理においてアイテムを識別するのに必要な id と function call に割り振られた call_id も設定します。関数の実行結果は output フィールドに格納します。 このイベントを作成することで、Realtime API に関数の結果を送信することができます。

送信すべき2つ目のイベントとして、response.create があります。関数の実行結果を送信しただけでは、それに伴うモデルの応答は生成されないことがあります。従って、関数呼び出しに伴ったモデルの応答生成を誘発するため、このイベントも Realtime API 側に送信する必要があります。

上記のように実装することで、ストリーミング処理の中に function calling を挟むことができます。 なお、今回は分かりやすさと手軽さを優先し、比較的ナイーブな実装になっています。非同期処理などを意識したより厳密な実装については、LangChain が公開している こちらのコード などが参考になるかも知れません。

システムの振る舞いに関する観察

Twilio の公開コード を参考に構築したアプリケーションは、 ngrok でトンネルを作成した上で uvicorn main:app --host 0.0.0.0 --port 5050 のコマンドで稼働させることができます。

何度か対話を試してみましたが、必要に応じて適切な関数を呼び出した処理が適切なタイミングで実行されました。 例えば日付のヒアリング場面では、こちらから「営業時間を教えてください」などの質問発話をすると、FAQ リストを取得する関数を呼び出し、その内容を踏まえた応答を生成してくれました。 質問回答が完了すると、きちんと予約ヒアリングの対話フローに戻り、柔軟な対話が実現されていることが体感できました。 この点について、個人的に印象の良かった実際の応答例をいくつかご紹介します。

柔軟な応答例
  • 予約可能日が複数あった場合、「予約可能な日付は1日から9日まで空いております。」という応答が見られた。日付を全て列挙するのではなく、効率的な情報伝達ができており、体験が良かった。
  • 一番早い日で予約したい、というこちらの要望に対して、「x月で予約可能な最も早い日はxx日になります。こちらで予約してよろしいでしょうか？」というとても自然な応答が見られた。

しかしながら、システム全体としては日本語だとまだまだ挙動が安定しない印象でした。良かった点も悪かった点も含め、以下に気づきを列挙します。

システム全体に関する感想
  • 明確な対話シナリオを用意しなくても、概ね問題のない柔軟な対話ができていた
  • 応答速度については、ほとんどストレスを感じないレベルでレスポンスが返ってきた
  • 発話区間検出について、今回はデフォルトの Server VAD モードを使用したが、テンポ良く対話を進めることができた
  • 項目のヒアリングについて、一度もミスをしない対話は少なかった（何かしらの項目において、ヒアリングミスが発生してしまった）
  • スロットとして対話状態を管理できるようにしたが、項目の修正は難しく、上手くいかないことが多かった
  • Realtime API から返却される応答情報として、テキストと音声の2つがあるが、音声は必ずしも返却されたテキスト通りの読み上げにはなっていないことがあった
  • 応答内容について都度生成するため、先に紹介したような体験の良い応答が常に返ってくる訳ではなく、UXにばらつきが生じやすいシステムになっていた
挙動に関するクセ
  • 一度会話が噛み合わなくなると、その後の会話も噛み合わないまま無理矢理な対話進行が行われることがあった
  • 電話番号については、正しくヒアリングされたことは一度もなかった (伝えた番号は無視され xxx-1234-5678 として認識されてしまった)
  • モデルの audio 出力について、漢字の読み間違いや不自然なイントネーションなど、気になる箇所がいくつかあった

上記はあくまで今回のデモシステムを通じた主観的な感想になります。 ひとまず動くものを簡単に作ることができますが、プロダクトとして使用するには不安要素が散在しているのが現状かと思われます。

また今回ご紹介した実装はかなりナイーブでシンプルだったので、システムプロンプトの推敲や function calling の適切な設計などで改善できる点も多くあるかと思います。 他の外部情報をリサーチしてみましたが、英語だと挙動が上手くいく項目（例えば電話番号の認識）も存在していたので、言語による振る舞いの違い等もあると考えられます。

おわりに

本記事では Realtime API と Twilio を組み合わせた、来店予約のデモシステムを構築しました。 Realtime API の使用により、とても気軽に音声アプリケーションを構築できるようになりました。 プロダクト利用としてはまだまだ難しい側面もありますが、引き続きこういった技術の適用可能性は AI Shift の AI チームとしても探り続けたいと思います。

PREV NEXT BACK TO LIST

CATEGORY

PICK UP

TAG