feat: added HITL feature

Author: kanakagrawal-crypto

app/ai/voice/agents/breeze_buddy/features/hitl/__init__.py

@@ -0,0 +1,6 @@
+from app.ai.voice.agents.breeze_buddy.features.hitl.manager import (
+    BreezeBuddyHITLManager,
+    get_hitl_manager,
+)
+
+__all__ = ["BreezeBuddyHITLManager", "get_hitl_manager"]

app/ai/voice/agents/breeze_buddy/features/hitl/manager.py

@@ -0,0 +1,287 @@
+"""HITL Manager for Breeze Buddy - handles voice-native confirmation requests."""
+
+import asyncio
+import uuid
+from dataclasses import dataclass, field
+from difflib import SequenceMatcher
+from typing import Any, Dict, List, Optional
+
+from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    TranscriptionFrame,
+    TTSSpeakFrame,
+    UserStoppedSpeakingFrame,
+)
+
+from app.ai.voice.agents.breeze_buddy.template.context import TemplateContext
+from app.ai.voice.agents.breeze_buddy.template.interruption import (
+    _apply_interruption_config,
+    _get_user_aggregator,
+)
+from app.ai.voice.agents.breeze_buddy.template.types import (
+    HITLConfig,
+    InterruptionConfig,
+    InterruptionMode,
+)
+from app.core.config.static import BREEZE_BUDDY_HITL_DEFAULT_TIMEOUT
+from app.core.logger import logger
+
+
+def _similarity(a: str, b: str) -> float:
+    return SequenceMatcher(None, a, b).ratio()
+
+
+def _matches_response(text: str, keywords: List[str]) -> bool:
+    """Match a single text chunk against keywords."""
+    text = text.lower().strip()
+
+    # Fast substring match
+    for keyword in keywords:
+        if keyword.lower() in text:
+            return True
+
+    # Fuzzy match (safer thresholds)
+    for keyword in keywords:
+        kw = keyword.lower()
+        if abs(len(text) - len(kw)) <= 2:
+            threshold = 0.9 if len(kw) <= 3 else 0.8
+            if _similarity(text, kw) >= threshold:
+                return True
+
+    return False
+
+
+@dataclass
+class PendingConfirmation:
+    confirmation_id: str
+    config: HITLConfig
+    function_name: str
+    arguments: Dict[str, Any]
+    event: asyncio.Event = field(default_factory=asyncio.Event)
+    response: Optional[Dict[str, Any]] = None
+    retry_count: int = 0
+    max_retries: int = 3
+    ask_again: bool = False
+
+
+class BreezeBuddyHITLManager:
+    def __init__(self):
+        self._pending: Dict[str, PendingConfirmation] = {}
+        self._active_confirmation_id: Optional[str] = None
+
+    def is_confirmation_active(self) -> bool:
+        return self._active_confirmation_id is not None
+
+    async def consume_transcription_if_hitl(self, text: str) -> bool:
+        """
+        Returns True if text is consumed by HITL (and should NOT go to LLM).
+        """
+        confirmation_id = self._active_confirmation_id
+        if not confirmation_id:
+            return False
+
+        pending = self._pending.get(confirmation_id)
+        if not pending:
+            return False
+
+        text = text.strip()
+        if not text:
+            return True
+
+        logger.info(f"HITL: Heard '{text}'")
+
+        if _matches_response(text, pending.config.accepted_responses):
+            pending.response = {"approved": True, "reason": "user_approved"}
+            pending.event.set()
+            return True
+
+        if _matches_response(text, pending.config.rejected_responses):
+            pending.response = {"approved": False, "reason": "user_rejected"}
+            pending.event.set()
+            return True
+
+        # Unmatched response - increment retry
+        pending.retry_count += 1
+        if pending.retry_count >= pending.max_retries:
+            pending.response = {"approved": False, "reason": "max_retries_exceeded"}
+            pending.event.set()
+            logger.warning(f"HITL: Max retries exceeded for {pending.function_name}")
+            return True
+
+        # Signal to re-prompt and continue listening
+        pending.ask_again = True
+        pending.event.set()
+        return True
+
+    async def request_confirmation(
+        self,
+        context: TemplateContext,
+        config: HITLConfig,
+        function_name: str,
+        arguments: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        confirmation_id = str(uuid.uuid4())
+
+        pending = PendingConfirmation(
+            confirmation_id=confirmation_id,
+            config=config,
+            function_name=function_name,
+            arguments=arguments,
+        )
+        self._pending[confirmation_id] = pending
+        self._active_confirmation_id = confirmation_id
+
+        timeout = float(config.timeout_seconds or BREEZE_BUDDY_HITL_DEFAULT_TIMEOUT)
+        logger.info(
+            f"HITL: Requesting confirmation for {function_name}, timeout={timeout}s"
+        )
+
+        try:
+            return await self._voice_confirmation(context, pending)
+        except asyncio.TimeoutError:
+            logger.warning(f"HITL timeout for {function_name}")
+            return {"approved": False, "reason": "timeout"}
+        finally:
+            if self._active_confirmation_id == confirmation_id:
+                self._active_confirmation_id = None
+            self._cleanup(confirmation_id)
+
+    async def _voice_confirmation(
+        self,
+        context: TemplateContext,
+        pending: PendingConfirmation,
+    ) -> Dict[str, Any]:
+        task = context.task
+        config = pending.config
+
+        if not task:
+            logger.error("No task available for HITL confirmation")
+            return {"approved": False, "reason": "no_task"}
+
+        message = config.confirmation_message or self._default_message(
+            pending.function_name, pending.arguments
+        )
+
+        timeout = float(config.timeout_seconds or BREEZE_BUDDY_HITL_DEFAULT_TIMEOUT)
+
+        logger.info(
+            f"HITL: Starting confirmation for {pending.function_name}, timeout={timeout}s"
+        )
+
+        task.add_reached_upstream_filter((TranscriptionFrame, UserStoppedSpeakingFrame))
+
+        user_aggregator = _get_user_aggregator(context.bot)
+        original_config = None
+
+        if user_aggregator:
+            original_config = getattr(user_aggregator, "interruption_config", None)
+            disable_config = InterruptionConfig(mode=InterruptionMode.DISABLED_DISCARD)
+
+            await _apply_interruption_config(
+                user_aggregator,
+                disable_config,
+                has_vad=context.bot.vad_analyzer is not None,
+                call_sid=context.call_sid or "unknown",
+                label="hitl_confirmation",
+                bot=context.bot,
+                user_speech_timeout=0.0,
+            )
+            logger.info(
+                f"HITL: Interruptions disabled (user_speech_timeout=0.0) for {timeout}s confirmation window"
+            )
+
+        # Two-phase TTS wait: BotStarted → BotStopped for THIS confirmation prompt
+        tts_complete = asyncio.Event()
+        bot_started = False
+
+        async def on_bot_tts_window(task_obj, frame):
+            nonlocal bot_started
+            if tts_complete.is_set():
+                return  # This handler's job is done; avoid stale reactions
+            if isinstance(frame, BotStartedSpeakingFrame):
+                bot_started = True
+            elif isinstance(frame, BotStoppedSpeakingFrame) and bot_started:
+                tts_complete.set()
+
+        # Ensure we receive BotStartedSpeakingFrame too
+        task.add_reached_downstream_filter((BotStartedSpeakingFrame,))
+        task.add_event_handler("on_frame_reached_downstream", on_bot_tts_window)
+
+        try:
+            await task.queue_frame(TTSSpeakFrame(text=message))
+
+            try:
+                await asyncio.wait_for(tts_complete.wait(), timeout=10)
+                logger.info("HITL: Confirmation prompt finished speaking")
+            except asyncio.TimeoutError:
+                logger.warning("HITL: TTS timeout")
+
+            logger.info("HITL: Listening for response...")
+
+            while pending.retry_count < pending.max_retries:
+                try:
+                    pending.event.clear()
+                    await asyncio.wait_for(pending.event.wait(), timeout=timeout)
+                except asyncio.TimeoutError:
+                    logger.warning("HITL: Response timeout")
+                    pending.response = {"approved": False, "reason": "timeout"}
+                    break
+
+                # Check if we need to ask again
+                if pending.ask_again:
+                    pending.ask_again = False
+                    retry_message = (
+                        f"I didn't understand. {message} Please say yes or no."
+                    )
+                    tts_complete.clear()
+                    bot_started = False
+                    await task.queue_frame(TTSSpeakFrame(text=retry_message))
+                    try:
+                        await asyncio.wait_for(tts_complete.wait(), timeout=10)
+                    except asyncio.TimeoutError:
+                        logger.warning("HITL: Retry TTS timeout")
+                    logger.info("HITL: Listening for retry response...")
+                    continue
+
+                # Response matched (approved or rejected)
+                break
+
+            result = pending.response or {"approved": False, "reason": "no_response"}
+            logger.info(f"HITL: Confirmation result={result}")
+            return result
+
+        finally:
+            # Critical: Restore interruptions AFTER HITL response is complete
+            # and has been delivered to LLM state machine
+            if user_aggregator and original_config:
+                logger.info(
+                    f"HITL: Restoring interruptions after confirmation "
+                    f"(result={pending.response})"
+                )
+                await _apply_interruption_config(
+                    user_aggregator,
+                    original_config,
+                    has_vad=context.bot.vad_analyzer is not None,
+                    call_sid=context.call_sid or "unknown",
+                    label="hitl_restore",
+                    bot=context.bot,
+                )
+                logger.info("HITL: Interruptions restored")
+
+    def _cleanup(self, confirmation_id: str):
+        self._pending.pop(confirmation_id, None)
+
+    def _default_message(self, function_name: str, arguments: Dict) -> str:
+        action = function_name.replace("_", " ")
+        return f"Should I {action}? Please say yes or no."
+
+
+_hitl_manager: Optional[BreezeBuddyHITLManager] = None
+
+
+def get_hitl_manager() -> BreezeBuddyHITLManager:
+    global _hitl_manager
+    if _hitl_manager is None:
+        _hitl_manager = BreezeBuddyHITLManager()
+    return _hitl_manager

app/ai/voice/agents/breeze_buddy/processors/transcription_gate.py

@@ -45,6 +45,7 @@
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
+from app.ai.voice.agents.breeze_buddy.features.hitl.manager import get_hitl_manager
 from app.ai.voice.agents.breeze_buddy.template.types import (
     KeywordFilterConfig,
     KeywordMatchType,
@@ -187,6 +188,23 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
 
         # ---- Transcription suppression logic -------------------------
         elif isinstance(frame, (TranscriptionFrame, InterimTranscriptionFrame)):
+            # HITL gate:
+            # - drop interim frames while HITL is active (avoid turn-start interruptions)
+            # - consume final transcription via HITL matcher
+            hitl_manager = get_hitl_manager()
+
+            if isinstance(frame, InterimTranscriptionFrame):
+                if hitl_manager.is_confirmation_active():
+                    logger.debug(
+                        "TranscriptionGate: dropping interim transcription (HITL active)"
+                    )
+                    return
+
+            if isinstance(frame, TranscriptionFrame):
+                consumed = await hitl_manager.consume_transcription_if_hitl(frame.text)
+                if consumed:
+                    logger.debug("TranscriptionGate: consumed transcription (HITL)")
+                    return
             # Mode 1: hard mute — drop unconditionally (both final and interim).
             # InterimTranscriptionFrame must also be dropped: TranscriptionUserTurnStartStrategy
             # with use_interim=True fires on interim frames, which triggers an interruption even

app/ai/voice/agents/breeze_buddy/template/builder.py

@@ -552,8 +552,8 @@ def _build_function_schema(self, func: FlowFunction) -> FlowsFunctionSchema:
             Pipecat FlowsFunctionSchema object with unified handler
         """
         logger.debug(
-            f"Building function schema for: {func.name}, "
-            f"transition_to={func.transition_to}, hooks={func.hooks}"
+            f"BUILDER: Building function schema for: {func.name}, "
+            f"transition_to={func.transition_to}, hooks={func.hooks}, hitl={func.hitl_config}"
         )
 
         # Get the wrapped unified handler from handler_map
@@ -566,19 +566,29 @@ def _build_function_schema(self, func: FlowFunction) -> FlowsFunctionSchema:
         hooks = [hook.model_dump() for hook in func.hooks] if func.hooks else []
         logger.debug(f"Using hooks for {func.name}: {hooks}")
 
-        # Create a wrapper handler matching FlowsFunctionSchema expected signature.
+        # Serialize HITL config if present
+        hitl_config = func.hitl_config.model_dump() if func.hitl_config else None
+        logger.info(f"BUILDER: hitl_config for {func.name} = {hitl_config}")
+
         # In flows 1.0, ConsolidatedFunctionResult is (FlowResult | None, NodeConfig | None);
         # the legacy str-node-name variant of next_node was removed.
+        # Create a wrapper handler matching FlowsFunctionSchema expected signature
+        # Signature: (llm_args: Dict[str, Any], flow_manager: FlowManager) -> Awaitable[FlowResult | tuple]
         async def wrapper_handler(
             llm_args: Dict[str, Any], _flow_manager: FlowManager
         ) -> FlowResult | tuple[FlowResult | None, NodeConfig | None]:
             # Call the wrapped unified transition handler
             # The with_context wrapper expects llm_args as first positional arg
+            logger.info(
+                f"DEBUG: Calling transition handler with hitl_config={hitl_config}"
+            )
+
             result = await cast(Callable[..., Any], wrapped_unified_handler)(
                 llm_args,
                 transition_to=func.transition_to,
                 hooks=hooks,
                 function_name=func.name,
+                hitl_config=hitl_config,
             )
             return result
 
@@ -590,6 +600,9 @@ async def wrapper_handler(
             handler=wrapper_handler,
             properties=func.properties,
             required=func.required,
+            timeout_secs=(
+                30.0 if func.hitl_config and func.hitl_config.enabled else None
+            ),
         )
 
     def _build_action(self, action: FlowAction) -> Dict[str, Any]:

app/ai/voice/agents/breeze_buddy/template/context.py

@@ -380,6 +380,7 @@ async def wrapper(*args, **kwargs):
             hooks = kwargs.pop("hooks", None)
             function_name = kwargs.pop("function_name", None)
             function_config = kwargs.pop("function_config", None)
+            hitl_config = kwargs.pop("hitl_config", None)
 
             is_transition_handler = hooks is not None or function_name is not None
             is_global_function_handler = function_config is not None
@@ -415,6 +416,7 @@ async def wrapper(*args, **kwargs):
                     transition_to=transition_to,
                     hooks=hooks,
                     function_name=function_name,
+                    hitl_config=hitl_config,
                 )
             else:
                 # Action handlers don't need hooks/function_name