通話路を覗くな、シグナリングを引け — crossbar_telepath 開発記 (2)

前回、TerraformでコールセンターをIaC構築し、公衆電話網からKinesis Video Streams(KVS)まで通話音声を引っ張ってくるところまでやった。MKVコンテナのCodecIDがA_AACを詐称していて中身は生PCMだった、という罠を突破して、自分の声をWAVで取り出したところで終わっている。

今回はその続きで、その声をリアルタイムに文字起こしして、ブラウザにチャットのように流すところまで。そして途中で「そもそも、いま処理しているこの音声はどの呼のものなのか?」という、電話屋なら最初に気づくべき問題に突き当たる。結論から言うと、そこがこの回の本題である。

まず「EC2を立てるか」問題

作業を始める前に迷ったのが、受信装置をどこに置くかだった。KVSから音声を吸い上げる常駐プロセスが要る。素直に考えればEC2かECSを立てる話になる。

だが結論としては何も立てなくてよかった。KVSのGetMediaはただのAWS APIで、認証情報さえあれば手元のMacから叩ける。網内に受信装置を置く必要はどこにもない。実際、この記事で書く作業は全部ローカルのコンテナで完結している。

クラウドに常設するのは「通話が来たら自動で受ける本番」になってからで十分だ。開発中にEC2へSSHして書き直しては再起動、というループを回すのは、単にイテレーションが遅くなるだけである。

リアルタイム文字起こしの配管

やることは4段の配管だ。

GetMediaでKVSからMKVバイト列を受ける
EBMLを逐次パースして話者別のPCMに分ける
電話帯域8kHzをOpenAIが要求する24kHzへ変換して流す
返ってきたテキストをWebSocketでブラウザへ

一番不確実だったのは3と4、つまりOpenAI側の受け口だった。ここは実装前に最小の実験で潰しておくに限る。

Realtime API の transcription セッション

OpenAI Realtime APIには「会話」ではなく「文字起こし専用」のセッションがある。WebSocketで?intent=transcriptionに繋ぎ、こう宣言する。

{
  "type": "session.update",
  "session": {
    "type": "transcription",
    "audio": {
      "input": {
        "format": {"type": "audio/pcm", "rate": 24000},
        "transcription": {"model": "gpt-4o-transcribe", "language": "ja"},
        "turn_detection": {"type": "server_vad", "silence_duration_ms": 500}
      }
    }
  }
}

あとは音声を100msずつinput_audio_buffer.appendで流し込むだけ。無音の検出(VAD)はOpenAI側がやってくれるので、こちらで発話の切れ目を判断する必要がない。前回の録音を流し込んでみると、こう返ってきた。

[speech_started] → [delta] 'もし' → [delta] 'もし' → [completed] 'もしもし'
[speech_started] → [delta] '以上' → [delta] '。'   → [completed] '以上。'

発話が2つに区切られ、文字が逐次流れて、確定する。狙いどおりの挙動だ。

謎解き1: RESTとRealtimeで結果が違う

実はこの実験の前に、同じ音声をRESTの/v1/audio/transcriptionsにも投げていた。そちらの答えはこうだった。

申します。以上。

「もしもし」が「申します」になっている。 同じOpenAI、同じ音声、同じ言語指定である。

理由は推測になるが、RESTは9.7秒のファイルを丸ごと一度に処理するのに対し、Realtimeはserver VADが発話区間を切り出してから処理する。電話帯域の8kHz音声は情報が薄いので、無音を含む長い区間より、発話だけを切り出したほうが当たりやすいのだろう。

ついでに分かったのは、8kHzのままでも日本語はちゃんと取れるということ。24kHzへのアップサンプルは精度のためではなく、単にAPIのフォーマット要件を満たすためだけの処理だった(実験では8kHzと24kHzで結果に差が出なかった)。電話の音質でAIがどこまで聞き取れるかは今回の企画の生命線だったので、ここが早々に確認できたのは大きい。

パーサをストリーミングに書き換える

前回作ったMKV抽出ツールは、ファイル全体をライブラリに渡して一気に解析する方式だった。リアルタイムではこれが使えない。届いたバイトだけを食わせて、取り出せた分だけ吐き出す作りに書き直す必要がある。

とはいえEBMLの構造は単純で、要素は「ID(可変長整数)+ サイズ(可変長整数)+ 中身」の繰り返しでしかない。降りていきたいマスター要素(Segment、Cluster、Tracks)だけ列挙して中に入り、それ以外はサイズ分読み飛ばす。音声の入ったSimpleBlockが完全に揃ったら1つ吐き出し、途中で足りなくなったら次のデータを待つ。

KVS特有の事情として、フラグメントごとにEBMLヘッダとSegmentが何度も再出現する。しかもSegmentは長さ未知(サイズのビットが全部1)で届く。ffmpegが以前吐いていた「Found unknown-length element with ID 0x18538067」という警告はこれだったわけだ。降りるべき要素として扱えば、何度現れても素直に処理できる。

書き上げたら、前回の抽出結果と照合する。ネットワーク受信を模して1〜5000バイトのランダム長で刻んで食わせてみた。

track_names: {1: 'AUDIO_TO_CUSTOMER', 2: 'AUDIO_FROM_CUSTOMER'}
track 1: 154,240 bytes, 9.6s   ← 前回のオフライン抽出と完全一致
track 2: 154,560 bytes, 9.7s   ← 同上

チャンクの切れ目がどこに来ても結果が変わらない。ついでに、前回は決め打ちしていたトラック番号と話者の対応も、MKVのTracks要素から読むようにした。前回まさにこの対応を取り違えて「無音側が自分の声」という結論を出しかけたので、二度と推測しないと決めた次第である。

動いた。が、片側しか喋らない

ブラウザまで繋いで、録音をリプレイしてみる。相手側の発言が「もしもし」「以上。」とチャットバブルで流れた。E2E成功である。

呼ごとのカードに話者別の文字起こしがチャット形式で表示された画面 — 呼単位のカードに、話者別の発言が流れる

ところがこちら側(TO_CUSTOMER)のバブルが一度も出ない。

謎解き2: 無音ループの罠

原因はコールフローにあった。前回作ったIVRはこういう流れになっている。

録音同意アナウンスを流す
メディアストリーミングを開始する
無音を1分×10回ループして通話を維持する
切断

お分かりだろうか。ストリーミングを開始した後、こちら側は永久に何も喋らない。 アナウンスは全部ストリーミング開始「前」に流し終わっている。つまりTO_CUSTOMERトラックには構造上、未来永劫デジタル無音しか乗らない。

「両話者の感情を可視化する」というプロダクトなのに、片方の音声チャネルが原理的に死んでいたわけだ。テストデータの都合ではなく設計の穴なので、ストリーミング開始直後にアナウンスを1つ挟むようフローを修正した。

本題: 呼が特定できない

さて、ここまでで動くものはできた。だが根本的な問題が残っていた。

SCPは、いま処理している音声がどの呼のものか知らない。

初版の実装はこうだった。KVSのストリーム一覧を3秒おきにポーリングし、見たことのない新しいストリームが現れたら「呼が来たらしい」と判断して受信を始める。

動きはする。だが電話網の言葉に翻訳すると、これは通話路(ベアラ)を覗いて呼の存在を推測している状態である。共通線信号でいうIAM(初期アドレスメッセージ)を受け取らないまま、通話チャネルに音が乗ったのを見て「誰か繋がったっぽい」と判断しているに等しい。交換機の設計として、これはあり得ない。

実害も3つあった。

通話の頭が取れない。 ポーリングで気づいてからStartSelectorType=NOWで繋ぐので、それまでの音声は永久に失われる
呼の情報が何も分からない。 発信者番号も、Connect側のコンタクトIDも、ストリーム名から推測するしかない
同時に複数の呼が来たら混乱する。 どのストリームがどの呼か対応づける根拠がない

シグナリングを引く

解決策は、電話網の常識どおり制御信号の線を別に引くことだった。

Amazon Connectのコールフローには「外部リソース呼び出し」ブロックがあり、Lambdaを起動できる。そしてメディアストリーミング開始後にこれを呼ぶと、Lambdaのイベントに呼の同定情報が入ってくる。

{
  "Details": {
    "ContactData": {
      "ContactId": "af330898-...",
      "CustomerEndpoint": {"Address": "+81..."},
      "MediaStreams": {"Customer": {"Audio": {
        "StreamARN": "arn:aws:kinesisvideo:...",
        "StartFragmentNumber": "91343852333181432392..."
      }}}
    }
  }
}

呼のIDと、その呼専用のストリームと、通話の先頭のフラグメント番号。推測ではなく、交換機からの通知である。

このLambdaがSQSへメッセージを投げ、消費サービス側はそれをロングポーリングで待つ。通話路(KVS)とシグナリング(SQS)が完全に分離された構成になった。

Amazon Connectから、シグナリング(Lambda→SQS)と通話路(KVS)の2経路が消費サービスへ入る構成図 — 左がシグナリング、右が通話路。交換機と同じく別の線として引く

SQSを選んだのは、ローカル開発でもFargate常設でも受け取り側のコードが一切変わらないからだ。HTTPで直接叩かせる方式だと、開発中は消費サービスが公開されていないので届かない。キューなら呼が溜まって待っていてくれる。

得られた効果は大きい。呼とストリームの対応が確定して取り違えようがなくなり、StartFragmentNumberから読むので通話の先頭から音声が取れるようになった。発信者番号も画面に出せる。そして複数の呼を独立したセッションとして並行処理できる。

実際にリプレイを時間差で2本走らせてみると、それぞれ別のコンタクトIDで並行処理され、発話が混ざらずに別カードへ振り分けられた。呼単位の管理が成立している。

今回のまとめ

KVSのGetMediaはただのAPI。受信のためにEC2を立てる必要はない
Realtime APIのtranscriptionセッションは、server VADに区切りを任せられて具合がいい。同じ音声でRESTより精度が良かった
電話帯域の8kHzでも日本語は十分に文字になる。24kHz変換はフォーマット要件を満たすためだけの処理
ストリーミング用のEBMLパーサは、降りる要素だけ列挙すれば素直に書ける
通話路を覗いて呼を推測してはいけない。 シグナリングを別に引くと、呼の同定・通話の先頭・発信者番号・複数呼の並行処理が一度に手に入る

最後の項目は、電話網をやっていた人間なら設計前に気づくべきことだった。クラウドのAPIとして眺めていると「ストリーム一覧をポーリングすればいいか」と流してしまうが、これは交換機の話なのだ。制御と通話路は別、という30年前から変わらない原則に引き戻されたのが、今回いちばんの収穫だった。

次回はいよいよ本命、相手がキレているかどうかをAIに判定させるフェーズに入る。文字起こしの配管がそのまま使えるので、音声を同じように流して「怒っているか」を返させればいい。カスハラ対策という当初の狙いに、ようやく手が届く。