Alpha108/GenerativeEngineOptimization
0
1"""2GEO Scoring Module3Analyzes content for Generative Engine Optimization (GEO) performance4"""5 6import json7from typing import Dict, Any, List8from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate9 10 11class GEOScorer:12 """Main class for calculating GEO scores and analysis"""13 14 def __init__(self, llm):15 self.llm = llm16 self.setup_prompts()17 18 def setup_prompts(self):19 """Initialize prompts for different types of analysis"""20 21 # Main GEO analysis prompt22 self.geo_analysis_prompt = (23 "You are a Generative Engine Optimization (GEO) Specialist. Your task is to critically analyze the input content for its effectiveness in AI-powered search engines and large language model (LLM) systems. "24 "Evaluate the content using the following GEO criteria, assigning a score from 1 to 10 for each: \n\n"25 "1. AI Search Visibility - How likely is the content to be surfaced by AI search engines?\n"26 "2. Query Intent Matching - How well does the content align with common user queries?\n"27 "3. Factual Accuracy & Authority - How trustworthy and authoritative is the information?\n"28 "4. Conversational Readiness - Is the content well-suited for AI chat responses?\n"29 "5. Semantic Richness - Does the content effectively use relevant semantic keywords?\n"30 "6. Context Completeness - Is the content self-contained and does it provide complete answers?\n"31 "7. Citation Worthiness - How likely is the content to be cited by AI systems?\n"32 "8. Multi-Query Coverage - Does the content address multiple related questions?\n\n"33 "Also provide:\n"34 "- Key topics and entities mentioned\n"35 "- Missing information or content gaps\n"36 "- Specific optimization opportunities\n"37 "- Actionable enhancement recommendations\n\n"38 "Respond strictly in JSON format using the structure below (double curly braces shown here to escape string formatting, do NOT include them in actual output):\n\n"39 "{{\n"40 " \"geo_scores\": {{\n"41 " \"ai_search_visibility\": 0.0,\n"42 " \"query_intent_matching\": 0.0,\n"43 " \"factual_accuracy\": 0.0,\n"44 " \"conversational_readiness\": 0.0,\n"45 " \"semantic_richness\": 0.0,\n"46 " \"context_completeness\": 0.0,\n"47 " \"citation_worthiness\": 0.0,\n"48 " \"multi_query_coverage\": 0.0\n"49 " }},\n"50 " \"overall_geo_score\": 0.0,\n"51 " \"primary_topics\": [\"topic1\", \"topic2\"],\n"52 " \"entities\": [\"entity1\", \"entity2\"],\n"53 " \"missing_gaps\": [\"gap1\", \"gap2\"],\n"54 " \"optimization_opportunities\": [\n"55 " {{\n"56 " \"type\": \"semantic_enhancement\",\n"57 " \"description\": \"Describe the improvement opportunity\",\n"58 " \"priority\": \"high\"\n"59 " }}\n"60 " ],\n"61 " \"recommendations\": [\n"62 " \"Write clear and specific suggestions to improve the content\"\n"63 " ]\n"64 "}}"65 )66 67 # Quick scoring prompt for faster analysis68 self.quick_score_prompt = (69 "You are an AI Search Optimization Analyst. Evaluate the given content and provide a quick scoring based on key criteria.\n"70 "Rate each of the following from 1 to 10:\n"71 "1. AI Search Visibility\n"72 "2. Query Intent Matching\n"73 "3. Conversational Readiness\n"74 "4. Citation Worthiness\n\n"75 "{{\n"76 " \"scores\": {{\n"77 " \"ai_search_visibility\": 0.0,\n"78 " \"query_intent_matching\": 0.0,\n"79 " \"conversational_readiness\": 0.0,\n"80 " \"citation_worthiness\": 0.0\n"81 " }},\n"82 " \"overall_score\": 0.0,\n"83 " \"top_recommendation\": \"Provide the most critical improvement needed\"\n"84 "}}"85 )86 87 # Competitive analysis prompt88 self.competitive_prompt = (89 "Compare these content pieces for GEO performance. Identify which performs better for AI search and why.\n"90 "Content A: {content_a}\n"91 "Content B: {content_b}\n"92 "Provide analysis in JSON:\n"93 "{{\n"94 " \"winner\": \"A\" or \"B\",\n"95 " \"score_comparison\": {{\n"96 " \"content_a_score\": 7.5,\n"97 " \"content_b_score\": 8.2\n"98 " }},\n"99 " \"key_differences\": [\"difference1\", \"difference2\"],\n"100 " \"improvement_suggestions\": {{\n"101 " \"content_a\": [\"suggestion1\"],\n"102 " \"content_b\": [\"suggestion1\"]\n"103 " }}\n"104 "}}"105 )106 107 def analyze_page_geo(self, content: str, title: str, detailed: bool = True) -> Dict[str, Any]:108 """109 Analyze a single page for GEO performance110 """111 try:112 # Choose prompt based on detail level113 if detailed:114 system_prompt = self.geo_analysis_prompt115 user_message = f"Title: {title}\n\nContent: {content[:8000]}"116 else:117 system_prompt = self.quick_score_prompt118 user_message = f"Title: {title}\n\nContent: {content[:4000]}"119 120 # Build prompt and run analysis121 prompt_template = ChatPromptTemplate.from_messages([122 SystemMessagePromptTemplate.from_template(system_prompt),123 HumanMessagePromptTemplate.from_template(user_message)124 ])125 # ("user", user_message)126 # ("system", system_prompt),127 chain = prompt_template | self.llm128 result = chain.invoke({}) # No variables needed129 130 # Extract and parse result131 result_content = result.content if hasattr(result, 'content') else str(result)132 parsed_result = self._parse_llm_response(result_content)133 134 # Add metadata135 parsed_result.update({136 'analyzed_title': title,137 'content_length': len(content),138 'word_count': len(content.split()),139 'analysis_type': 'detailed' if detailed else 'quick'140 })141 142 return parsed_result143 144 except Exception as e:145 return {'error': f"GEO analysis failed: {str(e)}"}146 147 def analyze_multiple_pages(self, pages_data: List[Dict[str, Any]], detailed: bool = True) -> List[Dict[str, Any]]:148 """149 Analyze multiple pages and return consolidated results150 151 Args:152 pages_data (List[Dict]): List of page data with content and metadata153 detailed (bool): Whether to perform detailed analysis154 155 Returns:156 List[Dict]: List of GEO analysis results157 """158 results = []159 160 for i, page_data in enumerate(pages_data):161 try:162 content = page_data.get('content', '')163 title = page_data.get('title', f'Page {i+1}')164 165 analysis = self.analyze_page_geo(content, title, detailed)166 167 # Add page-specific metadata168 analysis.update({169 'page_url': page_data.get('url', ''),170 'page_index': i,171 'source_word_count': page_data.get('word_count', 0)172 })173 174 results.append(analysis)175 176 except Exception as e:177 results.append({178 'page_index': i,179 'page_url': page_data.get('url', ''),180 'error': f"Analysis failed: {str(e)}"181 })182 183 return results184 185 def compare_content_geo(self, content_a: str, content_b: str, titles: tuple = None) -> Dict[str, Any]:186 """187 Compare two pieces of content for GEO performance188 189 Args:190 content_a (str): First content to compare191 content_b (str): Second content to compare 192 titles (tuple): Optional titles for the content pieces193 194 Returns:195 Dict: Comparison analysis results196 """197 try:198 title_a, title_b = titles if titles else ("Content A", "Content B")199 200 prompt_template = ChatPromptTemplate.from_messages([201 ("system", self.competitive_prompt),202 ("user", "")203 ])204 205 # Format the competitive analysis prompt206 formatted_prompt = self.competitive_prompt.format(207 content_a=f"Title: {title_a}\nContent: {content_a[:4000]}",208 content_b=f"Title: {title_b}\nContent: {content_b[:4000]}"209 )210 211 chain = ChatPromptTemplate.from_messages([212 ("system", formatted_prompt),213 ("user", "Perform the comparison analysis.")214 ]) | self.llm215 216 result = chain.invoke({})217 result_content = result.content if hasattr(result, 'content') else str(result)218 219 return self._parse_llm_response(result_content)220 221 except Exception as e:222 return {'error': f"Comparison analysis failed: {str(e)}"}223 224 def calculate_aggregate_scores(self, individual_results: List[Dict[str, Any]]) -> Dict[str, Any]:225 """226 Calculate aggregate GEO scores from multiple page analyses227 228 Args:229 individual_results (List[Dict]): List of individual page analysis results230 231 Returns:232 Dict: Aggregate scores and insights233 """234 try:235 valid_results = [r for r in individual_results if 'geo_scores' in r and not r.get('error')]236 237 if not valid_results:238 return {'error': 'No valid results to aggregate'}239 240 # Calculate average scores241 score_keys = list(valid_results[0]['geo_scores'].keys())242 avg_scores = {}243 244 for key in score_keys:245 scores = [r['geo_scores'][key] for r in valid_results if key in r['geo_scores']]246 avg_scores[key] = sum(scores) / len(scores) if scores else 0247 248 overall_avg = sum(avg_scores.values()) / len(avg_scores) if avg_scores else 0249 250 # Collect all recommendations and opportunities251 all_recommendations = []252 all_opportunities = []253 all_topics = []254 all_entities = []255 256 for result in valid_results:257 all_recommendations.extend(result.get('recommendations', []))258 all_opportunities.extend(result.get('optimization_opportunities', []))259 all_topics.extend(result.get('primary_topics', []))260 all_entities.extend(result.get('entities', []))261 262 # Remove duplicates and prioritize263 unique_recommendations = list(set(all_recommendations))264 unique_topics = list(set(all_topics))265 unique_entities = list(set(all_entities))266 267 # Find highest and lowest performing areas268 best_score = max(avg_scores.items(), key=lambda x: x[1]) if avg_scores else ('none', 0)269 worst_score = min(avg_scores.items(), key=lambda x: x[1]) if avg_scores else ('none', 0)270 271 return {272 'aggregate_scores': avg_scores,273 'overall_score': overall_avg,274 'pages_analyzed': len(valid_results),275 'best_performing_metric': {276 'metric': best_score[0],277 'score': best_score[1]278 },279 'lowest_performing_metric': {280 'metric': worst_score[0],281 'score': worst_score[1]282 },283 'consolidated_recommendations': unique_recommendations[:10],284 'all_topics': unique_topics,285 'all_entities': unique_entities,286 'high_priority_opportunities': [287 opp for opp in all_opportunities 288 if opp.get('priority') == 'high'289 ][:5],290 'score_distribution': self._calculate_score_distribution(avg_scores)291 }292 293 except Exception as e:294 return {'error': f"Aggregation failed: {str(e)}"}295 296 def generate_geo_report(self, analysis_results: Dict[str, Any], website_url: str = None) -> Dict[str, Any]:297 """298 Generate a comprehensive GEO report299 300 Args:301 analysis_results (Dict): Results from aggregate analysis302 website_url (str): Optional website URL for context303 304 Returns:305 Dict: Comprehensive GEO report306 """307 try:308 report = {309 'report_metadata': {310 'generated_at': self._get_timestamp(),311 'website_url': website_url,312 'analysis_type': 'GEO Performance Report'313 },314 'executive_summary': self._generate_executive_summary(analysis_results),315 'detailed_scores': analysis_results.get('aggregate_scores', {}),316 'performance_insights': self._generate_performance_insights(analysis_results),317 'actionable_recommendations': self._prioritize_recommendations(318 analysis_results.get('consolidated_recommendations', [])319 ),320 'optimization_roadmap': self._create_optimization_roadmap(analysis_results),321 'competitive_position': self._assess_competitive_position(analysis_results),322 'technical_details': {323 'pages_analyzed': analysis_results.get('pages_analyzed', 0),324 'overall_score': analysis_results.get('overall_score', 0),325 'score_distribution': analysis_results.get('score_distribution', {})326 }327 }328 329 return report330 331 except Exception as e:332 return {'error': f"Report generation failed: {str(e)}"}333 334 def _parse_llm_response(self, response_text: str) -> Dict[str, Any]:335 """Parse LLM response and extract JSON content"""336 try:337 # Find JSON content in the response338 json_start = response_text.find('{')339 json_end = response_text.rfind('}') + 1340 341 if json_start != -1 and json_end != -1:342 json_str = response_text[json_start:json_end]343 return json.loads(json_str)344 else:345 # If no JSON found, return the raw response346 return {'raw_response': response_text, 'parsing_error': 'No JSON found'}347 348 except json.JSONDecodeError as e:349 return {'raw_response': response_text, 'parsing_error': f'JSON decode error: {str(e)}'}350 except Exception as e:351 return {'raw_response': response_text, 'parsing_error': f'Unexpected error: {str(e)}'}352 353 def _calculate_score_distribution(self, scores: Dict[str, float]) -> Dict[str, Any]:354 """Calculate distribution of scores for insights"""355 if not scores:356 return {}357 358 score_values = list(scores.values())359 360 return {361 'highest_score': max(score_values),362 'lowest_score': min(score_values),363 'average_score': sum(score_values) / len(score_values),364 'score_range': max(score_values) - min(score_values),365 'scores_above_7': len([s for s in score_values if s >= 7.0]),366 'scores_below_5': len([s for s in score_values if s < 5.0])367 }368 369 def _generate_executive_summary(self, analysis_results: Dict[str, Any]) -> str:370 """Generate executive summary based on analysis results"""371 overall_score = analysis_results.get('overall_score', 0)372 pages_analyzed = analysis_results.get('pages_analyzed', 0)373 374 if overall_score >= 8.0:375 performance = "excellent"376 elif overall_score >= 6.5:377 performance = "good"378 elif overall_score >= 5.0:379 performance = "moderate"380 else:381 performance = "needs improvement"382 383 return f"Analysis of {pages_analyzed} pages shows {performance} GEO performance with an overall score of {overall_score:.1f}/10. Key opportunities exist in {analysis_results.get('lowest_performing_metric', {}).get('metric', 'multiple areas')}."384 385 def _generate_performance_insights(self, analysis_results: Dict[str, Any]) -> List[str]:386 """Generate performance insights based on analysis"""387 insights = []388 389 best_metric = analysis_results.get('best_performing_metric', {})390 worst_metric = analysis_results.get('lowest_performing_metric', {})391 392 if best_metric.get('score', 0) >= 8.0:393 insights.append(f"Strong performance in {best_metric.get('metric', 'unknown')} (score: {best_metric.get('score', 0):.1f})")394 395 if worst_metric.get('score', 10) < 6.0:396 insights.append(f"Significant improvement needed in {worst_metric.get('metric', 'unknown')} (score: {worst_metric.get('score', 0):.1f})")397 398 score_dist = analysis_results.get('score_distribution', {})399 if score_dist.get('score_range', 0) > 3.0:400 insights.append("High variability in scores indicates inconsistent optimization across metrics")401 402 return insights403 404 def _prioritize_recommendations(self, recommendations: List[str]) -> List[Dict[str, Any]]:405 """Prioritize recommendations based on impact potential"""406 prioritized = []407 408 # Simple prioritization based on keywords409 high_impact_keywords = ['semantic', 'structure', 'authority', 'factual']410 medium_impact_keywords = ['readability', 'clarity', 'format']411 412 for i, rec in enumerate(recommendations):413 priority = 'low'414 if any(keyword in rec.lower() for keyword in high_impact_keywords):415 priority = 'high'416 elif any(keyword in rec.lower() for keyword in medium_impact_keywords):417 priority = 'medium'418 419 prioritized.append({420 'recommendation': rec,421 'priority': priority,422 'order': i + 1423 })424 425 # Sort by priority426 priority_order = {'high': 1, 'medium': 2, 'low': 3}427 prioritized.sort(key=lambda x: priority_order[x['priority']])428 429 return prioritized430 431 def _create_optimization_roadmap(self, analysis_results: Dict[str, Any]) -> Dict[str, List[str]]:432 """Create a phased optimization roadmap"""433 roadmap = {434 'immediate_actions': [],435 'short_term_goals': [],436 'long_term_strategy': []437 }438 439 overall_score = analysis_results.get('overall_score', 0)440 worst_metric = analysis_results.get('lowest_performing_metric', {})441 442 # Immediate actions based on worst performing metric443 if worst_metric.get('score', 10) < 5.0:444 roadmap['immediate_actions'].append(f"Address critical issues in {worst_metric.get('metric', 'low-scoring areas')}")445 446 # Short-term goals447 if overall_score < 7.0:448 roadmap['short_term_goals'].append("Improve overall GEO score to above 7.0")449 roadmap['short_term_goals'].append("Enhance content structure and semantic richness")450 451 # Long-term strategy452 roadmap['long_term_strategy'].append("Establish consistent GEO optimization process")453 roadmap['long_term_strategy'].append("Monitor and track AI search performance")454 455 return roadmap456 457 def _assess_competitive_position(self, analysis_results: Dict[str, Any]) -> Dict[str, Any]:458 """Assess competitive position based on scores"""459 overall_score = analysis_results.get('overall_score', 0)460 461 if overall_score >= 8.5:462 position = "market_leader"463 description = "Content is highly optimized for AI search engines"464 elif overall_score >= 7.0:465 position = "competitive"466 description = "Content performs well but has room for improvement"467 elif overall_score >= 5.5:468 position = "average"469 description = "Content meets basic standards but lacks optimization"470 else:471 position = "needs_work"472 description = "Content requires significant optimization for AI search"473 474 return {475 'position': position,476 'description': description,477 'score': overall_score,478 'percentile_estimate': min(overall_score * 10, 100) # Rough percentile estimate479 }480 481 def _get_timestamp(self) -> str:482 """Get current timestamp"""483 from datetime import datetime484 return datetime.now().strftime('%Y-%m-%d %H:%M:%S')