From 59240c7b96e17eddeb066c6d1aa907e8ee592548 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 21 Dec 2024 14:35:45 -0800 Subject: [PATCH 1/8] delay gemini multimodal live websocket connect --- examples/foundational/07d-interruptible-elevenlabs.py | 3 ++- src/pipecat/services/gemini_multimodal_live/gemini.py | 7 ++++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/examples/foundational/07d-interruptible-elevenlabs.py b/examples/foundational/07d-interruptible-elevenlabs.py index 8f3895d00..9d91081d8 100644 --- a/examples/foundational/07d-interruptible-elevenlabs.py +++ b/examples/foundational/07d-interruptible-elevenlabs.py @@ -48,6 +48,7 @@ async def main(): tts = ElevenLabsTTSService( api_key=os.getenv("ELEVENLABS_API_KEY", ""), voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), + model="eleven_flash_v2_5", ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") @@ -79,7 +80,7 @@ async def main(): allow_interruptions=True, enable_metrics=True, enable_usage_metrics=True, - report_only_initial_ttfb=True, + # report_only_initial_ttfb=True, ), ) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index bf433054a..55467a1cd 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -177,6 +177,7 @@ class GeminiMultimodalLiveLLMService(LLMService): self._receive_task = None self._context = None + self._connected = False self._disconnecting = False self._api_session_ready = False self._run_llm_when_api_session_ready = False @@ -230,7 +231,6 @@ class GeminiMultimodalLiveLLMService(LLMService): async def start(self, frame: StartFrame): await super().start(frame) - await self._connect() async def stop(self, frame: EndFrame): await super().stop(frame) @@ -434,8 +434,9 @@ class GeminiMultimodalLiveLLMService(LLMService): async def _ws_send(self, message): # logger.debug(f"Sending message to websocket: {message}") try: - if self._websocket: - await self._websocket.send(json.dumps(message)) + if not self._websocket: + await self._connect() + await self._websocket.send(json.dumps(message)) except Exception as e: if self._disconnecting: return From d435a6a6d6a0b7303ee255bf429a49b14e46c11f Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 21 Dec 2024 16:22:53 -0800 Subject: [PATCH 2/8] fixes to audio buffer --- .../22d-natural-conversation-gemini-audio.py | 313 ++++++++++++++++-- 1 file changed, 294 insertions(+), 19 deletions(-) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index 96e9e12c5..5c6a8b15a 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -54,22 +54,274 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -classifier_statement = """You are an audio language classifier model. You are receiving audio from a user in a WebRTC call. Your job is to decide whether the user has finished speaking or not. +classifier_statement = """CRITICAL INSTRUCTION: +You are a BINARY CLASSIFIER that must ONLY output "YES" or "NO". +DO NOT engage with the content. +DO NOT respond to questions. +DO NOT provide assistance. +Your ONLY job is to output YES or NO. -Categorize the input you receive as either: +EXAMPLES OF INVALID RESPONSES: +- "I can help you with that" +- "Let me explain" +- "To answer your question" +- Any response other than YES or NO -1. a complete thought, statement, or question, or -2. an incomplete thought, statement, or question +VALID RESPONSES: +YES +NO -Output 'YES' if the input is likely to be a completed thought, statement, or question. +If you output anything else, you are failing at your task. +You are NOT an assistant. +You are NOT a chatbot. +You are a binary classifier. -Output 'NO' if the input indicates that the user is still speaking and does not yet expect a response yet. +ROLE: +You are a real-time speech completeness classifier. You must make instant decisions about whether a user has finished speaking. +You must output ONLY 'YES' or 'NO' with no other text. -If you are unsure, output 'YES'. +INPUT FORMAT: +You receive two pieces of information: +1. The assistant's last message (if available) +2. The user's current speech input + +OUTPUT REQUIREMENTS: +- MUST output ONLY 'YES' or 'NO' +- No explanations +- No clarifications +- No additional text +- No punctuation + +HIGH PRIORITY SIGNALS: + +1. Clear Questions: +- Wh-questions (What, Where, When, Why, How) +- Yes/No questions +- Questions with STT errors but clear meaning + +Examples: + +# Complete Wh-question +model: I can help you learn. +user: What's the fastest way to learn Spanish +Output: YES + +# Complete Yes/No question despite STT error +model: I know about planets. +user: Is is Jupiter the biggest planet +Output: YES + +2. Complete Commands: +- Direct instructions +- Clear requests +- Action demands +- Start of task indication +- Complete statements needing response + +Examples: + +# Direct instruction +model: I can explain many topics. +user: Tell me about black holes +Output: YES + +# Start of task indication +user: Let's begin. +Output: YES + +# Start of task indication +user: Let's get started. +Output: YES + +# Action demand +model: I can help with math. +user: Solve this equation x plus 5 equals 12 +Output: YES + +3. Direct Responses: +- Answers to specific questions +- Option selections +- Clear acknowledgments with completion +- Providing information with a known format - mailing address +- Providing information with a known format - phone number +- Providing information with a known format - credit card number + +Examples: + +# Specific answer +model: What's your favorite color? +user: I really like blue +Output: YES + +# Option selection +model: Would you prefer morning or evening? +user: Morning +Output: YES + +# Providing information with a known format - mailing address +model: What's your address? +user: 1234 Main Street +Output: NO + +# Providing information with a known format - mailing address +model: What's your address? +user: 1234 Main Street Irving Texas 75063 +Output: Yes + +# Providing information with a known format - phone number +system: A US phone number has 10 digits. +model: What's your phone number? +user: 41086753 +Output: NO + +# Providing information with a known format - phone number +system: A US phone number has 10 digits. +model: What's your phone number? +user: 4108675309 +Output: Yes + +# Providing information with a known format - phone number +system: A US phone number has 10 digits. +model: What's your phone number? +user: 220 +user: 111 +user: 8775 +Output: Yes + +# Providing information with a known format - credit card number +model: What's your phone number? +user: 5556 +Output: NO + +# Providing information with a known format - phone number +model: What's your phone number? +user: 5556710454680800 +Output: Yes + +MEDIUM PRIORITY SIGNALS: + +1. Speech Pattern Completions: +- Self-corrections reaching completion +- False starts with clear ending +- Topic changes with complete thought +- Mid-sentence completions + +Examples: + +# Self-correction reaching completion +model: What would you like to know? +user: Tell me about... no wait, explain how rainbows form +Output: YES + +# Topic change with complete thought +model: The weather is nice today. +user: Actually can you tell me who invented the telephone +Output: YES + +# Mid-sentence completion +model: Hello I'm ready. +user: What's the capital of? France +Output: YES + +2. Context-Dependent Brief Responses: +- Acknowledgments (okay, sure, alright) +- Agreements (yes, yeah) +- Disagreements (no, nah) +- Confirmations (correct, exactly) + +Examples: + +# Acknowledgment +model: Should we talk about history? +user: Sure +Output: YES + +# Disagreement with completion +model: Is that what you meant? +user: No not really +Output: YES + +LOW PRIORITY SIGNALS: + +1. STT Artifacts (Consider but don't over-weight): +- Repeated words +- Unusual punctuation +- Capitalization errors +- Word insertions/deletions + +Examples: + +# Word repetition but complete +model: I can help with that. +user: What what is the time right now +Output: YES + +# Missing punctuation but complete +model: I can explain that. +user: Please tell me how computers work +Output: YES + +2. Speech Features: +- Filler words (um, uh, like) +- Thinking pauses +- Word repetitions +- Brief hesitations + +Examples: + +# Filler words but complete +model: What would you like to know? +user: Um uh how do airplanes fly +Output: YES + +# Thinking pause but incomplete +model: I can explain anything. +user: Well um I want to know about the +Output: NO + +DECISION RULES: + +1. Return YES if: +- ANY high priority signal shows clear completion +- Medium priority signals combine to show completion +- Meaning is clear despite low priority artifacts + +2. Return NO if: +- No high priority signals present +- Thought clearly trails off +- Multiple incomplete indicators +- User appears mid-formulation + +3. When uncertain: +- If you can understand the intent → YES +- If meaning is unclear → NO +- Always make a binary decision +- Never request clarification + +Examples: + +# Incomplete despite corrections +model: What would you like to know about? +user: Can you tell me about +Output: NO + +# Complete despite multiple artifacts +model: I can help you learn. +user: How do you I mean what's the best way to learn programming +Output: YES + +# Trailing off incomplete +model: I can explain anything. +user: I was wondering if you could tell me why +Output: NO """ conversational_system_message = """You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. +If you know that a number string is a phone number from the context of the conversation, say it as a phone number. For example 210-333-4567. + +If you know that a number string is a credit card number, say it as a credit card number. For example 4111-1111-1111-1111. + Please be very concise in your responses. Unless you are explicitly asked to do otherwise, give me the shortest complete answer possible without unnecessary elaboration. Generally you should answer with a single sentence. """ @@ -79,13 +331,15 @@ class StatementJudgeAudioContextAccumulator(FrameProcessor): super().__init__(**kwargs) self._notifier = notifier self._audio_frames = [] - self._audio_frames = [] self._start_secs = 0.2 # this should match VAD start_secs (hardcoding for now) - self._user_speaking = False + self._max_buffer_size_secs = 30 + self._user_speaking_vad_state = False + self._user_speaking_utterance_state = False async def reset(self): self._audio_frames = [] - self._user_speaking = False + self._user_speaking_vad_state = False + self._user_speaking_utterance_state = False async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -99,22 +353,42 @@ class StatementJudgeAudioContextAccumulator(FrameProcessor): # but let's leave that as an exercise to the reader. :-) return if isinstance(frame, UserStartedSpeakingFrame): - self._user_speaking = True + self._user_speaking_vad_state = True + self._user_speaking_utterance_state = True + elif isinstance(frame, UserStoppedSpeakingFrame): + if self._audio_frames[-1]: + fr = self._audio_frames[-1] + frame_duration = len(fr.audio) / 2 * fr.num_channels / fr.sample_rate + + logger.debug( + f"!!! Frame duration: ({len(fr.audio)}) ({fr.num_channels}) ({fr.sample_rate}) {frame_duration}" + ) + + data = b"".join(frame.audio for frame in self._audio_frames) + logger.debug( + f"Processing audio buffer seconds: ({len(self._audio_frames)}) ({len(data)}) {len(data) / 2 / 16000}" + ) self._user_speaking = False context = GoogleLLMContext() context.set_messages([{"role": "system", "content": classifier_statement}]) context.add_audio_frames_message(audio_frames=self._audio_frames) await self.push_frame(OpenAILLMContextFrame(context=context)) elif isinstance(frame, InputAudioRawFrame): - if self._user_speaking: - self._audio_frames.append(frame) + # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest + # frames as necessary. + # Use a small buffer size when an utterance is not in progress. Just big enough to backfill the start_secs. + # Use a larger buffer size when an utterance is in progress. + # Assume all audio frames have the same duration. + self._audio_frames.append(frame) + frame_duration = len(frame.audio) / 2 * frame.num_channels / frame.sample_rate + buffer_duration = frame_duration * len(self._audio_frames) + # logger.debug(f"!!! Frame duration: {frame_duration}") + if self._user_speaking_utterance_state: + while buffer_duration > self._max_buffer_size_secs: + self._audio_frames.pop(0) + buffer_duration -= frame_duration else: - # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest - # frames as necessary. Assume all audio frames have the same duration. - 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 @@ -215,6 +489,7 @@ async def main(): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), vad_audio_passthrough=True, + audio_in_sample_rate=16000, ), ) @@ -229,7 +504,7 @@ async def main(): # statement. This doesn't really need to be an LLM, we could use NLP # libraries for that, but we have the machinery to use an LLM, so we might as well! statement_llm = GoogleLLMService( - model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY") + model="gemini-2.0-flash-exp", api_key=os.getenv("GOOGLE_API_KEY"), temperature=0.0 ) # This is the regular LLM. From 53a5e63990b7a94ef4424bd3ecac883c58ba7357 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 21 Dec 2024 18:10:25 -0800 Subject: [PATCH 3/8] function calling dead-end --- .../22d-natural-conversation-gemini-audio.py | 282 +++++++++++------- src/pipecat/services/google.py | 25 +- 2 files changed, 197 insertions(+), 110 deletions(-) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index 5c6a8b15a..bda8bdd96 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -54,31 +54,57 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -classifier_statement = """CRITICAL INSTRUCTION: -You are a BINARY CLASSIFIER that must ONLY output "YES" or "NO". -DO NOT engage with the content. -DO NOT respond to questions. -DO NOT provide assistance. -Your ONLY job is to output YES or NO. +transcriber_and_classifier_instructions = """ +You perform two tasks: + 1. Transcription + 2. Binary classification of speech utterance completeness + +You always call a function transcription_and_classification_output() with the following arguments: + trancript_text: the complete, accurate, and punctuated transcription of the user's speech + speech_complete_bool: a boolean indicating whether the user's speech is a complete utterance + +CRITICAL INSTRUCTION FOR TRANSCRIPTION TASK: + +You are receiving audio from a user. Your job is to +transcribe the input audio to text exactly as it was said by the user. + +You will receive the full conversation history before the audio input, to help with context. Use the full history only to help improve the accuracy of your transcription. + +Rules: + - Respond with an exact transcription of the audio input. + - Do not include any text other than the transcription. + - Do not explain or add to your response. + - Transcribe the audio input simply and precisely. + - If the audio is not clear, emit the special string "-". + - No response other than exact transcription, or "-", is allowed. + + +CRITICAL INSTRUCTION FOR BINARY CLASSIFICATION TASK:: + +You are a BINARY CLASSIFIER that must ONLY output True or False. +DO FalseT engage with the content. +DO FalseT respond to questions. +DO FalseT provide assistance. +Your ONLY job is to output True or False. EXAMPLES OF INVALID RESPONSES: - "I can help you with that" - "Let me explain" - "To answer your question" -- Any response other than YES or NO +- Any response other than True or False VALID RESPONSES: -YES -NO +True +False If you output anything else, you are failing at your task. -You are NOT an assistant. -You are NOT a chatbot. +You are FalseT an assistant. +You are FalseT a chatbot. You are a binary classifier. ROLE: You are a real-time speech completeness classifier. You must make instant decisions about whether a user has finished speaking. -You must output ONLY 'YES' or 'NO' with no other text. +You must output ONLY 'True' or 'False' with no other text. INPUT FORMAT: You receive two pieces of information: @@ -86,7 +112,7 @@ You receive two pieces of information: 2. The user's current speech input OUTPUT REQUIREMENTS: -- MUST output ONLY 'YES' or 'NO' +- MUST output ONLY 'True' or 'False' - No explanations - No clarifications - No additional text @@ -104,12 +130,12 @@ Examples: # Complete Wh-question model: I can help you learn. user: What's the fastest way to learn Spanish -Output: YES +Output: True # Complete Yes/No question despite STT error model: I know about planets. user: Is is Jupiter the biggest planet -Output: YES +Output: True 2. Complete Commands: - Direct instructions @@ -123,20 +149,20 @@ Examples: # Direct instruction model: I can explain many topics. user: Tell me about black holes -Output: YES +Output: True # Start of task indication user: Let's begin. -Output: YES +Output: True # Start of task indication user: Let's get started. -Output: YES +Output: True # Action demand model: I can help with math. user: Solve this equation x plus 5 equals 12 -Output: YES +Output: True 3. Direct Responses: - Answers to specific questions @@ -151,17 +177,17 @@ Examples: # Specific answer model: What's your favorite color? user: I really like blue -Output: YES +Output: True # Option selection model: Would you prefer morning or evening? user: Morning -Output: YES +Output: True # Providing information with a known format - mailing address model: What's your address? user: 1234 Main Street -Output: NO +Output: False # Providing information with a known format - mailing address model: What's your address? @@ -172,7 +198,7 @@ Output: Yes system: A US phone number has 10 digits. model: What's your phone number? user: 41086753 -Output: NO +Output: False # Providing information with a known format - phone number system: A US phone number has 10 digits. @@ -191,7 +217,7 @@ Output: Yes # Providing information with a known format - credit card number model: What's your phone number? user: 5556 -Output: NO +Output: False # Providing information with a known format - phone number model: What's your phone number? @@ -211,17 +237,17 @@ Examples: # Self-correction reaching completion model: What would you like to know? user: Tell me about... no wait, explain how rainbows form -Output: YES +Output: True # Topic change with complete thought model: The weather is nice today. user: Actually can you tell me who invented the telephone -Output: YES +Output: True # Mid-sentence completion model: Hello I'm ready. user: What's the capital of? France -Output: YES +Output: True 2. Context-Dependent Brief Responses: - Acknowledgments (okay, sure, alright) @@ -234,12 +260,12 @@ Examples: # Acknowledgment model: Should we talk about history? user: Sure -Output: YES +Output: True # Disagreement with completion model: Is that what you meant? user: No not really -Output: YES +Output: True LOW PRIORITY SIGNALS: @@ -254,12 +280,12 @@ Examples: # Word repetition but complete model: I can help with that. user: What what is the time right now -Output: YES +Output: True # Missing punctuation but complete model: I can explain that. user: Please tell me how computers work -Output: YES +Output: True 2. Speech Features: - Filler words (um, uh, like) @@ -272,29 +298,29 @@ Examples: # Filler words but complete model: What would you like to know? user: Um uh how do airplanes fly -Output: YES +Output: True # Thinking pause but incomplete model: I can explain anything. user: Well um I want to know about the -Output: NO +Output: False DECISION RULES: -1. Return YES if: +1. Return True if: - ANY high priority signal shows clear completion - Medium priority signals combine to show completion - Meaning is clear despite low priority artifacts -2. Return NO if: +2. Return False if: - No high priority signals present - Thought clearly trails off - Multiple incomplete indicators - User appears mid-formulation 3. When uncertain: -- If you can understand the intent → YES -- If meaning is unclear → NO +- If you can understand the intent → True +- If meaning is unclear → False - Always make a binary decision - Never request clarification @@ -303,33 +329,69 @@ Examples: # Incomplete despite corrections model: What would you like to know about? user: Can you tell me about -Output: NO +Output: False # Complete despite multiple artifacts model: I can help you learn. user: How do you I mean what's the best way to learn programming -Output: YES +Output: True # Trailing off incomplete model: I can explain anything. user: I was wondering if you could tell me why -Output: NO +Output: False """ -conversational_system_message = """You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. +conversational_system_message = """You are a helpful assistant participating in a voice converation. -If you know that a number string is a phone number from the context of the conversation, say it as a phone number. For example 210-333-4567. +Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. -If you know that a number string is a credit card number, say it as a credit card number. For example 4111-1111-1111-1111. +If you know that a number string is a phone number from the context of the conversation, write it as a phone number. For example 210-333-4567. -Please be very concise in your responses. Unless you are explicitly asked to do otherwise, give me the shortest complete answer possible without unnecessary elaboration. Generally you should answer with a single sentence. +If you know that a number string is a credit card number, write it as a credit card number. For example 4111-1111-1111-1111. + +Please be very concise in your responses. Unless you are explicitly asked to do otherwise, give me shortest complete answer possible without unnecessary elaboration. Generally you should answer with a single sentence. """ -class StatementJudgeAudioContextAccumulator(FrameProcessor): - def __init__(self, *, notifier: BaseNotifier, **kwargs): +async def transcription_and_classification_output(transcript_text: str, speech_complete_bool: bool): + print(f"TRANSCRIPT: {transcript_text}") + print("------") + print(f"COMPLETE: {speech_complete_bool}") + print("------") + return + + +tx_and_cl_tools = [ + { + "function_declarations": [ + { + "name": "transcription_and_classification_output", + "description": "Deliver the transcription and classification output to an external process.", + "parameters": { + "type": "object", + "properties": { + "transcription_text": { + "type": "string", + "description": "The complete, accurate, and punctuated transcription of the user's speech. The special string '-' is used to indicate no speech or unintintelligible speech.", + }, + "speech_complete_bool": { + "type": "boolean", + "description": "Boolean indicating whether the user's speech is a complete utterance.", + }, + }, + "required": ["transcription_text", "speech_complete_bool"], + }, + }, + ] + } +] + + +class AudioAccumulator(FrameProcessor): + def __init__(self, *, notifier: BaseNotifier = None, **kwargs): super().__init__(**kwargs) - self._notifier = notifier + # self._notifier = notifier self._audio_frames = [] self._start_secs = 0.2 # this should match VAD start_secs (hardcoding for now) self._max_buffer_size_secs = 30 @@ -371,7 +433,9 @@ class StatementJudgeAudioContextAccumulator(FrameProcessor): ) self._user_speaking = False context = GoogleLLMContext() - context.set_messages([{"role": "system", "content": classifier_statement}]) + context.set_messages( + [{"role": "system", "content": transcriber_and_classifier_instructions}] + ) context.add_audio_frames_message(audio_frames=self._audio_frames) await self.push_frame(OpenAILLMContextFrame(context=context)) elif isinstance(frame, InputAudioRawFrame): @@ -396,25 +460,28 @@ class StatementJudgeAudioContextAccumulator(FrameProcessor): await self.push_frame(frame, direction) -class CompletenessCheck(FrameProcessor): - def __init__( - self, notifier: BaseNotifier, audio_accumulator: StatementJudgeAudioContextAccumulator - ): - super().__init__() - self._notifier = notifier - self._audio_accumulator = audio_accumulator +# class ClAndTxContextCreator(FrameProcessor): - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - if isinstance(frame, TextFrame) and frame.text.startswith("YES"): - logger.debug("Completeness check YES") - await self.push_frame(UserStoppedSpeakingFrame()) - await self._audio_accumulator.reset() - await self._notifier.notify() - elif isinstance(frame, TextFrame): - if frame.text.strip(): - logger.debug(f"Completeness check NO - '{frame.text}'") +# class CompletenessCheck(FrameProcessor): +# def __init__( +# self, notifier: BaseNotifier, audio_accumulator: StatementJudgeAudioContextAccumulator +# ): +# super().__init__() +# self._notifier = notifier +# self._audio_accumulator = audio_accumulator + +# async def process_frame(self, frame: Frame, direction: FrameDirection): +# await super().process_frame(frame, direction) + +# if isinstance(frame, TextFrame) and frame.text.startswith("True"): +# logger.debug("Completeness check True") +# await self.push_frame(UserStoppedSpeakingFrame()) +# await self._audio_accumulator.reset() +# await self._notifier.notify() +# elif isinstance(frame, TextFrame): +# if frame.text.strip(): +# logger.debug(f"Completeness check False - '{frame.text}'") class OutputGate(FrameProcessor): @@ -493,50 +560,52 @@ async def main(): ), ) - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady ) - # This is the LLM that will be used to detect if the user has finished a - # statement. This doesn't really need to be an LLM, we could use NLP - # libraries for that, but we have the machinery to use an LLM, so we might as well! - statement_llm = GoogleLLMService( - model="gemini-2.0-flash-exp", api_key=os.getenv("GOOGLE_API_KEY"), temperature=0.0 + # This is the LLM that will classify and transcribe user speech. + tx_and_cl_llm = GoogleLLMService( + model="gemini-2.0-flash-exp", + api_key=os.getenv("GOOGLE_API_KEY"), + tools=tx_and_cl_tools, + temperature=0.0, + tool_config={ + "function_calling_config": { + "mode": "ANY", + "allowed_function_names": ["transcription_and_classification_output"], + }, + }, ) - # This is the regular LLM. - llm = GoogleLLMService(model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY")) + # This is the regular LLM that responds conversationally. + conversation_llm = GoogleLLMService( + model="gemini-2.0-flash-exp", + api_key=os.getenv("GOOGLE_API_KEY"), + system_instruction=conversational_system_message, + ) - messages = [ - { - "role": "system", - "content": conversational_system_message, - }, - ] + context = OpenAILLMContext() + context_aggregator = conversation_llm.create_context_aggregator(context) - context = OpenAILLMContext(messages) - context_aggregator = llm.create_context_aggregator(context) - - # We have instructed the LLM to return 'YES' if it thinks the user - # completed a sentence. So, if it's 'YES' we will return true in this + # We have instructed the LLM to return 'True' if it thinks the user + # completed a sentence. So, if it's 'True' we will return true in this # predicate which will wake up the notifier. async def wake_check_filter(frame): - return frame.text == "YES" + return frame.text == "True" # This is a notifier that we use to synchronize the two LLMs. notifier = EventNotifier() # This turns the LLM context into an inference request to classify the user's speech # as complete or incomplete. - statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier) + # statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier) # This sends a UserStoppedSpeakingFrame and triggers the notifier event - completeness_check = CompletenessCheck( - notifier=notifier, audio_accumulator=statement_judge_context_filter - ) + # completeness_check = CompletenessCheck( + # notifier=notifier, audio_accumulator=statement_judge_context_filter + # ) # # Notify if the user hasn't said anything. async def user_idle_notifier(frame): @@ -562,6 +631,7 @@ async def main(): pipeline = Pipeline( [ transport.input(), + AudioAccumulator(), ParallelPipeline( [ # Pass everything except UserStoppedSpeaking to the elements after @@ -569,24 +639,24 @@ async def main(): FunctionFilter(filter=block_user_stopped_speaking), ], [ - statement_judge_context_filter, - statement_llm, - completeness_check, - ], - [ - stt, - context_aggregator.user(), - # Block everything except OpenAILLMContextFrame and LLMMessagesFrame - FunctionFilter(filter=pass_only_llm_trigger_frames), - llm, - bot_output_gate, # Buffer all llm/tts output until notified. + # cl_and_tx_context_creator, + tx_and_cl_llm, + # completeness_check, + # context_aggregator.user(), ], + # [ + # # Block everything except OpenAILLMContextFrame and LLMMessagesFrame + # # FunctionFilter(filter=pass_only_llm_trigger_frames), + # audio_input_context_creator, + # llm, + # bot_output_gate, # Buffer all llm/tts output until notified. + # ], ), - tts, - user_idle, - transport.output(), - context_aggregator.assistant(), - ] + # tts, + # user_idle, + # transport.output(), + # context_aggregator.assistant(), + ], ) task = PipelineTask( diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 6bbf1d000..4c5d3c205 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -593,6 +593,8 @@ class GoogleLLMService(LLMService): model: str = "gemini-1.5-flash-latest", params: InputParams = InputParams(), system_instruction: Optional[str] = None, + tools: Optional[List[Dict[str, Any]]] = None, + tool_config: Optional[Dict[str, Any]] = None, **kwargs, ): super().__init__(**kwargs) @@ -607,6 +609,8 @@ class GoogleLLMService(LLMService): "top_p": params.top_p, "extra": params.extra if isinstance(params.extra, dict) else {}, } + self._tools = tools + self._tool_config = tool_config def can_generate_metrics(self) -> bool: return True @@ -625,12 +629,13 @@ class GoogleLLMService(LLMService): try: logger.debug( - f"Generating chat: {self._system_instruction} | {context.get_messages_for_logging()}" + # f"Generating chat: {self._system_instruction} | {context.get_messages_for_logging()}" + f"!! Generating chat: {context.get_messages_for_logging()}" ) messages = context.messages if context.system_message and self._system_instruction != context.system_message: - logger.debug(f"System instruction changed: {context.system_message}") + # logger.debug(f"System instruction changed: {context.system_message}") self._system_instruction = context.system_message self._create_client() @@ -649,10 +654,21 @@ class GoogleLLMService(LLMService): generation_config = GenerationConfig(**generation_params) if generation_params else None await self.start_ttfb_metrics() - tools = context.tools if context.tools else [] + tools = [] + if context.tools: + tools = context.tools + elif self._tools: + tools = self._tools + tool_config = None + if self._tool_config: + tool_config = self._tool_config response = await self._client.generate_content_async( - contents=messages, tools=tools, stream=True, generation_config=generation_config + contents=messages, + tools=tools, + stream=True, + generation_config=generation_config, + tool_config=tool_config, ) await self.stop_ttfb_metrics() @@ -671,6 +687,7 @@ class GoogleLLMService(LLMService): if c.text: await self.push_frame(TextFrame(c.text)) elif c.function_call: + logger.debug(f"!!! Function call: {c.function_call}") args = type(c.function_call).to_dict(c.function_call).get("args", {}) await self.call_function( context=context, From f3dd35bfd9bc177dfb520c248886c2b16eb742ee Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 21 Dec 2024 22:18:56 -0800 Subject: [PATCH 4/8] working but needs cleanup --- .../22d-natural-conversation-gemini-audio.py | 351 +++++++++++------- 1 file changed, 208 insertions(+), 143 deletions(-) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index bda8bdd96..fd99ca606 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -10,6 +10,7 @@ import sys import time import aiohttp +import google.ai.generativelanguage as glm from dotenv import load_dotenv from loguru import logger from runner import configure @@ -20,6 +21,8 @@ from pipecat.frames.frames import ( EndFrame, Frame, InputAudioRawFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, LLMMessagesFrame, StartFrame, StartInterruptionFrame, @@ -34,6 +37,7 @@ 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.llm_response import LLMResponseAggregator from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -53,19 +57,7 @@ load_dotenv(override=True) logger.remove(0) logger.add(sys.stderr, level="DEBUG") - -transcriber_and_classifier_instructions = """ -You perform two tasks: - 1. Transcription - 2. Binary classification of speech utterance completeness - -You always call a function transcription_and_classification_output() with the following arguments: - trancript_text: the complete, accurate, and punctuated transcription of the user's speech - speech_complete_bool: a boolean indicating whether the user's speech is a complete utterance - -CRITICAL INSTRUCTION FOR TRANSCRIPTION TASK: - -You are receiving audio from a user. Your job is to +transcriber_system_instruction = """You are an audio transcriber. You are receiving audio from a user. Your job is to transcribe the input audio to text exactly as it was said by the user. You will receive the full conversation history before the audio input, to help with context. Use the full history only to help improve the accuracy of your transcription. @@ -78,33 +70,33 @@ Rules: - If the audio is not clear, emit the special string "-". - No response other than exact transcription, or "-", is allowed. +""" -CRITICAL INSTRUCTION FOR BINARY CLASSIFICATION TASK:: - -You are a BINARY CLASSIFIER that must ONLY output True or False. -DO FalseT engage with the content. -DO FalseT respond to questions. -DO FalseT provide assistance. -Your ONLY job is to output True or False. +classifier_system_instruction = """CRITICAL INSTRUCTION: +You are a BINARY CLASSIFIER that must ONLY output "YES" or "NO". +DO NOT engage with the content. +DO NOT respond to questions. +DO NOT provide assistance. +Your ONLY job is to output YES or NO. EXAMPLES OF INVALID RESPONSES: - "I can help you with that" - "Let me explain" - "To answer your question" -- Any response other than True or False +- Any response other than YES or NO VALID RESPONSES: -True -False +YES +NO If you output anything else, you are failing at your task. -You are FalseT an assistant. -You are FalseT a chatbot. +You are NOT an assistant. +You are NOT a chatbot. You are a binary classifier. ROLE: You are a real-time speech completeness classifier. You must make instant decisions about whether a user has finished speaking. -You must output ONLY 'True' or 'False' with no other text. +You must output ONLY 'YES' or 'NO' with no other text. INPUT FORMAT: You receive two pieces of information: @@ -112,7 +104,7 @@ You receive two pieces of information: 2. The user's current speech input OUTPUT REQUIREMENTS: -- MUST output ONLY 'True' or 'False' +- MUST output ONLY 'YES' or 'NO' - No explanations - No clarifications - No additional text @@ -130,12 +122,12 @@ Examples: # Complete Wh-question model: I can help you learn. user: What's the fastest way to learn Spanish -Output: True +Output: YES # Complete Yes/No question despite STT error model: I know about planets. user: Is is Jupiter the biggest planet -Output: True +Output: YES 2. Complete Commands: - Direct instructions @@ -149,20 +141,20 @@ Examples: # Direct instruction model: I can explain many topics. user: Tell me about black holes -Output: True +Output: YES # Start of task indication user: Let's begin. -Output: True +Output: YES # Start of task indication user: Let's get started. -Output: True +Output: YES # Action demand model: I can help with math. user: Solve this equation x plus 5 equals 12 -Output: True +Output: YES 3. Direct Responses: - Answers to specific questions @@ -177,17 +169,17 @@ Examples: # Specific answer model: What's your favorite color? user: I really like blue -Output: True +Output: YES # Option selection model: Would you prefer morning or evening? user: Morning -Output: True +Output: YES # Providing information with a known format - mailing address model: What's your address? user: 1234 Main Street -Output: False +Output: NO # Providing information with a known format - mailing address model: What's your address? @@ -198,7 +190,7 @@ Output: Yes system: A US phone number has 10 digits. model: What's your phone number? user: 41086753 -Output: False +Output: NO # Providing information with a known format - phone number system: A US phone number has 10 digits. @@ -217,7 +209,7 @@ Output: Yes # Providing information with a known format - credit card number model: What's your phone number? user: 5556 -Output: False +Output: NO # Providing information with a known format - phone number model: What's your phone number? @@ -237,17 +229,17 @@ Examples: # Self-correction reaching completion model: What would you like to know? user: Tell me about... no wait, explain how rainbows form -Output: True +Output: YES # Topic change with complete thought model: The weather is nice today. user: Actually can you tell me who invented the telephone -Output: True +Output: YES # Mid-sentence completion model: Hello I'm ready. user: What's the capital of? France -Output: True +Output: YES 2. Context-Dependent Brief Responses: - Acknowledgments (okay, sure, alright) @@ -260,12 +252,12 @@ Examples: # Acknowledgment model: Should we talk about history? user: Sure -Output: True +Output: YES # Disagreement with completion model: Is that what you meant? user: No not really -Output: True +Output: YES LOW PRIORITY SIGNALS: @@ -280,12 +272,12 @@ Examples: # Word repetition but complete model: I can help with that. user: What what is the time right now -Output: True +Output: YES # Missing punctuation but complete model: I can explain that. user: Please tell me how computers work -Output: True +Output: YES 2. Speech Features: - Filler words (um, uh, like) @@ -298,29 +290,29 @@ Examples: # Filler words but complete model: What would you like to know? user: Um uh how do airplanes fly -Output: True +Output: YES # Thinking pause but incomplete model: I can explain anything. user: Well um I want to know about the -Output: False +Output: NO DECISION RULES: -1. Return True if: +1. Return YES if: - ANY high priority signal shows clear completion - Medium priority signals combine to show completion - Meaning is clear despite low priority artifacts -2. Return False if: +2. Return NO if: - No high priority signals present - Thought clearly trails off - Multiple incomplete indicators - User appears mid-formulation 3. When uncertain: -- If you can understand the intent → True -- If meaning is unclear → False +- If you can understand the intent → YES +- If meaning is unclear → NO - Always make a binary decision - Never request clarification @@ -329,20 +321,20 @@ Examples: # Incomplete despite corrections model: What would you like to know about? user: Can you tell me about -Output: False +Output: NO # Complete despite multiple artifacts model: I can help you learn. user: How do you I mean what's the best way to learn programming -Output: True +Output: YES # Trailing off incomplete model: I can explain anything. user: I was wondering if you could tell me why -Output: False +Output: NO """ -conversational_system_message = """You are a helpful assistant participating in a voice converation. +conversation_system_instruction = """You are a helpful assistant participating in a voice converation. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. @@ -354,44 +346,9 @@ Please be very concise in your responses. Unless you are explicitly asked to do """ -async def transcription_and_classification_output(transcript_text: str, speech_complete_bool: bool): - print(f"TRANSCRIPT: {transcript_text}") - print("------") - print(f"COMPLETE: {speech_complete_bool}") - print("------") - return - - -tx_and_cl_tools = [ - { - "function_declarations": [ - { - "name": "transcription_and_classification_output", - "description": "Deliver the transcription and classification output to an external process.", - "parameters": { - "type": "object", - "properties": { - "transcription_text": { - "type": "string", - "description": "The complete, accurate, and punctuated transcription of the user's speech. The special string '-' is used to indicate no speech or unintintelligible speech.", - }, - "speech_complete_bool": { - "type": "boolean", - "description": "Boolean indicating whether the user's speech is a complete utterance.", - }, - }, - "required": ["transcription_text", "speech_complete_bool"], - }, - }, - ] - } -] - - class AudioAccumulator(FrameProcessor): - def __init__(self, *, notifier: BaseNotifier = None, **kwargs): + def __init__(self, **kwargs): super().__init__(**kwargs) - # self._notifier = notifier self._audio_frames = [] self._start_secs = 0.2 # this should match VAD start_secs (hardcoding for now) self._max_buffer_size_secs = 30 @@ -433,10 +390,9 @@ class AudioAccumulator(FrameProcessor): ) self._user_speaking = False context = GoogleLLMContext() - context.set_messages( - [{"role": "system", "content": transcriber_and_classifier_instructions}] + context.add_audio_frames_message( + text="Audio to process", audio_frames=self._audio_frames ) - context.add_audio_frames_message(audio_frames=self._audio_frames) await self.push_frame(OpenAILLMContextFrame(context=context)) elif isinstance(frame, InputAudioRawFrame): # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest @@ -463,33 +419,49 @@ class AudioAccumulator(FrameProcessor): # class ClAndTxContextCreator(FrameProcessor): -# class CompletenessCheck(FrameProcessor): -# def __init__( -# self, notifier: BaseNotifier, audio_accumulator: StatementJudgeAudioContextAccumulator -# ): -# super().__init__() -# self._notifier = notifier -# self._audio_accumulator = audio_accumulator +class CompletenessCheck(FrameProcessor): + def __init__(self, notifier: BaseNotifier, audio_accumulator: AudioAccumulator, **kwargs): + super().__init__() + self._notifier = notifier + self._audio_accumulator = audio_accumulator -# async def process_frame(self, frame: Frame, direction: FrameDirection): -# await super().process_frame(frame, direction) + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) -# if isinstance(frame, TextFrame) and frame.text.startswith("True"): -# logger.debug("Completeness check True") -# await self.push_frame(UserStoppedSpeakingFrame()) -# await self._audio_accumulator.reset() -# await self._notifier.notify() -# elif isinstance(frame, TextFrame): -# if frame.text.strip(): -# logger.debug(f"Completeness check False - '{frame.text}'") + if isinstance(frame, TextFrame) and frame.text.startswith("YES"): + logger.debug("Completeness check YES") + await self.push_frame(UserStoppedSpeakingFrame()) + await self._audio_accumulator.reset() + await self._notifier.notify() + elif isinstance(frame, TextFrame): + if frame.text.strip(): + logger.debug(f"Completeness check NO - '{frame.text}'") + + +class TempPrinter(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if not isinstance(frame, InputAudioRawFrame): + logger.debug(f"!!! {frame}") + await self.push_frame(frame, direction) class OutputGate(FrameProcessor): - def __init__(self, notifier: BaseNotifier, **kwargs): + def __init__( + self, + notifier: BaseNotifier, + context: OpenAILLMContext, + user_transcription_buffer: "UserAggregatorBuffer", + **kwargs, + ): super().__init__(**kwargs) self._gate_open = False self._frames_buffer = [] self._notifier = notifier + self._context = context + self._transcription_buffer = user_transcription_buffer + + logger.debug("!!! OutputGate created") def close_gate(self): self._gate_open = False @@ -524,6 +496,7 @@ class OutputGate(FrameProcessor): self._frames_buffer.append((frame, direction)) async def _start(self): + logger.debug("!!! OutputGate start") self._frames_buffer = [] self._gate_task = self.get_event_loop().create_task(self._gate_task_handler()) @@ -533,14 +506,86 @@ class OutputGate(FrameProcessor): async def _gate_task_handler(self): while True: + logger.debug("!!! Waiting for notifier") try: await self._notifier.wait() + logger.debug("!!! Notified") + transcription = await self._transcription_buffer.wait_for_transcription() + + # logger.debug(f"!!! OutputGate got transcription: {transcription}") + # logger.debug( + # f"!!! OutputGate has messages: {self._context.get_messages_for_logging()}" + # ) + + last_message = self._context.messages[-1] + if last_message.role == "user": + last_message.parts = [glm.Part(text=transcription)] + + # logger.debug( + # f"!!! NOW OutputGate has messages: {self._context.get_messages_for_logging()}" + # ) + self.open_gate() for frame, direction in self._frames_buffer: await self.push_frame(frame, direction) self._frames_buffer = [] except asyncio.CancelledError: break + except Exception as e: + logger.error(f"!!! OutputGate error: {e}") + raise e + break + + +class ConversationAudioContextAssembler(FrameProcessor): + def __init__(self, context: OpenAILLMContext, **kwargs): + super().__init__(**kwargs) + self._context = context + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + # We must not block system frames. + if isinstance(frame, SystemFrame): + await self.push_frame(frame, direction) + return + + if isinstance(frame, OpenAILLMContextFrame): + GoogleLLMContext.upgrade_to_google(self._context) + last_message = frame.context.messages[-1] + self._context._messages.append(last_message) + logger.debug( + f"!!! ConversationAudioContextAssembler {self._context.get_messages_for_logging()}" + ) + await self.push_frame(OpenAILLMContextFrame(context=self._context)) + + +class UserAggregatorBuffer(LLMResponseAggregator): + def __init__(self, **kwargs): + super().__init__( + messages=None, + role=None, + start_frame=LLMFullResponseStartFrame, + end_frame=LLMFullResponseEndFrame, + accumulator_frame=TextFrame, + handle_interruptions=True, + expect_stripped_words=False, + ) + self._transcription = "" + + async def _push_aggregation(self): + if self._aggregation: + self._transcription = self._aggregation + self._aggregation = "" + + logger.debug(f"!!! UserAggregatorBuffer: {self._transcription}") + + async def wait_for_transcription(self): + while not self._transcription: + await asyncio.sleep(0.01) + tx = self._transcription + self._transcription = "" + return tx async def main(): @@ -565,25 +610,30 @@ async def main(): voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady ) - # This is the LLM that will classify and transcribe user speech. - tx_and_cl_llm = GoogleLLMService( + # This is the LLM that will transcribe user speech. + tx_llm = GoogleLLMService( + name="Transcriber", model="gemini-2.0-flash-exp", api_key=os.getenv("GOOGLE_API_KEY"), - tools=tx_and_cl_tools, temperature=0.0, - tool_config={ - "function_calling_config": { - "mode": "ANY", - "allowed_function_names": ["transcription_and_classification_output"], - }, - }, + system_instruction=transcriber_system_instruction, + ) + + # This is the LLM that will classify user speech as complete or incomplete. + classifier_llm = GoogleLLMService( + name="Classifier", + model="gemini-2.0-flash-exp", + api_key=os.getenv("GOOGLE_API_KEY"), + temperature=0.0, + system_instruction=classifier_system_instruction, ) # This is the regular LLM that responds conversationally. conversation_llm = GoogleLLMService( + name="Conversation", model="gemini-2.0-flash-exp", api_key=os.getenv("GOOGLE_API_KEY"), - system_instruction=conversational_system_message, + system_instruction=conversation_system_instruction, ) context = OpenAILLMContext() @@ -602,10 +652,11 @@ async def main(): # as complete or incomplete. # statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier) + audio_accumulater = AudioAccumulator() # This sends a UserStoppedSpeakingFrame and triggers the notifier event - # completeness_check = CompletenessCheck( - # notifier=notifier, audio_accumulator=statement_judge_context_filter - # ) + completeness_check = CompletenessCheck( + notifier=notifier, audio_accumulator=audio_accumulater + ) # # Notify if the user hasn't said anything. async def user_idle_notifier(frame): @@ -615,8 +666,6 @@ async def main(): # sentence, this will wake up the notifier if that happens. user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0) - bot_output_gate = OutputGate(notifier=notifier) - async def block_user_stopped_speaking(frame): return not isinstance(frame, UserStoppedSpeakingFrame) @@ -628,10 +677,18 @@ async def main(): or isinstance(frame, StopInterruptionFrame) ) + conversation_audio_context_assembler = ConversationAudioContextAssembler(context=context) + + user_aggregator_buffer = UserAggregatorBuffer() + + bot_output_gate = OutputGate( + notifier=notifier, context=context, user_transcription_buffer=user_aggregator_buffer + ) + pipeline = Pipeline( [ transport.input(), - AudioAccumulator(), + audio_accumulater, ParallelPipeline( [ # Pass everything except UserStoppedSpeaking to the elements after @@ -639,23 +696,31 @@ async def main(): FunctionFilter(filter=block_user_stopped_speaking), ], [ - # cl_and_tx_context_creator, - tx_and_cl_llm, - # completeness_check, - # context_aggregator.user(), + ParallelPipeline( + [ + classifier_llm, + completeness_check, + ], + [ + tx_llm, + user_aggregator_buffer, + ], + ) + ], + [ + # Block everything except OpenAILLMContextFrame and LLMMessagesFrame + # FunctionFilter(filter=pass_only_llm_trigger_frames), + conversation_audio_context_assembler, + conversation_llm, + bot_output_gate, # buffer output until notified. + # TempPrinter(), ], - # [ - # # Block everything except OpenAILLMContextFrame and LLMMessagesFrame - # # FunctionFilter(filter=pass_only_llm_trigger_frames), - # audio_input_context_creator, - # llm, - # bot_output_gate, # Buffer all llm/tts output until notified. - # ], ), - # tts, - # user_idle, - # transport.output(), - # context_aggregator.assistant(), + # wherefore art thou, user context aggregator? + tts, + user_idle, + transport.output(), + context_aggregator.assistant(), ], ) From f5f0de00e416cdc7947d503d820927d2a262610b Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 21 Dec 2024 23:04:00 -0800 Subject: [PATCH 5/8] still some cleanup to do --- .../22d-natural-conversation-gemini-audio.py | 94 ++++++++----------- src/pipecat/services/google.py | 2 +- 2 files changed, 42 insertions(+), 54 deletions(-) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index fd99ca606..78d1271e6 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -187,35 +187,35 @@ user: 1234 Main Street Irving Texas 75063 Output: Yes # Providing information with a known format - phone number -system: A US phone number has 10 digits. model: What's your phone number? user: 41086753 Output: NO # Providing information with a known format - phone number -system: A US phone number has 10 digits. model: What's your phone number? user: 4108675309 Output: Yes # Providing information with a known format - phone number -system: A US phone number has 10 digits. model: What's your phone number? user: 220 -user: 111 -user: 8775 -Output: Yes +Output: No # Providing information with a known format - credit card number -model: What's your phone number? +model: What's your credit card number? user: 5556 Output: NO # Providing information with a known format - phone number -model: What's your phone number? +model: What's your credit card number? user: 5556710454680800 Output: Yes +model: What's your credit card number? +user: 414067 +Output: NO + + MEDIUM PRIORITY SIGNALS: 1. Speech Pattern Completions: @@ -390,9 +390,7 @@ class AudioAccumulator(FrameProcessor): ) self._user_speaking = False context = GoogleLLMContext() - context.add_audio_frames_message( - text="Audio to process", audio_frames=self._audio_frames - ) + context.add_audio_frames_message(text="Audio follows", audio_frames=self._audio_frames) await self.push_frame(OpenAILLMContextFrame(context=context)) elif isinstance(frame, InputAudioRawFrame): # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest @@ -416,34 +414,50 @@ class AudioAccumulator(FrameProcessor): await self.push_frame(frame, direction) -# class ClAndTxContextCreator(FrameProcessor): - - class CompletenessCheck(FrameProcessor): + wait_time = 5.0 + def __init__(self, notifier: BaseNotifier, audio_accumulator: AudioAccumulator, **kwargs): super().__init__() self._notifier = notifier self._audio_accumulator = audio_accumulator + self._idle_task = None + self._wakeup_time = 0 async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, TextFrame) and frame.text.startswith("YES"): logger.debug("Completeness check YES") + if self._idle_task: + logger.debug(f"CompletenessCheck idle wait CANCEL") + self._idle_task.cancel() + self._idle_task = None await self.push_frame(UserStoppedSpeakingFrame()) await self._audio_accumulator.reset() await self._notifier.notify() elif isinstance(frame, TextFrame): if frame.text.strip(): logger.debug(f"Completeness check NO - '{frame.text}'") + # start timer to wake up if necessary + if self._wakeup_time: + self._wakeup_time = time.time() + self.wait_time + else: + logger.debug("CompletenessCheck idle wait START") + self._wakeup_time = time.time() + self.wait_time + self._idle_task = self.get_event_loop().create_task(self._idle_task_handler()) - -class TempPrinter(FrameProcessor): - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - if not isinstance(frame, InputAudioRawFrame): - logger.debug(f"!!! {frame}") - await self.push_frame(frame, direction) + async def _idle_task_handler(self): + try: + while time.time() < self._wakeup_time: + await asyncio.sleep(0.01) + logger.debug(f"CompletenessCheck idle wait OVER") + await self._notifier.notify() + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"CompletenessCheck idle wait error: {e}") + raise e class OutputGate(FrameProcessor): @@ -461,8 +475,6 @@ class OutputGate(FrameProcessor): self._context = context self._transcription_buffer = user_transcription_buffer - logger.debug("!!! OutputGate created") - def close_gate(self): self._gate_open = False @@ -496,7 +508,6 @@ class OutputGate(FrameProcessor): self._frames_buffer.append((frame, direction)) async def _start(self): - logger.debug("!!! OutputGate start") self._frames_buffer = [] self._gate_task = self.get_event_loop().create_task(self._gate_task_handler()) @@ -506,25 +517,17 @@ class OutputGate(FrameProcessor): async def _gate_task_handler(self): while True: - logger.debug("!!! Waiting for notifier") + # logger.debug("!!! Waiting for notifier") try: await self._notifier.wait() - logger.debug("!!! Notified") - transcription = await self._transcription_buffer.wait_for_transcription() - # logger.debug(f"!!! OutputGate got transcription: {transcription}") - # logger.debug( - # f"!!! OutputGate has messages: {self._context.get_messages_for_logging()}" - # ) + # logger.debug("!!! Notified") + transcription = await self._transcription_buffer.wait_for_transcription() last_message = self._context.messages[-1] if last_message.role == "user": last_message.parts = [glm.Part(text=transcription)] - # logger.debug( - # f"!!! NOW OutputGate has messages: {self._context.get_messages_for_logging()}" - # ) - self.open_gate() for frame, direction in self._frames_buffer: await self.push_frame(frame, direction) @@ -532,7 +535,7 @@ class OutputGate(FrameProcessor): except asyncio.CancelledError: break except Exception as e: - logger.error(f"!!! OutputGate error: {e}") + logger.error(f"OutputGate error: {e}") raise e break @@ -554,9 +557,6 @@ class ConversationAudioContextAssembler(FrameProcessor): GoogleLLMContext.upgrade_to_google(self._context) last_message = frame.context.messages[-1] self._context._messages.append(last_message) - logger.debug( - f"!!! ConversationAudioContextAssembler {self._context.get_messages_for_logging()}" - ) await self.push_frame(OpenAILLMContextFrame(context=self._context)) @@ -578,7 +578,7 @@ class UserAggregatorBuffer(LLMResponseAggregator): self._transcription = self._aggregation self._aggregation = "" - logger.debug(f"!!! UserAggregatorBuffer: {self._transcription}") + logger.debug(f"[Transcription] {self._transcription}") async def wait_for_transcription(self): while not self._transcription: @@ -658,14 +658,6 @@ async def main(): notifier=notifier, audio_accumulator=audio_accumulater ) - # # Notify if the user hasn't said anything. - async def user_idle_notifier(frame): - await notifier.notify() - - # Sometimes the LLM will fail detecting if a user has completed a - # sentence, this will wake up the notifier if that happens. - user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0) - async def block_user_stopped_speaking(frame): return not isinstance(frame, UserStoppedSpeakingFrame) @@ -708,17 +700,13 @@ async def main(): ) ], [ - # Block everything except OpenAILLMContextFrame and LLMMessagesFrame - # FunctionFilter(filter=pass_only_llm_trigger_frames), conversation_audio_context_assembler, conversation_llm, - bot_output_gate, # buffer output until notified. + bot_output_gate, # buffer output until notified, then flush frames and update context # TempPrinter(), ], ), - # wherefore art thou, user context aggregator? tts, - user_idle, transport.output(), context_aggregator.assistant(), ], diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 4c5d3c205..091c4df06 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -630,7 +630,7 @@ class GoogleLLMService(LLMService): try: logger.debug( # f"Generating chat: {self._system_instruction} | {context.get_messages_for_logging()}" - f"!! Generating chat: {context.get_messages_for_logging()}" + f"Generating chat: {context.get_messages_for_logging()}" ) messages = context.messages From ab5df1a236a93222a736dbd831cf019dfe88f3ba Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sun, 22 Dec 2024 11:19:02 -0800 Subject: [PATCH 6/8] feature complete gemini audio, transcription, and phrase endpointing demo --- .../22d-natural-conversation-gemini-audio.py | 181 +++++++++++------- src/pipecat/services/google.py | 13 +- 2 files changed, 117 insertions(+), 77 deletions(-) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index 78d1271e6..2c6d21f92 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -57,6 +57,14 @@ load_dotenv(override=True) logger.remove(0) logger.add(sys.stderr, level="DEBUG") +# TRANSCRIBER_MODEL = "gemini-1.5-flash-latest" +# CLASSIFIER_MODEL = "gemini-1.5-flash-latest" +# CONVERSATION_MODEL = "gemini-1.5-flash-latest" + +TRANSCRIBER_MODEL = "gemini-2.0-flash-exp" +CLASSIFIER_MODEL = "gemini-2.0-flash-exp" +CONVERSATION_MODEL = "gemini-2.0-flash-exp" + transcriber_system_instruction = """You are an audio transcriber. You are receiving audio from a user. Your job is to transcribe the input audio to text exactly as it was said by the user. @@ -347,6 +355,11 @@ Please be very concise in your responses. Unless you are explicitly asked to do class AudioAccumulator(FrameProcessor): + """Buffers user audio until the user stops speaking. + + Always pushes a fresh context with a single audio message. + """ + def __init__(self, **kwargs): super().__init__(**kwargs) self._audio_frames = [] @@ -376,14 +389,6 @@ class AudioAccumulator(FrameProcessor): self._user_speaking_utterance_state = True elif isinstance(frame, UserStoppedSpeakingFrame): - if self._audio_frames[-1]: - fr = self._audio_frames[-1] - frame_duration = len(fr.audio) / 2 * fr.num_channels / fr.sample_rate - - logger.debug( - f"!!! Frame duration: ({len(fr.audio)}) ({fr.num_channels}) ({fr.sample_rate}) {frame_duration}" - ) - data = b"".join(frame.audio for frame in self._audio_frames) logger.debug( f"Processing audio buffer seconds: ({len(self._audio_frames)}) ({len(data)}) {len(data) / 2 / 16000}" @@ -415,6 +420,12 @@ class AudioAccumulator(FrameProcessor): class CompletenessCheck(FrameProcessor): + """Checks the result of the classifier LLM to determine if the user has finished speaking. + + Triggers the notifier if the user has finished speaking. Also triggers the notifier if an + idle timeout is reached. + """ + wait_time = 5.0 def __init__(self, notifier: BaseNotifier, audio_accumulator: AudioAccumulator, **kwargs): @@ -427,12 +438,13 @@ class CompletenessCheck(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - if isinstance(frame, TextFrame) and frame.text.startswith("YES"): + if isinstance(frame, UserStartedSpeakingFrame): + if self._idle_task: + self._idle_task.cancel() + elif isinstance(frame, TextFrame) and frame.text.startswith("YES"): logger.debug("Completeness check YES") if self._idle_task: - logger.debug(f"CompletenessCheck idle wait CANCEL") self._idle_task.cancel() - self._idle_task = None await self.push_frame(UserStoppedSpeakingFrame()) await self._audio_accumulator.reset() await self._notifier.notify() @@ -443,7 +455,7 @@ class CompletenessCheck(FrameProcessor): if self._wakeup_time: self._wakeup_time = time.time() + self.wait_time else: - logger.debug("CompletenessCheck idle wait START") + # logger.debug("!!! CompletenessCheck idle wait START") self._wakeup_time = time.time() + self.wait_time self._idle_task = self.get_event_loop().create_task(self._idle_task_handler()) @@ -451,16 +463,87 @@ class CompletenessCheck(FrameProcessor): try: while time.time() < self._wakeup_time: await asyncio.sleep(0.01) - logger.debug(f"CompletenessCheck idle wait OVER") + # logger.debug(f"!!! CompletenessCheck idle wait OVER") + await self._audio_accumulator.reset() await self._notifier.notify() except asyncio.CancelledError: + # logger.debug(f"!!! CompletenessCheck idle wait CANCEL") pass except Exception as e: logger.error(f"CompletenessCheck idle wait error: {e}") raise e + finally: + # logger.debug(f"!!! CompletenessCheck idle wait FINALLY") + self._wakeup_time = 0 + self._idle_task = None + + +class UserAggregatorBuffer(LLMResponseAggregator): + """Buffers the output of the transcription LLM. Used by the bot output gate.""" + + def __init__(self, **kwargs): + super().__init__( + messages=None, + role=None, + start_frame=LLMFullResponseStartFrame, + end_frame=LLMFullResponseEndFrame, + accumulator_frame=TextFrame, + handle_interruptions=True, + expect_stripped_words=False, + ) + self._transcription = "" + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + # parent method pushes frames + if isinstance(frame, UserStartedSpeakingFrame): + self._transcription = "" + + async def _push_aggregation(self): + if self._aggregation: + self._transcription = self._aggregation + self._aggregation = "" + + logger.debug(f"[Transcription] {self._transcription}") + + async def wait_for_transcription(self): + while not self._transcription: + await asyncio.sleep(0.01) + tx = self._transcription + self._transcription = "" + return tx + + +class ConversationAudioContextAssembler(FrameProcessor): + """Takes the single-message context generated by the AudioAccumulator and adds it to the conversation LLM's context.""" + + def __init__(self, context: OpenAILLMContext, **kwargs): + super().__init__(**kwargs) + self._context = context + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + # We must not block system frames. + if isinstance(frame, SystemFrame): + await self.push_frame(frame, direction) + return + + if isinstance(frame, OpenAILLMContextFrame): + GoogleLLMContext.upgrade_to_google(self._context) + last_message = frame.context.messages[-1] + self._context._messages.append(last_message) + await self.push_frame(OpenAILLMContextFrame(context=self._context)) class OutputGate(FrameProcessor): + """Buffers output frames until the notifier is triggered. + + When the notifier fires, waits until a transcription is ready, then: + 1. Replaces the last user audio message with the transcription. + 2. Flushes the frames buffer. + """ + def __init__( self, notifier: BaseNotifier, @@ -501,6 +584,13 @@ class OutputGate(FrameProcessor): await self.push_frame(frame, direction) return + if isinstance(frame, LLMFullResponseStartFrame): + # Remove the audio message from the context. We will never need it again. + # If the completeness check fails, a new audio message will be appended to the context. + # If the completeness check succeeds, our notifier will fire and we will append the + # transcription to the context. + self._context._messages.pop() + if self._gate_open: await self.push_frame(frame, direction) return @@ -517,16 +607,13 @@ class OutputGate(FrameProcessor): async def _gate_task_handler(self): while True: - # logger.debug("!!! Waiting for notifier") try: await self._notifier.wait() - # logger.debug("!!! Notified") - transcription = await self._transcription_buffer.wait_for_transcription() - - last_message = self._context.messages[-1] - if last_message.role == "user": - last_message.parts = [glm.Part(text=transcription)] + transcription = await self._transcription_buffer.wait_for_transcription() or "-" + self._context._messages.append( + glm.Content(role="user", parts=[glm.Part(text=transcription)]) + ) self.open_gate() for frame, direction in self._frames_buffer: @@ -540,54 +627,6 @@ class OutputGate(FrameProcessor): break -class ConversationAudioContextAssembler(FrameProcessor): - def __init__(self, context: OpenAILLMContext, **kwargs): - super().__init__(**kwargs) - self._context = context - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # We must not block system frames. - if isinstance(frame, SystemFrame): - await self.push_frame(frame, direction) - return - - if isinstance(frame, OpenAILLMContextFrame): - GoogleLLMContext.upgrade_to_google(self._context) - last_message = frame.context.messages[-1] - self._context._messages.append(last_message) - await self.push_frame(OpenAILLMContextFrame(context=self._context)) - - -class UserAggregatorBuffer(LLMResponseAggregator): - def __init__(self, **kwargs): - super().__init__( - messages=None, - role=None, - start_frame=LLMFullResponseStartFrame, - end_frame=LLMFullResponseEndFrame, - accumulator_frame=TextFrame, - handle_interruptions=True, - expect_stripped_words=False, - ) - self._transcription = "" - - async def _push_aggregation(self): - if self._aggregation: - self._transcription = self._aggregation - self._aggregation = "" - - logger.debug(f"[Transcription] {self._transcription}") - - async def wait_for_transcription(self): - while not self._transcription: - await asyncio.sleep(0.01) - tx = self._transcription - self._transcription = "" - return tx - - async def main(): async with aiohttp.ClientSession() as session: (room_url, _) = await configure(session) @@ -613,7 +652,7 @@ async def main(): # This is the LLM that will transcribe user speech. tx_llm = GoogleLLMService( name="Transcriber", - model="gemini-2.0-flash-exp", + model=TRANSCRIBER_MODEL, api_key=os.getenv("GOOGLE_API_KEY"), temperature=0.0, system_instruction=transcriber_system_instruction, @@ -622,7 +661,7 @@ async def main(): # This is the LLM that will classify user speech as complete or incomplete. classifier_llm = GoogleLLMService( name="Classifier", - model="gemini-2.0-flash-exp", + model=CLASSIFIER_MODEL, api_key=os.getenv("GOOGLE_API_KEY"), temperature=0.0, system_instruction=classifier_system_instruction, @@ -631,7 +670,7 @@ async def main(): # This is the regular LLM that responds conversationally. conversation_llm = GoogleLLMService( name="Conversation", - model="gemini-2.0-flash-exp", + model=CONVERSATION_MODEL, api_key=os.getenv("GOOGLE_API_KEY"), system_instruction=conversation_system_instruction, ) diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 091c4df06..d724d2776 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -635,7 +635,7 @@ class GoogleLLMService(LLMService): messages = context.messages if context.system_message and self._system_instruction != context.system_message: - # logger.debug(f"System instruction changed: {context.system_message}") + logger.debug(f"System instruction changed: {context.system_message}") self._system_instruction = context.system_message self._create_client() @@ -673,15 +673,16 @@ class GoogleLLMService(LLMService): await self.stop_ttfb_metrics() if response.usage_metadata: + # Use only the prompt token count from the response object prompt_tokens = response.usage_metadata.prompt_token_count - completion_tokens = response.usage_metadata.candidates_token_count - total_tokens = response.usage_metadata.total_token_count + total_tokens = prompt_tokens async for chunk in response: if chunk.usage_metadata: - prompt_tokens += response.usage_metadata.prompt_token_count - completion_tokens += response.usage_metadata.candidates_token_count - total_tokens += response.usage_metadata.total_token_count + # Use only the completion_tokens from the chunks. Prompt tokens are already counted and + # are repeated here. + completion_tokens += chunk.usage_metadata.candidates_token_count + total_tokens += chunk.usage_metadata.candidates_token_count try: for c in chunk.parts: if c.text: From a5e985094b839baafdd2e3cd5d26cfaa0e074ce4 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sun, 22 Dec 2024 19:45:57 -0800 Subject: [PATCH 7/8] remove stray line --- src/pipecat/services/gemini_multimodal_live/gemini.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 55467a1cd..1fded9e1a 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -177,7 +177,6 @@ class GeminiMultimodalLiveLLMService(LLMService): self._receive_task = None self._context = None - self._connected = False self._disconnecting = False self._api_session_ready = False self._run_llm_when_api_session_ready = False From 1368d3db5ceb489deb94563f6ba588326cf938b3 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 23 Dec 2024 17:33:59 -0800 Subject: [PATCH 8/8] revert elevenlabs example changes --- examples/foundational/07d-interruptible-elevenlabs.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/examples/foundational/07d-interruptible-elevenlabs.py b/examples/foundational/07d-interruptible-elevenlabs.py index 9d91081d8..8f3895d00 100644 --- a/examples/foundational/07d-interruptible-elevenlabs.py +++ b/examples/foundational/07d-interruptible-elevenlabs.py @@ -48,7 +48,6 @@ async def main(): tts = ElevenLabsTTSService( api_key=os.getenv("ELEVENLABS_API_KEY", ""), voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), - model="eleven_flash_v2_5", ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") @@ -80,7 +79,7 @@ async def main(): allow_interruptions=True, enable_metrics=True, enable_usage_metrics=True, - # report_only_initial_ttfb=True, + report_only_initial_ttfb=True, ), )