Skip to content
Back to Blog
1 min read

Human Feedback Loops: Improving AI with User Input

I wrote “Human Feedback Loops: Improving AI with User Input” to share practical, production-minded guidance on this topic.

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.\n\n## Takeaways\n\nAdd a concise, personal takeaway and recommended next steps here.\n

Michael John Peña

Michael John Peña

Senior Data Engineer based in Sydney. Writing about data, cloud, and technology.