サーバー側での制御

サーバーからの GPT-Live セッションの制御

サーバーで会話イベントを受信したり、非公開ツールを実行したり、会話を更新したりする必要がある場合は、既存の GPT-Live の WebRTC または SIP セッションにアプリケーションサーバーを接続します。この 2 本目の接続を サイドバンド WebSocket と呼びます。どちらの接続も同じセッションを共有し、通話の音声は WebRTC または SIP で伝送します。

サイドバンドでイベントとコマンドをやり取りします。ツールの実行、認可のチェック、ビジネスルールの適用はアプリケーションが担います。API キーとツールの認証情報はサーバー上で保管してください。

サイドバンドの要否の判断

ブラウザアプリケーションで字幕を表示したり、ローカルの UI を更新したりするには、WebRTC データチャネルを使用します。ガードレールによるチェック、感情分析、先読みでのツール呼び出しなど、文字起こしをサーバーで処理する場合はサイドバンドを使用します。ブラウザの音声は引き続き WebRTC でやり取りしながら、サーバーでイベントを受信して同じセッションを直接制御できます。具体例は文字起こしの断片に応じた処理を参照してください。

バックエンドがメインの WebSocket 接続で音声をストリーミングする場合は、その接続でイベントを受信し、コマンドを送信します。

Responses への委譲はサイドバンドがなくても利用できます。ブラウザはデータチャネルで受信した関数呼び出しイベントを認証済みのバックエンドに転送し、そこで関数を実行できます。OpenAI がホストするツールは委譲先のバックエンドを通じて実行されるため、アプリケーション側にツールの実行機構は必要ありません。

既存のセッションへの接続

1. バックエンドで制御するセッションの ID を保存します。WebRTC または SIP の発信通話では、session.id を POST /v1/live/sessions の JSON レスポンスから取得します。SIP の着信では、まず着信を受け付けてから、その Webhook の data.session_id を使用します。ID はアプリケーションのユーザー情報および会話の記録とともに保存します。

2. 次の URL に保存した ID をそのまま入れ、サーバーから WebSocket 接続を開きます。セッションの作成時または着信の受付時と同じプロジェクトの認証情報を使い、Authorization: Bearer $OPENAI_API_KEY で認証します。セッションの作成時に必要な接続ヘッダーも同じように指定してください。

   wss://api.openai.com/v1/live/sessions/{session_id}/attach

3. 接続したソケットでイベントを受信し、コマンドを送信します。セッションはすでに実行中なので、session.start を再送しないでください。

セッション ID はプレフィックスも含めてそのまま使用し、アプリケーションにそのセッションへのアクセス権があることを確認してください。

イベントの監視とコマンドの送信

コマンドには、メインの接続と同じ検証ルールと委譲ルールが適用されます。セッション全体にコンテキストを追加する場合は delegation_id: null を使用します。null 以外の ID を指定する場合は、既存のクライアント委譲の ID でなければなりません。設定、関数の実行、コンテキスト追加の例は委譲とツールを参照してください。

ブラウザのセッションでは、マイク入力とスピーカー出力に、接続時にネゴシエーションした WebRTC メディアトラックを引き続き使用します。会話イベントの受信と制御にはサイドバンドを使い、再生状況は音声プレーヤーで追跡します。

入出力音声のコピーの受信

メインの接続で音声をリアルタイムに伝送している間、サイドバンドでも接続後の入出力音声のコピーを受信します。

メインの接続で使用する音声形式にかかわらず、どちらのペイロードも 24 kHz のモノラルの生 PCM16LE 音声を base64 でエンコードしたものです。どちらのイベントにも event_id はありません。入力音声のコピーには、入力ミュートを適用する前に受信した音声が含まれます。コピーを受信しても、モデルがそのサンプルを処理したことは確認できません。出力音声の区間には、フレームの欠落によって空白が生じることがあります。また、この区間から通話相手が実際に音声を聞いた時刻は分かりません。

マイク音声はメインの接続のみで送信してください。サイドバンドは音声のコピーを受信するために使用します。

アクションごとの担当の割り当て

各アクションをブラウザとバックエンドのどちらが担当するかを決めます。両方の接続で関数呼び出しイベントを受信しても、関数を実行するのは 1 回だけです。コンテキストの更新や、バックエンドでの処理継続を求めるリクエストにも同じルールを適用します。

バックエンドで会話を最初から監視する必要がある場合は、早めにサイドバンドを接続してください。接続前に収集した履歴も含め、文字起こしとツールの状態はアプリケーションに保存します。

サイドバンドの接続中も、ブラウザはセッションイベントを受信できます。機密性の高いツールの認証情報はバックエンドで管理し、認可の判断もそこで行います。返すのは会話に必要なコンテキストだけにしてください。

会話へのガードレールの適用

サーバーのメインの WebSocket 接続またはサイドバンドを使用して会話を監視し、リクエストがアプリケーションのポリシーに沿っているかをチェックします。チェックを実行するのはアプリケーションです。問題を検出したら、該当するアクションをブロックし、会話を修正するための指示を送信します。

会話と並行したチェックの実行

文字起こしの断片を受信するたびに処理する仕組みの用途の一つがガードレールです。同じストリームを使い、チェックと並行して先読み検索を開始したり、UI を更新したりすることもできます。

1. 文字起こしの監視。 session.input_transcript.delta の断片を蓄積し、ユーザーのリクエストにジェイルブレイクの試み、機密情報、ポリシー違反が含まれていないかをチェックします。session.output_transcript.delta を使用して、アシスタントの発話に裏付けのない主張や、アプリケーションの対応範囲を超える回答がないかをチェックします。各チェックは、評価対象の文字起こしとアプリケーションのリクエストに紐付けてください。
2. チェックの並行実行。 軽量で高速なモデルを使えば、会話を続けながらリクエストを評価できます。アプリケーションが処理に利用できるよう、{"triggered": true} のような簡潔な構造化データを結果として返します。承認が必要なアクションは、チェックに合格してから実行してください。チェックに失敗した場合やタイムアウトした場合は、ブロックを維持します。
3. 該当するアクションのブロック。 チェックで問題を検出したら、アプリケーションの状態に、そのリクエストがブロックされたことを記録します。すでにキューに入っている処理も含め、ツールを実行したり変更を確定したりする前に、この状態を確認してください。
4. 関連する処理の停止。 バックエンドがキャンセルに対応している場合は、アプリケーションが管理するジョブをキャンセルします。ブロック済みのリクエストや、新しいリクエストに置き換えられたものについては、後から届いた結果を破棄します。Responses への委譲では、該当するカスタム関数の実行を停止し、ブロックした処理を続行するための response.create は送信しないでください。これだけでは、ホスト側ですでに生成中のレスポンスはキャンセルされず、フロントエンドの発話も停止しません。
5. 判断の記録と会話の軌道修正。 該当するリクエスト ID と委譲 ID を判断結果とともにログに記録し、会話を修正するための指示を送信します。

断片の収集方法は文字起こしの差分を、バックエンドの結果を現在のタスクに合わせる方法は委譲とツールを参照してください。

会話の軌道修正

ガードレールに沿って会話を修正するには session.instructions.append を使用します。進行中の発話に割り込み、新しい指示を適用できます。たとえば、アプリケーションでリクエストをブロックした後に、次のように送信します。

1
2
3
4
5
6
7
8
9export function sendUpdate(connection) {
  connection.send({
    type: "session.instructions.append",
    event_id: "guardrail_block_17",
    delegation_id: null,
    content:
      "Stop speaking immediately. Do not continue or act on the last request. Refuse briefly, then wait.",
  });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15from openai.resources.live.live import AsyncLiveConnection
from openai.resources.live.sideband import AsyncSidebandConnection


async def send_update(
    connection: AsyncLiveConnection | AsyncSidebandConnection,
) -> None:
    await connection.session.instructions.append(
        event_id="guardrail_block_17",
        delegation_id=None,
        content=(
            "Stop speaking immediately. Do not continue or act on the last request. "
            "Refuse briefly, then wait."
        ),
    )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24import (
	"context"
	"github.com/openai/openai-go/v3"
	"github.com/openai/openai-go/v3/live"
	"github.com/openai/openai-go/v3/packages/param"
)

// SendSidebandUpdate uses the server connection attached to an existing session.
func SendSidebandUpdate(ctx context.Context, connection *live.SidebandConnection) error {
	return connection.Send(ctx, live.ConnectClientEventUnionParam{OfSessionInstructionsAppend: &live.InstructionsAppendEventParam{
		EventID:      openai.String("guardrail_block_17"),
		DelegationID: param.Null[string](),
		Content:      "Stop speaking immediately. Do not continue or act on the last request. Refuse briefly, then wait.",
	}})
}

// SendUpdate requires a started Live session.
func SendUpdate(ctx context.Context, connection *live.LiveConnection) error {
	return connection.Send(ctx, live.ClientEventUnionParam{OfSessionInstructionsAppend: &live.InstructionsAppendEventParam{
		EventID:      openai.String("guardrail_block_17"),
		DelegationID: param.Null[string](),
		Content:      "Stop speaking immediately. Do not continue or act on the last request. Refuse briefly, then wait.",
	}})
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27import com.openai.core.JsonField;
import com.openai.helpers.LiveConnection;
import com.openai.helpers.LiveSidebandConnection;
import com.openai.models.live.*;

// Use the server connection attached to an existing session.
public static void sendUpdate(LiveSidebandConnection connection) {
  connection.send(
      InstructionsAppendEvent.builder()
          .eventId("guardrail_block_17")
          .delegationId(JsonField.<String>ofNullable(null))
          .content(
              "Stop speaking immediately. Do not continue or act on the last request. Refuse briefly, then wait.")
          .build());
}

// Use a primary connection after its session has started.
public static void sendUpdate(LiveConnection connection) {
  connection.send(
      ClientEvent.ofSessionInstructionsAppend(
          InstructionsAppendEvent.builder()
              .eventId("guardrail_block_17")
              .delegationId(JsonField.<String>ofNullable(null))
              .content(
                  "Stop speaking immediately. Do not continue or act on the last request. Refuse briefly, then wait.")
              .build()));
}

1
2
3def send_update(connection)
  connection.send_event(type: "session.instructions.append", event_id: "guardrail_block_17", delegation_id: nil, content: "Stop speaking immediately. Do not continue or act on the last request. Refuse briefly, then wait.")
end

指示の内容はアプリケーション側で作成してください。信頼できないユーザー入力をコピーして指示として使わないでください。この修正はセッション全体に適用するため delegation_id: null を使用し、content は 500 トークン以内に収めます。

session.instructions.appended は client_event_id を使って送信したコマンドと対応付けます。この確認応答は、コンテキストへの反映が完了したと推定される時点より後に届きます。通話相手に音声が届くのを止めるには、以下の再生制御を使用してください。

特定の文言で告知を読み上げる必要がある場合も、指示を使用します。具体例と再生時の注意点は告知の読み上げを参照してください。

必要に応じた再生制御

アプリケーションでモデルの音声をブロックする必要がある場合は、クライアントまたはメディアリレーで再生を制御します。出力を一時的にミュートするか破棄し、ローカルの再生キューに残っている音声を破棄して、修正指示を送信します。古い音声を取り除いてから、アプリケーションの復旧ポリシーに従って再生を再開してください。指示への確認応答は再生再開の合図ではありません。サイドバンドだけでは再生を制御できず、通話相手がすでに聞いた音声は修正指示を送っても取り消せません。

session.input_audio.mute が制御するのは通話相手のマイク入力です。モデルの出力をミュートしたり、委譲した処理をキャンセルしたりすることはできません。

再生前の発話チェック

ほとんどのアプリケーションでは、会話を続けながらユーザーとアシスタントの文字起こしを監視します。ガードレールが作動したら、アプリケーションは該当するアクションをブロックしたり、修正指示を送信したりできます。

アシスタントの発話を再生前にチェックする必要がある場合は、通話相手に送る前に、プレーヤーまたはメディアブリッジで音声をバッファリングします。チェックの実行中も音声と文字起こしのイベントを受信し続け、再生とは独立して文字起こしを読み取ってください。

1. 音声と文字起こしの収集。 アプリケーションに出力側の音声区間検出（VAD）を実装するか、メディアフレームワークのノイズゲートを使用して、チェック対象となる発話区間の候補を特定します。GPT-Live は、この処理に使える出力側の VAD やノイズゲートを提供していません。各区間のチェックに必要な文字起こしが届くまで待ちます。
2. 承認済み音声の再生。 区間がチェックに合格したら、バッファに保存しておいたその区間の元の音声を再生キューに追加します。チェックに失敗した場合、文字起こしがない場合、またはチェックがタイムアウトした場合は、その区間を破棄し、アプリケーションで定めた安全な代替処理を行います。
3. 割り込みへの対応。 音声、文字起こし、チェック結果、再生状態を、アプリケーションが生成した ID に紐付けます。割り込みによってその発話がキャンセルされたら、バッファと再生キューから該当する音声を削除し、後からその発話の承認が届いても無視します。

モデルが回答を生成している途中でも、発話の途切れ目を手がかりに、チェック対象の区間を仮に区切ることができます。ポリシーで回答全体のチェックが必要な場合は、回答の完了をアプリケーションでどう判定するかを定めてください。音声区間検出だけでは、回答全体が完了したかどうかは判断できません。

バッファリングすると再生までの遅延が増えます。破棄した発話もモデルの会話コンテキストには残るため、音声を通話相手に届けなかった場合や代替処理を行った場合に、その後の会話がどう進むかをテストしてください。

ガードレール作動時の動作テスト

許可されるリクエストとブロックされるリクエスト、誤検知、チェックの遅延や失敗、発話中やツール実行中のガードレール作動、キャンセルした処理の結果が遅れて届く場合をテストします。アクションのブロック、アプリケーションの状態、修正指示に応じた発話、実際の再生は、それぞれ個別に検証してください。出力を制御する場合は、再生キューに残っている音声の扱いと再生の復旧もテストに含めます。音声エージェント評価の Cookbook を使用して、タスクの成功率と音声応答までの時間を比較してください。

セッションの終了処理

ツールの処理が完了し、セッションから最終的な使用量が報告されるまで、イベントを受信し続けます。session.closed のハンドラーを登録してから session.close を送信し、保留中の処理がすべて完了するまで WebRTC 接続、データチャネル、サイドバンドを開いたままにします。リソースを解放する前に、セッションの最終使用量と、Responses イベントで受信したバックエンドの使用量を保存してください。最終イベントが届く前に接続が途切れた場合は、終了処理が未完了であることを記録します。終了手順はセッションの管理を参照してください。

Realtime API では、クライアントから WebRTC または SIP を介して API サーバーに直接接続できます。ただし、ツールの実行やその他のビジネスロジックは、多くの場合、アプリケーションサーバーに置くのが適しています。こうすることで、ロジックを非公開に保ち、クライアントに依存せずに実行できます。

「サイドバンド」と呼ばれる制御用チャネルで接続すると、ツールの実行やビジネスロジックなどの実装をサーバー側で安全に管理できます。サイドバンドは、SIP 接続と WebRTC 接続のどちらでも利用できます。

サイドバンドを使用すると、同じ Realtime セッションに対して、ユーザーのクライアントからの接続とアプリケーションサーバーからの接続の 2 本が同時に有効になります。サーバー側の接続では、セッションの監視、指示の更新、ツール呼び出しへの応答ができます。

WebRTC での接続

1. ピア接続を確立する際は、Realtime API にリクエストを送り、返された SDP レスポンスを使って接続を設定します。WebRTC ガイドのサンプルコードを使用した場合は、次のようになります。

1
2
3
4
5
6
7
8
9const baseUrl = "https://api.openai.com/v1/realtime/calls";
const sdpResponse = await fetch(baseUrl, {
  method: "POST",
  body: offer.sdp,
  headers: {
    Authorization: `Bearer ${EPHEMERAL_KEY}`,
    "Content-Type": "application/sdp",
  },
});

2. 取得したレスポンスの Location ヘッダーには、一意の通話 ID が含まれています。サーバーはこの ID を使い、同じ Realtime セッションへの WebSocket 接続を確立できます。

1
2
3
4// Location: /v1/realtime/calls/rtc_123456
const location = sdpResponse.headers.get("Location");
const callId = location?.split("/").pop();
console.log(callId);

3. サーバーでは、通常の Realtime API の WebSocket 接続と同じように、イベントの受信とセッションの設定ができます。次の例のように、通話 ID を URL wss://api.openai.com/v1/realtime?call_id=rtc_xxxxx に指定します。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30import WebSocket from "ws";
const callId = "rtc_u1_9c6574da8b8a41a18da9308f4ad974ce";

// Connect to a WebSocket for the in-progress call
const url = "wss://api.openai.com/v1/realtime?call_id=" + callId;
const ws = new WebSocket(url, {
  headers: {
    Authorization: "Bearer " + process.env.OPENAI_API_KEY,
  },
});

ws.on("open", function open() {
  console.log("Connected to server.");

  // Send client events over the WebSocket once connected
  ws.send(
    JSON.stringify({
      type: "session.update",
      session: {
        type: "realtime",
        instructions: "Be extra nice today!",
      },
    })
  );
});

// Listen for and parse server events
ws.on("message", function incoming(message) {
  console.log(JSON.parse(message.toString()));
});

Go では、サーバー側でセッションを更新する関数に通話 ID を渡します。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28func updateSession(ctx context.Context, client openai.Client, callID string) error {
	conn, err := client.Realtime.Connect(ctx, realtime.RealtimeConnectionOptions{}, option.WithQuery("call_id", callID))
	if err != nil {
		return err
	}
	defer conn.Close()
	if err := conn.Send(ctx, realtime.RealtimeClientEventParamOfSessionUpdate(realtime.RealtimeSessionCreateRequestParam{
		Instructions: openai.String("Be extra nice today!"),
	})); err != nil {
		return err
	}
	for {
		event, err := conn.Recv(ctx)
		if err != nil {
			return err
		}
		if event.Type == "error" {
			return fmt.Errorf("Realtime API error: %s", event.OfRealtimeErrorEvent.Error.Message)
		}
		if event.Type == "session.updated" {
			if event.OfSessionUpdatedEvent.Session.OfRealtimeSessionCreateRequest.Instructions != "Be extra nice today!" {
				return fmt.Errorf("session update was not applied")
			}
			fmt.Println("session.updated")
			return nil
		}
	}
}

こうすることで、ツールの追加、セッションの監視、ビジネスロジックの実行をサーバー側で行えます。クライアント側でこれらの処理を設定する必要はありません。

SIP での接続

1. ユーザーが電話から SIP 経由で OpenAI に接続します。
2. OpenAI は、アプリケーションサーバーの Webhook URL に Webhook を送信し、セッションの状態をアプリケーションに通知します。Webhook の例を以下に示します。

123456789101112131415161718192021POST https://my_website.com/webhook_endpoint
user-agent: OpenAI/1.0 (+https://platform.openai.com/docs/webhooks)
content-type: application/json
webhook-id: wh_685342e6c53c8190a1be43f081506c52 # unique id for idempotency
webhook-timestamp: 1750287078 # timestamp of delivery attempt
webhook-signature: v1,K5oZfzN95Z9UVu1EsfQmfVNQhnkZ2pj9o9NDN/H/pI4= # signature to verify authenticity from OpenAI

{
  "object": "event",
  "id": "evt_685343a1381c819085d44c354e1b330e",
  "type": "realtime.call.incoming",
  "created_at": 1750287018, // Unix timestamp
  "data": {
    "call_id": "some_unique_id",
    "sip_headers": [
      { "name": "From", "value": "sip:+142555512112@sip.example.com" },
      { "name": "To", "value": "sip:+18005551212@sip.example.com" },
      { "name": "Call-ID", "value": "03782086-4ce9-44bf-8b0d-4e303d2cc590"}
    ]
  }
}

3. アプリケーションサーバーは、Webhook で受け取った call_id の値を使い、wss://api.openai.com/v1/realtime?call_id={callId} のような URL で Realtime API への WebSocket 接続を開きます。この WebSocket 接続は、SIP 通話が終了するまで維持されます。

以降は、WebSocket 接続でセッションを開始した場合と同様に、この接続でイベントを送受信して通話を制御できます。通話の監視、指示の動的な更新、ツール呼び出しへの応答などが可能です。