local_voice

local_voice

Fully local realtime voice: Whisper → Ollama → Kokoro, nothing leaves.

The other two providers stream the raw microphone to Google or OpenAI; this one keeps the whole loop on the machine. It is honest about what that buys and what it costs: turn-based with barge-in rather than full duplex, first spoken word ~1.5–2.5 s after the user's last syllable (measured: STT ~1 s for a 4 s French utterance, qwen3.5:9b first token 0.47 s warm, Kokoro RTF 0.28) — against ~0.5 s for Gemini Live. In exchange: no key, no account, and the factory's local-only guard can finally let voice through instead of refusing it wholesale.

Structure: send_audio only buffers and detects turn boundaries; the response pipeline (transcribe → stream tokens → speak sentence by sentence) runs as a cancellable task, because barge-in is nothing more than cancelling it. The three stages are injectable callables so the turn logic is testable without a microphone, a model server or a vocoder.

Classes

LocalVoiceSession

LocalVoiceSession(
    *,
    model: str = "",
    voice: str = "",
    instructions: str = "",
    language: str = "",
    api_key: Optional[str] = None,
    enable_tools: bool = True,
    max_tool_steps: int = 12,
    allowed_tools: Optional[Sequence[str]] = None,
    stt: Optional[Callable[[bytes], str]] = None,
    llm: Optional[Callable[[List[dict]], Any]] = None,
    tts: Optional[Callable[[str], bytes]] = None,
    tool_executor: Optional[
        Callable[[str, dict], dict]
    ] = None,
    sur_echange: Optional[
        Callable[[str, str], None]
    ] = None,
)

Bases: RealtimeVoiceSession

Turn-based local voice with barge-in, behind the realtime contract.

Source code in src/diapason/speech/realtime/local_voice.py

def __init__(
    self,
    *,
    model: str = "",
    voice: str = "",
    instructions: str = "",
    language: str = "",
    api_key: Optional[str] = None,  # accepted, unused: nothing to unlock
    enable_tools: bool = True,  # same allow-listed tools as Gemini
    max_tool_steps: int = 12,
    allowed_tools: Optional[Sequence[str]] = None,
    stt: Optional[Callable[[bytes], str]] = None,
    llm: Optional[Callable[[List[dict]], Any]] = None,
    tts: Optional[Callable[[str], bytes]] = None,
    tool_executor: Optional[Callable[[str, dict], dict]] = None,
    sur_echange: Optional[Callable[[str, str], None]] = None,
) -> None:
    self._model = model or DEFAULT_MODEL
    self._voice = voice or DEFAULT_VOICE
    self._instructions = instructions
    self._language = language
    self._stt = stt
    self._llm = llm
    self._tts = tts
    self._enable_tools = bool(enable_tools)
    self._allowed_tools = list(allowed_tools) if allowed_tools else None
    from diapason.speech.realtime.tools import VoiceToolBudget

    self._budget = VoiceToolBudget(max_tool_steps)
    self._tool_executor = tool_executor
    self._queue: asyncio.Queue[Optional[SessionEvent]] = asyncio.Queue()
    self._buffer = bytearray()
    self._preroll = bytearray()
    self._speech_samples = 0
    self._silence_samples = 0
    self._in_speech = False
    self._respond_task: Optional[asyncio.Task[None]] = None
    self._warm_task: Optional[asyncio.Task[None]] = None
    # L'historique des échanges vocaux (traces.db, agent='voice') — la
    # mémoire nocturne ne relit que ce qui est écrit quelque part. Résolu
    # paresseusement ; ici et pas dans connect() : les bancs d'essai des
    # tests contournent connect() (leçon du 23 août 2026).
    self._magasin_traces_obj: Any = None
    self._magasin_traces_resolu = False
    # Le raccord vers la mémoire vivante (24 août 2026) : appelé à
    # chaque échange abouti, en plus de la trace. Sans lui, un fait
    # confié à l'oral attendait la consolidation du lendemain 3h30.
    self._sur_echange = sur_echange
    # Engagé dès la construction : le premier tour d'une session vient de
    # quelqu'un qui a cliqué « Démarrer » — il s'adresse à nous.
    self._engagee_jusqua = time.monotonic() + ADDRESS_WINDOW_S
    self._voice_lock: Optional[bool] = None
    # (buffered byte count, transcription task) — valid only while the
    # buffer has not grown past the snapshot it was taken from.
    self._speculative: Optional[tuple[int, asyncio.Task[str]]] = None
    # Réponse spéculative : la GÉNÉRATION lancée pendant le silence de
    # fin de tour, dès que la transcription complète est connue. Sûre par
    # construction : les outils ne s'exécutent et la voix ne part que
    # dans la boucle de drainage de _respond_to_text — une file qu'on
    # remplit sans la drainer ne peut ni agir ni parler.
    self._spec_llm: Optional[_SpecTurn] = None
    # Première phrase déjà synthétisée par la spéculation : (texte après
    # speakable, tâche de synthèse). Consommé une seule fois, apparié sur
    # le texte exact — un raté d'appariement coûte une synthèse normale,
    # jamais un mauvais audio.
    self._spec_audio: Optional[tuple[str, "asyncio.Task[bytes]"]] = None
    # Les postes de latence du tour en cours, remplis par le producteur
    # LLM et les jalons du tour ; snapshotés dans traces.db à la fin.
    self._mesures: dict[str, Any] = {}
    self._response_started: Optional[float] = None
    self._first_audio_logged = True
    self._ack_suivant = 0
    # Transcription affichée PENDANT qu'on parle. Distincte du
    # spéculatif, qui sert à répondre plus tôt et ne s'exécute que dans
    # le silence : celle-ci tourne au milieu de la phrase, et son seul
    # rôle est que l'écran suive la voix.
    self._partial: Optional[asyncio.Task[str]] = None
    self._partial_mark = 0
    self._partial_text = ""
    # Longueur du tampon à la dernière trame PARLÉE : la couverture d'un
    # partiel se juge contre la fin de la parole, pas celle du tampon,
    # qui continue de grossir avec le silence.
    self._speech_end_mark = 0
    # When, on OUR clock, the audio already shipped to the client will
    # finish playing. Synthesis outruns playback, so the respond task is
    # usually long done while the user is still hearing the answer — this
    # clock is what lets speech interrupt a playback with no task left to
    # cancel. That gap was exactly the reported "il ne s'arrête pas".
    self._speaking_until = 0.0
    self._history: List[dict] = []
    self._closed = False

Functions:

classify_endpoint

classify_endpoint(text: str) -> str

« complete », « hesitation » ou « neutral » pour une fin de tour.

Conservateur par construction : « complete » exige une ponctuation terminale — c'est le seul verdict qui RACCOURCIT l'attente, donc le seul qui puisse couper quelqu'un. « hesitation » ne fait qu'attendre plus, l'erreur y est bon marché. Tout le reste garde le délai normal.

Source code in src/diapason/speech/realtime/local_voice.py

def classify_endpoint(text: str) -> str:
    """« complete », « hesitation » ou « neutral » pour une fin de tour.

    Conservateur par construction : « complete » exige une ponctuation
    terminale — c'est le seul verdict qui RACCOURCIT l'attente, donc le seul
    qui puisse couper quelqu'un. « hesitation » ne fait qu'attendre plus,
    l'erreur y est bon marché. Tout le reste garde le délai normal.
    """
    t = (text or "").strip().rstrip("»\"' ")
    if not t:
        return "neutral"
    # « … » est l'orthographe même du trailing-off : Whisper l'émet quand la
    # voix retombe sans conclure. C'est le contraire d'une phrase finie.
    if t.endswith(("...", "…")):
        return "hesitation"
    if t[-1] in ".!?":
        return "complete"
    if t[-1] in ",;:":
        return "hesitation"
    dernier = re.split(r"[\s']+", t.lower())[-1]
    if dernier in _TRAILING_INCOMPLETE:
        return "hesitation"
    return "neutral"

speakable

speakable(text: str) -> str

Strip what a voice cannot say; collapse the leftover whitespace.

Returns "" when nothing pronounceable is left — an emoji-only chunk leaves its punctuation behind ("👍👍." → "."), and a vocoder handed a bare period says "point" out loud.

Source code in src/diapason/speech/realtime/local_voice.py

def speakable(text: str) -> str:
    """Strip what a voice cannot say; collapse the leftover whitespace.

    Returns "" when nothing pronounceable is left — an emoji-only chunk
    leaves its punctuation behind ("👍👍." → "."), and a vocoder handed a
    bare period says "point" out loud.
    """
    cleaned = _UNSPEAKABLE.sub("", text or "")
    # 21/09/2026 : « Mark Carney [2] » se prononçait « Mark Carney deux » ;
    # un numéro de source ou une adresse web ne se disent pas.
    cleaned = _CITATION_ECRITE.sub("", cleaned)
    cleaned = re.sub(r"^[\s\-•]+", "", cleaned)
    cleaned = re.sub(r"[ \t]{2,}", " ", cleaned).strip()
    if not re.search(r"[\w]", cleaned, re.UNICODE):
        return ""
    return cleaned

french_now

french_now(now=None) -> str

The machine's local date and time, spelled out in French.

Hand-rolled rather than strftime with a locale: setlocale is process-wide state and this runs inside a server thread pool. Injected into the system prompt at every turn — a model has no clock, and "quelle heure est-il" answered with "je ne peux pas lire l'heure" was a reported failure, on a machine that obviously knows.

Source code in src/diapason/speech/realtime/local_voice.py

def french_now(now=None) -> str:
    """The machine's local date and time, spelled out in French.

    Hand-rolled rather than strftime with a locale: setlocale is process-wide
    state and this runs inside a server thread pool. Injected into the system
    prompt at every turn — a model has no clock, and "quelle heure est-il"
    answered with "je ne peux pas lire l'heure" was a reported failure, on a
    machine that obviously knows.
    """
    import datetime

    if now is None:
        now = datetime.datetime.now()
    day = _FRENCH_DAYS[now.weekday()]
    month = _FRENCH_MONTHS[now.month - 1]
    return f"{day} {now.day} {month} {now.year}, {now.hour} h {now.minute:02d}"

french_today

french_today(now=None) -> str

La date seule, sans l'heure — stable toute la journée.

L'horloge à la minute près, recollée au prompt à chaque appel, invalidait le cache de préfixe d'Ollama dès que la minute changeait : les six mille jetons de prompt et de schémas d'outils étaient relus en entier, plusieurs secondes par tour. La voix dispose maintenant de current_time : quand l'heure compte, le modèle la LIT — c'est plus juste qu'une heure figée au début du tour, et le préfixe, lui, ne bouge plus qu'à minuit.

Source code in src/diapason/speech/realtime/local_voice.py

def french_today(now=None) -> str:
    """La date seule, sans l'heure — stable toute la journée.

    L'horloge à la minute près, recollée au prompt à chaque appel, invalidait
    le cache de préfixe d'Ollama dès que la minute changeait : les six mille
    jetons de prompt et de schémas d'outils étaient relus en entier, plusieurs
    secondes par tour. La voix dispose maintenant de ``current_time`` : quand
    l'heure compte, le modèle la LIT — c'est plus juste qu'une heure figée au
    début du tour, et le préfixe, lui, ne bouge plus qu'à minuit.
    """
    import datetime

    if now is None:
        now = datetime.datetime.now()
    day = _FRENCH_DAYS[now.weekday()]
    month = _FRENCH_MONTHS[now.month - 1]
    return f"{day} {now.day} {month} {now.year}"

ollama_reachable

ollama_reachable(timeout_s: float = 1.5) -> bool

True when the local model server answers. Never raises.

Source code in src/diapason/speech/realtime/local_voice.py

def ollama_reachable(timeout_s: float = 1.5) -> bool:
    """True when the local model server answers. Never raises."""
    try:
        from diapason.core.local_mode import assert_may_leave

        assert_may_leave("the Ollama health request", destination=_ollama_base())
        with urllib.request.urlopen(
            f"{_ollama_base()}/api/tags", timeout=timeout_s
        ) as response:
            return response.status == 200
    except Exception:  # noqa: BLE001 - unreachable is a normal state
        return False

local_voice_readiness

local_voice_readiness(
    timeout_s: float = 1.5,
) -> tuple[bool, str]

Return whether every local voice runtime component is available.

Keep this check cheap: model construction belongs to connect(), but a missing optional extra or system phonemizer must disable Start instead of letting the WebSocket claim readiness and fail a few seconds later.

Source code in src/diapason/speech/realtime/local_voice.py

def local_voice_readiness(timeout_s: float = 1.5) -> tuple[bool, str]:
    """Return whether every local voice runtime component is available.

    Keep this check cheap: model construction belongs to ``connect()``, but a
    missing optional extra or system phonemizer must disable Start instead of
    letting the WebSocket claim readiness and fail a few seconds later.
    """
    required_modules = ("faster_whisper", "kokoro", "soundfile")
    if any(importlib.util.find_spec(name) is None for name in required_modules):
        return False, "missing-dependencies"
    # The voice-local extra installs espeakng-loader, which supplies a bundled
    # phonemizer even when launchd's minimal PATH cannot see Homebrew's binary.
    has_phonemizer = (
        shutil.which("espeak-ng") is not None
        or shutil.which("espeak") is not None
        or importlib.util.find_spec("espeakng_loader") is not None
    )
    if not has_phonemizer:
        return False, "missing-phonemizer"
    if not ollama_reachable(timeout_s=timeout_s):
        return False, "ollama-unavailable"
    return True, "ready"

polish_transcript

polish_transcript(text: str) -> str

Apply the user's dictation dictionary to a voice transcript.

The dictation path earns its accuracy partly AFTER Whisper: the personal dictionary fixes the words the recognizer keeps getting wrong ("App Store", proper nouns). Voice transcripts deserve the same corrections — same user, same vocabulary, same mistakes. bump_usage=False: voice hits must not skew the dictation dictionary's learning statistics.

Source code in src/diapason/speech/realtime/local_voice.py

def polish_transcript(text: str) -> str:
    """Apply the user's dictation dictionary to a voice transcript.

    The dictation path earns its accuracy partly AFTER Whisper: the personal
    dictionary fixes the words the recognizer keeps getting wrong ("App
    Store", proper nouns). Voice transcripts deserve the same corrections —
    same user, same vocabulary, same mistakes. bump_usage=False: voice hits
    must not skew the dictation dictionary's learning statistics.
    """
    cleaned = (text or "").strip()
    if not cleaned:
        return cleaned
    try:
        from diapason.speech.dictate_polish import reparer_homophones_de_commande

        cleaned = reparer_homophones_de_commande(cleaned)
    except Exception:  # noqa: BLE001 - la réparation est un bonus, jamais une porte
        pass
    try:
        from diapason.speech.dictation_dictionary import apply_dictionary

        return apply_dictionary(cleaned, bump_usage=False)
    except Exception:  # noqa: BLE001 - the dictionary is a bonus, never a gate
        return cleaned

mentions_assistant_name

mentions_assistant_name(text: str) -> bool

Vrai si un mot — ou DEUX mots adjacents recollés — ressemble au nom.

La transcription coupe le nom en deux : « Dia pasons, quelle heure… », constaté en session réelle. Un seul mot ne suffit donc pas ; les paires adjacentes se recollent avant la comparaison.

Source code in src/diapason/speech/realtime/local_voice.py

def mentions_assistant_name(text: str) -> bool:
    """Vrai si un mot — ou DEUX mots adjacents recollés — ressemble au nom.

    La transcription coupe le nom en deux : « Dia pasons, quelle heure… »,
    constaté en session réelle. Un seul mot ne suffit donc pas ; les paires
    adjacentes se recollent avant la comparaison.
    """
    import difflib

    mots = _NOM_RE.findall(str(text or "").casefold())
    candidats = [m for m in mots if len(m) >= 6]
    candidats += [a + b for a, b in zip(mots, mots[1:]) if len(a + b) >= 6]
    return any(
        difflib.SequenceMatcher(None, c, "diapason").ratio() >= 0.75 for c in candidats
    )

strip_assistant_name

strip_assistant_name(text: str) -> str

Retire l'appel initial — « Diapason, ouvre… » → « ouvre… ».

Source code in src/diapason/speech/realtime/local_voice.py

def strip_assistant_name(text: str) -> str:
    """Retire l'appel initial — « Diapason, ouvre… » → « ouvre… »."""
    return (
        re.sub(
            # « Diapason, », « diapasant » — et « Dia pasons, », le nom coupé en
            # deux par la transcription.
            r"^\W*(?:[a-zà-ÿ]{2,4}\s+)?[a-zà-ÿ]*(?:diapa|pason|pazon)\w*[\s,.:!?]*",
            "",
            str(text or ""),
            flags=re.IGNORECASE,
        ).strip()
        or str(text or "").strip()
    )