nordabiz/feedback_learning_service.py

#!/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: Maciej Pienczyn, InPi sp. z o.o.
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