Spaces:
Running
Running
File size: 2,258 Bytes
53ea588 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 |
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Blingfire-based text aggregator for sentence detection."""
import blingfire as bf
from loguru import logger
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
class BlingfireTextAggregator(BaseTextAggregator):
"""This is a text aggregator that uses blingfire for sentence detection.
It aggregates text until a complete sentence is detected using blingfire's
sentence segmentation capabilities.
"""
def __init__(self):
"""Initialize the BlingfireTextAggregator with an empty text buffer."""
self._text = ""
logger.debug("BlingfireTextAggregator: Initialized new instance")
@property
def text(self) -> str:
"""Return the currently aggregated text.
Returns:
str: The text currently being aggregated.
"""
return self._text
async def aggregate(self, text: str) -> str | None:
"""Aggregate text and return a complete sentence when detected.
Args:
text (str): The text to be aggregated.
Returns:
str | None: A complete sentence if one is detected, otherwise None.
"""
result: str | None = None
self._text += text
logger.debug(f"BlingfireTextAggregator: Aggregating text: '{self._text}'")
# Use blingfire to split text into sentences
sentences_text = bf.text_to_sentences(self._text)
sentences = [s.strip() for s in sentences_text.split("\n") if s.strip()]
# If we have multiple sentences, return the first complete one
if len(sentences) > 1:
result = sentences[0]
if self._text.lstrip().startswith(sentences[0].strip()):
self._text = self._text[len(sentences[0]) :]
return result
async def handle_interruption(self):
"""Handle interruption by clearing the aggregated text buffer."""
logger.debug("BlingfireTextAggregator: Handling interruption")
self._text = ""
async def reset(self):
"""Reset the aggregated text buffer to empty."""
logger.debug("BlingfireTextAggregator: Resetting text buffer")
self._text = ""
|