Adolfo Reyna
44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
"""Stop the agent from hearing itself.
|
|
|
|
On laptop speakers with no acoustic echo cancellation, the microphone picks up
|
|
whatever Kokoro just said. The recognizer transcribes it, the aggregator treats
|
|
it as a user turn, and the agent answers its own greeting — which is exactly
|
|
what happens without this.
|
|
|
|
Pipecat's ``AlwaysUserMuteStrategy`` mutes while the bot is speaking, but
|
|
unmutes the instant playback ends, and the tail of that audio is still in the
|
|
input buffer. This keeps the mute up for a short while afterwards.
|
|
"""
|
|
|
|
import time
|
|
|
|
from pipecat.frames.frames import BotStartedSpeakingFrame, BotStoppedSpeakingFrame, Frame
|
|
from pipecat.turns.user_mute.base_user_mute_strategy import BaseUserMuteStrategy
|
|
|
|
|
|
class EchoGuardUserMuteStrategy(BaseUserMuteStrategy):
|
|
"""Mute the user while the bot speaks, plus a tail to let echo drain.
|
|
|
|
Args:
|
|
tail_secs: How long to stay muted after playback ends.
|
|
"""
|
|
|
|
def __init__(self, *, tail_secs: float = 0.4):
|
|
super().__init__()
|
|
self._tail_secs = tail_secs
|
|
self._bot_speaking = False
|
|
self._stopped_at = 0.0
|
|
|
|
async def process_frame(self, frame: Frame) -> bool:
|
|
await super().process_frame(frame)
|
|
|
|
if isinstance(frame, BotStartedSpeakingFrame):
|
|
self._bot_speaking = True
|
|
elif isinstance(frame, BotStoppedSpeakingFrame):
|
|
self._bot_speaking = False
|
|
self._stopped_at = time.monotonic()
|
|
|
|
if self._bot_speaking:
|
|
return True
|
|
return time.monotonic() - self._stopped_at < self._tail_secs
|