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:
- Commit too early and OpenAI receives incomplete thoughts.
- Wait too long and the companion feels unresponsive.
- Retry carelessly and the character answers twice.
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.
What we are building
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.
The state contract
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:
- There is at most one active request identifier.
- A callback is accepted only if it matches the active request and state.
- Reconnection creates a new session epoch, so old transcript callbacks cannot become new turns.
Create the project
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
Reproduce the races before connecting a microphone
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), [])
# 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.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:
- Two final segments can still represent one utterance.
- Duplicate recognition callbacks do not create duplicate model calls.
- Best-effort cancellation is backed by stale-result rejection.
- Reconnection invalidates callbacks from the previous media session.
- A partial transcript cannot leave the application waiting forever.
Connect the effects to your live application
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:
# Classify the actual stage in production rather than using
# one broad exception handler.
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:
- Join the RTC conversation and start the chosen recognition path.
- Forward every transcript revision to
on_transcript(). - Call
tick()from your application timer. - Execute emitted effects through provider adapters.
- Call
on_user_speech_started()when your turn-detection policy confirms a new user turn. - Call
on_speech_finished()only after playback completion is confirmed. - On transport loss, call
on_disconnect()before attempting to rejoin. - Call
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.
Choosing the quiet window
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 pauses 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.
Failure behavior users can understand
Recognition never marks the segment final
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.
The LLM completes after interruption
Provider cancellation may fail or arrive too late. The request-ID check is the authoritative defense: a callback without current ownership produces no speech.
Moderation is unavailable
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.
Speech synthesis fails after text was approved
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.
The network reconnects while callbacks are queued
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.
The process restarts
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.
Verification checklist for staging
Run these drills with real audio and your configured providers:
- [ ] Pause between two clauses and confirm only one LLM request is logged.
- [ ] Deliver the same final transcript revision twice and confirm one commit.
- [ ] Interrupt during generation and confirm late output never reaches synthesis.
- [ ] Interrupt during playback and confirm old completion callbacks do not reset the new turn.
- [ ] Disconnect after a final segment but before the quiet window expires.
- [ ] Reconnect and inject a callback carrying the previous epoch.
- [ ] Make moderation time out and confirm the draft is not spoken.
- [ ] Make synthesis fail and confirm the UI does not claim the answer was delivered.
- [ ] Confirm logs carry one request identifier through LLM, moderation, and synthesis.
- [ ] Confirm microphone, transcript, and generated-content handling match your consent and privacy policy.
The useful decision behind the “better AI pet” discussion
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.
Top comments (0)