diff --git a/examples/foundational/49a-thinking-anthropic.py b/examples/foundational/49a-thinking-anthropic.py index 45066527e..80477ff14 100644 --- a/examples/foundational/49a-thinking-anthropic.py +++ b/examples/foundational/49a-thinking-anthropic.py @@ -12,16 +12,16 @@ from loguru import logger from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import LLMRunFrame, ThoughtTranscriptionMessage, TranscriptionMessage +from pipecat.frames.frames import LLMRunFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( + AssistantThoughtMessage, LLMContextAggregatorPair, LLMUserAggregatorParams, ) -from pipecat.processors.transcript_processor import TranscriptProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.anthropic.llm import AnthropicLLMService @@ -74,8 +74,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) - transcript = TranscriptProcessor(process_thoughts=True) - messages = [ { "role": "system", @@ -93,17 +91,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) + user_aggregator = context_aggregator.user() + assistant_aggregator = context_aggregator.assistant() + pipeline = Pipeline( [ transport.input(), # Transport user input stt, - transcript.user(), # User transcripts - context_aggregator.user(), # User responses + user_aggregator, # User responses llm, # LLM tts, # TTS transport.output(), # Transport bot output - transcript.assistant(), # Assistant transcripts (including thoughts) - context_aggregator.assistant(), # Assistant spoken responses + assistant_aggregator, # Assistant spoken responses ] ) @@ -143,14 +142,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Client disconnected") await task.cancel() - # Register event handler for transcript updates - @transcript.event_handler("on_transcript_update") - async def on_transcript_update(processor, frame): - for msg in frame.messages: - if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)): - timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" - role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role - logger.info(f"Transcript: {timestamp}{role}: {msg.content}") + @assistant_aggregator.event_handler("on_assistant_thought") + async def on_assistant_thought(aggregator, message: AssistantThoughtMessage): + logger.info(f"Thought (timestamp: {message.timestamp}): {message.content}") runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) diff --git a/examples/foundational/49b-thinking-google.py b/examples/foundational/49b-thinking-google.py index d3aa6134f..e0c9dcd6e 100644 --- a/examples/foundational/49b-thinking-google.py +++ b/examples/foundational/49b-thinking-google.py @@ -9,20 +9,19 @@ import os from dotenv import load_dotenv from loguru import logger -from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import LLMRunFrame, ThoughtTranscriptionMessage, TranscriptionMessage +from pipecat.frames.frames import LLMRunFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( + AssistantThoughtMessage, LLMContextAggregatorPair, LLMUserAggregatorParams, ) -from pipecat.processors.transcript_processor import TranscriptProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService @@ -80,8 +79,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) - transcript = TranscriptProcessor(process_thoughts=True) - messages = [ { "role": "system", @@ -99,17 +96,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) + user_aggregator = context_aggregator.user() + assistant_aggregator = context_aggregator.assistant() + pipeline = Pipeline( [ transport.input(), # Transport user input stt, - transcript.user(), # User transcripts - context_aggregator.user(), # User responses + user_aggregator, # User responses llm, # LLM tts, # TTS transport.output(), # Transport bot output - transcript.assistant(), # Assistant transcripts (including thoughts) - context_aggregator.assistant(), # Assistant spoken responses + assistant_aggregator, # Assistant spoken responses ] ) @@ -150,14 +148,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Client disconnected") await task.cancel() - # Register event handler for transcript updates - @transcript.event_handler("on_transcript_update") - async def on_transcript_update(processor, frame): - for msg in frame.messages: - if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)): - timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" - role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role - logger.info(f"Transcript: {timestamp}{role}: {msg.content}") + @assistant_aggregator.event_handler("on_assistant_thought") + async def on_assistant_thought(aggregator, message: AssistantThoughtMessage): + logger.info(f"Thought (timestamp: {message.timestamp}): {message.content}") runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) diff --git a/examples/foundational/49c-thinking-functions-anthropic.py b/examples/foundational/49c-thinking-functions-anthropic.py index 93fc33524..6480c7390 100644 --- a/examples/foundational/49c-thinking-functions-anthropic.py +++ b/examples/foundational/49c-thinking-functions-anthropic.py @@ -10,20 +10,19 @@ from dotenv import load_dotenv from loguru import logger from pipecat.adapters.schemas.tools_schema import ToolsSchema -from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import LLMRunFrame, ThoughtTranscriptionMessage, TranscriptionMessage +from pipecat.frames.frames import LLMRunFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( + AssistantThoughtMessage, LLMContextAggregatorPair, LLMUserAggregatorParams, ) -from pipecat.processors.transcript_processor import TranscriptProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.anthropic.llm import AnthropicLLMService @@ -101,8 +100,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): tools = ToolsSchema(standard_tools=[check_flight_status, book_taxi]) - transcript = TranscriptProcessor(process_thoughts=True) - messages = [ { "role": "system", @@ -120,17 +117,17 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) + user_aggregator = context_aggregator.user() + assistant_aggregator = context_aggregator.assistant() + pipeline = Pipeline( [ transport.input(), # Transport user input stt, - transcript.user(), # User transcripts - context_aggregator.user(), # User responses + user_aggregator, # User responses llm, # LLM tts, # TTS - transport.output(), # Transport bot output - transcript.assistant(), # Assistant transcripts (including thoughts) - context_aggregator.assistant(), # Assistant spoken responses + assistant_aggregator, # Assistant spoken responses ] ) @@ -169,13 +166,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Client disconnected") await task.cancel() - @transcript.event_handler("on_transcript_update") - async def on_transcript_update(processor, frame): - for msg in frame.messages: - if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)): - timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" - role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role - logger.info(f"Transcript: {timestamp}{role}: {msg.content}") + @assistant_aggregator.event_handler("on_assistant_thought") + async def on_assistant_thought(aggregator, message: AssistantThoughtMessage): + logger.info(f"Thought (timestamp: {message.timestamp}): {message.content}") runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) diff --git a/examples/foundational/49d-thinking-functions-google.py b/examples/foundational/49d-thinking-functions-google.py index 356fe806b..794d4e4f7 100644 --- a/examples/foundational/49d-thinking-functions-google.py +++ b/examples/foundational/49d-thinking-functions-google.py @@ -10,7 +10,6 @@ from dotenv import load_dotenv from loguru import logger from pipecat.adapters.schemas.tools_schema import ToolsSchema -from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams @@ -20,6 +19,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( + AssistantThoughtMessage, LLMContextAggregatorPair, LLMUserAggregatorParams, ) @@ -106,8 +106,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): tools = ToolsSchema(standard_tools=[check_flight_status, book_taxi]) - transcript = TranscriptProcessor(process_thoughts=True) - messages = [ { "role": "system", @@ -125,17 +123,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) + user_aggregator = context_aggregator.user() + assistant_aggregator = context_aggregator.assistant() + pipeline = Pipeline( [ transport.input(), # Transport user input stt, - transcript.user(), # User transcripts - context_aggregator.user(), # User responses + user_aggregator, # User responses llm, # LLM tts, # TTS transport.output(), # Transport bot output - transcript.assistant(), # Assistant transcripts (including thoughts) - context_aggregator.assistant(), # Assistant spoken responses + assistant_aggregator, # Assistant spoken responses ] ) @@ -174,13 +173,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Client disconnected") await task.cancel() - @transcript.event_handler("on_transcript_update") - async def on_transcript_update(processor, frame): - for msg in frame.messages: - if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)): - timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" - role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role - logger.info(f"Transcript: {timestamp}{role}: {msg.content}") + @assistant_aggregator.event_handler("on_assistant_thought") + async def on_assistant_thought(aggregator, message: AssistantThoughtMessage): + logger.info(f"Thought (timestamp: {message.timestamp}): {message.content}") runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)