wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "Merge pull request #3004 from pipecat-ai/mb/improve-concat-aggregated-text"

Browse Source 
This reverts commit 5e7f59a0b0, reversing
changes made to 2ad4122b77.
This commit is contained in:
Mark Backman
2025-11-10 09:55:46 -05:00
parent 16e2d5b998
commit 2300941bb8
3 changed files with 6 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
tests/test_transcript_processor.py
View File

@@ -479,103 +479,3 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
self.assertEqual(message.role, "assistant")
# Should be properly joined without extra spaces
self.assertEqual(message.content, "Hello there! How's it going?")
async def test_openai_realtime_syllable_fragments(self):
"""Test OpenAI Realtime syllable-by-syllable output with standalone punctuation
OpenAI Realtime can output single words as syllable fragments with punctuation
as a separate fragment. Example: ["Met", "am", "orph", "osis", "."]
This should be concatenated without spaces to form "Metamorphosis."
"""
processor = AssistantTranscriptProcessor()
received_updates = []
@processor.event_handler("on_transcript_update")
async def handle_update(proc, frame: TranscriptionUpdateFrame):
received_updates.append(frame)
# Simulate OpenAI Realtime syllable-by-syllable output
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(),
TTSTextFrame(text="Met"),
TTSTextFrame(text="am"),
TTSTextFrame(text="orph"),
TTSTextFrame(text="osis"),
TTSTextFrame(text="."), # Standalone punctuation fragment
BotStoppedSpeakingFrame(),
]
expected_down_frames = [
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
TTSTextFrame,
TTSTextFrame,
TTSTextFrame,
TTSTextFrame,
TTSTextFrame,
TranscriptionUpdateFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
)
# Verify syllables are concatenated without spaces
self.assertEqual(len(received_updates), 1)
message = received_updates[0].messages[0]
self.assertEqual(message.role, "assistant")
self.assertEqual(message.content, "Metamorphosis.")
async def test_gemini_live_syllable_fragments_with_newline(self):
"""Test Gemini Live syllable-by-syllable output with trailing newline
Gemini Live can output syllable fragments where the last fragment contains
trailing whitespace like newlines. Example: ["Met", "amo", "rph", "osi", "s.\\n"]
This should be concatenated without spaces to form "Metamorphosis."
"""
processor = AssistantTranscriptProcessor()
received_updates = []
@processor.event_handler("on_transcript_update")
async def handle_update(proc, frame: TranscriptionUpdateFrame):
received_updates.append(frame)
# Simulate Gemini Live syllable-by-syllable output with trailing newline
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(),
TTSTextFrame(text="Met"),
TTSTextFrame(text="amo"),
TTSTextFrame(text="rph"),
TTSTextFrame(text="osi"),
TTSTextFrame(text="s.\n"), # Last fragment with trailing newline
BotStoppedSpeakingFrame(),
]
expected_down_frames = [
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
TTSTextFrame,
TTSTextFrame,
TTSTextFrame,
TTSTextFrame,
TTSTextFrame,
TranscriptionUpdateFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
)
# Verify syllables are concatenated without spaces and newline is stripped
self.assertEqual(len(received_updates), 1)
message = received_updates[0].messages[0]
self.assertEqual(message.role, "assistant")
self.assertEqual(message.content, "Metamorphosis.")