Voice-companion demos invite an obvious question: Which model makes the smartest or most charming character?
That is rarely the first production decision. A capable model still feels broken if your application sends it half a sentence, repeats a finalized transcript, or plays a response after the user has already started a new turn.
The practical tension is between responsiveness and certainty:
This tutorial builds a small Python utterance commit gate for a Tencent RTC Conversational AI application. It waits for a configurable quiet window, correlates every downstream operation, and makes recovery explicit. The result is not a smarter model. It is a voice pipeline that gives the model coherent turns.
Keep the media and AI responsibilities separate:
Microphone
│
▼
Tencent RTC media transport
│
▼
Speech recognition ── partial/final segments
│
▼
Utterance commit gate ← this tutorial
│
├── LLM request
├── output moderation
└── speech synthesis
│
▼
RTC audio playback
Tencent RTC's Conversational AI scenario supports real-time voice interaction with multiple LLM providers. Its LLM configuration documentation also describes connecting OpenAI-compatible models and carrying request identifiers for routing and observability:
The Python types below are deliberately application-owned. They are not Tencent RTC SDK API names. Your adapter translates actual speech, model, moderation, synthesis, and RTC callbacks into these events.
Our controller recognizes six states:
| State | Meaning | Permitted next step |
|---|---|---|
LISTENING |
Collecting partial or final transcript segments | Wait or begin settling |
SETTLING |
All known segments are final; quiet timer is running | Commit or accept another segment |
THINKING |
One LLM request owns the turn | Moderate its result or cancel it |
MODERATING |
The generated draft is being checked | Speak it or enter recovery |
SPEAKING |
Approved audio is being presented | Finish or be interrupted |
RECOVERING |
The current operation has an uncertain or failed outcome | Resume explicitly |
Three invariants matter more than the exact state names:
This example uses only the Python standard library and Python 3.11 or newer.
mkdir rtc-turn-commit
cd rtc-turn-commit
touch companion.py test_companion.py
Add the controller to 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
Add deterministic tests to test_companion.py:
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), [])
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)],
)
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.mode, Mode.LISTENING)
def test_partial_transcript_eventually_enters_recovery(self) -> None:
c = TurnCommitter(
settle_ms=100,
transcript_timeout_ms=1_000,
)
c.on_transcript(
Transcript(c.epoch, "a", 1, "unfinished", False),
now_ms=0,
)
self.assertEqual(
c.tick(1_000),
[Effect("status", text="transcript_timeout")],
)
self.assertEqual(c.mode, Mode.RECOVERING)
if __name__ == "__main__":
unittest.main()
Run the suite:
python -m unittest -v
You should see five passing tests. More importantly, the tests prove behavioral properties rather than relying on real-time sleeps:
Keep provider-specific code outside the controller. A simplified dispatcher might look like this:
async def apply_effect(effect, services, controller):
try:
if effect.kind == "generate":
draft = await services.llm.generate(
prompt=effect.text,
correlation_id=effect.request_id,
)
for next_effect in controller.on_llm_result(
effect.request_id,
draft,
):
await apply_effect(next_effect, services, controller)
elif effect.kind == "moderate":
result = await services.moderation.check(effect.text)
next_effects = controller.on_moderation_result(
effect.request_id,
allowed=result.allowed,
approved_text=result.approved_text,
)
for next_effect in next_effects:
await apply_effect(next_effect, services, controller)
elif effect.kind == "speak":
await services.speech.play(
text=effect.text,
correlation_id=effect.request_id,
)
elif effect.kind == "cancel_generation":
await services.llm.cancel(effect.request_id)
elif effect.kind == "stop_speech":
await services.speech.stop(effect.request_id)
elif effect.kind == "status":
services.ui.show_status(effect.text)
except Exception:
for recovery_effect in controller.on_stage_error(
effect.request_id,
effect.kind,
):
services.ui.show_status(recovery_effect.text)
services.llm, services.moderation, and services.speech are ports owned by your application. Map correlation_id to the request-identifier mechanism supported by your configured integration; do not assume the pseudocode argument name is an SDK field.
The live event flow is then:
on_transcript(). tick() from your application timer.on_user_speech_started() when your turn-detection policy confirms a new user turn.on_speech_finished() only after playback completion is confirmed.on_disconnect() before attempting to rejoin.resume() after the media and recognition path is ready again.
The Tencent RTC Social Entertainment solution includes AI virtual companions and character dialogue among its scenarios. The commit gate applies equally to a playful character, an assistant, or an AI host because it controls turns rather than personality.
Do not copy 350 ms into production and call it solved. It is an initial configuration value, not a universal latency target.
Choose it with a replay set containing your application's actual conversational patterns:
| Observation | Likely adjustment | Trade-off |
|---|---|---|
| Clause-ending s create separate requests | Increase the settling window | Slower response start |
| Complete commands feel delayed | Decrease the settling window | More premature commits |
| Recognition emits several final segments per sentence | Keep segment aggregation enabled | Requires stable ordering and IDs |
| Very long partials never finalize | Review recognition behavior and timeout policy | Recovery may ask the user to repeat |
Record at least these timestamps per request identifier:
first_transcript_at
last_final_segment_at
utterance_committed_at
llm_started_at
llm_finished_at
moderation_finished_at
speech_playback_started_at
speech_playback_finished_at
This separates distinct delays. A single “AI latency” number cannot tell you whether time was spent waiting for a turn boundary, generating text, checking the output, synthesizing speech, or starting playback.
The transcript timeout enters RECOVERING and emits transcript_timeout. Show a visible and, when possible, accessible prompt such as “I didn't catch the end of that.” Do not submit the partial text as though it were confirmed.
Provider cancellation may fail or arrive too late. The request-ID check is the authoritative defense: a callback without current ownership produces no speech.
Do not silently bypass the stage. Move to recovery, keep the draft out of synthesis, and let the user retry. For a social companion, moderation and user-visible control belong in the runtime design, not only in the prompt.
The generated answer may be valid while its delivery outcome is unknown. Show a failure state and return to listening only through an explicit recovery action. Automatically generating a new answer could duplicate content.
Increment the epoch before rejoining. Old ASR callbacks are then structurally incapable of forming a new request. Start a fresh turn rather than pretending an interrupted utterance continued seamlessly.
This in-memory example intentionally does not resume an in-flight turn. A restarted service should create a new epoch and expose the interruption. If you later persist state, store the epoch, active request identifier, mode, and committed transcript together; restoring only the transcript can replay a turn without restoring its ownership rules.
Run these drills with real audio and your configured providers:
A model can demonstrably generate engaging character dialogue. That does not demonstrate that a live companion can identify a completed turn, recover from a partition, or respect an interruption.
The human decision underneath the model comparison is where to spend the next engineering cycle. Before changing models or expanding the character prompt, inspect ten imperfect voice sessions. If you find split sentences, duplicate requests, or stale replies, add a commit gate and correlation logging first. Model evaluation becomes more meaningful once every candidate receives the same coherent input turns.
Disclosure: This article was produced in connection with Tencent RTC. Official Tencent RTC documentation was used as the implementation reference.