echovoice

Sentence-level TTS audio cache for pipecat voice pipelines. In-memory only for now — audio captured on one call is served on every later call in the same process; no external service, no persistence.

Layout

echovoice/
  defaults.py   every SDK-wide tunable default, in one place
  common/       shared dataclasses (CachedAudio, AudioMeta) + logger shim
  cache/        keying (normalization + cache keys) and the in-memory store
  pipecat_tts/  pipeline integration — mixin.py (read), capture.py (write),
                pending.py (read→write handoff), bracket_filter.py (merges
                per-sentence Started/Stopped into turn-level brackets so the
                transport behaves like stock); needs pipecat-ai
  remote/       dormant echo-service HTTP path (client + config)

How it works

Two pieces, wired around the host's existing TTS service:

• Read — EnableTTSCache (echovoice.pipecat_tts.mixin): drop-in replacement for instantiating the provider class. Every sentence's run_tts first checks the shared in-memory store: a hit replays the cached PCM as TTSAudioRawFrames (zero provider traffic, barge-in aware), a miss synthesizes live and marks the sentence for capture.
• Write — TTSCacheCapture (echovoice.pipecat_tts.capture): a pass-through processor placed right after the TTS service in the pipeline. It buffers each miss-marked sentence's audio and stores the finished sentence in the shared store (echovoice.cache.store.AUDIO_STORE, a bounded LRU map — 500 entries / 200 MB by default). Interruptions discard partial buffers, so truncated audio is never stored.

Cache keys (echovoice.cache.keying) hash normalized text + provider + voice

• model + sample rate, so whitespace/casing never causes a false miss and two voices in one process can never serve each other's audio. Normalization is for key matching only — synthesis always uses the real text.

ElevenLabs specifics handled by the wrapper (validated against the multi-context websocket API):

• One provider context per sentence, closed right after its text is sent (close is graceful — the provider flushes the audio, then sends the final).
• A client-side gate caps simultaneous contexts (default 4 — ElevenLabs kills the socket at >5; tune via EnableTTSCache(cache_config={"max_concurrent_context": ..., "sentence_wait_timeout_s": ...}), all defaults live in echovoice/defaults.py). Slots free on the context's final, interruption, or websocket reconnect.
• Cache hits take no context and no gate slot; the replayed context is completed explicitly so the sentence serializer never stalls.

Install

pip install -e /path/to/echovoice             # core (no pipecat needed)
pip install -e "/path/to/echovoice[pipecat]"  # with the pipecat wrapper

With uv, in the host app's pyproject.toml:

dependencies = ["echovoice"]

[tool.uv.sources]
echovoice = { path = "../echovoice", editable = true }

Usage (fw-aisha style)

from echovoice.pipecat_tts import EnableTTSCache, TTSCacheCapture

# read side — instead of CustomElevenLabsTTSService(...):
tts = EnableTTSCache(CustomElevenLabsTTSService,
    url="wss://api.in.residency.elevenlabs.io",
    api_key=ELEVENLABS_API_KEY,
    settings=ElevenLabsTTSService.Settings(model=..., voice=...),
)

# write side + bracket filter — right after the TTS service in the pipeline:
pipeline = Pipeline([..., tts, TTSCacheCapture(), TTSTurnBracketFilter(),
                     transport.output(), ...])

No configuration needed. The wrapped service exposes get_and_reset_turn_stats() and log_cache_summary(call_sid=...) for call logs. Skip both pieces and the provider runs fully stock.

echovoice.remote holds the HTTP client for a future server-backed cache (echo-service); it is dormant — nothing on the active path uses it.

Tests

uv sync && uv run pytest