My Voice Router That Refuses to Think: Pattern‑First Multi‑Agent Orchestration for Sub‑Second Latency

flowchart TD
  voiceRoute[Voice route] --> voiceProcessor[VoiceAgentProcessor]
  voiceProcessor --> orchestrator[AgentOrchestrator]
  orchestrator --> router[RouterAgent]
  router --> locationAgent[LocationAgent]
  router --> searchAgent[SearchAgent]
  router --> m365Agent[M365Agent]
  router --> crmAgent[CrmAgent]
  locationAgent --> searchAgent
  searchAgent --> response[Voice-optimized response]
  m365Agent --> response
  crmAgent --> response

import re
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class Intent(str, Enum):
    SEARCH = "search"
    LOCATION_SEARCH = "location_search"
    M365 = "m365"
    CRM = "crm"
    AMBIGUOUS = "ambiguous"


@dataclass
class RoutingDecision:
    intent: Intent
    agents: list[str]
    confidence: float
    reason: str
    used_fallback: bool = False
    latency_ms: float = 0.0


# --- Pattern definitions (coarse on purpose) ---

LOCATION_PATTERNS = [
    re.compile(r"\b(?:in|from|near|around)\s+[A-Z][a-z]+", re.IGNORECASE),
    re.compile(
        r"\b(?:northeast|southeast|midwest|west coast|east coast"
        r"|california|texas|new york|florida|illinois)\b",
        re.IGNORECASE,
    ),
]

SEARCH_KEYWORDS = re.compile(
    r"\b(?:how many|find|search|list|show|count|look up"
    r"|tickets|orders|inventory|items|records)\b",
    re.IGNORECASE,
)

M365_KEYWORDS = re.compile(
    r"\b(?:email|calendar|schedule|meeting|outlook|teams"
    r"|send mail|book a meeting|invite)\b",
    re.IGNORECASE,
)

CRM_KEYWORDS = re.compile(
    r"\b(?:crm|call log|deal|pipeline|contact list"
    r"|sales call|account notes)\b",
    re.IGNORECASE,
)


class RouterAgent:
    """Deterministic-first intent router with LLM fallback."""

    def __init__(self, llm_client=None, llm_timeout: float = 2.0):
        self.llm_client = llm_client
        self.llm_timeout = llm_timeout

    async def route(self, transcript: str) -> RoutingDecision:
        start = time.perf_counter()
        text = transcript.strip().lower()

        # --- deterministic pass ---
        has_location = any(p.search(transcript) for p in LOCATION_PATTERNS)
        has_search = bool(SEARCH_KEYWORDS.search(text))
        has_m365 = bool(M365_KEYWORDS.search(text))
        has_crm = bool(CRM_KEYWORDS.search(text))

        decision: Optional[RoutingDecision] = None

        if has_location and has_search:
            decision = RoutingDecision(
                intent=Intent.LOCATION_SEARCH,
                agents=["LocationAgent", "SearchAgent"],
                confidence=0.95,
                reason="location + search keywords detected",
            )
        elif has_search and not has_m365 and not has_crm:
            decision = RoutingDecision(
                intent=Intent.SEARCH,
                agents=["SearchAgent"],
                confidence=0.90,
                reason="search keywords detected",
            )
        elif has_m365 and not has_crm:
            decision = RoutingDecision(
                intent=Intent.M365,
                agents=["M365Agent"],
                confidence=0.90,
                reason="M365 keywords detected",
            )
        elif has_crm and not has_m365:
            decision = RoutingDecision(
                intent=Intent.CRM,
                agents=["CrmAgent"],
                confidence=0.90,
                reason="CRM keywords detected",
            )

        if decision is not None:
            decision.latency_ms = (time.perf_counter() - start) * 1000
            return decision

        # --- LLM fallback for genuinely ambiguous queries ---
        decision = await self._llm_classify(transcript)
        decision.latency_ms = (time.perf_counter() - start) * 1000
        return decision

    async def _llm_classify(self, transcript: str) -> RoutingDecision:
        """Call the LLM classifier. Returns a decision with used_fallback=True."""
        if self.llm_client is None:
            return RoutingDecision(
                intent=Intent.AMBIGUOUS,
                agents=["SearchAgent"],  # safe default
                confidence=0.30,
                reason="no LLM client configured; defaulting to search",
                used_fallback=True,
            )

        import asyncio

        try:
            resp = await asyncio.wait_for(
                self.llm_client.classify(transcript),
                timeout=self.llm_timeout,
            )
            return RoutingDecision(
                intent=Intent(resp["intent"]),
                agents=resp["agents"],
                confidence=resp.get("confidence", 0.70),
                reason=f"LLM classification: {resp.get('reason', 'n/a')}",
                used_fallback=True,
            )
        except (asyncio.TimeoutError, Exception) as exc:
            return RoutingDecision(
                intent=Intent.AMBIGUOUS,
                agents=["SearchAgent"],
                confidence=0.20,
                reason=f"LLM fallback failed ({type(exc).__name__}); defaulting",
                used_fallback=True,
            )

import asyncio
import time
import logging
from dataclasses import dataclass
from typing import Any, Optional

logger = logging.getLogger(__name__)


@dataclass
class AgentResult:
    agent: str
    success: bool
    data: Any = None
    error: Optional[str] = None
    latency_ms: float = 0.0


@dataclass
class OrchestrationResult:
    results: list[AgentResult]
    text: str  # voice-ready summary
    total_ms: float = 0.0


class AgentOrchestrator:
    """Coordinates agents based on RouterAgent decisions.

    Runs chains (e.g. LocationAgent -> SearchAgent) sequentially
    and independent agents concurrently.  Enforces per-agent timeouts.
    """

    def __init__(
        self,
        agents: dict,  # name -> agent instance (each has async .run(query, context))
        router: "RouterAgent",
        agent_timeout: float = 3.0,
        total_timeout: float = 5.0,
    ):
        self.agents = agents
        self.router = router
        self.agent_timeout = agent_timeout
        self.total_timeout = total_timeout

    async def handle(self, transcript: str, context: dict | None = None) -> OrchestrationResult:
        wall_start = time.perf_counter()
        context = context or {}

        # 1. Route
        decision = await self.router.route(transcript)
        logger.info(
            "routed intent=%s agents=%s fallback=%s latency=%.1fms",
            decision.intent,
            decision.agents,
            decision.used_fallback,
            decision.latency_ms,
        )

        # 2. Execute agent chain
        agent_results: list[AgentResult] = []
        chain_context = {**context, "transcript": transcript}

        for agent_name in decision.agents:
            agent = self.agents.get(agent_name)
            if agent is None:
                agent_results.append(
                    AgentResult(agent=agent_name, success=False, error="agent not registered")
                )
                continue

            result = await self._run_agent(agent_name, agent, transcript, chain_context)
            agent_results.append(result)

            # Feed successful output forward so the next agent in the chain can use it
            if result.success and result.data is not None:
                chain_context[agent_name] = result.data

        # 3. Build voice-ready text
        text = self._build_response_text(agent_results, decision)

        total_ms = (time.perf_counter() - wall_start) * 1000
        return OrchestrationResult(results=agent_results, text=text, total_ms=total_ms)

    async def _run_agent(
        self, name: str, agent: Any, transcript: str, context: dict
    ) -> AgentResult:
        start = time.perf_counter()
        try:
            data = await asyncio.wait_for(
                agent.run(transcript, context),
                timeout=self.agent_timeout,
            )
            return AgentResult(
                agent=name,
                success=True,
                data=data,
                latency_ms=(time.perf_counter() - start) * 1000,
            )
        except asyncio.TimeoutError:
            logger.warning("%s timed out after %.1fs", name, self.agent_timeout)
            return AgentResult(
                agent=name,
                success=False,
                error="timeout",
                latency_ms=(time.perf_counter() - start) * 1000,
            )
        except Exception as exc:
            logger.exception("%s failed: %s", name, exc)
            return AgentResult(
                agent=name,
                success=False,
                error=str(exc),
                latency_ms=(time.perf_counter() - start) * 1000,
            )

    @staticmethod
    def _build_response_text(results: list[AgentResult], decision) -> str:
        """Merge agent outputs into a single voice-friendly string."""
        parts: list[str] = []
        for r in results:
            if r.success and r.data:
                # Each agent is expected to return a dict with a "summary" key
                summary = r.data.get("summary") if isinstance(r.data, dict) else str(r.data)
                if summary:
                    parts.append(summary)
            elif not r.success:
                parts.append(f"I wasn't able to complete the {r.agent} step.")

        return " ".join(parts) if parts else "I'm sorry, I couldn't find an answer for that."

import asyncio
import logging
import hashlib
from dataclasses import dataclass
from typing import Optional

logger = logging.getLogger(__name__)

MAX_VOICE_CHARS = 300  # TTS engines get sluggish with long strings
VOICE_TIMEOUT = 4.5    # seconds — hard ceiling for a single voice turn


@dataclass
class VoiceResponse:
    text: str
    success: bool
    latency_ms: float = 0.0
    truncated: bool = False


class VoiceAgentProcessor:
    """Bridge between the voice transport layer and the agent orchestrator.

    Callers use `handle_voice_query` and get back a VoiceResponse.
    Everything about routing, agent selection, and timeout enforcement
    is hidden behind this interface.
    """

    def __init__(self, orchestrator: "AgentOrchestrator", timeout: float = VOICE_TIMEOUT):
        self.orchestrator = orchestrator
        self.timeout = timeout

    async def handle_voice_query(
        self,
        transcript: str,
        user_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ) -> VoiceResponse:
        """Main entry point for voice routes.

        Args:
            transcript: Raw speech-to-text output.
            user_id: Opaque, redacted user identifier (never an email).
            session_id: Voice session ID for log correlation.
        """
        import time

        start = time.perf_counter()

        context = {
            "user_id": self._redact_id(user_id) if user_id else "anon",
            "session_id": session_id or "unknown",
            "channel": "voice",
        }

        try:
            result = await asyncio.wait_for(
                self.orchestrator.handle(transcript, context),
                timeout=self.timeout,
            )
            text = self._format_for_voice(result.text)
            latency = (time.perf_counter() - start) * 1000

            return VoiceResponse(
                text=text,
                success=True,
                latency_ms=latency,
                truncated=len(result.text) > MAX_VOICE_CHARS,
            )

        except asyncio.TimeoutError:
            latency = (time.perf_counter() - start) * 1000
            logger.warning(
                "voice turn timed out after %.0fms session=%s",
                latency,
                session_id,
            )
            return VoiceResponse(
                text="I'm still working on that. Let me get back to you in a moment.",
                success=False,
                latency_ms=latency,
            )

        except Exception as exc:
            latency = (time.perf_counter() - start) * 1000
            logger.exception("voice query failed session=%s: %s", session_id, exc)
            return VoiceResponse(
                text="Something went wrong. Could you try asking again?",
                success=False,
                latency_ms=latency,
            )

    @staticmethod
    def _format_for_voice(text: str) -> str:
        """Trim and clean text for TTS output."""
        text = text.strip()
        if len(text) <= MAX_VOICE_CHARS:
            return text

        # Cut at the last sentence boundary within the limit
        truncated = text[:MAX_VOICE_CHARS]
        last_period = truncated.rfind(".")
        if last_period > MAX_VOICE_CHARS // 2:
            return truncated[: last_period + 1]
        return truncated.rstrip() + "..."

    @staticmethod
    def _redact_id(user_id: str) -> str:
        """One-way hash so logs never contain raw identifiers."""
        return hashlib.sha256(user_id.encode()).hexdigest()[:12]