03Conversation
Custom turn-taking
Add your own controllers and change the rules in src/turns.py.
config.cuelo picks from the built-in controllers. For anything else, use
src/turns.py. Cuelo generates it once and never overwrites it.
from cuelo.controllers.interface import Controller
from cuelo.policies.turn_taking import StandardTurnTaking, TurnTakingPolicy
def controllers(defaults: list[Controller]) -> list[Controller]:
"""The controllers for one call. `defaults` come from config.cuelo."""
return defaults
def policy() -> TurnTakingPolicy:
"""Decides what to do with controller signals."""
return StandardTurnTaking()Add a controller
A controller consumes audio and transcripts and returns signals. This one stops the agent when the caller says "stop":
from cuelo.controllers.interface import ControlEvent, Controller, Signal
class StopWord(Controller):
emits = frozenset({Signal.INTERRUPT})
def push_user(self, audio):
return []
def push_transcript(self, transcript):
if self.agent_speaking and "stop" in transcript.text.lower().split():
return [ControlEvent(Signal.INTERRUPT, t=transcript.end or 0.0)]
return []
def controllers(defaults):
return [*defaults, StopWord()]What to know:
- Synchronous and threaded. Controllers are plain sync code. The harness runs each one on its own worker thread and feeds it in order.
- Audio time.
tis seconds of caller audio consumed, not wall clock, so the same audio always produces the same signals and calls replay exactly. emitsdeclares the signals a controller produces. The harness fills gaps with the other controllers.- Duplex controllers set
duplex = Trueand also receive the agent's audio throughpush_agent(). Others can readself.agent_speaking.
The full interface is in the Python API. To package a new model as a controller, with pinned weights and a catalog entry, see Add a turn-taking model.
Change the policy
StandardTurnTaking takes three knobs:
| Argument | Default | Meaning |
|---|---|---|
hold_ms | 200 | wait this long after interrupt before stopping, in case it was a backchannel |
dedupe_s | 0.3 | turn_ends from several controllers within this window count as one |
reply_wait_s | 3.0 | no backchannels from deciding to respond until the reply finishes, or this long if it never starts |
def policy():
return StandardTurnTaking(hold_ms=100) # stop faster on interruptionsFor different rules, subclass TurnTakingPolicy. It receives every
ControlEvent in on_event(), gets tick(t) as audio time advances (for
timers), and returns Decisions: RESPOND, STOP, BACKCHANNEL or
CONTINUE.
from cuelo.policies.turn_taking import Action, Decision, StandardTurnTaking
class NoInterruptionsDuringDisclosure(StandardTurnTaking):
"""Ignore interruptions while a legal disclosure plays."""
disclosure = False
def on_event(self, event):
decisions = super().on_event(event)
if self.disclosure:
return [d for d in decisions if d.action != Action.STOP]
return decisionsEvery decision is recorded as a policy.decision event, so the dashboard's
events tab shows exactly why the agent did what it did.