Add process_thought constructor argument to TranscriptProcessor to control whether to handle thoughts in addition to assistant utterances. Defaults to False.

This commit is contained in:
Paul Kompfner
2025-12-08 10:27:36 -05:00
parent 8ccc2cbf31
commit 61674d7758
4 changed files with 32 additions and 21 deletions

View File

@@ -125,7 +125,7 @@ async def run_bot(
tools = ToolsSchema(standard_tools=[check_flight_status, book_taxi]) tools = ToolsSchema(standard_tools=[check_flight_status, book_taxi])
transcript = TranscriptProcessor() transcript = TranscriptProcessor(process_thoughts=True)
messages = [ messages = [
{ {

View File

@@ -99,7 +99,7 @@ async def run_bot(
else: else:
raise ValueError(f"Unsupported LLM provider: {llm_provider}") raise ValueError(f"Unsupported LLM provider: {llm_provider}")
transcript = TranscriptProcessor() transcript = TranscriptProcessor(process_thoughts=True)
messages = [ messages = [
{ {

View File

@@ -101,14 +101,16 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
- The pipeline ends (EndFrame, CancelFrame) - The pipeline ends (EndFrame, CancelFrame)
""" """
def __init__(self, **kwargs): def __init__(self, *, process_thoughts: bool = False, **kwargs):
"""Initialize processor with aggregation state. """Initialize processor with aggregation state.
Args: Args:
process_thoughts: Whether to process LLM thought frames. Defaults to False.
**kwargs: Additional arguments passed to parent class. **kwargs: Additional arguments passed to parent class.
""" """
super().__init__(**kwargs) super().__init__(**kwargs)
self._process_thoughts = process_thoughts
self._current_assistant_text_parts: List[TextPartForConcatenation] = [] self._current_assistant_text_parts: List[TextPartForConcatenation] = []
self._assistant_text_start_time: Optional[str] = None self._assistant_text_start_time: Optional[str] = None
@@ -188,18 +190,19 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
# Emit accumulated text and thought with interruptions # Emit accumulated text and thought with interruptions
await self._emit_aggregated_assistant_text() await self._emit_aggregated_assistant_text()
if self._thought_active: if self._process_thoughts and self._thought_active:
await self._emit_aggregated_thought() await self._emit_aggregated_thought()
elif isinstance(frame, LLMThoughtStartFrame): elif isinstance(frame, LLMThoughtStartFrame):
# Start a new thought # Start a new thought
self._thought_active = True if self._process_thoughts:
self._thought_start_time = time_now_iso8601() self._thought_active = True
self._current_thought_parts = [] self._thought_start_time = time_now_iso8601()
self._current_thought_parts = []
# Push frame. # Push frame.
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
elif isinstance(frame, LLMThoughtTextFrame): elif isinstance(frame, LLMThoughtTextFrame):
# Aggregate thought text if we have an active thought # 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( self._current_thought_parts.append(
TextPartForConcatenation( TextPartForConcatenation(
frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces
@@ -209,7 +212,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
elif isinstance(frame, LLMThoughtEndFrame): elif isinstance(frame, LLMThoughtEndFrame):
# Emit accumulated thought when thought ends # Emit accumulated thought when thought ends
if self._thought_active: if self._process_thoughts and self._thought_active:
await self._emit_aggregated_thought() await self._emit_aggregated_thought()
# Push frame. # Push frame.
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
@@ -230,7 +233,7 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
# Emit accumulated text when bot finishes speaking or pipeline ends. # Emit accumulated text when bot finishes speaking or pipeline ends.
await self._emit_aggregated_assistant_text() await self._emit_aggregated_assistant_text()
# Emit accumulated thought at pipeline end if still active # 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() await self._emit_aggregated_thought()
# Push frame. # Push frame.
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
@@ -268,8 +271,14 @@ class TranscriptProcessor:
print(f"New messages: {frame.messages}") print(f"New messages: {frame.messages}")
""" """
def __init__(self): def __init__(self, *, process_thoughts: bool = False):
"""Initialize factory.""" """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._user_processor = None
self._assistant_processor = None self._assistant_processor = None
self._event_handlers = {} self._event_handlers = {}
@@ -304,7 +313,9 @@ class TranscriptProcessor:
The assistant transcript processor instance. The assistant transcript processor instance.
""" """
if self._assistant_processor is None: 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 # Apply any registered event handlers
for event_name, handler in self._event_handlers.items(): for event_name, handler in self._event_handlers.items():

View File

@@ -496,7 +496,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase):
async def test_basic_thought_transcription(self): async def test_basic_thought_transcription(self):
"""Test basic thought frame processing""" """Test basic thought frame processing"""
processor = AssistantTranscriptProcessor() processor = AssistantTranscriptProcessor(process_thoughts=True)
received_updates: List[TranscriptionUpdateFrame] = [] received_updates: List[TranscriptionUpdateFrame] = []
@@ -533,7 +533,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase):
async def test_thought_aggregation(self): async def test_thought_aggregation(self):
"""Test that thought text frames are properly aggregated""" """Test that thought text frames are properly aggregated"""
processor = AssistantTranscriptProcessor() processor = AssistantTranscriptProcessor(process_thoughts=True)
received_updates: List[TranscriptionUpdateFrame] = [] received_updates: List[TranscriptionUpdateFrame] = []
@@ -575,7 +575,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase):
async def test_thought_with_interruption(self): async def test_thought_with_interruption(self):
"""Test that thoughts are properly captured when interrupted""" """Test that thoughts are properly captured when interrupted"""
processor = AssistantTranscriptProcessor() processor = AssistantTranscriptProcessor(process_thoughts=True)
received_updates: List[TranscriptionUpdateFrame] = [] received_updates: List[TranscriptionUpdateFrame] = []
@@ -613,7 +613,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase):
async def test_thought_with_cancel(self): async def test_thought_with_cancel(self):
"""Test that thoughts are properly captured when cancelled""" """Test that thoughts are properly captured when cancelled"""
processor = AssistantTranscriptProcessor() processor = AssistantTranscriptProcessor(process_thoughts=True)
received_updates: List[TranscriptionUpdateFrame] = [] received_updates: List[TranscriptionUpdateFrame] = []
@@ -649,7 +649,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase):
async def test_thought_with_end_frame(self): async def test_thought_with_end_frame(self):
"""Test that thoughts are captured when pipeline ends normally""" """Test that thoughts are captured when pipeline ends normally"""
processor = AssistantTranscriptProcessor() processor = AssistantTranscriptProcessor(process_thoughts=True)
received_updates: List[TranscriptionUpdateFrame] = [] received_updates: List[TranscriptionUpdateFrame] = []
@@ -683,7 +683,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase):
async def test_multiple_thoughts(self): async def test_multiple_thoughts(self):
"""Test multiple separate thoughts in sequence""" """Test multiple separate thoughts in sequence"""
processor = AssistantTranscriptProcessor() processor = AssistantTranscriptProcessor(process_thoughts=True)
received_updates: List[TranscriptionUpdateFrame] = [] received_updates: List[TranscriptionUpdateFrame] = []
@@ -735,7 +735,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase):
async def test_empty_thought_handling(self): async def test_empty_thought_handling(self):
"""Test that empty thoughts are not emitted""" """Test that empty thoughts are not emitted"""
processor = AssistantTranscriptProcessor() processor = AssistantTranscriptProcessor(process_thoughts=True)
received_updates: List[TranscriptionUpdateFrame] = [] received_updates: List[TranscriptionUpdateFrame] = []
@@ -768,7 +768,7 @@ class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase):
async def test_thought_without_start_frame(self): async def test_thought_without_start_frame(self):
"""Test that thought text without start frame is ignored""" """Test that thought text without start frame is ignored"""
processor = AssistantTranscriptProcessor() processor = AssistantTranscriptProcessor(process_thoughts=True)
received_updates: List[TranscriptionUpdateFrame] = [] received_updates: List[TranscriptionUpdateFrame] = []