こんにちは。この記事では、RAG と TTS を組み合わせたリアルタイムシステムでのレイテンシ削減にフォーカスし、実運用で使える「バンドル・オーディオ・キャッシュ」手法を紹介します。コード例と運用上の注意点も交えて分かりやすく説明します。
オーディオを単独で走らせないでください。RAG の応答をキャッシュする際、テキストと一緒にオーディオ（Base64）を同じペイロードにまとめてください。
小さな工夫で、大きな効果が期待できます。
1. 問題点（The Problem）
RAG と音声（TTS）を組み合わせた一般的なフローは次のようになります：
バックエンドでは通常、次の手順が発生します：
- テキスト用のキャッシュを問い合わせる
- キャッシュミスなら RAG を実行する
- TTS API を呼んでオーディオを生成する
- テキストをキャッシュする
- （あれば）オーディオをキャッシュする
結果として、ネットワークの往復が 2 回、Redis への問い合わせが 2 回発生することになり、オーディオは常にテキストより遅くなります。リアルタイム用途では 20〜50ms の差も問題になります。
2. 音声は本当に「別個」にする必要があるか？
同じ入力に対してテキストとオーディオが決定論的に同じであれば、なぜ別々にキャッシュする必要があるのでしょうか。ここで「バンドル（まとめて）キャッシュ」戦略が意味を持ちます。
3. 解決策：バンドル・オーディオ・キャッシュ（All-in-One Payload）
アイデア
RAG の応答をキャッシュする際、オーディオ（Base64）を同じオブジェクトに含めます。1 つのキャッシュキー — 1 つのペイロードですべてを返せるようにします。
ペイロード例
{
"raw": "RAG の生の応答",
"script": "TTS 用にフォーマットしたテキスト",
"audio_data": "BASE64_ENCODEドされたオーディオ",
"schema_version": 1
}
フロントエンドは完全なペイロードを受け取るため、追加の TTS 呼び出しは不要になります。
4. ビフォー / アフターの比較
従来（分離キャッシュ）では、テキストとオーディオを別々に問い合わせます：
往復が 2 回発生します。
バンドル方式では：
オーディオはテキストに "同乗" するため、オーディオ側の追加レイテンシはほぼゼロになります（キャッシュヒット時）。
5. コード例（Python）
以下は簡潔な runnable な例です。synthesize_tts はプレースホルダなので実際の TTS クライアントに置き換えてください。
import base64
import json
import time
import redis
import requests
# --- config ---
REDIS_URL = "redis://localhost:6379/0"
redis_cli = redis.from_url(REDIS_URL)
TTL_SECONDS = 60 * 60 * 24 # 1 day
def synthesize_tts(text: str) -> bytes:
"""仮の TTS 呼び出し。実環境では実際の TTS API を使用する。"""
resp = requests.post("https://example-tts/synthesize", json={"text": text})
resp.raise_for_status()
return resp.content
def make_bundled_payload(raw: str, script: str, audio_bytes: bytes) -> str:
audio_b64 = base64.b64encode(audio_bytes).decode("ascii")
payload = {
"raw": raw,
"script": script,
"audio_data": audio_b64,
"schema_version": 1,
}
return json.dumps(payload)
def store_bundled(key: str, raw: str, script: str):
audio = synthesize_tts(script)
payload = make_bundled_payload(raw, script, audio)
redis_cli.set(key, payload, ex=TTL_SECONDS)
def get_bundled(key: str):
cached = redis_cli.get(key)
if not cached:
return None
data = json.loads(cached)
audio_b64 = data.get("audio_data")
if audio_b64:
audio_bytes = base64.b64decode(audio_b64)
else:
audio_bytes = None
return {
"raw": data.get("raw"),
"script": data.get("script"),
"audio_bytes": audio_bytes,
}
# --- usage ---
if __name__ == "__main__":
k = "answer:how-to-cache-kem:ja"
# store_bundled(k, raw_text, formatted_script)
start = time.perf_counter()
res = get_bundled(k)
elapsed_ms = (time.perf_counter() - start) * 1000
print(f"Got bundled? {bool(res)} - latency {elapsed_ms:.1f}ms")
注記：
- Base64 はバイナリに比べて約 33% 増加します。音声を小さく保つために Opus 等で圧縮することを検討してください。
- 大きなオーディオはオブジェクトストレージに置き、ペイロードには小さなプレビューや URL を含める運用も現実的です。
6. 得られる効果（概算）
| 項目 | 以前 | 以後（バンドル） |
|---|---|---|
| Redis GET | 2 | 1 |
| TTS 呼び出し | 必要 | 不要（キャッシュヒット時） |
| オーディオレイテンシ | 100–500ms | 約 0ms（バンドルヒット時） |
| バックエンドの複雑さ | 高い | 低い |
実際の数値はネットワーク状況、オーディオサイズ、キャッシュヒット率に依存します。
7. いつ使うべきか
適しているケース：
- 同じ質問が頻出する
- オーディオが決定論的（ランタイムコンテキストに依存しない）
- リアルタイム／音声優先のアプリケーション
- ストレージ増加を許容してでもレイテンシを削りたい
避けるべきケース：
- オーディオがユーザーごとの声質や感情に依存する
- キャッシュストレージが極端に制限されている
- オーディオをストリーミングで生成する必要がある
8. 注意点
重要：バンドルでキャッシュサイズは増加します。対策として：
- 適切な TTL を設定する
- 適切な eviction policy（LRU/LFU など）を採用する
- キャッシュサイズの監視を行う
-
schema_versionをペイロードに入れて後方互換性を確保する
9. まとめ
同じ答えを何度も出すユースケースでは、オーディオをテキストと一緒にキャッシュすること（Bundled Audio Caching）は小さな設計変更で大きなレイテンシ改善をもたらします。ただし、ストレージコストと運用負荷は増えるため、測定と監視を忘れないでください。