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