5a31f52310
Services: Admin-Lehrer, Backend-Lehrer, Studio v2, Website, Klausur-Service, School-Service, Voice-Service, Geo-Service, BreakPilot Drive, Agent-Core Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
369 lines
11 KiB
Python
369 lines
11 KiB
Python
"""
|
|
Intent Router - Voice Command Classification
|
|
Routes detected intents to appropriate handlers
|
|
|
|
Supports all use case groups:
|
|
1. Kurze Notizen (Autofahrt)
|
|
2. Arbeitsblatt-Generierung (Zug)
|
|
3. Situatives Arbeiten (Schule)
|
|
4. Canvas-Editor
|
|
5. Korrektur & RAG-Assistenz
|
|
6. Follow-up über Tage
|
|
"""
|
|
import structlog
|
|
import re
|
|
from typing import Optional, List, Dict, Any
|
|
from dataclasses import dataclass
|
|
|
|
from config import settings
|
|
from models.task import TaskType
|
|
from models.session import TranscriptMessage
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class DetectedIntent:
|
|
"""Detected intent with confidence and parameters."""
|
|
type: TaskType
|
|
confidence: float
|
|
parameters: Dict[str, Any]
|
|
is_actionable: bool
|
|
|
|
|
|
# Pattern-based intent detection rules
|
|
INTENT_PATTERNS = {
|
|
# Gruppe 1: Kurze Notizen
|
|
TaskType.STUDENT_OBSERVATION: [
|
|
r"notiz\s+zu\s+(\w+)",
|
|
r"beobachtung\s+(\w+)",
|
|
r"(\w+)\s+hat\s+(gestoert|gestört)",
|
|
r"(\w+)\s+braucht",
|
|
],
|
|
TaskType.REMINDER: [
|
|
r"erinner\s+mich",
|
|
r"morgen\s+(\d+:\d+)",
|
|
r"reminder",
|
|
r"nicht\s+vergessen",
|
|
],
|
|
TaskType.HOMEWORK_CHECK: [
|
|
r"hausaufgabe\s+kontrollieren",
|
|
r"(\w+)\s+mathe\s+hausaufgabe",
|
|
r"ha\s+check",
|
|
],
|
|
TaskType.CONFERENCE_TOPIC: [
|
|
r"thema\s+(lehrerkonferenz|konferenz)",
|
|
r"fuer\s+die\s+konferenz",
|
|
r"konferenzthema",
|
|
],
|
|
TaskType.CORRECTION_NOTE: [
|
|
r"aufgabe\s+(\d+)",
|
|
r"haeufiger\s+fehler",
|
|
r"naechste\s+stunde\s+erklaeren",
|
|
r"korrekturnotiz",
|
|
],
|
|
|
|
# Gruppe 2: Arbeitsblatt-Generierung
|
|
TaskType.WORKSHEET_GENERATE: [
|
|
r"arbeitsblatt\s+(erstellen|machen|generieren)",
|
|
r"nimm\s+vokabeln",
|
|
r"mach\s+(\d+)\s+lueckentexte",
|
|
r"uebungsblatt",
|
|
],
|
|
TaskType.WORKSHEET_DIFFERENTIATE: [
|
|
r"differenzierung",
|
|
r"zwei\s+schwierigkeitsstufen",
|
|
r"basis\s+und\s+plus",
|
|
r"leichtere\s+version",
|
|
],
|
|
|
|
# Gruppe 3: Situatives Arbeiten
|
|
TaskType.QUICK_ACTIVITY: [
|
|
r"(\d+)\s+minuten\s+einstieg",
|
|
r"schnelle\s+aktivitaet",
|
|
r"warming\s*up",
|
|
r"einstiegsaufgabe",
|
|
],
|
|
TaskType.QUIZ_GENERATE: [
|
|
r"vokabeltest",
|
|
r"quiz\s+(erstellen|generieren)",
|
|
r"(\d+)-minuten\s+test",
|
|
r"kurzer\s+test",
|
|
],
|
|
TaskType.PARENT_LETTER: [
|
|
r"elternbrief\s+wegen",
|
|
r"elternbrief",
|
|
r"brief\s+an\s+eltern",
|
|
r"wegen\s+wiederholter?\s+(stoerungen|störungen)",
|
|
r"wegen\s+(stoerungen|störungen)",
|
|
r"mitteilung\s+an\s+eltern",
|
|
],
|
|
TaskType.CLASS_MESSAGE: [
|
|
r"nachricht\s+an\s+(\d+\w+)",
|
|
r"klassen\s*nachricht",
|
|
r"info\s+an\s+die\s+klasse",
|
|
],
|
|
|
|
# Gruppe 4: Canvas-Editor
|
|
TaskType.CANVAS_EDIT: [
|
|
r"ueberschriften?\s+(groesser|kleiner|größer)",
|
|
r"bild\s+(\d+)\s+(nach|auf)",
|
|
r"pfeil\s+(von|auf)",
|
|
r"kasten\s+(hinzufuegen|einfügen)",
|
|
],
|
|
TaskType.CANVAS_LAYOUT: [
|
|
r"auf\s+eine\s+seite",
|
|
r"drucklayout\s+a4",
|
|
r"layout\s+(aendern|ändern)",
|
|
r"alles\s+auf\s+a4",
|
|
],
|
|
|
|
# Gruppe 5: Korrektur & RAG
|
|
TaskType.OPERATOR_CHECKLIST: [
|
|
r"operatoren[-\s]*checkliste",
|
|
r"welche\s+operatoren",
|
|
r"operatoren\s+fuer\s+diese\s+aufgabe",
|
|
],
|
|
TaskType.EH_PASSAGE: [
|
|
r"erwartungshorizont",
|
|
r"eh\s*passage",
|
|
r"was\s+steht\s+im\s+eh",
|
|
],
|
|
TaskType.FEEDBACK_SUGGEST: [
|
|
r"feedback\s*(vorschlag|vorschlagen)",
|
|
r"wie\s+formuliere\s+ich",
|
|
r"rueckmeldung\s+geben",
|
|
],
|
|
|
|
# Gruppe 6: Follow-up
|
|
TaskType.REMINDER_SCHEDULE: [
|
|
r"erinner\s+mich\s+morgen",
|
|
r"in\s+(\d+)\s+(stunden|tagen)",
|
|
r"naechste\s+woche",
|
|
],
|
|
TaskType.TASK_SUMMARY: [
|
|
r"offenen?\s+(aufgaben|tasks)",
|
|
r"was\s+steht\s+noch\s+an",
|
|
r"zusammenfassung",
|
|
r"fasse.+zusammen",
|
|
r"diese[rn]?\s+woche",
|
|
],
|
|
}
|
|
|
|
|
|
class IntentRouter:
|
|
"""
|
|
Routes voice commands to appropriate task types.
|
|
|
|
Uses a combination of:
|
|
1. Pattern matching for common phrases
|
|
2. LLM-based classification for complex queries
|
|
3. Context from previous messages for disambiguation
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._compiled_patterns: Dict[TaskType, List[re.Pattern]] = {}
|
|
self._compile_patterns()
|
|
|
|
def _compile_patterns(self):
|
|
"""Pre-compile regex patterns for performance."""
|
|
for task_type, patterns in INTENT_PATTERNS.items():
|
|
self._compiled_patterns[task_type] = [
|
|
re.compile(pattern, re.IGNORECASE | re.UNICODE)
|
|
for pattern in patterns
|
|
]
|
|
|
|
async def detect_intent(
|
|
self,
|
|
text: str,
|
|
context: List[TranscriptMessage] = None,
|
|
) -> Optional[DetectedIntent]:
|
|
"""
|
|
Detect intent from text with optional context.
|
|
|
|
Args:
|
|
text: Input text (transcript)
|
|
context: Previous messages for disambiguation
|
|
|
|
Returns:
|
|
DetectedIntent or None if no clear intent
|
|
"""
|
|
# Normalize text
|
|
normalized = self._normalize_text(text)
|
|
|
|
# Try pattern matching first
|
|
pattern_result = self._pattern_match(normalized)
|
|
if pattern_result and pattern_result.confidence > 0.6:
|
|
logger.info(
|
|
"Intent detected via pattern",
|
|
type=pattern_result.type.value,
|
|
confidence=pattern_result.confidence,
|
|
)
|
|
return pattern_result
|
|
|
|
# Fall back to LLM classification
|
|
if settings.fallback_llm_provider != "none":
|
|
llm_result = await self._llm_classify(normalized, context)
|
|
if llm_result and llm_result.confidence > 0.5:
|
|
logger.info(
|
|
"Intent detected via LLM",
|
|
type=llm_result.type.value,
|
|
confidence=llm_result.confidence,
|
|
)
|
|
return llm_result
|
|
|
|
# Check for context-based disambiguation
|
|
if context:
|
|
context_result = self._context_disambiguate(normalized, context)
|
|
if context_result:
|
|
logger.info(
|
|
"Intent detected via context",
|
|
type=context_result.type.value,
|
|
)
|
|
return context_result
|
|
|
|
logger.debug("No intent detected", text=text[:50])
|
|
return None
|
|
|
|
def _normalize_text(self, text: str) -> str:
|
|
"""Normalize text for matching."""
|
|
# Convert umlauts
|
|
text = text.lower()
|
|
text = text.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue")
|
|
text = text.replace("ß", "ss")
|
|
# Remove extra whitespace
|
|
text = " ".join(text.split())
|
|
return text
|
|
|
|
def _pattern_match(self, text: str) -> Optional[DetectedIntent]:
|
|
"""Match text against known patterns."""
|
|
best_match = None
|
|
best_confidence = 0.0
|
|
|
|
for task_type, patterns in self._compiled_patterns.items():
|
|
for pattern in patterns:
|
|
match = pattern.search(text)
|
|
if match:
|
|
# Calculate confidence based on match quality
|
|
match_ratio = len(match.group()) / len(text)
|
|
confidence = min(0.95, 0.6 + match_ratio * 0.4)
|
|
|
|
if confidence > best_confidence:
|
|
# Extract parameters from groups
|
|
parameters = self._extract_parameters(task_type, match, text)
|
|
|
|
best_match = DetectedIntent(
|
|
type=task_type,
|
|
confidence=confidence,
|
|
parameters=parameters,
|
|
is_actionable=self._is_actionable(task_type),
|
|
)
|
|
best_confidence = confidence
|
|
|
|
return best_match
|
|
|
|
def _extract_parameters(
|
|
self,
|
|
task_type: TaskType,
|
|
match: re.Match,
|
|
full_text: str,
|
|
) -> Dict[str, Any]:
|
|
"""Extract parameters from regex match."""
|
|
params = {}
|
|
|
|
# Extract named groups or positional groups
|
|
if match.groups():
|
|
groups = match.groups()
|
|
|
|
# Task-specific parameter extraction
|
|
if task_type == TaskType.STUDENT_OBSERVATION:
|
|
params["student_name"] = groups[0] if groups else None
|
|
|
|
elif task_type == TaskType.HOMEWORK_CHECK:
|
|
params["subject"] = "mathe" if "mathe" in full_text else None
|
|
|
|
elif task_type == TaskType.QUICK_ACTIVITY:
|
|
params["duration_minutes"] = int(groups[0]) if groups else 10
|
|
|
|
elif task_type == TaskType.QUIZ_GENERATE:
|
|
params["duration_minutes"] = int(groups[0]) if groups and groups[0].isdigit() else 10
|
|
|
|
elif task_type == TaskType.CLASS_MESSAGE:
|
|
params["class_name"] = groups[0] if groups else None
|
|
|
|
# Extract time references
|
|
time_match = re.search(r"(\d{1,2}):?(\d{2})?", full_text)
|
|
if time_match:
|
|
params["time"] = time_match.group()
|
|
|
|
# Extract content after colon
|
|
colon_match = re.search(r":\s*(.+)$", full_text)
|
|
if colon_match:
|
|
params["content"] = colon_match.group(1).strip()
|
|
|
|
return params
|
|
|
|
def _is_actionable(self, task_type: TaskType) -> bool:
|
|
"""Check if intent type creates an actionable task."""
|
|
# All task types are actionable except queries
|
|
query_types = [
|
|
TaskType.OPERATOR_CHECKLIST,
|
|
TaskType.EH_PASSAGE,
|
|
TaskType.TASK_SUMMARY,
|
|
]
|
|
return task_type not in query_types
|
|
|
|
async def _llm_classify(
|
|
self,
|
|
text: str,
|
|
context: List[TranscriptMessage] = None,
|
|
) -> Optional[DetectedIntent]:
|
|
"""Use LLM for intent classification."""
|
|
from services.fallback_llm_client import FallbackLLMClient
|
|
|
|
llm = FallbackLLMClient()
|
|
result = await llm.detect_intent(text)
|
|
|
|
if result.get("type") == "unknown":
|
|
return None
|
|
|
|
try:
|
|
task_type = TaskType(result["type"])
|
|
return DetectedIntent(
|
|
type=task_type,
|
|
confidence=result.get("confidence", 0.5),
|
|
parameters=result.get("parameters", {}),
|
|
is_actionable=result.get("is_actionable", True),
|
|
)
|
|
except ValueError:
|
|
logger.warning("Unknown task type from LLM", type=result.get("type"))
|
|
return None
|
|
|
|
def _context_disambiguate(
|
|
self,
|
|
text: str,
|
|
context: List[TranscriptMessage],
|
|
) -> Optional[DetectedIntent]:
|
|
"""Disambiguate intent using conversation context."""
|
|
if not context:
|
|
return None
|
|
|
|
# Look for continuation patterns
|
|
continuation_words = ["ja", "genau", "richtig", "okay", "mach das", "bitte"]
|
|
|
|
if any(word in text.lower() for word in continuation_words):
|
|
# Find the last assistant message with a suggestion
|
|
for msg in reversed(context):
|
|
if msg.role == "assistant" and msg.intent:
|
|
try:
|
|
return DetectedIntent(
|
|
type=TaskType(msg.intent),
|
|
confidence=0.6,
|
|
parameters={},
|
|
is_actionable=True,
|
|
)
|
|
except ValueError:
|
|
pass
|
|
|
|
return None
|