Spaces:
Running
Running
File size: 9,795 Bytes
534b431 dfedf76 534b431 dfedf76 534b431 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 | """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."""
|