Human Feedback Loops: Improving AI with User Input

Human feedback is essential for improving AI systems. Here’s how to implement effective feedback loops.

Human Feedback Implementation

from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime
from enum import Enum

class FeedbackType(Enum):
    THUMBS_UP = "positive"
    THUMBS_DOWN = "negative"
    CORRECTION = "correction"
    RATING = "rating"

@dataclass
class UserFeedback:
    interaction_id: str
    feedback_type: FeedbackType
    value: any
    comment: Optional[str] = None
    timestamp: datetime = None

class HumanFeedbackSystem:
    def __init__(self):
        self.feedback_store = FeedbackStore()
        self.aggregator = FeedbackAggregator()

    async def collect_feedback(self, feedback: UserFeedback):
        """Collect and process user feedback."""
        feedback.timestamp = datetime.now()

        # Store feedback
        await self.feedback_store.save(feedback)

        # Update aggregated metrics
        await self.aggregator.update(feedback)

        # Trigger improvement workflows if needed
        await self.check_improvement_triggers(feedback)

    async def check_improvement_triggers(self, feedback: UserFeedback):
        """Check if feedback triggers improvement actions."""
        # Get recent feedback for this query pattern
        recent = await self.feedback_store.get_similar_feedback(
            feedback.interaction_id,
            limit=10
        )

        negative_rate = sum(1 for f in recent if f.feedback_type == FeedbackType.THUMBS_DOWN) / len(recent)

        if negative_rate > 0.3:
            await self.trigger_review(feedback, recent)

    async def trigger_review(self, feedback: UserFeedback, related: List[UserFeedback]):
        """Trigger human review for problematic patterns."""
        review_task = {
            "trigger_feedback": feedback,
            "related_feedback": related,
            "priority": "high" if len(related) > 5 else "medium",
            "suggested_action": self.suggest_action(related)
        }

        await self.create_review_task(review_task)

    def suggest_action(self, feedback_list: List[UserFeedback]) -> str:
        """Suggest improvement action based on feedback patterns."""
        corrections = [f for f in feedback_list if f.feedback_type == FeedbackType.CORRECTION]

        if corrections:
            return "Update knowledge base with corrections"

        return "Review and update prompt/instructions"

    async def get_improvement_insights(self) -> Dict:
        """Generate insights for AI improvement."""
        feedback = await self.feedback_store.get_all_recent(days=30)

        return {
            "total_feedback": len(feedback),
            "positive_rate": sum(1 for f in feedback if f.feedback_type == FeedbackType.THUMBS_UP) / len(feedback),
            "common_issues": self.identify_common_issues(feedback),
            "improvement_opportunities": self.identify_opportunities(feedback),
            "user_corrections": self.extract_corrections(feedback)
        }

    def identify_common_issues(self, feedback: List[UserFeedback]) -> List[Dict]:
        """Identify common issues from negative feedback."""
        negative = [f for f in feedback if f.feedback_type == FeedbackType.THUMBS_DOWN]

        # Cluster by similarity
        clusters = self.cluster_feedback(negative)

        return [
            {
                "issue": cluster["representative"],
                "frequency": len(cluster["items"]),
                "sample_comments": [f.comment for f in cluster["items"][:3] if f.comment]
            }
            for cluster in clusters
        ]

    async def apply_corrections(self, corrections: List[Dict]):
        """Apply user corrections to improve system."""
        for correction in corrections:
            if correction["type"] == "knowledge":
                await self.update_knowledge_base(correction)
            elif correction["type"] == "behavior":
                await self.update_instructions(correction)

Systematic human feedback collection drives continuous AI improvement.