Upload 2 files

enhanced_prompt_builder.py

@@ -0,0 +1,131 @@
+from enhanced_retriever import EnhancedRetriever
+from enhanced_knowledge_graph import EnhancedKnowledgeGraph
+from feedback_analyzer import FeedbackAnalyzer
+from typing import List, Dict
+
+class EnhancedPromptBuilder:
+    """Enhanced prompt builder with all advanced features integrated"""
+    
+    def __init__(self):
+        self.retriever = EnhancedRetriever()
+        self.knowledge_graph = EnhancedKnowledgeGraph()
+        self.feedback_analyzer = FeedbackAnalyzer()
+        
+    def build_adaptive_prompt(self, ad_text: str, tone: str, platforms: List[str]) -> str:
+        """Build an adaptive prompt using all enhancement layers"""
+        
+        # 1. Get enhanced RAG results with relevance scores
+        rag_results = self.retriever.retrieve_with_relevance(tone, platforms)
+        formatted_guidance = self.retriever.format_guidance_with_scores(rag_results)
+        
+        # 2. Get knowledge graph insights with traversal
+        kg_insights = []
+        
+        # Get recommendations for each platform
+        for platform in platforms:
+            recommendations = self.knowledge_graph.get_recommendations(tone, platform)
+            kg_insights.append(f"\n{platform} Insights:")
+            kg_insights.append(f"  - Compatibility Score: {recommendations['compatibility_score']:.2f}")
+            
+            if recommendations['suggested_elements']:
+                kg_insights.append("  - Suggestions: " + ", ".join(recommendations['suggested_elements']))
+            
+            if recommendations['warnings']:
+                kg_insights.append("  - ⚠️ Warnings: " + ", ".join(recommendations['warnings']))
+                
+            if recommendations['creative_types']:
+                kg_insights.append("  - Recommended Creative Types: " + ", ".join(recommendations['creative_types']))
+            
+            # Add relationship explanations
+            relationship = self.knowledge_graph.explain_relationship(tone, platform)
+            kg_insights.append(f"  - Relationship: {relationship}")
+        
+        kg_insights_str = "\n".join(kg_insights)
+        
+        # 3. Get adaptive weights from feedback analysis
+        weights = self.feedback_analyzer.get_adaptive_weights()
+        
+        # 4. Add performance insights if available
+        analysis = self.feedback_analyzer.analyze_patterns()
+        performance_notes = []
+        
+        if analysis.get("recommendations"):
+            relevant_recs = [rec for rec in analysis["recommendations"] 
+                           if any(p.lower() in rec.lower() for p in platforms) or tone.lower() in rec.lower()]
+            if relevant_recs:
+                performance_notes.append("\nHistorical Performance Notes:")
+                performance_notes.extend([f"  - {rec}" for rec in relevant_recs[:3]])
+        
+        performance_str = "\n".join(performance_notes) if performance_notes else ""
+        
+        # 5. Build the enhanced prompt
+        platform_str = ", ".join(platforms)
+        
+        # Apply adaptive weights to emphasize better-performing combinations
+        weight_notes = []
+        for platform in platforms:
+            combo_key = f"{tone}_{platform}"
+            weight = weights.get(combo_key, 1.0)
+            if weight > 0.8:
+                weight_notes.append(f"  - {platform}: High confidence (historical success)")
+            elif weight < 0.6:
+                weight_notes.append(f"  - {platform}: Needs improvement (based on feedback)")
+        
+        weight_str = "\n".join(weight_notes) if weight_notes else ""
+        
+        prompt = f"""
+You are an expert ad copywriter with access to advanced AI assistance.
+
+TASK: Rewrite the following ad text in a {tone} tone and optimize it individually for: {platform_str}
+
+ORIGINAL AD TEXT: "{ad_text}"
+
+=== ENHANCED GUIDANCE (with Relevance Scores) ===
+{formatted_guidance}
+
+=== KNOWLEDGE GRAPH INSIGHTS ===
+{kg_insights_str}
+
+=== ADAPTIVE LEARNING INSIGHTS ===
+{weight_str}
+{performance_str}
+
+=== INSTRUCTIONS ===
+1. Maintain the core message and key information from the original ad
+2. Adapt the tone and style according to the guidelines above
+3. Consider the compatibility scores and warnings for each platform
+4. Use suggested elements where appropriate
+5. Avoid any warned elements or approaches
+6. Apply lessons from historical performance data
+
+OUTPUT FORMAT:
+Provide a separate, optimized version for each platform:
+
+{''.join([f'{p}:\n<Your rewritten ad text here>\n\n' for p in platforms])}
+
+Remember: Each platform version should be uniquely tailored while maintaining brand consistency.
+"""
+        
+        return prompt
+    
+    def get_improvement_suggestions(self) -> List[str]:
+        """Get suggestions for improving the system based on feedback"""
+        analysis = self.feedback_analyzer.analyze_patterns()
+        suggestions = []
+        
+        # Check overall performance
+        avg_rating = analysis.get("average_rating", 0)
+        if avg_rating < 3.5:
+            suggestions.append("Consider updating tone guidelines based on feedback patterns")
+            
+        # Check for problematic combinations
+        for pattern in analysis.get("low_performing_patterns", []):
+            tone, platform = pattern["pattern"].split("_")
+            suggestions.append(f"Review and update guidelines for {tone} tone on {platform}")
+            
+        # Suggest new relationships for KG
+        high_performers = analysis.get("high_performing_patterns", [])
+        if high_performers:
+            suggestions.append("Consider strengthening KG relationships for high-performing combinations")
+            
+        return suggestions 

feedback_analyzer.py

@@ -0,0 +1,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