895 lines
33 KiB
Python
895 lines
33 KiB
Python
#
|
||
# Copyright (c) 2024–2025, Daily
|
||
#
|
||
# SPDX-License-Identifier: BSD 2-Clause License
|
||
#
|
||
|
||
import argparse
|
||
import asyncio
|
||
import collections
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from typing import Deque
|
||
|
||
from dotenv import load_dotenv
|
||
from loguru import logger
|
||
from pipecat_flows import (
|
||
ContextStrategy,
|
||
ContextStrategyConfig,
|
||
FlowArgs,
|
||
FlowManager,
|
||
FlowResult,
|
||
FlowsFunctionSchema,
|
||
NodeConfig,
|
||
)
|
||
|
||
from pipecat.audio.vad.silero import SileroVADAnalyzer
|
||
from pipecat.audio.vad.vad_analyzer import VADParams
|
||
from pipecat.frames.frames import (
|
||
BotInterruptionFrame,
|
||
CancelFrame,
|
||
EndFrame,
|
||
EndTaskFrame,
|
||
Frame,
|
||
FunctionCallInProgressFrame,
|
||
FunctionCallResultFrame,
|
||
InputAudioRawFrame,
|
||
LLMMessagesFrame,
|
||
StartFrame,
|
||
StartInterruptionFrame,
|
||
SystemFrame,
|
||
TranscriptionFrame,
|
||
TTSSpeakFrame,
|
||
TTSTextFrame,
|
||
UserStartedSpeakingFrame,
|
||
UserStoppedSpeakingFrame,
|
||
VADUserStartedSpeakingFrame,
|
||
)
|
||
from pipecat.observers.base_observer import BaseObserver, FramePushed
|
||
from pipecat.pipeline.parallel_pipeline import ParallelPipeline
|
||
from pipecat.pipeline.pipeline import Pipeline
|
||
from pipecat.pipeline.runner import PipelineRunner
|
||
from pipecat.pipeline.task import PipelineParams, PipelineTask
|
||
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
|
||
from pipecat.processors.filters.function_filter import FunctionFilter
|
||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||
from pipecat.processors.transcript_processor import TranscriptProcessor
|
||
from pipecat.services.cartesia.tts import CartesiaTTSService
|
||
from pipecat.services.deepgram.stt import DeepgramSTTService
|
||
from pipecat.services.google.llm import GoogleLLMContext, GoogleLLMService
|
||
from pipecat.services.llm_service import FunctionCallParams
|
||
from pipecat.sync.event_notifier import EventNotifier
|
||
from pipecat.transports.services.daily import DailyParams, DailyTransport
|
||
|
||
load_dotenv(override=True)
|
||
|
||
logger.remove(0)
|
||
logger.add(sys.stderr, level="DEBUG")
|
||
|
||
daily_api_key = os.getenv("DAILY_API_KEY", "")
|
||
daily_api_url = os.getenv("DAILY_API_URL", "https://api.daily.co/v1")
|
||
|
||
use_prebuilt = True
|
||
|
||
# Simple constants for model states
|
||
VOICEMAIL_MODE = "voicemail"
|
||
HUMAN_MODE = "human"
|
||
MUTE_MODE = "mute"
|
||
|
||
VOICEMAIL_CONFIDENCE_THRESHOLD = 0.6
|
||
HUMAN_CONFIDENCE_THRESHOLD = 0.6
|
||
|
||
|
||
def warm():
|
||
"""
|
||
Warm up function to ensure the bot is ready to handle requests.
|
||
This function can be called periodically to keep the bot warm.
|
||
"""
|
||
logger.info("Warming up the bot...")
|
||
# Perform any necessary warm-up tasks here
|
||
# For example, you can load models, initialize connections, etc.
|
||
# This is just a placeholder for demonstration purposes
|
||
pass
|
||
|
||
|
||
# ------------ FLOW MANAGER SETUP ------------
|
||
|
||
|
||
# ------------ PIPECAT FLOWS FOR HUMAN CONVERSATION ------------
|
||
|
||
|
||
# Type definitions for flows
|
||
class GreetingResult(FlowResult):
|
||
greeting_complete: bool
|
||
|
||
|
||
class ConversationResult(FlowResult):
|
||
message: str
|
||
|
||
|
||
class EndConversationResult(FlowResult):
|
||
status: str
|
||
|
||
|
||
# Flow function handlers - updated to return (result, next_node) tuple
|
||
async def handle_greeting(
|
||
args: FlowArgs, flow_manager: FlowManager
|
||
) -> tuple[GreetingResult, NodeConfig]:
|
||
"""Handle initial greeting to human."""
|
||
logger.debug("handle_greeting executing")
|
||
|
||
result = GreetingResult(greeting_complete=True)
|
||
|
||
# Return the next node config directly instead of using set_node
|
||
next_node = create_conversation_node()
|
||
return result, next_node
|
||
|
||
|
||
async def handle_conversation(
|
||
args: FlowArgs, flow_manager: FlowManager
|
||
) -> tuple[ConversationResult, NodeConfig]:
|
||
"""Handle ongoing conversation with human."""
|
||
message = args.get("message", "")
|
||
logger.debug(f"handle_conversation executing with message: {message}")
|
||
|
||
result = ConversationResult(message=message)
|
||
|
||
# Return the same conversation node to continue the chat
|
||
next_node = create_conversation_node()
|
||
return result, next_node
|
||
|
||
|
||
async def handle_end_conversation(
|
||
args: FlowArgs, flow_manager: FlowManager
|
||
) -> tuple[EndConversationResult, NodeConfig]:
|
||
"""Handle ending the conversation."""
|
||
logger.debug("handle_end_conversation executing")
|
||
|
||
result = EndConversationResult(status="completed")
|
||
|
||
# Return the end node config directly
|
||
next_node = create_end_node()
|
||
return result, next_node
|
||
|
||
|
||
# Node configurations for human conversation flow
|
||
def create_greeting_node() -> NodeConfig:
|
||
"""Create the initial greeting node for human conversation."""
|
||
return {
|
||
"name": "greeting",
|
||
"role_messages": [
|
||
{
|
||
"role": "system",
|
||
"content": """You are a friendly chatbot. Your responses will be
|
||
converted to audio, so avoid special characters.
|
||
Be conversational and helpful. The user will have just replied to your greeting and question asking if they are Tim.
|
||
If they say yes, proceed to the conversation node.""",
|
||
}
|
||
],
|
||
"task_messages": [
|
||
{
|
||
"role": "system",
|
||
"content": """Decide if the user is Tim based on their response.
|
||
If they say yes, call handle_greeting to proceed to the conversation.""",
|
||
}
|
||
],
|
||
"respond_immediately": False,
|
||
"functions": [
|
||
FlowsFunctionSchema(
|
||
name="handle_greeting",
|
||
description="Mark that greeting is complete and proceed to conversation",
|
||
properties={},
|
||
required=[],
|
||
handler=handle_greeting,
|
||
)
|
||
],
|
||
}
|
||
|
||
|
||
def create_conversation_node() -> NodeConfig:
|
||
"""Create the main conversation node."""
|
||
return {
|
||
"name": "conversation",
|
||
"task_messages": [
|
||
{
|
||
"role": "system",
|
||
"content": (
|
||
"You are having a friendly conversation with a human. "
|
||
"Listen to what they say and respond helpfully. "
|
||
"Keep your responses brief and conversational. "
|
||
"If they indicate they want to end the conversation (saying goodbye, "
|
||
"thanks, that's all, etc.), call handle_end_conversation. "
|
||
"Otherwise, use handle_conversation to continue the chat."
|
||
),
|
||
}
|
||
],
|
||
"functions": [
|
||
FlowsFunctionSchema(
|
||
name="handle_conversation",
|
||
description="Continue the conversation with the human",
|
||
properties={"message": {"type": "string", "description": "The response message"}},
|
||
required=["message"],
|
||
handler=handle_conversation,
|
||
),
|
||
FlowsFunctionSchema(
|
||
name="handle_end_conversation",
|
||
description="End the conversation when the human is ready to finish",
|
||
properties={},
|
||
required=[],
|
||
handler=handle_end_conversation,
|
||
),
|
||
],
|
||
}
|
||
|
||
|
||
def create_end_node() -> NodeConfig:
|
||
"""Create the final conversation end node."""
|
||
return {
|
||
"name": "end",
|
||
"task_messages": [
|
||
{
|
||
"role": "system",
|
||
"content": (
|
||
"Thank the person for the conversation and say goodbye. "
|
||
"Keep it brief and friendly."
|
||
),
|
||
}
|
||
],
|
||
"functions": [], # Required by FlowManager, even if empty
|
||
"post_actions": [{"type": "end_conversation"}],
|
||
}
|
||
|
||
|
||
# ------------ SIMPLIFIED CLASSES ------------
|
||
|
||
|
||
class DebugClass(FrameProcessor):
|
||
"""A simple debug class to log frames."""
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
|
||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
logger.debug(f"DebugClass received frame: {frame} in direction: {direction}")
|
||
await self.push_frame(frame, direction)
|
||
|
||
|
||
class VoicemailDetectionObserver(BaseObserver):
|
||
"""Observes voicemail speaking patterns to know when voicemail is done."""
|
||
|
||
def __init__(self, timeout: float = 5.0):
|
||
super().__init__()
|
||
self._processed_frames = set()
|
||
self._timeout = timeout
|
||
self._last_turn_time = 0
|
||
self._voicemail_speaking = False
|
||
|
||
async def on_push_frame(self, data: FramePushed):
|
||
if data.frame.id in self._processed_frames:
|
||
return
|
||
self._processed_frames.add(data.frame.id)
|
||
|
||
if isinstance(data.frame, UserStartedSpeakingFrame):
|
||
self._voicemail_speaking = True
|
||
self._last_turn_time = 0
|
||
elif isinstance(data.frame, UserStoppedSpeakingFrame):
|
||
self._last_turn_time = time.time()
|
||
|
||
async def wait_for_voicemail(self):
|
||
"""Wait for voicemail to finish speaking."""
|
||
while self._voicemail_speaking:
|
||
logger.debug("📩️ Waiting for voicemail to finish")
|
||
if self._last_turn_time:
|
||
diff_time = time.time() - self._last_turn_time
|
||
self._voicemail_speaking = diff_time < self._timeout
|
||
if self._voicemail_speaking:
|
||
await asyncio.sleep(0.5)
|
||
|
||
|
||
class VADPrebufferProcessor(FrameProcessor):
|
||
"""
|
||
This processor buffers a specified number of audio frames before speech is
|
||
detected.
|
||
|
||
When a VADUserStartedSpeakingFrame is received, it first replays the
|
||
buffered audio frames in the correct order, ensuring that the very
|
||
beginning of the user's speech is not missed. After replaying the buffer,
|
||
all subsequent frames are passed through immediately.
|
||
|
||
This is useful for preventing the initial part of a user's utterance from
|
||
being cut off by the Voice Activity Detection (VAD).
|
||
|
||
Args:
|
||
prebuffer_frame_count (int): The number of InputAudioRawFrames to buffer before speech.
|
||
Defaults to 33.
|
||
direction (FrameDirection): The direction of frames to process (UPSTREAM or DOWNSTREAM).
|
||
Defaults to DOWNSTREAM.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
prebuffer_frame_count: int = 33,
|
||
direction: FrameDirection = FrameDirection.DOWNSTREAM,
|
||
):
|
||
super().__init__()
|
||
self._direction = direction
|
||
self._speech_started = False
|
||
self._prebuffer_frame_count = prebuffer_frame_count
|
||
|
||
# A deque with a maxlen is a highly efficient fixed-size buffer.
|
||
# When it's full, adding a new item automatically discards the oldest item.
|
||
self._audio_buffer: Deque[InputAudioRawFrame] = collections.deque(
|
||
maxlen=prebuffer_frame_count
|
||
)
|
||
|
||
def _should_passthrough_frame(self, frame: Frame, direction: FrameDirection) -> bool:
|
||
"""Determines if a frame should bypass the buffering logic entirely."""
|
||
return isinstance(frame, (SystemFrame, EndFrame)) or direction != self._direction
|
||
|
||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
|
||
# Let system/end frames and frames in the wrong direction pass through immediately.
|
||
if self._should_passthrough_frame(frame, direction):
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
|
||
# If speech has already started, the gate is open. Let all frames through.
|
||
if self._speech_started:
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
|
||
# --- Speech has NOT started yet ---
|
||
|
||
# The VAD frame is the trigger to release the buffered audio.
|
||
if isinstance(frame, VADUserStartedSpeakingFrame):
|
||
logger.debug(
|
||
f"Initial VAD Detected. Replaying {len(self._audio_buffer)} buffered audio frames."
|
||
)
|
||
# 1. Set the flag so all future frames pass through immediately.
|
||
self._speech_started = True
|
||
|
||
# 2. Push all the buffered audio frames downstream in order.
|
||
for buffered_frame in self._audio_buffer:
|
||
await self.push_frame(buffered_frame, direction)
|
||
|
||
# 3. Clear the buffer now that it's been sent.
|
||
self._audio_buffer.clear()
|
||
|
||
# 4. Finally, push the VAD frame itself so downstream processors know speech has started.
|
||
await self.push_frame(frame, direction)
|
||
|
||
# If it's an audio frame, add it to our buffer. It won't be pushed downstream yet.
|
||
elif isinstance(frame, InputAudioRawFrame):
|
||
self._audio_buffer.append(frame)
|
||
|
||
# Any other frames that arrive before speech (e.g., TextFrame) will be
|
||
# ignored by this processor, as they don't match the conditions above.
|
||
|
||
|
||
class BlockAudioFrames(FrameProcessor):
|
||
"""Blocks audio frames from being processed further, conditionally based on mode."""
|
||
|
||
def __init__(self, mode_checker, allowed_modes):
|
||
super().__init__()
|
||
self._mode_checker = mode_checker
|
||
self._allowed_modes = allowed_modes if isinstance(allowed_modes, list) else [allowed_modes]
|
||
|
||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
|
||
# Block audio frames based on current mode
|
||
if (
|
||
isinstance(frame, InputAudioRawFrame)
|
||
or isinstance(frame, UserStartedSpeakingFrame)
|
||
or isinstance(frame, UserStoppedSpeakingFrame)
|
||
or isinstance(frame, TranscriptionFrame)
|
||
):
|
||
current_mode = self._mode_checker()
|
||
# logger.debug(f"Current mode: {current_mode}, allowed modes: {self._allowed_modes}")
|
||
if current_mode in self._allowed_modes:
|
||
await self.push_frame(frame, direction)
|
||
# If current mode is not in allowed modes, just return (block the frame)
|
||
return
|
||
|
||
# Pass all other frames through
|
||
await self.push_frame(frame, direction)
|
||
|
||
|
||
class OutputGate(FrameProcessor):
|
||
"""Simple gate that opens when notified."""
|
||
|
||
def __init__(self, notifier, start_open: bool = False):
|
||
super().__init__()
|
||
self._gate_open = start_open
|
||
self._frames_buffer = []
|
||
self._notifier = notifier
|
||
self._gate_task = None
|
||
|
||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||
await super().process_frame(frame, direction)
|
||
|
||
# Always pass system frames and function call frames
|
||
if isinstance(
|
||
frame,
|
||
(
|
||
SystemFrame,
|
||
EndFrame,
|
||
FunctionCallInProgressFrame,
|
||
FunctionCallResultFrame,
|
||
),
|
||
):
|
||
if isinstance(frame, StartFrame):
|
||
await self._start()
|
||
elif isinstance(frame, (CancelFrame, EndFrame)):
|
||
await self._stop()
|
||
elif isinstance(frame, StartInterruptionFrame):
|
||
self._frames_buffer = []
|
||
self._gate_open = False
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
|
||
# Only gate downstream frames
|
||
if direction != FrameDirection.DOWNSTREAM:
|
||
await self.push_frame(frame, direction)
|
||
return
|
||
|
||
if self._gate_open:
|
||
await self.push_frame(frame, direction)
|
||
else:
|
||
# Buffer frames until gate opens
|
||
self._frames_buffer.append((frame, direction))
|
||
|
||
async def _start(self):
|
||
self._frames_buffer = []
|
||
if not self._gate_task:
|
||
self._gate_task = self.create_task(self._gate_task_handler())
|
||
|
||
async def _stop(self):
|
||
if self._gate_task:
|
||
await self.cancel_task(self._gate_task)
|
||
self._gate_task = None
|
||
|
||
async def _gate_task_handler(self):
|
||
"""Wait for notification to open gate."""
|
||
while True:
|
||
try:
|
||
await self._notifier.wait()
|
||
self._gate_open = True
|
||
# Flush buffered frames
|
||
for frame, direction in self._frames_buffer:
|
||
await self.push_frame(frame, direction)
|
||
self._frames_buffer = []
|
||
break # Gate stays open
|
||
except asyncio.CancelledError:
|
||
break
|
||
|
||
|
||
class UserAudioCollector(FrameProcessor):
|
||
"""Collects audio frames for the LLM context."""
|
||
|
||
def __init__(self, context, user_context_aggregator):
|
||
super().__init__()
|
||
self._context = context
|
||
self._user_context_aggregator = user_context_aggregator
|
||
self._audio_frames = []
|
||
self._start_secs = 0.2
|
||
self._user_speaking = False
|
||
|
||
async def process_frame(self, frame, direction):
|
||
await super().process_frame(frame, direction)
|
||
|
||
if isinstance(frame, TranscriptionFrame):
|
||
return
|
||
elif isinstance(frame, UserStartedSpeakingFrame):
|
||
self._user_speaking = True
|
||
elif isinstance(frame, UserStoppedSpeakingFrame):
|
||
self._user_speaking = False
|
||
self._context.add_audio_frames_message(audio_frames=self._audio_frames)
|
||
await self._user_context_aggregator.push_frame(
|
||
self._user_context_aggregator.get_context_frame()
|
||
)
|
||
elif isinstance(frame, InputAudioRawFrame):
|
||
if self._user_speaking:
|
||
self._audio_frames.append(frame)
|
||
else:
|
||
# Maintain rolling buffer
|
||
self._audio_frames.append(frame)
|
||
frame_duration = len(frame.audio) / 16 * frame.num_channels / frame.sample_rate
|
||
buffer_duration = frame_duration * len(self._audio_frames)
|
||
while buffer_duration > self._start_secs:
|
||
self._audio_frames.pop(0)
|
||
buffer_duration -= frame_duration
|
||
|
||
await self.push_frame(frame, direction)
|
||
|
||
|
||
# ------------ MAIN FUNCTION ------------
|
||
|
||
|
||
async def run_bot(room_url: str, token: str, body: dict) -> None:
|
||
"""Run the voice bot with parallel pipeline architecture."""
|
||
|
||
# ------------ SETUP ------------
|
||
logger.info(f"Starting bot with room: {room_url}")
|
||
|
||
body_data = json.loads(body)
|
||
dialout_settings = body_data["dialout_settings"]
|
||
phone_number = dialout_settings["phone_number"]
|
||
caller_id = dialout_settings.get("caller_id")
|
||
|
||
# Simple state tracking
|
||
current_mode = MUTE_MODE
|
||
is_voicemail = False
|
||
|
||
# Notifier for human conversation gate
|
||
human_notifier = EventNotifier()
|
||
|
||
# Observer for voicemail detection
|
||
voicemail_observer = VoicemailDetectionObserver()
|
||
|
||
# ------------ FUNCTION HANDLERS ------------
|
||
|
||
async def voicemail_detected(params: FunctionCallParams):
|
||
nonlocal current_mode, is_voicemail
|
||
|
||
confidence = params.arguments["confidence"]
|
||
reasoning = params.arguments["reasoning"]
|
||
|
||
logger.info(f"Voicemail detected - confidence: {confidence}, reasoning: {reasoning}")
|
||
|
||
if confidence >= VOICEMAIL_CONFIDENCE_THRESHOLD and current_mode == MUTE_MODE:
|
||
logger.info(f"🔄 MODE CHANGE: {current_mode} -> {VOICEMAIL_MODE}")
|
||
current_mode = VOICEMAIL_MODE
|
||
is_voicemail = True
|
||
|
||
await voicemail_observer.wait_for_voicemail()
|
||
|
||
# Generate voicemail message
|
||
message = "Hello, this is a message for Pipecat example user. This is Chatbot. Please call back on 123-456-7891. Thank you."
|
||
logger.info(f"🎤 SENDING VOICEMAIL MESSAGE: {message}")
|
||
await voicemail_tts.queue_frame(TTSSpeakFrame(text=message))
|
||
await voicemail_tts.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
|
||
|
||
await params.result_callback({"confidence": f"{confidence}", "reasoning": reasoning})
|
||
|
||
async def human_detected(params: FunctionCallParams):
|
||
nonlocal current_mode, is_voicemail
|
||
|
||
confidence = params.arguments["confidence"]
|
||
reasoning = params.arguments["reasoning"]
|
||
|
||
logger.info(f"Human detected - confidence: {confidence}, reasoning: {reasoning}")
|
||
|
||
if confidence >= HUMAN_CONFIDENCE_THRESHOLD and current_mode == MUTE_MODE:
|
||
logger.info(f"🔄 MODE CHANGE: {current_mode} -> {HUMAN_MODE}")
|
||
current_mode = HUMAN_MODE
|
||
is_voicemail = False
|
||
|
||
await human_notifier.notify()
|
||
message = "Hello, this is virtual agent John. Am I speaking to Tim?"
|
||
logger.info(f"🎤 SENDING HUMAN MESSAGE: {message}")
|
||
await voicemail_tts.queue_frame(TTSSpeakFrame(text=message))
|
||
|
||
await params.result_callback({"confidence": f"{confidence}", "reasoning": reasoning})
|
||
|
||
# async def terminate_call(params: FunctionCallParams):
|
||
# logger.info("Terminating call")
|
||
# await asyncio.sleep(3) # Brief delay before termination
|
||
# await params.llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
|
||
# await params.result_callback({"status": "call terminated"})
|
||
|
||
# ------------ TRANSPORT & SERVICES ------------
|
||
|
||
transport = DailyTransport(
|
||
room_url,
|
||
token,
|
||
"Voicemail Detection Bot",
|
||
DailyParams(
|
||
api_url=daily_api_url,
|
||
api_key=daily_api_key,
|
||
audio_in_enabled=True,
|
||
audio_out_enabled=True,
|
||
video_out_enabled=False,
|
||
vad_analyzer=SileroVADAnalyzer(
|
||
sample_rate=16000,
|
||
params=VADParams(start_secs=0.1, confidence=0.4, min_volume=0.4),
|
||
),
|
||
),
|
||
)
|
||
|
||
# TTS services
|
||
voicemail_tts = CartesiaTTSService(
|
||
api_key=os.getenv("CARTESIA_API_KEY", ""),
|
||
voice_id="b7d50908-b17c-442d-ad8d-810c63997ed9",
|
||
)
|
||
|
||
human_tts = CartesiaTTSService(
|
||
api_key=os.getenv("CARTESIA_API_KEY", ""),
|
||
voice_id="b7d50908-b17c-442d-ad8d-810c63997ed9",
|
||
)
|
||
|
||
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
|
||
|
||
# ------------ LLM SETUP ------------
|
||
|
||
detection_tools = [
|
||
{
|
||
"function_declarations": [
|
||
{
|
||
"name": "voicemail_detected",
|
||
"description": "Signals that a voicemail greeting has been detected by the LLM.",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"confidence": {
|
||
"type": "number",
|
||
"description": "The LLM's confidence score (ranging from 0.0 to 1.0) that a voicemail greeting was detected.",
|
||
},
|
||
"reasoning": {
|
||
"type": "string",
|
||
"description": "The LLM's textual explanation for why it believes a voicemail was detected, often citing specific phrases from the transcript.",
|
||
},
|
||
},
|
||
"required": ["confidence", "reasoning"],
|
||
},
|
||
},
|
||
{
|
||
"name": "human_detected",
|
||
"description": "Signals that a human attempting to communicate has been detected by the LLM.",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"confidence": {
|
||
"type": "number",
|
||
"description": "The LLM's confidence score (ranging from 0.0 to 1.0) that a human conversation has been detected.",
|
||
},
|
||
"reasoning": {
|
||
"type": "string",
|
||
"description": "The LLM's textual explanation for why it believes a human communication was detected, often citing specific phrases from the transcript.",
|
||
},
|
||
},
|
||
"required": ["confidence", "reasoning"],
|
||
},
|
||
},
|
||
]
|
||
}
|
||
]
|
||
|
||
detection_system_instructions = """
|
||
You are an AI Call Analyzer. Your primary function is to determine if the initial audio from an incoming call is a voicemail system/answering machine or a live human attempting to engage in conversation.
|
||
|
||
You will be provided with a transcript of the first few seconds of an audio interaction.
|
||
|
||
Based on your analysis of this transcript, you MUST decide to call ONE of the following two functions:
|
||
|
||
1. voicemail_detected
|
||
* Call this function if the transcript strongly indicates a pre-recorded voicemail greeting, an answering machine message, or instructions to leave a message.
|
||
* Keywords and phrases to look for: "you've reached," "not available," "leave a message," "at the tone/beep," "sorry I missed your call," "please leave your name and number."
|
||
* Also consider if the speech sounds like a monologue without expecting an immediate response.
|
||
* Keep in mind that the beep noise from a typical pre-recorded voicemail greeting comes after the greeting and not before.
|
||
|
||
2. human_detected
|
||
* Call this function if the transcript indicates a human is present and actively trying to communicate or expecting an immediate response.
|
||
* Keywords and phrases to look for: "Hello?", "Hi," "[Company Name], how can I help you?", "Speaking.", or any direct question aimed at initiating a dialogue.
|
||
* Consider if the speech sounds like the beginning of a two-way conversation.
|
||
|
||
**Decision Guidelines:**
|
||
|
||
* **Prioritize Human:** If there's ambiguity but a slight indication of a human trying to speak (e.g., a simple "Hello?" followed by a pause, which could be either), err on the side of `human_detected` to avoid missing a live interaction. Only call `voicemail_detected` if there are clear, strong indicators of a voicemail system.
|
||
* **Focus on Intent:** Is the speaker *delivering information* (likely voicemail) or *seeking interaction* (likely human)?
|
||
* **Brevity:** Voicemail greetings are often concise and formulaic. Human openings can be more varied."""
|
||
|
||
detection_llm = GoogleLLMService(
|
||
model="models/gemini-2.0-flash-lite",
|
||
api_key=os.getenv("GOOGLE_API_KEY"),
|
||
system_instruction=detection_system_instructions,
|
||
tools=detection_tools,
|
||
)
|
||
|
||
human_llm = GoogleLLMService(
|
||
model="models/gemini-2.0-flash-001",
|
||
api_key=os.getenv("GOOGLE_API_KEY"),
|
||
)
|
||
|
||
# ------------ CONTEXTS & FUNCTIONS ------------
|
||
|
||
detection_context = GoogleLLMContext()
|
||
detection_context_aggregator = detection_llm.create_context_aggregator(detection_context)
|
||
|
||
human_context = GoogleLLMContext()
|
||
human_context_aggregator = human_llm.create_context_aggregator(human_context)
|
||
|
||
# Register functions
|
||
detection_llm.register_function("voicemail_detected", voicemail_detected)
|
||
detection_llm.register_function("human_detected", human_detected)
|
||
|
||
# ------------ PROCESSORS ------------
|
||
|
||
def get_current_mode():
|
||
"""Get the current conversation mode."""
|
||
return current_mode
|
||
|
||
audio_collector = UserAudioCollector(detection_context, detection_context_aggregator.user())
|
||
voicemail_audio_blocker = BlockAudioFrames(get_current_mode, [VOICEMAIL_MODE, MUTE_MODE])
|
||
human_audio_blocker = BlockAudioFrames(get_current_mode, [HUMAN_MODE])
|
||
|
||
_VADPrebufferProcessor = VADPrebufferProcessor()
|
||
|
||
# Filter functions
|
||
async def voicemail_filter(frame) -> bool:
|
||
result = current_mode == VOICEMAIL_MODE or current_mode == MUTE_MODE
|
||
if hasattr(frame, "text") and frame.text:
|
||
logger.debug(
|
||
f"🎯 VOICEMAIL FILTER: mode={current_mode}, allowing={result}, frame={type(frame).__name__}"
|
||
)
|
||
return result
|
||
|
||
async def human_filter(frame) -> bool:
|
||
result = current_mode == HUMAN_MODE
|
||
if hasattr(frame, "text") and frame.text:
|
||
logger.debug(
|
||
f"🎯 HUMAN FILTER: mode={current_mode}, allowing={result}, frame={type(frame).__name__}"
|
||
)
|
||
return result
|
||
|
||
debug_processor = DebugClass()
|
||
|
||
transcript = TranscriptProcessor()
|
||
|
||
@transcript.event_handler("on_transcript_update")
|
||
async def handle_update(processor, frame):
|
||
for message in frame.messages:
|
||
logger.info(f"📝 TRANSCRIPT {message.role}: {message.content}")
|
||
|
||
# Add debug logging for TTS frames
|
||
class TTSDebugProcessor(FrameProcessor):
|
||
"""Debug processor to track TTS frames."""
|
||
|
||
def __init__(self, name):
|
||
super().__init__()
|
||
self._name = name
|
||
|
||
async def process_frame(self, frame, direction):
|
||
await super().process_frame(frame, direction)
|
||
# Log all frame types for comprehensive debugging
|
||
frame_type = type(frame).__name__
|
||
if isinstance(frame, TTSTextFrame):
|
||
logger.info(f"🔊 TTS DEBUG ({self._name}): {frame_type} - {frame.text}")
|
||
await self.push_frame(frame, direction)
|
||
|
||
voicemail_tts_debug = TTSDebugProcessor("VOICEMAIL")
|
||
human_tts_debug = TTSDebugProcessor("HUMAN")
|
||
|
||
# Debug processor to see what makes it past transport.output()
|
||
post_transport_debug = TTSDebugProcessor("POST_TRANSPORT")
|
||
|
||
# ------------ PIPELINE ------------
|
||
|
||
pipeline = Pipeline(
|
||
[
|
||
transport.input(),
|
||
ParallelPipeline(
|
||
# Voicemail detection branch
|
||
[
|
||
voicemail_audio_blocker, # Allows audio at the start to detect voicemail, and while in voicemail mode. Is blocked when LLM detects human.
|
||
_VADPrebufferProcessor,
|
||
audio_collector,
|
||
detection_context_aggregator.user(),
|
||
detection_llm,
|
||
FunctionFilter(voicemail_filter),
|
||
],
|
||
[
|
||
voicemail_tts,
|
||
transcript.assistant(), # Capture voicemail TTS frames
|
||
],
|
||
[
|
||
# Human conversation branch
|
||
human_audio_blocker, # Allows audio when in human mode, blocks when voicemail is detected or when deciding if human or voicemail.
|
||
stt,
|
||
transcript.user(), # Place after STT
|
||
human_context_aggregator.user(),
|
||
human_llm,
|
||
FunctionFilter(human_filter),
|
||
human_tts,
|
||
transcript.assistant(), # Capture human TTS frame
|
||
human_context_aggregator.assistant(),
|
||
],
|
||
),
|
||
transport.output(),
|
||
]
|
||
)
|
||
|
||
pipeline_task = PipelineTask(
|
||
pipeline,
|
||
idle_timeout_secs=90,
|
||
params=PipelineParams(
|
||
allow_interruptions=True,
|
||
enable_metrics=True,
|
||
enable_usage_metrics=True,
|
||
audio_in_sample_rate=16000,
|
||
audio_out_sample_rate=16000,
|
||
),
|
||
cancel_on_idle_timeout=False,
|
||
observers=[voicemail_observer],
|
||
)
|
||
|
||
flow_manager = FlowManager(
|
||
task=pipeline_task,
|
||
tts=human_tts,
|
||
llm=human_llm,
|
||
context_aggregator=human_context_aggregator,
|
||
transport=transport,
|
||
)
|
||
|
||
# ------------ EVENT HANDLERS ------------
|
||
|
||
@transport.event_handler("on_joined")
|
||
async def on_joined(transport, data):
|
||
await flow_manager.initialize(create_greeting_node())
|
||
|
||
if not use_prebuilt:
|
||
dialout_params = {"phoneNumber": phone_number}
|
||
if caller_id:
|
||
dialout_params["callerId"] = caller_id
|
||
await transport.start_dialout(dialout_params)
|
||
|
||
@transport.event_handler("on_participant_updated")
|
||
async def on_participant_updated(transport, participant):
|
||
logger.debug(f"Participant updated: {participant}")
|
||
|
||
@transport.event_handler("on_dialout_answered")
|
||
async def on_dialout_answered(transport, data):
|
||
logger.debug(f"Call answered: {data}")
|
||
await transport.capture_participant_transcription(data["sessionId"])
|
||
|
||
@transport.event_handler("on_dialout_error")
|
||
async def on_dialout_error(transport, data):
|
||
logger.error(f"Dialout error: {data}")
|
||
await pipeline_task.queue_frame(EndFrame())
|
||
|
||
@transport.event_handler("on_participant_left")
|
||
async def on_participant_left(transport, participant, reason):
|
||
await pipeline_task.queue_frame(EndFrame())
|
||
|
||
# Remove the problematic on_pipeline_started handler
|
||
# The context will be initialized naturally when frames flow through the pipeline
|
||
|
||
# ------------ RUN ------------
|
||
|
||
runner = PipelineRunner()
|
||
logger.info("Starting simplified parallel pipeline bot")
|
||
|
||
try:
|
||
await runner.run(pipeline_task)
|
||
except Exception as e:
|
||
logger.error(f"Pipeline error: {e}")
|
||
import traceback
|
||
|
||
logger.error(traceback.format_exc())
|
||
|
||
|
||
# ------------ ENTRY POINT ------------
|
||
|
||
|
||
async def main():
|
||
parser = argparse.ArgumentParser(description="Simplified Parallel Pipeline Bot")
|
||
parser.add_argument("-u", "--url", type=str, help="Room URL")
|
||
parser.add_argument("-t", "--token", type=str, help="Room Token")
|
||
parser.add_argument("-b", "--body", type=str, help="JSON config")
|
||
|
||
args = parser.parse_args()
|
||
if not all([args.url, args.token, args.body]):
|
||
logger.error("All arguments required")
|
||
parser.print_help()
|
||
sys.exit(1)
|
||
|
||
await run_bot(args.url, args.token, args.body)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|