From 61674d7758e6b4940623acd1f766f025db37b07e Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 8 Dec 2025 10:27:36 -0500 Subject: [PATCH] Add `process_thought` constructor argument to `TranscriptProcessor` to control whether to handle thoughts in addition to assistant utterances. Defaults to `False`. --- .../foundational/49-thinking-functions.py | 2 +- examples/foundational/49-thinking.py | 2 +- .../processors/transcript_processor.py | 33 ++++++++++++------- tests/test_transcript_processor.py | 16 ++++----- 4 files changed, 32 insertions(+), 21 deletions(-) diff --git a/examples/foundational/49-thinking-functions.py b/examples/foundational/49-thinking-functions.py index d24adbf3e..411316efe 100644 --- a/examples/foundational/49-thinking-functions.py +++ b/examples/foundational/49-thinking-functions.py @@ -125,7 +125,7 @@ async def run_bot( tools = ToolsSchema(standard_tools=[check_flight_status, book_taxi]) - transcript = TranscriptProcessor() + transcript = TranscriptProcessor(process_thoughts=True) messages = [ { diff --git a/examples/foundational/49-thinking.py b/examples/foundational/49-thinking.py index 756134076..74da17c66 100644 --- a/examples/foundational/49-thinking.py +++ b/examples/foundational/49-thinking.py @@ -99,7 +99,7 @@ async def run_bot( else: raise ValueError(f"Unsupported LLM provider: {llm_provider}") - transcript = TranscriptProcessor() + transcript = TranscriptProcessor(process_thoughts=True) messages = [ { diff --git a/src/pipecat/processors/transcript_processor.py b/src/pipecat/processors/transcript_processor.py index 150d6c5c8..0dd59f1b3 100644 --- a/src/pipecat/processors/transcript_processor.py +++ b/src/pipecat/processors/transcript_processor.py @@ -101,14 +101,16 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor): - The pipeline ends (EndFrame, CancelFrame) """ - def __init__(self, **kwargs): + def __init__(self, *, process_thoughts: bool = False, **kwargs): """Initialize processor with aggregation state. Args: + process_thoughts: Whether to process LLM thought frames. Defaults to False. **kwargs: Additional arguments passed to parent class. """ super().__init__(**kwargs) + self._process_thoughts = process_thoughts self._current_assistant_text_parts: List[TextPartForConcatenation] = [] self._assistant_text_start_time: Optional[str] = None @@ -188,18 +190,19 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor): await self.push_frame(frame, direction) # Emit accumulated text and thought with interruptions await self._emit_aggregated_assistant_text() - if self._thought_active: + if self._process_thoughts and self._thought_active: await self._emit_aggregated_thought() elif isinstance(frame, LLMThoughtStartFrame): # Start a new thought - self._thought_active = True - self._thought_start_time = time_now_iso8601() - self._current_thought_parts = [] + if self._process_thoughts: + self._thought_active = True + self._thought_start_time = time_now_iso8601() + self._current_thought_parts = [] # Push frame. await self.push_frame(frame, direction) elif isinstance(frame, LLMThoughtTextFrame): # Aggregate thought text if we have an active thought - if self._thought_active: + if self._process_thoughts and self._thought_active: self._current_thought_parts.append( TextPartForConcatenation( frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces @@ -209,7 +212,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor): await self.push_frame(frame, direction) elif isinstance(frame, LLMThoughtEndFrame): # Emit accumulated thought when thought ends - if self._thought_active: + if self._process_thoughts and self._thought_active: await self._emit_aggregated_thought() # Push frame. await self.push_frame(frame, direction) @@ -230,7 +233,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor): # Emit accumulated text when bot finishes speaking or pipeline ends. await self._emit_aggregated_assistant_text() # Emit accumulated thought at pipeline end if still active - if isinstance(frame, EndFrame) and self._thought_active: + if isinstance(frame, EndFrame) and self._process_thoughts and self._thought_active: await self._emit_aggregated_thought() # Push frame. await self.push_frame(frame, direction) @@ -268,8 +271,14 @@ class TranscriptProcessor: print(f"New messages: {frame.messages}") """ - def __init__(self): - """Initialize factory.""" + def __init__(self, *, process_thoughts: bool = False): + """Initialize factory. + + Args: + process_thoughts: Whether the assistant processor should handle LLM thought + frames. Defaults to False. + """ + self._process_thoughts = process_thoughts self._user_processor = None self._assistant_processor = None self._event_handlers = {} @@ -304,7 +313,9 @@ class TranscriptProcessor: The assistant transcript processor instance. """ if self._assistant_processor is None: - self._assistant_processor = AssistantTranscriptProcessor(**kwargs) + self._assistant_processor = AssistantTranscriptProcessor( + process_thoughts=self._process_thoughts, **kwargs + ) # Apply any registered event handlers for event_name, handler in self._event_handlers.items(): diff --git a/tests/test_transcript_processor.py b/tests/test_transcript_processor.py index a5e0702a0..c8e15eb24 100644 --- a/tests/test_transcript_processor.py +++ b/tests/test_transcript_processor.py @@ -496,7 +496,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): async def test_basic_thought_transcription(self): """Test basic thought frame processing""" - processor = AssistantTranscriptProcessor() + processor = AssistantTranscriptProcessor(process_thoughts=True) received_updates: List[TranscriptionUpdateFrame] = [] @@ -533,7 +533,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): async def test_thought_aggregation(self): """Test that thought text frames are properly aggregated""" - processor = AssistantTranscriptProcessor() + processor = AssistantTranscriptProcessor(process_thoughts=True) received_updates: List[TranscriptionUpdateFrame] = [] @@ -575,7 +575,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): async def test_thought_with_interruption(self): """Test that thoughts are properly captured when interrupted""" - processor = AssistantTranscriptProcessor() + processor = AssistantTranscriptProcessor(process_thoughts=True) received_updates: List[TranscriptionUpdateFrame] = [] @@ -613,7 +613,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): async def test_thought_with_cancel(self): """Test that thoughts are properly captured when cancelled""" - processor = AssistantTranscriptProcessor() + processor = AssistantTranscriptProcessor(process_thoughts=True) received_updates: List[TranscriptionUpdateFrame] = [] @@ -649,7 +649,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): async def test_thought_with_end_frame(self): """Test that thoughts are captured when pipeline ends normally""" - processor = AssistantTranscriptProcessor() + processor = AssistantTranscriptProcessor(process_thoughts=True) received_updates: List[TranscriptionUpdateFrame] = [] @@ -683,7 +683,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): async def test_multiple_thoughts(self): """Test multiple separate thoughts in sequence""" - processor = AssistantTranscriptProcessor() + processor = AssistantTranscriptProcessor(process_thoughts=True) received_updates: List[TranscriptionUpdateFrame] = [] @@ -735,7 +735,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): async def test_empty_thought_handling(self): """Test that empty thoughts are not emitted""" - processor = AssistantTranscriptProcessor() + processor = AssistantTranscriptProcessor(process_thoughts=True) received_updates: List[TranscriptionUpdateFrame] = [] @@ -768,7 +768,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): async def test_thought_without_start_frame(self): """Test that thought text without start frame is ignored""" - processor = AssistantTranscriptProcessor() + processor = AssistantTranscriptProcessor(process_thoughts=True) received_updates: List[TranscriptionUpdateFrame] = []