1  """Fast-path handler for simple deterministic queries.
 2  
 3  Intercepts common queries and responds directly without LLM invocation.
 4  Runs as a Pipecat FrameProcessor between STT and the LLM context aggregator.
 5  
 6  Handles:
 7  - Time/date queries
 8  - Basic weather (from cached data)
 9  - System status (quick summary)
10  
11  If no fast-path match, passes through to LLM normally.
12  """
13  
14  import json
15  import os
16  import re
17  import urllib.request
18  from datetime import datetime
19  
20  from loguru import logger
21  
22  from pipecat.frames.frames import (
23      Frame,
24      TextFrame,
25      TranscriptionFrame,
26  )
27  from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
28  
29  FAST_PATH_ENABLED = os.getenv("FAST_PATH_ENABLED", "true").lower() == "true"
30  WEATHER_CACHE_URL = os.getenv("WEATHER_CACHE_URL", "")  # Optional: cached weather endpoint
31  
32  # Time/date patterns
33  TIME_PATTERNS = [
34      r'\bwhat time\b', r'\bwhat\'s the time\b', r'\bcurrent time\b',
35      r'\btell me the time\b',
36  ]
37  DATE_PATTERNS = [
38      r'\bwhat(?:\'s| is) (?:the |today\'s )?date\b', r'\bwhat day\b',
39      r'\btoday\'s date\b', r'\bwhat is today\b',
40  ]
41  
42  # Compile patterns
43  _time_re = re.compile('|'.join(TIME_PATTERNS), re.IGNORECASE)
44  _date_re = re.compile('|'.join(DATE_PATTERNS), re.IGNORECASE)
45  
46  
47  def check_fast_path(text: str) -> str | None:
48      """Check if the text matches a fast-path pattern. Returns response or None."""
49      text_lower = text.lower().strip()
50  
51      # Time
52      if _time_re.search(text_lower):
53          now = datetime.now()
54          return f"It's {now.strftime('%I:%M %p').lstrip('0')}."
55  
56      # Date
57      if _date_re.search(text_lower):
58          now = datetime.now()
59          return f"Today is {now.strftime('%A, %B %d, %Y')}."
60  
61      return None
62  
63  
64  class FastPathProcessor(FrameProcessor):
65      """Intercepts simple queries and responds without LLM invocation."""
66  
67      def __init__(self, *, tts_service=None, **kwargs):
68          super().__init__(**kwargs)
69          self._tts = tts_service
70  
71      async def process_frame(self, frame: Frame, direction: FrameDirection):
72          await super().process_frame(frame, direction)
73  
74          if not FAST_PATH_ENABLED:
75              await self.push_frame(frame, direction)
76              return
77  
78          if isinstance(frame, TranscriptionFrame):
79              text = frame.text
80              # Strip speaker tag if present
81              if text.startswith("["):
82                  idx = text.find("]")
83                  if idx > 0:
84                      text = text[idx+1:].strip()
85  
86              response = check_fast_path(text)
87              if response:
88                  logger.info(f"Fast-path response: '{text}' → '{response}'")
89                  # Push the response as a TextFrame directly to TTS
90                  # Skip the LLM entirely
91                  await self.push_frame(TextFrame(text=response), direction)
92                  return
93  
94          # No fast-path match — pass through to LLM
95          await self.push_frame(frame, direction)