定型応答のための音声キャッシュ

from cartesia import Cartesia
import os

CACHE_DIR = "phrase-cache"
MODEL = "sonic-3.5"  # 本番ではスナップショットに固定: "sonic-3.5-2026-05-04"
VOICE_ID = "47c38ca4-5f35-497b-b1a3-415245fb35e1" # ユースケースに合ったボイスを play.cartesia.ai で選択してください
OUTPUT_FORMAT = {"container": "raw", "encoding": "pcm_s16le", "sample_rate": 24000} # 本番の「ライブ」TTS 生成で使用する音声フォーマットと一致させる必要があります

CUSTOMER_NAME="Elon"

# キャッシュする定型フレーズ。
# ブランド向けにカスタムの単語発音を作成する必要がある場合は、
# https://docs.cartesia.ai/build-with-cartesia/capability-guides/custom-pronunciations を参照してください
PHRASES = {
    "greeting": "Thanks for calling Acme support. How can I help you today?",
    "hold": f"Please hold while I look that up for you {CUSTOMER_NAME}.",
    "closing": "Is there anything else I can help you with?",
}

os.makedirs(CACHE_DIR, exist_ok=True)

client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])

# TTS 音声フレーズを生成します。
with client.tts.websocket_connect() as ws:
    for phrase_name, phrase in PHRASES.items():
        ctx = ws.context(
            model_id=MODEL,
            voice={"mode": "id", "id": VOICE_ID},
            output_format=OUTPUT_FORMAT,
            language="en",
        )
        ctx.push(phrase)
        ctx.no_more_inputs()

        audio = bytearray()
        for response in ctx.receive():
            if response.type == "chunk" and response.audio:
                audio += response.audio
            elif response.type == "done":
                break

        path = os.path.join(CACHE_DIR, f"{phrase_name}.raw")
        with open(path, "wb") as f:
            f.write(audio)
        print(f"Cached '{phrase_name}': {len(audio)} bytes -> {path}")

from cartesia import Cartesia
import os

client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])

# build_cache.py の設定と一致させる必要があります
CACHE_DIR = "phrase-cache"
MODEL = "sonic-3.5"
VOICE_ID = "47c38ca4-5f35-497b-b1a3-415245fb35e1"
OUTPUT_FORMAT = {"container": "raw", "encoding": "pcm_s16le", "sample_rate": 24000}

CUSTOMER_NAME="Elon"

def load_cached_raw_file(name: str) -> bytes:
    with open(os.path.join(CACHE_DIR, f"{name}.raw"), "rb") as f:
        return f.read()


def call_llm(index: int, context: str = "") -> str:
    """本番では、会話コンテキストとともに推論用 LLM を呼び出します。
    デモではハードコードされた応答を返します。"""
    responses = [
        "Ok, sure. Let me pull up order number AKQ 4245 for you.",
        f"Ok! Good news {CUSTOMER_NAME}. Your order shipped yesterday and should arrive tomorrow.",
    ]
    return responses[index]


def call_cartesia_tts(ws, text: str) -> bytes:
    ctx = ws.context(
        model_id=MODEL,
        voice={"mode": "id", "id": VOICE_ID},
        output_format=OUTPUT_FORMAT,
        language="en",
    )
    ctx.push(text)
    ctx.no_more_inputs()

    audio = bytearray()
    for response in ctx.receive():
        if response.type == "chunk" and response.audio:
            audio += response.audio
        elif response.type == "done":
            break
    return bytes(audio)


# "cached" = ローカルキャッシュから読み込み（クレジット消費ゼロ）、"live" = Cartesia TTS で生成（クレジット消費あり）
interleaved_sequence: list[dict[str, str]] = [
    {"cached": "greeting"},
    {"live": call_llm(0)},
    {"cached": "hold"},
    {"live": call_llm(1)},
    {"cached": "closing"},
]

output = bytearray()

# キャッシュ済みフレーズとライブフレーズのシーケンスから音声を連結します。
with client.tts.websocket_connect() as ws:
    for item in interleaved_sequence:
        if "cached" in item:
            output += load_cached_raw_file(item["cached"])  # クレジット消費ゼロ
        else:
            output += call_cartesia_tts(ws, item["live"])  # クレジット消費あり

# 連結した PCM を試聴できるように WAV として書き出します。
import wave
os.makedirs("./output", exist_ok=True)
with wave.open("./output/spliced.wav", "wb") as w:
    w.setnchannels(1)
    w.setsampwidth(2)  # 16 ビット pcm_s16le
    w.setframerate(OUTPUT_FORMAT["sample_rate"])
    w.writeframes(output)

print(f"Wrote spliced.wav ({len(output)} bytes)")