Module agents.eou

Classes

class EOU (threshold: float = 0.7)

Expand source code

class EOU(EventEmitter[Literal["error"]]):
    """Base class for End of Utterance Detection implementations"""
    
    def __init__(self, threshold: float = 0.7) -> None:
        super().__init__()
        self._label = f"{type(self).__module__}.{type(self).__name__}"
        self._threshold = threshold

    @property
    def label(self) -> str:
        """Get the EOU provider label"""
        return self._label

    @property
    def threshold(self) -> float:
        """Get the EOU detection threshold"""
        return self._threshold

    @abstractmethod
    def get_eou_probability(self, chat_context: ChatContext) -> float:
        """
        Get the probability score for end of utterance detection.
        
        Args:
            chat_context: Chat context to analyze
            
        Returns:
            float: Probability score (0.0 to 1.0)
        """
        raise NotImplementedError

    def detect_end_of_utterance(self, chat_context: ChatContext, threshold: Optional[float] = None) -> bool:
        """
        Detect if the given chat context represents an end of utterance.
        
        Args:
            chat_context: Chat context to analyze
            threshold: Optional threshold override
            
        Returns:
            bool: True if end of utterance is detected, False otherwise
        """
        if threshold is None:
            threshold = self._threshold
        
        probability = self.get_eou_probability(chat_context)
        return probability >= threshold

    def set_threshold(self, threshold: float) -> None:
        """Update the EOU detection threshold"""
        self._threshold = threshold
    
    async def aclose(self) -> None:
        """Cleanup resources - should be overridden by subclasses to cleanup models"""
        logger.info(f"Cleaning up EOU: {self._label}")
        
        try:
            import gc
            gc.collect()
            logger.info(f"EOU garbage collection completed: {self._label}")
        except Exception as e:
            logger.error(f"Error during EOU garbage collection: {e}")
        
        logger.info(f"EOU cleanup completed: {self._label}")
    
    async def cleanup(self) -> None:
        """Cleanup resources - calls aclose for compatibility"""
        await self.aclose()

Base class for End of Utterance Detection implementations

Ancestors

• EventEmitter
• typing.Generic

Instance variables

prop label : str

Expand source code

@property
def label(self) -> str:
    """Get the EOU provider label"""
    return self._label

Get the EOU provider label

prop threshold : float

Expand source code

@property
def threshold(self) -> float:
    """Get the EOU detection threshold"""
    return self._threshold

Get the EOU detection threshold

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    """Cleanup resources - should be overridden by subclasses to cleanup models"""
    logger.info(f"Cleaning up EOU: {self._label}")
    
    try:
        import gc
        gc.collect()
        logger.info(f"EOU garbage collection completed: {self._label}")
    except Exception as e:
        logger.error(f"Error during EOU garbage collection: {e}")
    
    logger.info(f"EOU cleanup completed: {self._label}")

Cleanup resources - should be overridden by subclasses to cleanup models

async def cleanup(self) ‑> None

Expand source code

async def cleanup(self) -> None:
    """Cleanup resources - calls aclose for compatibility"""
    await self.aclose()

Cleanup resources - calls aclose for compatibility

def detect_end_of_utterance(self, chat_context: ChatContext, threshold: Optional[float] = None) ‑> bool

Expand source code

def detect_end_of_utterance(self, chat_context: ChatContext, threshold: Optional[float] = None) -> bool:
    """
    Detect if the given chat context represents an end of utterance.
    
    Args:
        chat_context: Chat context to analyze
        threshold: Optional threshold override
        
    Returns:
        bool: True if end of utterance is detected, False otherwise
    """
    if threshold is None:
        threshold = self._threshold
    
    probability = self.get_eou_probability(chat_context)
    return probability >= threshold

Detect if the given chat context represents an end of utterance.

Args

chat_context

Chat context to analyze

threshold

Optional threshold override

Returns

bool

True if end of utterance is detected, False otherwise

def get_eou_probability(self, chat_context: ChatContext) ‑> float

Expand source code

@abstractmethod
def get_eou_probability(self, chat_context: ChatContext) -> float:
    """
    Get the probability score for end of utterance detection.
    
    Args:
        chat_context: Chat context to analyze
        
    Returns:
        float: Probability score (0.0 to 1.0)
    """
    raise NotImplementedError

Get the probability score for end of utterance detection.

Args

chat_context

Chat context to analyze

Returns

float

Probability score (0.0 to 1.0)

def set_threshold(self, threshold: float) ‑> None

Expand source code

def set_threshold(self, threshold: float) -> None:
    """Update the EOU detection threshold"""
    self._threshold = threshold

Update the EOU detection threshold