Update app.py

app.py

@@ -459,59 +459,133 @@ def _fallback_image_analysis(prompt: str, review_guidelines: str) -> str:
         "original_prompt": prompt
     })
 
-def generate_and_review_marketing_image(prompt: str, style: str = "realistic", review_guidelines: str = "") -> str:
+def generate_and_review_marketing_image(prompt: str, style: str = "realistic", review_guidelines: str = "", max_retries: int = 3, review_threshold: float = 0.8) -> str:
     """
-    Complete workflow: Generate a marketing image and provide quality review.
+    Complete workflow: Generate a marketing image and provide quality review with iterations.
     
     Args:
         prompt (str): Description of the marketing image to generate
         style (str): Art style for the image (realistic, artistic, cartoon, photographic, illustration)
         review_guidelines (str): Specific guidelines for marketing review
+        max_retries (int): Maximum number of generation attempts
+        review_threshold (float): Minimum quality score required for approval
         
     Returns:
         str: JSON string containing image, review, and recommendations
     """
     logger.info(f"🎭 Starting complete marketing workflow for: {prompt}")
+    logger.info(f"🔄 Max retries: {max_retries}, Review threshold: {review_threshold}")
+    
+    workflow_history = []
+    best_result = None
+    best_score = 0.0
     
     try:
-        # Step 1: Generate the image
-        generation_response = generate_marketing_image(prompt, style)
-        generation_data = json.loads(generation_response)
-        
-        if not generation_data.get("success", False):
-            return generation_response  # Return error
-        
-        # Step 2: Analyze the generated image with Gemini Vision
-        image_url = generation_data.get("image_url", "")
-        analysis_response = analyze_marketing_image_with_gemini(image_url, prompt, review_guidelines)
-        analysis_data = json.loads(analysis_response)
+        for iteration in range(1, max_retries + 1):
+            logger.info(f"🔄 Iteration {iteration} of {max_retries}")
+            
+            # Step 1: Generate the image
+            generation_response = generate_marketing_image(prompt, style)
+            generation_data = json.loads(generation_response)
+            
+            if not generation_data.get("success", False):
+                logger.error(f"Generation failed on iteration {iteration}: {generation_data.get('error')}")
+                workflow_history.append({
+                    "iteration": iteration,
+                    "generation_status": "failed",
+                    "review_score": 0.0,
+                    "error": generation_data.get('error', 'Unknown error')
+                })
+                continue
+            
+            # Step 2: Analyze the generated image with Gemini Vision
+            image_url = generation_data.get("image_url", "")
+            analysis_response = analyze_marketing_image_with_gemini(image_url, prompt, review_guidelines)
+            analysis_data = json.loads(analysis_response)
+            
+            current_score = analysis_data.get("overall_score", 0.0)
+            logger.info(f"📊 Iteration {iteration} score: {current_score:.2f} (threshold: {review_threshold})")
+            
+            # Record this iteration
+            workflow_history.append({
+                "iteration": iteration,
+                "generation_status": "success",
+                "review_score": current_score,
+                "review_method": analysis_data.get("analysis_method", "unknown"),
+                "recommendations": analysis_data.get("recommendations", [])[:3]  # Top 3 for history
+            })
+            
+            # Create result for this iteration
+            current_result = {
+                "generation_data": generation_data,
+                "analysis_data": analysis_data,
+                "image_url": image_url,
+                "score": current_score,
+                "iteration": iteration
+            }
+            
+            # Keep track of best result
+            if current_score > best_score:
+                best_result = current_result
+                best_score = current_score
+            
+            # Check if threshold is met
+            if current_score >= review_threshold:
+                logger.info(f"✅ Quality threshold met on iteration {iteration}! Score: {current_score:.2f}")
+                best_result = current_result  # Use this result since it passes threshold
+                break
+            else:
+                logger.info(f"⚠️ Score {current_score:.2f} below threshold {review_threshold}. {'Retrying...' if iteration < max_retries else 'Max attempts reached.'}")
+                
+                # Enhance prompt for next iteration based on feedback
+                if iteration < max_retries:
+                    missing_elements = analysis_data.get("missing_elements", [])
+                    violations = analysis_data.get("violations", [])
+                    if missing_elements:
+                        prompt += f" Including: {', '.join(missing_elements[:2])}"
+                    if violations and "english" in review_guidelines.lower():
+                        prompt += " with English signage and text"
+        
+        # Use best result if we have one
+        if not best_result:
+            return json.dumps({
+                "success": False,
+                "error": "All generation attempts failed",
+                "workflow_history": workflow_history
+            })
+        
+        # Build final result
+        final_passed = best_result["score"] >= review_threshold
+        final_status = "passed" if final_passed else "needs_improvement"
         
-        # Combine results
         workflow_result = {
             "success": True,
             "image": {
-                "url": generation_data.get("image_url", ""),
-                "data": generation_data.get("image_url", ""),
+                "url": best_result["image_url"],
+                "data": best_result["image_url"],
                 "prompt": prompt,
                 "style": style
             },
             "review": {
-                "quality_score": analysis_data.get("overall_score", 0.7),
-                "final_status": "passed" if analysis_data.get("overall_score", 0) > 0.7 else "needs_improvement",
-                "iterations": 1,
-                "passed": analysis_data.get("overall_score", 0) > 0.7,
-                "recommendations": analysis_data.get("recommendations", []),
-                "analysis_details": analysis_data
+                "quality_score": best_result["score"],
+                "final_status": final_status,
+                "iterations": len(workflow_history),
+                "passed": final_passed,
+                "recommendations": best_result["analysis_data"].get("recommendations", []),
+                "analysis_details": best_result["analysis_data"],
+                "workflow_history": workflow_history
             },
             "metadata": {
-                "generation_method": generation_data.get("generation_method", "unknown"),
-                "real_ai_generation": generation_data.get("real_ai_generation", False),
-                "review_method": analysis_data.get("analysis_method", "unknown"),
-                "workflow_type": "gradio_mcp_server"
+                "generation_method": best_result["generation_data"].get("generation_method", "unknown"),
+                "real_ai_generation": best_result["generation_data"].get("real_ai_generation", False),
+                "review_method": best_result["analysis_data"].get("analysis_method", "unknown"),
+                "workflow_type": "gradio_mcp_server",
+                "best_iteration": best_result["iteration"],
+                "threshold_met": final_passed
             }
         }
         
-        logger.info("✅ Complete marketing workflow successful!")
+        logger.info(f"✅ Complete marketing workflow successful! Best score: {best_score:.2f} from iteration {best_result['iteration']}")
         return json.dumps(workflow_result)
         
     except Exception as e:
@@ -618,14 +692,20 @@ def process_generated_image_and_results(api_response_str: str) -> Tuple[Image.Im
     except Exception as e:
         return None, f"❌ Error processing results: {str(e)}"
 
-def gradio_generate_marketing_image(prompt: str, style: str, review_guidelines: str) -> Tuple[Image.Image, str]:
-    """Gradio interface wrapper for complete marketing image generation"""
+def gradio_generate_marketing_image(prompt: str, style: str, max_retries: int, review_threshold: float, review_guidelines: str) -> Tuple[Image.Image, str]:
+    """Gradio interface wrapper for complete marketing image generation with iterations"""
     if not prompt.strip():
         return None, "⚠️ Please enter a prompt to generate an image."
     
     try:
-        # Use the complete workflow function
-        result_json = generate_and_review_marketing_image(prompt, style, review_guidelines)
+        # Use the complete workflow function with iteration parameters
+        result_json = generate_and_review_marketing_image(
+            prompt=prompt, 
+            style=style, 
+            review_guidelines=review_guidelines,
+            max_retries=max_retries,
+            review_threshold=review_threshold
+        )
         return process_generated_image_and_results(result_json)
     except Exception as e:
         error_message = f"❌ Error: {str(e)}"