Spaces:
Running
Running
| """Manager for orchestrating transcript analyzers and executing reactions.""" | |
| from __future__ import annotations | |
| import time | |
| import asyncio | |
| import logging | |
| from typing import Any, List | |
| from .base import EntityMatch, TriggerMatch, ReactionConfig | |
| from lyon_chatbox.tools.core_tools import ToolDependencies | |
| logger = logging.getLogger(__name__) | |
| class TranscriptAnalysisManager: | |
| """Orchestrates transcript analyzers and dispatches reaction callbacks. | |
| Owns: | |
| - Analyzer creation (partitions ReactionConfigs by trigger type) | |
| - Per-reaction deduplication (each reaction fires at most once per turn) | |
| - Callback dispatch with TriggerMatch + params | |
| """ | |
| DEBOUNCE_INTERVAL = 0.4 # minimum seconds between partial analyses | |
| def __init__(self, reactions: list[ReactionConfig], deps: ToolDependencies): | |
| """Initialize with reaction configs and tool dependencies.""" | |
| self.deps = deps | |
| self.reactions = {r.name: r for r in reactions} | |
| self._pending_tasks: List[asyncio.Task[Any]] = [] | |
| self._last_analysis_time = 0.0 | |
| self.triggered_reactions: set[str] = set() | |
| # For repeatable entity reactions: track (reaction_name, entity_text) already fired | |
| self._triggered_entity_keys: set[tuple[str, str]] = set() | |
| # Partition reactions by trigger type and build analyzers | |
| # _all_reaction_groups maps real name → list of synthetic group names | |
| self._all_reaction_groups: dict[str, list[str]] = {} | |
| self.keyword_analyzer = self._build_keyword_analyzer(reactions) | |
| self.entity_analyzer = self._build_entity_analyzer(reactions) | |
| # Map entity label → reaction names for dispatch | |
| self.entity_reaction_map: dict[str, list[str]] = {} | |
| for r in reactions: | |
| for label in r.trigger.entities: | |
| self.entity_reaction_map.setdefault(label, []).append(r.name) | |
| logger.info(f"TranscriptAnalysisManager initialized with {len(reactions)} reactions") | |
| def _build_keyword_analyzer(self, reactions: list[ReactionConfig]) -> Any | None: | |
| """Build KeywordAnalyzer from reactions with word triggers.""" | |
| from .keyword_analyzer import KeywordAnalyzer | |
| reaction_words: dict[str, list[str]] = {} | |
| for r in reactions: | |
| # Simple word triggers | |
| if r.trigger.words: | |
| reaction_words[r.name] = r.trigger.words | |
| # Boolean `all` triggers: register each group as a synthetic entry | |
| if r.trigger.all: | |
| group_names: list[str] = [] | |
| for i, sub_trigger in enumerate(r.trigger.all): | |
| if sub_trigger.words: | |
| synthetic_name = f"{r.name}__all_{i}" | |
| reaction_words[synthetic_name] = sub_trigger.words | |
| group_names.append(synthetic_name) | |
| if group_names: | |
| self._all_reaction_groups[r.name] = group_names | |
| if not reaction_words: | |
| return None | |
| return KeywordAnalyzer(reaction_words) | |
| def _build_entity_analyzer(self, reactions: list[ReactionConfig]) -> Any | None: | |
| """Build EntityAnalyzer from reactions with entity triggers.""" | |
| all_labels: set[str] = set() | |
| for r in reactions: | |
| all_labels.update(r.trigger.entities) | |
| if not all_labels: | |
| return None | |
| try: | |
| from .entity_analyzer import EntityAnalyzer | |
| return EntityAnalyzer(sorted(all_labels)) | |
| except ImportError: | |
| logger.warning( | |
| "GLiNER not installed, skipping entity analyzer. " | |
| "Install with: pip install 'lyon_chatbox[cascade_gliner]'" | |
| ) | |
| return None | |
| async def analyze_partial(self, text: str) -> None: | |
| """Analyze partial transcript with debouncing. Fire-and-forget.""" | |
| current_time = time.time() | |
| if current_time - self._last_analysis_time < self.DEBOUNCE_INTERVAL: | |
| return | |
| self._last_analysis_time = current_time | |
| logger.info(f"Analyzing PARTIAL transcript: '{text[:100]}...' ({len(text)} chars)") | |
| task = asyncio.create_task(self._analyze_and_dispatch(text, is_final=False)) | |
| self._pending_tasks.append(task) | |
| async def analyze_final(self, text: str) -> None: | |
| """Analyze final transcript. Blocks until analysis completes.""" | |
| logger.info(f"Analyzing FINAL transcript: '{text[:100]}...' ({len(text)} chars)") | |
| await self._analyze_and_dispatch(text, is_final=True) | |
| async def _analyze_and_dispatch(self, text: str, is_final: bool) -> None: | |
| """Run analyzers and dispatch triggered reactions.""" | |
| try: | |
| # Run keyword and entity analyzers in parallel | |
| tasks = [] | |
| if self.keyword_analyzer: | |
| tasks.append(self.keyword_analyzer.analyze(text, is_final)) | |
| if self.entity_analyzer: | |
| tasks.append(self.entity_analyzer.analyze(text, is_final)) | |
| if not tasks: | |
| return | |
| results = await asyncio.gather(*tasks, return_exceptions=True) | |
| idx = 0 | |
| # Process keyword results | |
| keyword_matches: dict[str, list[str]] = {} | |
| if self.keyword_analyzer: | |
| kw_result = results[idx] | |
| if isinstance(kw_result, BaseException): | |
| logger.warning(f"Keyword analyzer error: {kw_result}") | |
| else: | |
| keyword_matches = kw_result | |
| idx += 1 | |
| # Process entity results | |
| entity_matches: list[EntityMatch] = [] | |
| if self.entity_analyzer: | |
| ent_result = results[idx] | |
| if isinstance(ent_result, BaseException): | |
| logger.warning(f"Entity analyzer error: {ent_result}") | |
| else: | |
| entity_matches = ent_result | |
| # Evaluate boolean `all` triggers: merge synthetic groups into real reactions | |
| for real_name, group_names in self._all_reaction_groups.items(): | |
| real_reaction = self.reactions[real_name] | |
| if not real_reaction.repeatable and real_name in self.triggered_reactions: | |
| # Already fired — just strip synthetic entries | |
| for g in group_names: | |
| keyword_matches.pop(g, None) | |
| continue | |
| if all(g in keyword_matches for g in group_names): | |
| merged_words: list[str] = [] | |
| for g in group_names: | |
| merged_words.extend(keyword_matches.pop(g)) | |
| keyword_matches[real_name] = merged_words | |
| else: | |
| # Not all groups matched — strip synthetic entries | |
| for g in group_names: | |
| keyword_matches.pop(g, None) | |
| # Dispatch keyword-triggered reactions | |
| for reaction_name, matched_words in keyword_matches.items(): | |
| if reaction_name in self.triggered_reactions: | |
| continue | |
| reaction = self.reactions[reaction_name] | |
| if not reaction.repeatable: | |
| self.triggered_reactions.add(reaction_name) | |
| match = TriggerMatch(words=matched_words) | |
| logger.info(f"Reaction triggered: {reaction_name} (words: {matched_words})") | |
| asyncio.create_task(self._execute(reaction, match)) | |
| # Dispatch entity-triggered reactions | |
| for em in entity_matches: | |
| for reaction_name in self.entity_reaction_map.get(em.label, []): | |
| if reaction_name in self.triggered_reactions: | |
| continue | |
| reaction = self.reactions[reaction_name] | |
| if reaction.repeatable: | |
| # Deduplicate by (reaction, entity_text) so each unique entity fires once | |
| entity_key = (reaction_name, em.text.lower()) | |
| if entity_key in self._triggered_entity_keys: | |
| continue | |
| self._triggered_entity_keys.add(entity_key) | |
| else: | |
| self.triggered_reactions.add(reaction_name) | |
| match = TriggerMatch(entities=[em]) | |
| logger.info(f"Reaction triggered: {reaction_name} (entity: {em.text} [{em.label}])") | |
| asyncio.create_task(self._execute(reaction, match)) | |
| except Exception as e: | |
| logger.exception(f"Error in transcript analysis: {e}") | |
| async def _execute(self, reaction: ReactionConfig, match: TriggerMatch) -> None: | |
| """Execute a reaction callback with deps, match, and params.""" | |
| try: | |
| await reaction.callback(self.deps, match, **reaction.params) | |
| except Exception as e: | |
| logger.exception(f"Error executing reaction '{reaction.name}': {e}") | |
| def reset(self) -> None: | |
| """Reset deduplication state for next conversation turn.""" | |
| logger.debug(f"Resetting manager ({len(self.triggered_reactions)} reactions triggered)") | |
| self.triggered_reactions.clear() | |
| self._triggered_entity_keys.clear() | |
| for task in self._pending_tasks: | |
| if not task.done(): | |
| task.cancel() | |
| self._pending_tasks.clear() | |
| self._last_analysis_time = 0.0 | |
| class NoOpTranscriptManager: | |
| """No-op manager when transcript analysis is disabled.""" | |
| async def analyze_partial(self, text: str) -> None: | |
| """No-op.""" | |
| async def analyze_final(self, text: str) -> None: | |
| """No-op.""" | |
| def reset(self) -> None: | |
| """No-op.""" | |