Spaces:
Sleeping
Sleeping
File size: 9,194 Bytes
02e306a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 |
import json
from datetime import datetime
from typing import Dict, List, Tuple
from collections import defaultdict
import statistics
class FeedbackAnalyzer:
"""Analyze feedback patterns and provide improvement recommendations"""
def __init__(self, feedback_file: str = "feedback_store.json"):
self.feedback_file = feedback_file
self.feedback_data = self._load_feedback()
def _load_feedback(self) -> List[Dict]:
"""Load feedback from JSON file"""
try:
with open(self.feedback_file, 'r', encoding='utf-8') as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return []
def analyze_patterns(self) -> Dict[str, any]:
"""Analyze patterns in feedback data"""
if not self.feedback_data:
return {"error": "No feedback data available"}
analysis = {
"total_feedback": len(self.feedback_data),
"average_rating": 0,
"tone_performance": defaultdict(list),
"platform_performance": defaultdict(list),
"tone_platform_combo": defaultdict(list),
"low_performing_patterns": [],
"high_performing_patterns": [],
"recommendations": []
}
# Collect ratings by different dimensions
all_ratings = []
for entry in self.feedback_data:
rating = entry.get("rating", 0)
tone = entry.get("tone", "unknown")
platforms = entry.get("platforms", [])
all_ratings.append(rating)
analysis["tone_performance"][tone].append(rating)
for platform in platforms:
analysis["platform_performance"][platform].append(rating)
combo_key = f"{tone}_{platform}"
analysis["tone_platform_combo"][combo_key].append(rating)
# Calculate averages
analysis["average_rating"] = statistics.mean(all_ratings) if all_ratings else 0
# Analyze tone performance
tone_stats = {}
for tone, ratings in analysis["tone_performance"].items():
if ratings:
avg = statistics.mean(ratings)
tone_stats[tone] = {
"average": avg,
"count": len(ratings),
"std_dev": statistics.stdev(ratings) if len(ratings) > 1 else 0
}
analysis["tone_stats"] = tone_stats
# Analyze platform performance
platform_stats = {}
for platform, ratings in analysis["platform_performance"].items():
if ratings:
avg = statistics.mean(ratings)
platform_stats[platform] = {
"average": avg,
"count": len(ratings),
"std_dev": statistics.stdev(ratings) if len(ratings) > 1 else 0
}
analysis["platform_stats"] = platform_stats
# Find patterns
combo_stats = {}
for combo, ratings in analysis["tone_platform_combo"].items():
if ratings:
avg = statistics.mean(ratings)
combo_stats[combo] = {
"average": avg,
"count": len(ratings),
"ratings": ratings
}
# Identify low and high performers
if avg < 2.5 and len(ratings) >= 2:
analysis["low_performing_patterns"].append({
"pattern": combo,
"average_rating": avg,
"sample_size": len(ratings)
})
elif avg >= 4.0 and len(ratings) >= 2:
analysis["high_performing_patterns"].append({
"pattern": combo,
"average_rating": avg,
"sample_size": len(ratings)
})
analysis["combo_stats"] = combo_stats
# Generate recommendations
analysis["recommendations"] = self._generate_recommendations(analysis)
return analysis
def _generate_recommendations(self, analysis: Dict) -> List[str]:
"""Generate actionable recommendations based on analysis"""
recommendations = []
# Tone recommendations
tone_stats = analysis.get("tone_stats", {})
for tone, stats in tone_stats.items():
if stats["average"] < 3.0:
recommendations.append(
f"Consider adjusting '{tone}' tone guidelines - average rating is {stats['average']:.2f}"
)
elif stats["average"] > 4.5:
recommendations.append(
f"'{tone}' tone is performing excellently (avg: {stats['average']:.2f}) - use as reference"
)
# Platform recommendations
platform_stats = analysis.get("platform_stats", {})
for platform, stats in platform_stats.items():
if stats["std_dev"] > 1.5:
recommendations.append(
f"{platform} shows high variance (σ={stats['std_dev']:.2f}) - consider more consistent approach"
)
# Combo recommendations
for pattern in analysis.get("low_performing_patterns", []):
tone, platform = pattern["pattern"].split("_")
recommendations.append(
f"{tone} tone on {platform} is underperforming (avg: {pattern['average_rating']:.2f}) - needs revision"
)
for pattern in analysis.get("high_performing_patterns", []):
tone, platform = pattern["pattern"].split("_")
recommendations.append(
f"{tone} tone on {platform} is a winning combination (avg: {pattern['average_rating']:.2f})"
)
# General recommendations
overall_avg = analysis.get("average_rating", 0)
if overall_avg < 3.5:
recommendations.append(
"Overall performance needs improvement - consider reviewing prompt templates"
)
if len(analysis.get("feedback_data", [])) < 10:
recommendations.append(
"Limited feedback data - collect more samples for reliable patterns"
)
return recommendations
def get_adaptive_weights(self) -> Dict[str, float]:
"""Generate adaptive weights for prompt building based on feedback"""
analysis = self.analyze_patterns()
weights = {}
# Base weights
default_weight = 1.0
# Adjust weights based on performance
for combo, stats in analysis.get("combo_stats", {}).items():
if stats["count"] >= 2: # Only adjust if we have enough data
performance_ratio = stats["average"] / 5.0 # Normalize to 0-1
weights[combo] = 0.5 + (performance_ratio * 0.5) # Scale between 0.5-1.0
else:
weights[combo] = default_weight
return weights
def get_time_based_trends(self) -> Dict[str, any]:
"""Analyze trends over time"""
if not self.feedback_data:
return {"error": "No feedback data available"}
# Sort by timestamp
sorted_feedback = sorted(
self.feedback_data,
key=lambda x: datetime.fromisoformat(x.get("timestamp", "2024-01-01"))
)
# Group by day
daily_ratings = defaultdict(list)
for entry in sorted_feedback:
timestamp = datetime.fromisoformat(entry.get("timestamp", "2024-01-01"))
day = timestamp.date().isoformat()
daily_ratings[day].append(entry.get("rating", 0))
# Calculate daily averages
trends = {}
for day, ratings in daily_ratings.items():
trends[day] = {
"average_rating": statistics.mean(ratings),
"count": len(ratings)
}
return trends
def export_insights(self, output_file: str = "feedback_insights.json"):
"""Export analysis insights to a file"""
analysis = self.analyze_patterns()
trends = self.get_time_based_trends()
weights = self.get_adaptive_weights()
insights = {
"generated_at": datetime.now().isoformat(),
"analysis": analysis,
"trends": trends,
"adaptive_weights": weights
}
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(insights, f, indent=2)
return insights |