import redis import json import logging from typing import Dict, List, Tuple import os import sys # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class RalphOptimizer: """ Analyzes completed Ralph stories, generates optimization recommendations, and saves findings to a markdown report. """ def __init__(self, redis_host: str = 'localhost', redis_port: int = 6379, redis_db: int = 0, redis_hash_name: str = 'genesis:ralph_completed', output_path: str = '/mnt/e/genesis-system/data/ralph_analysis.md'): """ Initializes the RalphOptimizer with Redis connection details and output path. """ self.redis_host = redis_host self.redis_port = redis_port self.redis_db = redis_db self.redis_hash_name = redis_hash_name self.output_path = output_path self.redis_client = None # Initialize redis_client to None def connect_to_redis(self) -> None: """ Connects to the Redis server. """ try: self.redis_client = redis.Redis(host=self.redis_host, port=self.redis_port, db=self.redis_db, decode_responses=True) self.redis_client.ping() # Check if the connection is successful logging.info("Successfully connected to Redis.") except redis.exceptions.ConnectionError as e: logging.error(f"Could not connect to Redis: {e}") raise def fetch_completed_stories(self) -> Dict[str, str]: """ Fetches all completed stories from Redis. Returns: A dictionary where keys are story IDs and values are JSON strings representing the story data. """ if self.redis_client is None: raise ValueError("Redis client not initialized. Call connect_to_redis() first.") try: stories = self.redis_client.hgetall(self.redis_hash_name) logging.info(f"Fetched {len(stories)} completed stories from Redis.") return stories except redis.exceptions.RedisError as e: logging.error(f"Error fetching stories from Redis: {e}") return {} def analyze_stories(self, stories: Dict[str, str]) -> Tuple[float, float, Dict[str, int], Dict[str, int]]: """ Analyzes the completed stories to calculate success rate, average cost, failure patterns, and story type distribution. Args: stories: A dictionary of completed stories. Returns: A tuple containing the success rate, average cost per story, failure patterns, and story type distribution. """ total_stories = len(stories) successful_stories = 0 total_cost = 0.0 failure_reasons: Dict[str, int] = {} story_types: Dict[str, int] = {} for story_id, story_json in stories.items(): try: story = json.loads(story_json) if story.get("passes", False): successful_stories += 1 total_cost += float(story.get("cost", 0.0)) failure_reason = story.get("failure_reason", "N/A") failure_reasons[failure_reason] = failure_reasons.get(failure_reason, 0) + 1 story_type = story.get("story_type", "Unknown") story_types[story_type] = story_types.get(story_type, 0) + 1 except json.JSONDecodeError as e: logging.error(f"Error decoding JSON for story {story_id}: {e}") except Exception as e: logging.error(f"Error processing story {story_id}: {e}") success_rate = (successful_stories / total_stories) * 100 if total_stories > 0 else 0.0 average_cost = total_cost / total_stories if total_stories > 0 else 0.0 return success_rate, average_cost, failure_reasons, story_types def generate_recommendations(self, success_rate: float, average_cost: float, failure_reasons: Dict[str, int], story_types: Dict[str, int]) -> List[str]: """ Generates recommendations to improve the Ralph system prompt based on the analysis. Args: success_rate: The success rate of the stories. average_cost: The average cost per story. failure_reasons: A dictionary of failure reasons and their counts. story_types: A dictionary of story types and their counts. Returns: A list of recommendations. """ recommendations = [] if success_rate < 80.0: recommendations.append(f"Success rate is {success_rate:.2f}%. Improve prompt clarity to reduce ambiguity.") if average_cost > 1.0: recommendations.append(f"Average cost per story is ${average_cost:.2f}. Optimize task decomposition for efficiency.") if failure_reasons: top_failures = sorted(failure_reasons.items(), key=lambda item: item[1], reverse=True)[:3] for reason, count in top_failures: recommendations.append(f"Top failure reason: '{reason}' ({count} occurrences). Adjust prompt to mitigate this.") else: recommendations.append("No failure reasons found. This is unusual; ensure failure tracking is working correctly.") if story_types: recommendations.append(f"Story type distribution: {story_types}") # Based on story type distribution, suggest focusing on the most common type for optimization most_common_story_type = max(story_types, key=story_types.get) recommendations.append(f"Consider focusing optimization efforts on the most common story type: '{most_common_story_type}'.") else: recommendations.append("No story types found. Ensure story type tracking is working correctly.") return recommendations def save_report(self, success_rate: float, average_cost: float, failure_reasons: Dict[str, int], story_types: Dict[str, int], recommendations: List[str]) -> None: """ Saves the analysis report to a markdown file. Args: success_rate: The success rate of the stories. average_cost: The average cost per story. failure_reasons: A dictionary of failure reasons and their counts. story_types: A dictionary of story types and their counts. recommendations: A list of recommendations. """ try: with open(self.output_path, "w") as f: f.write("# Ralph System Analysis Report\n\n") f.write(f"**Success Rate:** {success_rate:.2f}%\n\n") f.write(f"**Average Cost per Story:** ${average_cost:.2f}\n\n") f.write("## Failure Reasons\n") for reason, count in failure_reasons.items(): f.write(f"- {reason}: {count}\n") f.write("\n") f.write("## Story Types\n") for story_type, count in story_types.items(): f.write(f"- {story_type}: {count}\n") f.write("\n") f.write("## Recommendations\n") for recommendation in recommendations: f.write(f"- {recommendation}\n") logging.info(f"Report saved to {self.output_path}") except Exception as e: logging.error(f"Error saving report: {e}") def run_analysis(self) -> None: """ Runs the complete analysis and generates the report. """ try: self.connect_to_redis() stories = self.fetch_completed_stories() success_rate, average_cost, failure_reasons, story_types = self.analyze_stories(stories) recommendations = self.generate_recommendations(success_rate, average_cost, failure_reasons, story_types) self.save_report(success_rate, average_cost, failure_reasons, story_types, recommendations) except Exception as e: logging.error(f"Analysis failed: {e}")