Stop Sending Half-Sentences to OpenAI in a Tencent RTC Voice Companion

Microphone │ ▼ Tencent RTC media transport │ ▼ Speech recognition ── partial/final segments │ ▼ Utterance commit gate ← this tutorial │ ├── LLM request ├── output moderation └── speech synthesis │ ▼ RTC audio playback 

mkdir rtc-turn-commit cd rtc-turn-commit touch companion.py test_companion.py 

from __future__ import annotations from dataclasses import dataclass from enum import Enum, auto from typing import Optional class Mode(Enum): LISTENING = auto() SETTLING = auto() THINKING = auto() MODERATING = auto() SPEAKING = auto() RECOVERING = auto() STOPPED = auto() @dataclass(frozen=True) class Transcript: epoch: int segment_id: str revision: int text: str final: bool @dataclass class StoredSegment: revision: int text: str final: bool order: int @dataclass(frozen=True) class Effect: kind: str request_id: Optional[str] = None text: Optional[str] = None class TurnCommitter: def __init__( self, settle_ms: int = 350, transcript_timeout_ms: int = 8_000, ) -> None: if settle_ms < 0: raise ValueError("settle_ms must be non-negative") if transcript_timeout_ms <= settle_ms: raise ValueError("transcript timeout must exceed settle time") self.settle_ms = settle_ms self.transcript_timeout_ms = transcript_timeout_ms self.mode = Mode.LISTENING self.epoch = 1 self._segments: dict[str, StoredSegment] = {} self._consumed_segment_ids: set[str] = set() self._next_order = 0 self._first_segment_at: Optional[int] = None self._settle_deadline: Optional[int] = None self._turn_number = 0 self._active_request: Optional[str] = None @property def active_request(self) -> Optional[str]: return self._active_request def on_transcript(self, event: Transcript, now_ms: int) -> list[Effect]: if self.mode not in (Mode.LISTENING, Mode.SETTLING): return [] if event.epoch != self.epoch: return [] if event.segment_id in self._consumed_segment_ids: return [] previous = self._segments.get(event.segment_id) if previous and event.revision <= previous.revision: return [] if previous is None: order = self._next_order self._next_order += 1 else: order = previous.order self._segments[event.segment_id] = StoredSegment( revision=event.revision, text=event.text.strip(), final=event.final, order=order, ) if self._first_segment_at is None: self._first_segment_at = now_ms if self._segments and all(s.final for s in self._segments.values()): self.mode = Mode.SETTLING self._settle_deadline = now_ms + self.settle_ms else: self.mode = Mode.LISTENING self._settle_deadline = None return [] def tick(self, now_ms: int) -> list[Effect]: if ( self._first_segment_at is not None and now_ms - self._first_segment_at >= self.transcript_timeout_ms and self.mode in (Mode.LISTENING, Mode.SETTLING) ): self._clear_transcript() self.mode = Mode.RECOVERING return [Effect("status", text="transcript_timeout")] if ( self.mode != Mode.SETTLING or self._settle_deadline is None or now_ms < self._settle_deadline ): return [] ordered = sorted(self._segments.items(), key=lambda item: item[1].order) text = " ".join(segment.text for _, segment in ordered if segment.text) text = " ".join(text.split()) for segment_id, _ in ordered: self._consumed_segment_ids.add(segment_id) self._clear_transcript() if not text: self.mode = Mode.LISTENING return [] self._turn_number += 1 request_id = f"session-{self.epoch}-turn-{self._turn_number}" self._active_request = request_id self.mode = Mode.THINKING return [Effect("generate", request_id=request_id, text=text)] def on_llm_result(self, request_id: str, draft: str) -> list[Effect]: if self.mode != Mode.THINKING or request_id != self._active_request: return [] self.mode = Mode.MODERATING return [Effect("moderate", request_id=request_id, text=draft)] def on_moderation_result( self, request_id: str, allowed: bool, approved_text: str = "", ) -> list[Effect]: if self.mode != Mode.MODERATING or request_id != self._active_request: return [] if not allowed: self._active_request = None self.mode = Mode.RECOVERING return [Effect("status", text="reply_blocked")] self.mode = Mode.SPEAKING return [ Effect("speak", request_id=request_id, text=approved_text) ] def on_user_speech_started(self) -> list[Effect]: effects: list[Effect] = [] if self.mode in (Mode.THINKING, Mode.MODERATING): effects.append(Effect("cancel_generation", self._active_request)) elif self.mode == Mode.SPEAKING: effects.append(Effect("stop_speech", self._active_request)) if self.mode in (Mode.THINKING, Mode.MODERATING, Mode.SPEAKING): self._active_request = None self.mode = Mode.LISTENING return effects def on_speech_finished(self, request_id: str) -> list[Effect]: if self.mode != Mode.SPEAKING or request_id != self._active_request: return [] self._active_request = None self.mode = Mode.LISTENING return [] def on_stage_error(self, request_id: str, stage: str) -> list[Effect]: if request_id != self._active_request: return [] self._active_request = None self.mode = Mode.RECOVERING return [Effect("status", text=f"{stage}_failed")] def on_disconnect(self) -> list[Effect]: effects: list[Effect] = [] if self.mode in (Mode.THINKING, Mode.MODERATING): effects.append(Effect("cancel_generation", self._active_request)) elif self.mode == Mode.SPEAKING: effects.append(Effect("stop_speech", self._active_request)) self.epoch += 1 self._active_request = None self._consumed_segment_ids.clear() self._clear_transcript() self.mode = Mode.RECOVERING effects.append(Effect("status", text="connection_lost")) return effects def resume(self) -> None: if self.mode == Mode.RECOVERING: self.mode = Mode.LISTENING def stop(self) -> None: self._active_request = None self._clear_transcript() self.mode = Mode.STOPPED def _clear_transcript(self) -> None: self._segments.clear() self._first_segment_at = None self._settle_deadline = None 

import unittest from companion import Effect, Mode, Transcript, TurnCommitter class TurnCommitterTests(unittest.TestCase): def test_two_final_segments_become_one_model_request(self) -> None: c = TurnCommitter(settle_ms=300) c.on_transcript( Transcript(c.epoch, "a", 1, "Could you explain", True), now_ms=0, ) self.assertEqual(c.tick(299), []) # A second final segment restarts the settling window.  c.on_transcript( Transcript(c.epoch, "b", 1, "Python generators?", True), now_ms=250, ) self.assertEqual(c.tick(549), []) self.assertEqual( c.tick(550), [ Effect( "generate", "session-1-turn-1", "Could you explain Python generators?", ) ], ) self.assertEqual(c.mode, Mode.THINKING) def test_duplicate_revision_cannot_create_a_second_turn(self) -> None: c = TurnCommitter(settle_ms=100) event = Transcript(c.epoch, "segment-1", 2, "Hello", True) c.on_transcript(event, now_ms=0) first = c.tick(100) self.assertEqual(len(first), 1) c.on_user_speech_started() c.on_transcript(event, now_ms=200) self.assertEqual(c.tick(500), []) def test_interruption_invalidates_late_model_output(self) -> None: c = TurnCommitter(settle_ms=100) c.on_transcript( Transcript(c.epoch, "a", 1, "Tell me a story", True), now_ms=0, ) request = c.tick(100)[0].request_id self.assertEqual( c.on_user_speech_started(), [Effect("cancel_generation", request)], ) # Cancellation is best-effort. The old callback may still arrive.  self.assertEqual(c.on_llm_result(request, "Once upon a time"), []) self.assertEqual(c.mode, Mode.LISTENING) def test_reconnect_rejects_old_transcript_callbacks(self) -> None: c = TurnCommitter(settle_ms=100) old_epoch = c.epoch c.on_disconnect() c.resume() c.on_transcript( Transcript(old_epoch, "late", 1, "Old audio", True), now_ms=0, ) self.assertEqual(c.tick(500), []) self.assertEqual(c.modeSource：https://dev.to/susiewang/stop-sending-half-sentences-to-openai-in-a-tencent-rtc-voice-companion-3ni6
My favorites