8bd781242a
The CallEndCoordinator.observe() decremented _tracked_speeches on every BotStoppedSpeakingFrame, not just the one corresponding to tracked (end-node) speech. When a preceding LLM utterance finished between track_speech() and the actual fixed-speech playback, the counter was consumed prematurely, ending the call before the end-node message played. Gate the decrement behind a _current_speech_is_tracked flag that is set only when BotStartedSpeakingFrame follows a pending tracked-speech start. Co-Authored-By: Claude <noreply@anthropic.com>
130 lines
4.9 KiB
Python
130 lines
4.9 KiB
Python
"""Shared call termination timing for prompt tools and workflow end nodes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections import deque
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
from loguru import logger
|
|
from pipecat.frames.frames import BotStartedSpeakingFrame, BotStoppedSpeakingFrame
|
|
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
|
|
|
|
|
class CallEndCoordinator:
|
|
"""End immediately or after the currently armed closing speech finishes."""
|
|
|
|
def __init__(self, queue_end: Callable[[str], Awaitable[None]]):
|
|
self._queue_end = queue_end
|
|
self._ending = False
|
|
self._armed = False
|
|
self._speaking = False
|
|
self._speech_stopped = asyncio.Event()
|
|
self._speech_stopped.set()
|
|
self._response_speech_started = False
|
|
self._tracked_speeches = 0
|
|
self._tracked_speech_completions: deque[asyncio.Future[None]] = deque()
|
|
self._finish_after_tracked_speech = False
|
|
self._finished = False
|
|
self._reason = "completed"
|
|
# Only BotStoppedSpeakingFrame that follows a tracked BotStartedSpeakingFrame
|
|
# decrements the tracked-speech counter. This prevents a stale or cross-talk
|
|
# stop frame (e.g. from a preceding LLM utterance) from consuming the counter
|
|
# meant for a fixed end-node message.
|
|
self._pending_tracked_starts = 0
|
|
self._current_speech_is_tracked = False
|
|
|
|
@property
|
|
def ending(self) -> bool:
|
|
return self._ending
|
|
|
|
@property
|
|
def speaking(self) -> bool:
|
|
"""Whether transport output is currently producing bot speech."""
|
|
return self._speaking
|
|
|
|
async def wait_until_silent(self) -> None:
|
|
"""Wait for the transport-owned bot speech boundary."""
|
|
await self._speech_stopped.wait()
|
|
|
|
def begin(self, reason: str) -> None:
|
|
self._ending = True
|
|
self._reason = reason or "completed"
|
|
|
|
def begin_response(self) -> None:
|
|
"""Start tracking speech produced by one LLM response."""
|
|
self._response_speech_started = False
|
|
|
|
def arm_after_speech(self) -> None:
|
|
"""Wait for the next observed bot speech to finish."""
|
|
self._armed = True
|
|
|
|
def track_speech(self) -> Awaitable[None]:
|
|
"""Register fixed speech and return its transport completion signal."""
|
|
completion = asyncio.get_running_loop().create_future()
|
|
self._tracked_speech_completions.append(completion)
|
|
self._tracked_speeches += 1
|
|
self._pending_tracked_starts += 1
|
|
return completion
|
|
|
|
async def arm_after_tracked_speech(self) -> None:
|
|
"""Finish after every already queued fixed utterance has played."""
|
|
self._finish_after_tracked_speech = True
|
|
if self._tracked_speeches == 0:
|
|
await self.finish()
|
|
|
|
async def finish_after_current_speech(self, *, has_text: bool) -> None:
|
|
"""Finish now if speech is absent/done, otherwise wait for its stop."""
|
|
if not has_text:
|
|
await self.finish()
|
|
return
|
|
if self._response_speech_started and not self._speaking:
|
|
await self.finish()
|
|
return
|
|
self._armed = True
|
|
|
|
async def finish(self) -> None:
|
|
if self._finished:
|
|
return
|
|
self._finished = True
|
|
await self._queue_end(self._reason)
|
|
|
|
async def observe(self, frame) -> None:
|
|
if isinstance(frame, BotStartedSpeakingFrame):
|
|
self._speaking = True
|
|
self._speech_stopped.clear()
|
|
self._response_speech_started = True
|
|
if self._pending_tracked_starts > 0:
|
|
self._pending_tracked_starts -= 1
|
|
self._current_speech_is_tracked = True
|
|
elif isinstance(frame, BotStoppedSpeakingFrame) and self._speaking:
|
|
self._speaking = False
|
|
self._speech_stopped.set()
|
|
if self._current_speech_is_tracked:
|
|
self._current_speech_is_tracked = False
|
|
if self._tracked_speeches > 0:
|
|
self._tracked_speeches -= 1
|
|
completion = self._tracked_speech_completions.popleft()
|
|
if not completion.done():
|
|
completion.set_result(None)
|
|
if (
|
|
self._finish_after_tracked_speech
|
|
and self._tracked_speeches == 0
|
|
):
|
|
logger.info("所有工作流结束语播报完毕,挂断通话")
|
|
await self.finish()
|
|
elif self._armed:
|
|
logger.info("结束语播报完毕,挂断通话")
|
|
await self.finish()
|
|
|
|
|
|
class EndCallAfterSpeechProcessor(FrameProcessor):
|
|
def __init__(self, coordinator: CallEndCoordinator):
|
|
super().__init__()
|
|
self._coordinator = coordinator
|
|
|
|
async def process_frame(self, frame, direction: FrameDirection):
|
|
await super().process_frame(frame, direction)
|
|
await self.push_frame(frame, direction)
|
|
await self._coordinator.observe(frame)
|