遺失物受付AI 実装詳細

Gemini 3.1 Flash Live API をベースとした、タスク指向リアルタイム音声対話システムの技術解説。 Function Calling による構造化状態管理を中心に、音声処理・相槌・電話番号リピートの実装を解説する。

Overview概要

このシステムは、遺失物受付窓口の対話をAIで代替するデモアプリケーションである。 単なる自由対話ではなく、受付票の5項目（落とした物・特徴・場所・日時・連絡先）を漏れなく収集し、完了を確実に検知する タスク指向対話（Task-Oriented Dialogue; TOD）として設計されている。

従来のTODシステムはルールベースの状態機械か、Fine-tuningされた専用モデルで実装されることが多かった。 本システムは Gemini の Function Calling（FC） を活用することで、 自由な雑談と構造化された情報収集を単一の大規模言語モデルで両立している。

Archシステム構成

通信プロトコルはステートフル WebSocket。 ブラウザは AudioWorklet（pcm-worklet.js）で16kHz PCMバイナリを連続送信し、 FastAPI は Gemini Live とのセッションを中継する。 Gemini からはPCM音声バイナリ・テキストトランスクリプト・Function Callの3種類のイベントが届く。

CoreFunction Calling（FC）

FCとは何か

Function Calling（ツール呼び出し）とは、LLMが応答を生成する代わりに 「この関数をこの引数で呼んでほしい」というJSONを出力する機能である。 OpenAI が2023年6月に GPT-4 向けに実装して以降、Gemini・Claude・Llama 等の主要モデル全てが対応する業界標準機能となった。

仕組みの詳細

セッション開始時に、利用可能なツールをOpenAPIスキーマ形式で宣言する。 モデルは会話の文脈から「今ツールを呼ぶべきか」「どの引数で呼ぶか」を自律的に判断する。

# ツール宣言（セッション設定に含める）
UPDATE_REPORT_DECL = {
    "name": "update_report",
    "description": "遺失物受付票の1項目を記録する。confirmed=falseで仮記録、"
                  "confirmed=trueで確定記録。同フィールドの再呼び出しで上書き。",
    "parameters": {
        "type": "object",
        "properties": {
            "field": {
                "type": "string",
                "enum": ["item", "features", "location", "date", "contact"],
            },
            "value":     {"type": "string"},
            "confirmed": {"type": "boolean"},
        },
        "required": ["field", "value", "confirmed"],
    },
}

モデルが「利用者から情報が確定した」と判断すると、 テキスト応答の代わりに（または応答に加えて）tool_callイベントを発火する。 アプリ側はこれを受け取って処理し、send_tool_response()で結果を返す。

# send_loop 内での tool_call ハンドリング
if resp.tool_call and report_state is not None:
    fn_responses = []
    for fc in resp.tool_call.function_calls:
        result = handle_tool(fc.name, dict(fc.args), report_state)
        fn_responses.append(types.FunctionResponse(
            id=fc.id, name=fc.name, response=result,
        ))
    await session.send_tool_response(function_responses=fn_responses)
    # フロントに受付票の現在状態を通知
    await websocket.send_text(json.dumps({
        "type": "state_update", "state": report_state,
    }))

本システムでの実装：2つのツール

confirmedフラグの2段階設計が重要なポイントである。 利用者が情報を述べた段階では confirmed=false（仮記録）、 AIが読み上げて「よろしいですか？」と確認し利用者が同意した段階で confirmed=true（確定）となる。 missing_fieldsはconfirmed=trueの項目のみを「収集済み」とカウントするため、 確認をスキップした項目は必ず再確認が促される。

def handle_tool(name: str, args: dict, state: dict) -> dict:
    if name == "update_report":
        field     = args.get("field")
        value     = args.get("value", "")
        confirmed = bool(args.get("confirmed", False))
        if field == "contact":
            value = normalize_phone(value)  # 電話番号を正規化
        state[field] = {"value": value, "confirmed": confirmed}
        missing = [
            f for f in REQUIRED_FIELDS
            if state.get(f) is None or not state[f].get("confirmed", False)
        ]
        return {
            "result":        "確定記録" if confirmed else "仮記録",
            "missing_fields": missing,
            "all_complete":   len(missing) == 0,
        }

対話フロー：FCによるスロット充填

FCの他の活用例

FCは「自然言語 → 構造化アクション」の汎用変換機構であり、対話システム以外でも広く応用できる。

Sessionセッション管理

本システムはマルチセッション対応のステートフルWebSocket APIとして動作する。 「マルチテナント」（複数組織のデータ分離）とは異なる概念であることに注意。

# ws_endpoint はリクエストごとに独立したコルーチンとして動作
async def ws_endpoint(websocket: WebSocket, ...):
    report_state = make_report_state()  # ← ローカル変数: 接続ごとに独立
    #   グローバル共有ストレージは一切使わない
    async with client.aio.live.connect(...) as session:
        rt = asyncio.create_task(recv_loop())  # マイク受信タスク
        st = asyncio.create_task(send_loop())  # Gemini→ブラウザ送信タスク
        await asyncio.wait([rt, st], return_when=asyncio.FIRST_COMPLETED)

各接続で recv_loop（マイク音声受信）と send_loop（Gemini応答配信）が 非同期タスクとして並走する。一方が切断されると他方もキャンセルされる設計。 uvicorn の asyncio イベントループが多数の接続を1スレッドで並列処理するため、 CPUバウンドでない限りスケールする。

Audio音声パイプライン

入力は AudioWorklet（pcm-worklet.js）で16kHz・16bit PCMとして抽出し、 WebSocket バイナリフレームで連続送信する。 AudioContext のサンプルレートを16kHzに固定することで再サンプリングコストを排除している。

出力は Gemini から返る24kHz Int16LE PCMをFloat32に変換し、 AudioBufferSourceNodeをチェーンして継ぎ目のない再生を実現。 playbackTime を累積することで音声チャンク間のギャップを防いでいる。

UX相槌（あいづち）システム

AIが応答を生成・送信する前の数百ミリ秒の無音を埋めるため、 事前生成した短い相槌音声（「はい」「なるほど」等）をフロントで即時再生する。 これにより会話のリズムが自然になる。

相槌は2つのトリガーで発火する：

発話の文脈（疑問形・共感・挨拶・汎用）を正規表現で分類し、 カテゴリに合った相槌クリップを選択する。 電話番号収集フェーズでは相槌を完全に抑制し、 AIの即時オウム返しとの干渉を防ぐ設計としている。

Feature電話番号リアルタイムリピート

NTTのデモで確認された「電話番号を区切りながら言うと即座にオウム返しする」体験を実装。 技術的には Gemini Live のターン検出（300ms無音でターン終了）を活用することで、 特別な実装なく自然に実現できる。

# システムプロンプトによる制御
"""
▼ 利用者が番号の途中で止まった場合：
  その部分だけをすぐそのまま繰り返す。余計な言葉を足さない。
  例）利用者「080の」→ AI「080の、」

▼ 全桁揃ったら区切りを付けて確認：
  携帯なら 3-4-4: 「080-1234-5678 でよろしいですか？」
  固定電話なら市外局番ルールで区切り
"""

バックエンドでは normalize_phone() が電話番号の桁数に応じて 自動的にハイフン区切りにフォーマットする。 11桁携帯（090/080/070）は3-4-4、 10桁固定電話（03/06など）は2-4-4のルールを適用する。

def normalize_phone(raw: str) -> str:
    digits = re.sub(r"\D", "", raw)
    if len(digits) == 11:
        if digits[:2] in ("09", "08", "07"):
            return f"{digits[:3]}-{digits[3:7]}-{digits[7:]}"
    if len(digits) == 10 and digits[:2] in ("03", "06", ...):
        return f"{digits[:2]}-{digits[2:6]}-{digits[6:]}"
    return raw