eou_detection

eou_detection

MultilingualModel

eou_detection

# Example: EOU detection using LiveKit turn detector (no AgentSession)
# Note: This uses the MultilingualModel from livekit.plugins.turn_detector.multilingual
# Make sure you have the livekit plugins package installed and accessible.

from typing import List, Dict, Tuple
from livekit.plugins.turn_detector.multilingual import MultilingualModel

class EouDetector:
    def __init__(self, threshold: float = 0.7, window_size: int = 4):
        """
        threshold: confidence threshold to treat model output as EOU
        window_size: number of recent turns to keep for context (default 4)
        """
        self.model = MultilingualModel()
        self.threshold = threshold
        self.window_size = window_size

        # conversation history as list of (speaker, text) for last N turns
        self.history: List[Tuple[str, str]] = []

        # buffer for current speaker partials (accumulate until they become a final segment)
        self.current_buffers: Dict[str, str] = {}

    def _update_history_with_final(self, speaker: str, text: str):
        # Append finalized turn and truncate to window_size
        self.history.append((speaker, text))
        if len(self.history) > self.window_size:
            self.history = self.history[-self.window_size :]

    def _build_context_for_model(self, incoming_speaker: str, incoming_text: str) -> List[Dict]:
        """
        Build the model input context: last turns plus the current trailing partial.
        The exact input shape will depend on the model API; adapt as needed.
        Here we prepare a list of turn dicts: {"speaker": ..., "text": ...}
        """
        ctx = [{"speaker": s, "text": t} for (s, t) in self.history]
        # include the current speaker's in-progress text as the trailing turn
        ctx.append({"speaker": incoming_speaker, "text": incoming_text})
        return ctx

    def on_transcript_update(self, speaker: str, text: str, is_final: bool) -> Dict:
        """
        Called for every transcript update from STT.
        - speaker: identifier for the speaker (e.g., "user" or user id)
        - text: newest transcript text (interim or final)
        - is_final: whether this transcript segment is finalized by STT
        Returns a dict containing:
        - eou_detection: bool
        - confidence: float from model
        - reason: optional string describing decision
        """
        # Update current buffer for the speaker
        if is_final:
            # Finalized: commit the full text for this speaker as a turn
            full_text = (self.current_buffers.get(speaker, "") + " " + text).strip() if self.current_buffers.get(speaker) else text
            self.current_buffers[speaker] = ""  # clear buffer
            self._update_history_with_final(speaker, full_text)
            incoming_text = full_text
        else:
            # Interim: update buffer but don't add to history yet
            existing = self.current_buffers.get(speaker, "")
            combined = (existing + " " + text).strip() if existing else text
            self.current_buffers[speaker] = combined
            incoming_text = combined

        # Build context including trailing partial/final for incoming speaker
        context = self._build_context_for_model(speaker, incoming_text)

        # Call the model - MultilingualModel API will accept the context and return a confidence
        # Adjust this call to match the plugin's actual method signature.
        # For example purpose we assume `predict` returns a dict with 'confidence' float in [0,1].
        result = self.model.predict(context)  # adapt if method name is different
        confidence = float(result.get("confidence", 0.0))

        # Map confidence to boolean EOU detection
        is_eou = confidence >= self.threshold

        # Optionally, add simple business rules to avoid false positives for structured data
        # For example, if trailing text looks like a phone number and is not final, suppress EOU
        # (Implement heuristics here as needed.)

        return {
            "eou_detection": is_eou,
            "confidence": confidence,
            "context": context,
            "is_final_segment": is_final,
        }

# Usage example:
detector = EouDetector(threshold=0.75)

# Interim partial arrives
resp = detector.on_transcript_update(speaker="user_1", text="my phone number is 555", is_final=False)
print(resp)

# Interim updated with more words
resp = detector.on_transcript_update(speaker="user_1", text="555 1234 and", is_final=False)
print(resp)

# Finalized segment comes from STT
resp = detector.on_transcript_update(speaker="user_1", text="my phone number is 555 1234 and", is_final=True)
print(resp)

MultilingualModel