Skip to content

fix(amd): start SIP detection timeout after answer #6580

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chenghao-mou wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(amd): start SIP detection timeout after answer #6580

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 32 additions & 18 deletions examples/telephony/amd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import asyncio
import logging
import os

Expand Down Expand Up @@ -49,6 +50,16 @@ async def entrypoint(ctx: JobContext):
room=ctx.room,
)

async def hangup():
await ctx.api.room.delete_room(
api.DeleteRoomRequest(
room=ctx.room.name,
)
)

# Register before dialing so failed calls also delete the room.
ctx.add_shutdown_callback(hangup)

phone_number = os.getenv("SIP_PHONE_NUMBER")
participant_identity = os.getenv("SIP_PARTICIPANT_IDENTITY")
outbound_trunk_id = os.getenv("SIP_OUTBOUND_TRUNK_ID")
Expand All @@ -69,16 +80,28 @@ async def entrypoint(ctx: JobContext):
# start running amd before the SIP participant joins to avoid audio loss
if phone_number and outbound_trunk_id and participant_identity:
logger.info(f"creating SIP participant for {participant_identity}")
await ctx.api.sip.create_sip_participant(
api.CreateSIPParticipantRequest(
room_name=ctx.room.name,
sip_trunk_id=outbound_trunk_id,
sip_call_to=phone_number,
participant_identity=participant_identity,
wait_until_answered=True,
# The API timeout must outlast the ring window; AMD's timeout starts after answer.
try:
await ctx.api.sip.create_sip_participant(
api.CreateSIPParticipantRequest(
room_name=ctx.room.name,
sip_trunk_id=outbound_trunk_id,
sip_call_to=phone_number,
participant_identity=participant_identity,
wait_until_answered=True,
),
timeout=45,
)
)
participant = await ctx.wait_for_participant(identity=participant_identity)
except (api.SipCallError, asyncio.TimeoutError) as e:
logger.info(f"call was not answered: {e}")
ctx.shutdown("call not answered")
return
# The call may end just before wait_until_answered returns.
participant = ctx.room.remote_participants.get(participant_identity)
if participant is None:
logger.info("SIP participant missing, ending")
ctx.shutdown("participant missing")
return
logger.info(
"participant joined",
extra={
Expand Down Expand Up @@ -127,15 +150,6 @@ async def entrypoint(ctx: JobContext):

ctx.shutdown("mailbox unavailable")

async def hangup():
await ctx.api.room.delete_room(
api.DeleteRoomRequest(
room=ctx.room.name,
)
)

ctx.add_shutdown_callback(hangup)


if __name__ == "__main__":
cli.run_app(server)
16 changes: 16 additions & 0 deletions livekit-agents/livekit/agents/utils/participant.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,10 @@ async def wait_for_track_publication(

When `include_local` is True, tracks published by the local participant are also considered;
local publications resolve on publish and ignore ``wait_for_subscription``.

When ``identity`` is set, raises :class:`RuntimeError` if that participant
disconnects before a matching publication is found, so callers don't wait
on a participant that is gone.
"""
if not room.isconnected():
raise RuntimeError("room is not connected")
Expand Down Expand Up @@ -295,6 +299,14 @@ def _on_local_track_published(
if (identity is None or local.identity == identity) and kind_match(publication.kind):
fut.set_result(publication)

def _on_participant_disconnected(p: rtc.RemoteParticipant) -> None:
if p.identity == identity and not fut.done():
fut.set_exception(
RuntimeError(
f"participant {identity!r} disconnected while waiting for track publication"
)
)

def _on_connection_state_changed(state: int) -> None:
if state == rtc.ConnectionState.CONN_DISCONNECTED and not fut.done():
fut.set_exception(RuntimeError("room disconnected while waiting for track publication"))
Expand All @@ -305,6 +317,8 @@ def _on_connection_state_changed(state: int) -> None:
room.on("track_published", _on_track_published)
if include_local:
room.on("local_track_published", _on_local_track_published)
if identity is not None:
room.on("participant_disconnected", _on_participant_disconnected)

room.on("connection_state_changed", _on_connection_state_changed)

Expand Down Expand Up @@ -332,4 +346,6 @@ def _on_connection_state_changed(state: int) -> None:
room.off("track_published", _on_track_published)
if include_local:
room.off("local_track_published", _on_local_track_published)
if identity is not None:
room.off("participant_disconnected", _on_participant_disconnected)
room.off("connection_state_changed", _on_connection_state_changed)
25 changes: 8 additions & 17 deletions livekit-agents/livekit/agents/voice/amd/classifier.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,7 @@ def start_detection_timer(self) -> None:
self._detection_timeout_timer = asyncio.get_running_loop().call_later(
self._timeout,
functools.partial(
self._on_timeout,
self.settle,
category=AMDCategory.UNCERTAIN,
reason="detection_timeout",
),
Expand All @@ -178,7 +178,7 @@ def start_listening(self) -> None:
self._no_speech_timer = asyncio.get_running_loop().call_later(
self._no_speech_threshold,
functools.partial(
self._on_timeout,
self.settle,
category=AMDCategory.UNCERTAIN,
reason="no_speech_timeout",
),
Expand Down Expand Up @@ -222,7 +222,7 @@ def on_user_speech_ended(self, silence_duration: float) -> None:
self._silence_timer = asyncio.get_running_loop().call_later(
max(0, self._human_silence_threshold - silence_duration),
functools.partial(
self._on_timeout,
self.settle,
category=AMDCategory.HUMAN,
reason="short_greeting",
speech_duration=speech_duration,
Expand Down Expand Up @@ -342,25 +342,16 @@ def _on_silence_reached(self) -> None:
self._try_emit_result()

@log_exceptions(logger=logger)
def _on_timeout(
def settle(
self,
category: AMDCategory,
reason: str,
speech_duration: float | None = None,
) -> None:
"""A timeout (detection budget, no-speech, short greeting) fired.

Commit a fallback verdict if none exists, then try to emit. This only
decides *what* the verdict is; ``_can_emit`` decides *when* it is
released. End-of-turn is forced here only when there is nothing left to
wait for: no speech was heard, or we are not waiting for the greeting to
finish. When ``wait_until_finished`` is set and speech was heard, the
fallback is still committed but its release stays gated on end-of-turn
(the real signal or the backstop timer), so we don't cut the greeting
short with an ``uncertain`` result.

Not gated by ``_listening_guard``: detection_timeout must still fire
when the call never reaches listening (e.g. sip never answered).
"""Commit a fallback verdict and attempt emission.

This can run before listening begins. After speech, ``wait_until_finished``
keeps emission gated on end-of-turn so a fallback cannot cut the greeting short.
"""
if self._closed:
return
Expand Down
61 changes: 32 additions & 29 deletions livekit-agents/livekit/agents/voice/amd/detector.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,10 +101,10 @@ class AMD(EventEmitter[Literal["amd_prediction"]]):
- ``machine-unavailable``: the mailbox is full or not set up; leaving a message is not possible.
- ``uncertain``: the transcript is ambiguous and could not be classified.

AMD should be started before the SIP participant is created so no audio
is missed. The overall detection-timeout budget starts when the
participant's audio track is subscribed (so AMD cannot hang if the call
never connects).
Start AMD before creating the SIP participant so no audio is missed. Its
detection timeout begins only after listening starts; SIP settings bound
the pre-answer phase. If the call ends before audio arrives, AMD settles
immediately with an ``uncertain`` verdict (``reason="participant_missing"``).

For SIP participants, the no-speech timer and
audio/transcript processing are deferred until ``sip.callStatus ==
Expand All @@ -114,8 +114,7 @@ class AMD(EventEmitter[Literal["amd_prediction"]]):
The recommended pattern is the async context manager::

async with AMD(session, llm="openai/gpt-4.1-mini") as detector:
await ctx.api.sip.create_sip_participant(...)
await ctx.wait_for_participant(identity=participant_identity)
await ctx.api.sip.create_sip_participant(...) # wait_until_answered=True
result = await detector.execute()

Args:
Expand All @@ -132,8 +131,9 @@ class AMD(EventEmitter[Literal["amd_prediction"]]):
ivr_detection: If ``True`` (default), automatically start IVR
navigation when a ``machine-ivr`` result is returned.
participant_identity: If set, AMD listens only to this participant's
audio track. If omitted, the first remote audio track wins and
the publisher is resolved from the track sid.
audio track, and settles immediately if that participant
disconnects before publishing audio. If omitted, the first remote
audio track wins and the publisher is resolved from the track sid.
stt: STT used for transcript generation. Accepts an :class:`STT`
instance or an inference model string (e.g.
``"cartesia/ink-whisper"``). When omitted, AMD auto-selects:
Expand Down Expand Up @@ -400,26 +400,21 @@ async def _setup(self, session: AgentSession) -> None:
logger.warning(
"session room_io unavailable, starting amd timers immediately as fallback"
)
if self._classifier:
self._classifier.start_detection_timer()
self._classifier.start_listening()
self._start_listening()
else:
# Start the outer budget before waiting for a publication, so AMD
# can settle even if the participant never publishes audio.
if self._classifier:
self._classifier.start_detection_timer()
room = session._room_io.room
publication = await wait_for_track_publication(
room=room,
identity=self._participant_identity or None,
kind=rtc.TrackKind.KIND_AUDIO,
wait_for_subscription=True,
)
try:
publication = await wait_for_track_publication(
room=room,
identity=self._participant_identity or None,
kind=rtc.TrackKind.KIND_AUDIO,
wait_for_subscription=True,
)
except RuntimeError as e:
self._settle_participant_missing(str(e))
return
if self._closed or not self._classifier:
return
# Reset the budget at track-up so normal AMD timing runs from the
# subscribed publication, matching the active audio source.
self._classifier.start_detection_timer()

if self._participant_identity:
publisher = room.remote_participants.get(self._participant_identity)
Expand Down Expand Up @@ -452,8 +447,18 @@ async def _setup(self, session: AgentSession) -> None:
def _start_listening(self) -> None:
if self._closed or not self._classifier:
return
self._classifier.start_detection_timer()
self._classifier.start_listening()
logger.debug("call has been answered, AMD starts listening")
logger.debug("AMD starts listening")

def _settle_participant_missing(self, error: str) -> None:
if self._closed or not self._classifier:
return
logger.debug(
"AMD: call ended before detection could run, settling",
extra={"error": error},
)
self._classifier.settle(AMDCategory.UNCERTAIN, reason="participant_missing")

async def _wait_for_sip_answer(self, room: rtc.Room, identity: str) -> None:
try:
Expand All @@ -464,10 +469,8 @@ async def _wait_for_sip_answer(self, room: rtc.Room, identity: str) -> None:
value=_SIP_CALL_STATUS_ACTIVE,
)
except RuntimeError as e:
# SIP participant disconnected before going active, default to detection timeout
logger.debug(
"AMD: SIP answer wait failed; starting to listen", extra={"reason": str(e)}
)
self._settle_participant_missing(str(e))
return

if not self._closed:
self._start_listening()
Expand Down
Loading