#!/usr/bin/env python3 """ Feedback Learning Service for NordaBiz AI Chat =============================================== Implements few-shot learning from user feedback to improve AI responses. Features: - Collects positive feedback examples for few-shot learning - Identifies negative patterns to avoid - Provides curated seed examples for cold start - Caches learning context for performance Author: Norda Biznes Development Team Created: 2026-01-11 """ import json import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple from dataclasses import dataclass from sqlalchemy import desc, func from sqlalchemy.orm import Session from database import ( SessionLocal, AIChatMessage, AIChatFeedback, AIChatConversation ) logger = logging.getLogger(__name__) @dataclass class FeedbackExample: """Represents a feedback example for learning""" query: str response: str rating: int # 1=negative, 2=positive feedback_comment: Optional[str] = None companies_mentioned: Optional[List[str]] = None class FeedbackLearningService: """ Service for AI learning from user feedback. Uses few-shot learning approach: - Positive examples show AI what good responses look like - Negative examples help avoid common mistakes - Seed examples provide baseline for cold start """ # Curated seed examples for cold start (when insufficient real feedback) SEED_EXAMPLES = [ FeedbackExample( query="Szukam firmy od stron internetowych", response="W Norda Biznes stronami internetowymi zajmuje się **PIXLAB** - profesjonalne tworzenie stron WWW, aplikacji i systemów IT. Kontakt: tel. 58 888 99 00, www.pixlab.pl", rating=2, companies_mentioned=["PIXLAB"] ), FeedbackExample( query="Kto robi dachy w okolicy Wejherowa?", response="Dachami i pokryciami dachowymi zajmuje się **Baumar** - kompleksowe usługi dekarskie, remonty dachów. Kontakt: tel. 58 672 15 20, www.baumar.pl. Również **Pro-Bud** oferuje usługi budowlane w tym pokrycia dachowe.", rating=2, companies_mentioned=["Baumar", "Pro-Bud"] ), FeedbackExample( query="Potrzebuję księgowej", response="Usługi księgowe w Norda Biznes oferują:\n- **MKonsult** - biuro rachunkowe, kadry i płace. Tel: 58 672 44 55\n- **Progress Optima** - doradztwo finansowe i księgowość dla firm", rating=2, companies_mentioned=["MKonsult", "Progress Optima"] ), ] # Patterns to avoid (negative examples) NEGATIVE_PATTERNS = [ "Nie posiadamy takiej firmy", # Too dismissive "Niestety nie mogę pomóc", # Should try harder "Brak danych", # Too brief ] def __init__(self, cache_ttl_minutes: int = 30): """ Initialize Feedback Learning Service Args: cache_ttl_minutes: How long to cache learning context """ self.cache_ttl = timedelta(minutes=cache_ttl_minutes) self._cache: Optional[Dict] = None self._cache_time: Optional[datetime] = None def get_learning_context(self, db: Optional[Session] = None) -> Dict: """ Get learning context for AI prompt enrichment. Returns cached context or builds new one if expired. Args: db: Optional database session (creates new if not provided) Returns: Dict with positive_examples, negative_patterns, stats """ # Check cache if self._cache and self._cache_time: if datetime.now() - self._cache_time < self.cache_ttl: return self._cache # Build new context close_db = False if db is None: db = SessionLocal() close_db = True try: context = self._build_learning_context(db) # Update cache self._cache = context self._cache_time = datetime.now() return context finally: if close_db: db.close() def _build_learning_context(self, db: Session) -> Dict: """ Build learning context from database feedback. Args: db: Database session Returns: Learning context dict """ # Get positive examples from feedback positive_examples = self._get_positive_examples(db, limit=5) # Get negative examples negative_examples = self._get_negative_examples(db, limit=3) # Calculate stats stats = self._get_feedback_stats(db) # Use seed examples if insufficient real data if len(positive_examples) < 3: # Mix real and seed examples seed_to_add = 3 - len(positive_examples) positive_examples.extend(self.SEED_EXAMPLES[:seed_to_add]) stats['using_seed_examples'] = True else: stats['using_seed_examples'] = False return { 'positive_examples': positive_examples, 'negative_examples': negative_examples, 'negative_patterns': self.NEGATIVE_PATTERNS, 'stats': stats, 'generated_at': datetime.now().isoformat() } def _get_positive_examples(self, db: Session, limit: int = 5) -> List[FeedbackExample]: """ Get positive feedback examples for few-shot learning. Prioritizes: 1. Most recent positive feedback 2. With comments (more context) 3. Diverse queries (different topics) Args: db: Database session limit: Max examples to return Returns: List of FeedbackExample objects """ examples = [] # Query positive feedback (rating=2 = thumbs up) positive_messages = db.query(AIChatMessage).filter( AIChatMessage.role == 'assistant', AIChatMessage.feedback_rating == 2 ).order_by(desc(AIChatMessage.feedback_at)).limit(limit * 2).all() for msg in positive_messages: if len(examples) >= limit: break # Get the user query that preceded this response user_query = db.query(AIChatMessage).filter( AIChatMessage.conversation_id == msg.conversation_id, AIChatMessage.id < msg.id, AIChatMessage.role == 'user' ).order_by(desc(AIChatMessage.id)).first() if user_query: # Extract company names mentioned (simple heuristic) companies = self._extract_company_names(msg.content) examples.append(FeedbackExample( query=user_query.content, response=msg.content, rating=2, feedback_comment=msg.feedback_comment, companies_mentioned=companies )) return examples def _get_negative_examples(self, db: Session, limit: int = 3) -> List[FeedbackExample]: """ Get negative feedback examples to learn what to avoid. Args: db: Database session limit: Max examples to return Returns: List of FeedbackExample objects (with rating=1) """ examples = [] # Query negative feedback (rating=1 = thumbs down) negative_messages = db.query(AIChatMessage).filter( AIChatMessage.role == 'assistant', AIChatMessage.feedback_rating == 1 ).order_by(desc(AIChatMessage.feedback_at)).limit(limit).all() for msg in negative_messages: # Get the user query user_query = db.query(AIChatMessage).filter( AIChatMessage.conversation_id == msg.conversation_id, AIChatMessage.id < msg.id, AIChatMessage.role == 'user' ).order_by(desc(AIChatMessage.id)).first() if user_query: examples.append(FeedbackExample( query=user_query.content, response=msg.content[:200] + "..." if len(msg.content) > 200 else msg.content, rating=1, feedback_comment=msg.feedback_comment )) return examples def _get_feedback_stats(self, db: Session) -> Dict: """ Get feedback statistics. Args: db: Database session Returns: Stats dict """ total_responses = db.query(AIChatMessage).filter( AIChatMessage.role == 'assistant' ).count() with_feedback = db.query(AIChatMessage).filter( AIChatMessage.feedback_rating.isnot(None) ).count() positive_count = db.query(AIChatMessage).filter( AIChatMessage.feedback_rating == 2 ).count() negative_count = db.query(AIChatMessage).filter( AIChatMessage.feedback_rating == 1 ).count() return { 'total_responses': total_responses, 'with_feedback': with_feedback, 'positive_count': positive_count, 'negative_count': negative_count, 'feedback_rate': round(with_feedback / total_responses * 100, 1) if total_responses > 0 else 0, 'positive_rate': round(positive_count / with_feedback * 100, 1) if with_feedback > 0 else 0 } def _extract_company_names(self, text: str) -> List[str]: """ Extract company names from response text. Simple heuristic: looks for **bold** text which typically marks company names. Args: text: Response text Returns: List of company names """ import re # Find text between ** markers (markdown bold) pattern = r'\*\*([^*]+)\*\*' matches = re.findall(pattern, text) # Filter out non-company text (too short, common words) return [m for m in matches if len(m) > 2 and not m.lower() in ['kontakt', 'tel', 'www', 'email']] def format_for_prompt(self, context: Optional[Dict] = None) -> str: """ Format learning context as text for inclusion in AI prompt. Args: context: Learning context (fetches if not provided) Returns: Formatted string for prompt injection """ if context is None: context = self.get_learning_context() lines = [] # Add positive examples section if context['positive_examples']: lines.append("\n📚 PRZYKŁADY DOBRYCH ODPOWIEDZI (ucz się z nich):") for i, ex in enumerate(context['positive_examples'][:3], 1): lines.append(f"\nPrzykład {i}:") lines.append(f"Pytanie: {ex.query}") lines.append(f"Odpowiedź: {ex.response[:300]}{'...' if len(ex.response) > 300 else ''}") # Add negative patterns to avoid if context['negative_patterns']: lines.append("\n\n⚠️ UNIKAJ takich odpowiedzi:") for pattern in context['negative_patterns']: lines.append(f"- {pattern}") # Add guidance based on negative feedback if context['negative_examples']: lines.append("\n\n❌ Użytkownicy ocenili negatywnie odpowiedzi typu:") for ex in context['negative_examples'][:2]: if ex.feedback_comment: lines.append(f"- Pytanie: '{ex.query[:50]}...' - komentarz: {ex.feedback_comment}") return '\n'.join(lines) def invalidate_cache(self): """Force cache refresh on next request""" self._cache = None self._cache_time = None def record_feedback_used(self, message_id: int, examples_used: List[int]): """ Record which feedback examples were used for a response. This helps track the effectiveness of few-shot learning. Args: message_id: The AI response message ID examples_used: List of message IDs that were used as examples """ # TODO: Implement tracking in separate table for analytics logger.info(f"Feedback examples {examples_used} used for message {message_id}") # Global service instance _feedback_service: Optional[FeedbackLearningService] = None def get_feedback_learning_service() -> FeedbackLearningService: """Get or create global FeedbackLearningService instance""" global _feedback_service if _feedback_service is None: _feedback_service = FeedbackLearningService() return _feedback_service