examples(foundational): update 49 series with on_assistant_thought

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)
 

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)
 

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)
 

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)