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

Code review fixes

Browse Source
This commit is contained in:
Mark Backman
2025-01-20 20:23:07 -05:00
parent c2fe8e7fdb
commit e1430be9f9
1 changed files with 7 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/pipecat/processors/transcript_processor.py
View File

@@ -4,8 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License # SPDX-License-Identifier: BSD 2-Clause License
# #
from datetime import datetime, timezone from typing import List, Optional
from typing import List
from loguru import logger from loguru import logger
@@ -20,6 +19,7 @@ from pipecat.frames.frames import (
TTSTextFrame, TTSTextFrame,
) )
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.time import time_now_iso8601
class BaseTranscriptProcessor(FrameProcessor): class BaseTranscriptProcessor(FrameProcessor):
@@ -86,22 +86,18 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
"""Initialize processor with aggregation state.""" """Initialize processor with aggregation state."""
super().__init__(**kwargs) super().__init__(**kwargs)
self._current_text_parts: List[str] = [] self._current_text_parts: List[str] = []
self._aggregation_start_time: datetime | None = None self._aggregation_start_time: Optional[str] | None = None
async def _emit_aggregated_text(self): async def _emit_aggregated_text(self):
"""Emit aggregated text as a transcript message.""" """Emit aggregated text as a transcript message."""
if self._current_text_parts and self._aggregation_start_time: if self._current_text_parts and self._aggregation_start_time:
content = " ".join(self._current_text_parts).strip() content = " ".join(self._current_text_parts).strip()
if content: if content:
# Format timestamp with 3 decimal places
formatted_timestamp = (
self._aggregation_start_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "+00:00"
)
logger.debug(f"Emitting aggregated assistant message: {content}") logger.debug(f"Emitting aggregated assistant message: {content}")
message = TranscriptionMessage( message = TranscriptionMessage(
role="assistant", role="assistant",
content=content, content=content,
timestamp=formatted_timestamp, timestamp=self._aggregation_start_time,
) )
await self._emit_update([message]) await self._emit_update([message])
else: else:
@@ -129,16 +125,12 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
if isinstance(frame, TTSTextFrame): if isinstance(frame, TTSTextFrame):
# Start timestamp on first text part # Start timestamp on first text part
if not self._aggregation_start_time: if not self._aggregation_start_time:
self._aggregation_start_time = datetime.now(timezone.utc) self._aggregation_start_time = time_now_iso8601()
self._current_text_parts.append(frame.text) self._current_text_parts.append(frame.text)
elif isinstance(frame, BotStoppedSpeakingFrame): elif isinstance(frame, (BotStoppedSpeakingFrame, StartInterruptionFrame)):
# Emit accumulated text when bot finishes speaking # Emit accumulated text when bot finishes speaking or is interrupted
await self._emit_aggregated_text()
elif isinstance(frame, StartInterruptionFrame):
# Emit any pending text when interrupted
await self._emit_aggregated_text() await self._emit_aggregated_text()
elif isinstance(frame, EndFrame): elif isinstance(frame, EndFrame):