""" RelevanceProfile Model. Definiert das Relevanzprofil eines Nutzers für die Alerts-Filterung. Lernt über Zeit durch Feedback. """ from dataclasses import dataclass, field from datetime import datetime from typing import Optional import uuid @dataclass class PriorityItem: """Ein Prioritäts-Thema im Profil.""" label: str # z.B. "Inklusion", "Datenschutz Schule" weight: float = 0.5 # 0.0 - 1.0, höher = wichtiger keywords: list = field(default_factory=list) # Zusätzliche Keywords description: Optional[str] = None # Kontext für LLM def to_dict(self) -> dict: return { "label": self.label, "weight": self.weight, "keywords": self.keywords, "description": self.description, } @classmethod def from_dict(cls, data: dict) -> "PriorityItem": return cls(**data) @dataclass class RelevanceProfile: """ Nutzerprofil für Relevanz-Scoring. Das Profil wird verwendet, um Alerts auf Relevanz zu prüfen. Es enthält: - Prioritäten: Themen die wichtig sind (mit Gewichtung) - Ausschlüsse: Themen die ignoriert werden sollen - Positive Beispiele: URLs/Titel die relevant waren - Negative Beispiele: URLs/Titel die irrelevant waren - Policies: Zusätzliche Regeln (z.B. nur deutsche Quellen) """ # Identifikation id: str = field(default_factory=lambda: str(uuid.uuid4())) user_id: Optional[str] = None # Falls benutzerspezifisch # Relevanz-Kriterien priorities: list = field(default_factory=list) # List[PriorityItem] exclusions: list = field(default_factory=list) # Keywords zum Ausschließen # Beispiele für Few-Shot Learning positive_examples: list = field(default_factory=list) # Relevante Alerts negative_examples: list = field(default_factory=list) # Irrelevante Alerts # Policies policies: dict = field(default_factory=dict) # Metadaten created_at: datetime = field(default_factory=datetime.utcnow) updated_at: datetime = field(default_factory=datetime.utcnow) # Statistiken total_scored: int = 0 total_kept: int = 0 total_dropped: int = 0 accuracy_estimate: Optional[float] = None # Geschätzte Genauigkeit def add_priority(self, label: str, weight: float = 0.5, **kwargs) -> None: """Füge ein Prioritäts-Thema hinzu.""" self.priorities.append(PriorityItem( label=label, weight=weight, **kwargs )) self.updated_at = datetime.utcnow() def add_exclusion(self, keyword: str) -> None: """Füge ein Ausschluss-Keyword hinzu.""" if keyword not in self.exclusions: self.exclusions.append(keyword) self.updated_at = datetime.utcnow() def add_positive_example(self, title: str, url: str, reason: str = "") -> None: """Füge ein positives Beispiel hinzu (für Few-Shot Learning).""" self.positive_examples.append({ "title": title, "url": url, "reason": reason, "added_at": datetime.utcnow().isoformat(), }) # Begrenze auf letzte 20 Beispiele self.positive_examples = self.positive_examples[-20:] self.updated_at = datetime.utcnow() def add_negative_example(self, title: str, url: str, reason: str = "") -> None: """Füge ein negatives Beispiel hinzu.""" self.negative_examples.append({ "title": title, "url": url, "reason": reason, "added_at": datetime.utcnow().isoformat(), }) # Begrenze auf letzte 20 Beispiele self.negative_examples = self.negative_examples[-20:] self.updated_at = datetime.utcnow() def update_from_feedback(self, alert_title: str, alert_url: str, is_relevant: bool, reason: str = "") -> None: """ Aktualisiere Profil basierend auf Nutzer-Feedback. Args: alert_title: Titel des Alerts alert_url: URL des Alerts is_relevant: True wenn der Nutzer den Alert als relevant markiert hat reason: Optional - Grund für die Entscheidung """ if is_relevant: self.add_positive_example(alert_title, alert_url, reason) self.total_kept += 1 else: self.add_negative_example(alert_title, alert_url, reason) self.total_dropped += 1 self.total_scored += 1 # Aktualisiere Accuracy-Schätzung (vereinfacht) if self.total_scored > 10: # Hier könnte eine komplexere Berechnung erfolgen # basierend auf Vergleich von Vorhersage vs. tatsächlichem Feedback pass def get_prompt_context(self) -> str: """ Generiere Kontext für LLM-Prompt. Dieser Text wird in den System-Prompt des Relevanz-Scorers eingefügt. """ lines = ["## Relevanzprofil des Nutzers\n"] # Prioritäten if self.priorities: lines.append("### Prioritäten (Themen von Interesse):") for p in self.priorities: if isinstance(p, dict): p = PriorityItem.from_dict(p) weight_label = "Sehr wichtig" if p.weight > 0.7 else "Wichtig" if p.weight > 0.4 else "Interessant" lines.append(f"- **{p.label}** ({weight_label})") if p.description: lines.append(f" {p.description}") if p.keywords: lines.append(f" Keywords: {', '.join(p.keywords)}") lines.append("") # Ausschlüsse if self.exclusions: lines.append("### Ausschlüsse (ignorieren):") lines.append(f"Themen mit diesen Keywords: {', '.join(self.exclusions)}") lines.append("") # Positive Beispiele if self.positive_examples: lines.append("### Beispiele für relevante Alerts:") for ex in self.positive_examples[-5:]: # Letzte 5 lines.append(f"- \"{ex['title']}\"") if ex.get("reason"): lines.append(f" Grund: {ex['reason']}") lines.append("") # Negative Beispiele if self.negative_examples: lines.append("### Beispiele für irrelevante Alerts:") for ex in self.negative_examples[-5:]: # Letzte 5 lines.append(f"- \"{ex['title']}\"") if ex.get("reason"): lines.append(f" Grund: {ex['reason']}") lines.append("") # Policies if self.policies: lines.append("### Zusätzliche Regeln:") for key, value in self.policies.items(): lines.append(f"- {key}: {value}") return "\n".join(lines) def to_dict(self) -> dict: """Konvertiere zu Dictionary.""" return { "id": self.id, "user_id": self.user_id, "priorities": [p.to_dict() if isinstance(p, PriorityItem) else p for p in self.priorities], "exclusions": self.exclusions, "positive_examples": self.positive_examples, "negative_examples": self.negative_examples, "policies": self.policies, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), "total_scored": self.total_scored, "total_kept": self.total_kept, "total_dropped": self.total_dropped, "accuracy_estimate": self.accuracy_estimate, } @classmethod def from_dict(cls, data: dict) -> "RelevanceProfile": """Erstelle RelevanceProfile aus Dictionary.""" # Parse Timestamps for field_name in ["created_at", "updated_at"]: if field_name in data and isinstance(data[field_name], str): data[field_name] = datetime.fromisoformat(data[field_name]) # Parse Priorities if "priorities" in data: data["priorities"] = [ PriorityItem.from_dict(p) if isinstance(p, dict) else p for p in data["priorities"] ] return cls(**data) @classmethod def create_default_education_profile(cls) -> "RelevanceProfile": """ Erstelle ein Standard-Profil für Bildungsthemen. Dieses Profil ist für Lehrkräfte/Schulpersonal optimiert. """ profile = cls() # Bildungs-relevante Prioritäten profile.add_priority( "Inklusion", weight=0.9, keywords=["inklusiv", "Förderbedarf", "Behinderung", "Barrierefreiheit"], description="Inklusive Bildung, Förderschulen, Nachteilsausgleich" ) profile.add_priority( "Datenschutz Schule", weight=0.85, keywords=["DSGVO", "Schülerfotos", "Einwilligung", "personenbezogene Daten"], description="DSGVO in Schulen, Datenschutz bei Klassenfotos" ) profile.add_priority( "Schulrecht Bayern", weight=0.8, keywords=["BayEUG", "Schulordnung", "Kultusministerium", "Bayern"], description="Bayerisches Schulrecht, Verordnungen" ) profile.add_priority( "Digitalisierung Schule", weight=0.7, keywords=["DigitalPakt", "Tablet-Klasse", "Lernplattform"], description="Digitale Medien im Unterricht" ) profile.add_priority( "Elternarbeit", weight=0.6, keywords=["Elternbeirat", "Elternabend", "Kommunikation"], description="Zusammenarbeit mit Eltern" ) # Standard-Ausschlüsse profile.exclusions = [ "Stellenanzeige", "Praktikum gesucht", "Werbung", "Pressemitteilung", # Oft generisch ] # Policies profile.policies = { "prefer_german_sources": True, "max_age_days": 30, # Ältere Alerts ignorieren "min_content_length": 100, # Sehr kurze Snippets ignorieren } return profile def __repr__(self) -> str: return f"RelevanceProfile(id={self.id[:8]}, priorities={len(self.priorities)}, examples={len(self.positive_examples) + len(self.negative_examples)})"