title: "音声読み上げの回答を「3文ずつ」にチャンク分割する実装【Python】"
tags: Python, FastAPI, 音声認識, 介護AI, チャンキング
published: false

全体像はnoteに → https://note.com/yamashita_aidev/n/n7654a6de99cf

環境

Python 3.11+
FastAPI 0.111+

問題

音声AIシステムで長い手順を「次」コマンドで分割して返す機能を実装していた。初期設計では「次」という発話も毎回 LLM API を呼ぶ設計になっていた。

# 「次」コマンドの旧フロー（問題のある設計）
マイク入力
  → STT API (Whisper 等)       ← 1〜2秒 / 毎回課金
  → サーバーでコマンド判定
  → LLM API (step_offset 付き) ← 2〜3秒 / 毎回課金
  → TTS API
  → 音声再生

連続ナビゲーション5回で、API コストは初回質問の6倍になっていた。

解決策

初回質問で全回答を生成 → chunk_answer() でチャンク分割してキャッシュ → 「次」では TTS のみ実行する。

chunk_answer() の実装

def chunk_answer(text: str, sentences_per_chunk: int = 3) -> list[str]:
    """回答テキストを「。」で分割し、指定文数ずつチャンク化する。

    Args:
        text: LLM が生成した回答テキスト（マークダウン記号なし前提）
        sentences_per_chunk: 1チャンクあたりの文数。デフォルト3文。

    Returns:
        チャンク文字列のリスト。空入力の場合は空リストを返す。
    """
    # 「。」で分割し、各文に「。」を再付加する
    sentences = [s + "。" for s in text.split("。") if s.strip()]
    if not sentences:
        # 入力が空またはスペースのみの場合は空リストを返す
        return [text] if text.strip() else []

    chunks: list[str] = []
    for i in range(0, len(sentences), sentences_per_chunk):
        # sentences_per_chunk 文ごとにひとつのチャンクにまとめる
        chunk = "".join(sentences[i : i + sentences_per_chunk])
        chunks.append(chunk)
    return chunks if chunks else [text]

使用例

text = "まず手を洗います。次にグローブを着けます。患者さんに声をかけます。ベッドの角度を調整します。"
result = chunk_answer(text, sentences_per_chunk=3)
# → ["まず手を洗います。次にグローブを着けます。患者さんに声をかけます。",
#    "ベッドの角度を調整します。"]

4文の入力が「3文」と「1文」に分割される。

「3文ずつ」の根拠

音声読み上げに適したチャンクサイズの選定はユーザビリティテストで決めた。

チャンクサイズ	評価
2文	短すぎて「次」の頻度が増える
3文	聞き取れる上限。現場スタッフが支持
4文	デバイス本体スピーカーで追いきれない

騒がしい介護施設フロアでは、1チャンクあたり3文が「全部聞き取れる最大量」だった。

セッションコンテキストへの格納

チャンク分割後のデータは FastAPI の app.state で管理するインメモリ辞書に保存する。

# インメモリのセッション辞書 — WebSocket セッション単位で保持する
session_contexts: dict[str, dict] = {}

# --- 初回質問処理後のコンテキスト更新 ---
answer_chunks = chunk_answer(answer_text)  # 全回答をチャンク分割

session_contexts[session_id] = {
    "original_query": original_query,       # デバッグ・再質問用
    "context_chunks": scored_chunks,        # RAG 検索結果（再利用用）
    "answer_chunks": answer_chunks,         # 分割済みチャンクリスト
    "current_chunk_index": 0,               # 現在何チャンク目まで返したか
    "full_answer": answer_text,             # 全回答テキスト（デバッグ用）
}

「次」コマンド処理 — STT/RAG/LLM を一切呼ばない

async def _process_next_command(
    websocket: WebSocket,
    session_id: str,
    tts_service: TTSService,
) -> None:
    """「次」コマンド処理 — STT/RAG/LLM を一切呼ばない。

    セッションコンテキストからチャンクを取り出し、TTS のみ実行する。
    外部 API への通信は TTS のみ発生する。
    """
    ctx = session_contexts.get(session_id)
    if ctx is None:
        # セッション切れ: クライアントへのエラー通知処理をここに追加する
        return

    answer_chunks = ctx["answer_chunks"]
    next_index = ctx["current_chunk_index"] + 1

    if next_index >= len(answer_chunks):
        # 全チャンク読み上げ完了 → 完了メッセージを返してセッションを破棄する
        complete_text = "以上で全ての内容をお伝えしました。"
        audio_bytes, duration_ms = await tts_service.synthesize(complete_text)
        session_contexts.pop(session_id, None)  # セッションを破棄する
        return

    # 次のチャンクを取得して TTS のみ実行 — API コストが発生するのはここだけ
    chunk_text = answer_chunks[next_index]
    ctx["current_chunk_index"] = next_index
    audio_bytes, duration_ms = await tts_service.synthesize(chunk_text)
    # audio_bytes を WebSocket でクライアントに送信する処理を続ける

Before / After の比較

処理	初回質問	「次」Before	「次」After
STT API	実行 (1-2秒)	実行（無駄）	スキップ
RAG 検索	実行	スキップ	スキップ
LLM 呼び出し	実行 (2-3秒)	実行（無駄）	スキップ
TTS 合成	実行 (1秒)	実行	実行（チャンクのみ）
合計レイテンシ	4-6秒	4-6秒（無駄）	1秒以下

連続ナビゲーション5回を含むセッションで、API コストは約90%削減になる。

LLM プロンプトの変更点

チャンキング方式に切り替える際、LLM へのプロンプトも変更が必要になる。

# 旧プロンプト（毎回 LLM を呼ぶ設計）
「手順のステップ {step_offset} 番目以降から回答してください。」

# 新プロンプト（全回答を1回で生成する設計）
「手順がある場合はすべてのステップを含めて完全に回答してください。
回答は音声読み上げ専用のため、マークダウン記号は使用しないでください。
接続詞として『まず』『次に』『それから』『最後に』を使ってください。」

マークダウン禁止を入れ忘れると、TTS が「シャープシャープステップ1 アスタリスク...」と読み上げる。音声出力では記号がそのまま読まれるため注意が必要だ。

注意点・ハマりどころ

chunk_answer() は「。」以外の区切りに非対応

現実装は「。」のみを区切り文字として使用している。英語混じりの文章や「！」「？」で終わる文がある場合は、区切りパターンを正規表現に拡張する必要がある。

import re

def split_sentences(text: str) -> list[str]:
    """日本語・英語混在テキストの文分割（拡張版）。

    「。」「！」「？」「.」「!」「?」を区切りとして分割する。
    """
    # 区切り文字の直後で分割するが、区切り文字自体は保持する
    pattern = re.compile(r"(?<=[。！？.!?])")
    parts = pattern.split(text)
    return [p for p in parts if p.strip()]

セッション TTL（有効期限）を設定すること

本コードにはセッション TTL が含まれていない。本番運用では WebSocket の disconnect イベントで明示的に破棄すること。

@app.websocket("/ws/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str) -> None:
    await websocket.accept()
    try:
        # ... メインループ ...
        pass
    finally:
        # 切断時にセッションコンテキストを必ず破棄する
        session_contexts.pop(session_id, None)

まとめ

chunk_answer() は「。」で文分割し、sentences_per_chunk 文ずつ結合するだけのシンプルな実装
デフォルトの3文はユーザビリティテスト由来。用途によって変更する
チャンキングとセッションキャッシュを組み合わせることで「次」コマンドの LLM/STT 呼び出しをゼロにできる
LLM プロンプトの「マークダウン禁止」指示は音声出力では必須

全体像はnoteに → https://note.com/yamashita_aidev/n/n7654a6de99cf