import gradio as gr import anthropic import os import json import asyncio from typing import Dict, List from datetime import datetime from PIL import Image import io import base64 from dotenv import load_dotenv from mcpserver import MCPOrchestrator from llamaindex_rag import LlamaIndexEnvironmentalRAG load_dotenv() # Initialize Anthropic client and MCP Orchestrator client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY")) mcp = MCPOrchestrator(api_key=os.environ.get("ANTHROPIC_API_KEY")) llamaindex_rag = LlamaIndexEnvironmentalRAG() # ===== AUTONOMOUS AGENT SYSTEM ===== class EcoAgent: """Autonomous agent with planning, reasoning, and execution phases""" def __init__(self): self.execution_log = [] self.plan = [] def plan_assessment(self, product_name: str) -> List[str]: """PHASE 1: PLANNING - Agent creates execution plan""" # Optimized: Return static plan to save time self.plan = [ "Identify product category", "Analyze materials and composition", "Calculate lifecycle carbon footprint", "Assess environmental issues", "Find sustainable alternatives", "Generate recommendations" ] self.execution_log.append(f"Plan created: {len(self.plan)} steps") return self.plan def reason_about_product(self, product_name: str) -> str: """PHASE 2: REASONING - Agent reasons about product context""" reasoning_prompt = f"""As an environmental expert, reason about this product: {product_name} Analyze: 1. What category does this product belong to? 2. What are the likely materials and manufacturing processes? 3. What environmental concerns are most critical? 4. What lifecycle stage has the most impact? Provide a concise reasoning summary (3-4 sentences).""" try: message = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=600, messages=[{"role": "user", "content": reasoning_prompt}] ) reasoning = message.content[0].text.strip() self.execution_log.append(f"Reasoning completed") return reasoning except Exception as e: return f"Product requires environmental assessment with focus on materials and lifecycle." # ===== RAG COMPONENT: RETRIEVAL AUGMENTED GENERATION ===== def retrieve_environmental_knowledge(product_name: str, reasoning: str) -> str: """RAG: Retrieve relevant environmental knowledge using LlamaIndex semantic search""" print(f"🔍 LlamaIndex searching for: {product_name}") # Use LlamaIndex for advanced semantic search knowledge = llamaindex_rag.retrieve_knowledge(product_name, top_k=2) return f"**LlamaIndex Retrieved Knowledge:**\n\n{knowledge}" # ===== IMPROVED MCP INTEGRATION: VISION ANALYSIS ===== def analyze_image_with_mcp(image) -> tuple: """Use MCP Vision Server for deep image analysis with improved fallback""" if image is None: return "", None try: img_copy = image.copy() max_size = (1568, 1568) img_copy.thumbnail(max_size, Image.Resampling.LANCZOS) buffered = io.BytesIO() img_copy.save(buffered, format="JPEG", quality=85) image_base64 = base64.b64encode(buffered.getvalue()).decode() # Try MCP first try: mcp_result = mcp.call_mcp_tool( "vision", "analyze_product_image", image_base64=image_base64, query="Identify product and materials for environmental assessment" ) if mcp_result.get("status") == "success": analysis = mcp_result.get("analysis", {}) product_type = analysis.get("product_type", "") # Check if we got a valid product name (not generic placeholder) if product_type and product_type.lower() not in ["unknown", "unknown product", "product", "item"]: return product_type, mcp_result except: pass # Always use direct Claude Vision as primary method for reliability message = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=500, messages=[{ "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": "image/jpeg", "data": image_base64, }, }, { "type": "text", "text": """What is this product? Identify it clearly and specifically. Examples of good responses: - "wireless headphones" - "plastic water bottle" - "laptop computer" - "smartphone" - "coffee mug" - "running shoes" - "cotton t-shirt" Return ONLY the product name in 2-5 words. Be specific about the actual product, not generic terms.""" } ] }] ) product_name = message.content[0].text.strip().strip('"').strip("'").lower() # Create simple vision data for direct Claude response vision_data = { "status": "success", "analysis": { "product_type": product_name, "detection_method": "direct_vision" } } return product_name, vision_data except Exception as e: return f"Error: {str(e)}", None # ===== MCP INTEGRATION: LIFECYCLE ASSESSMENT ===== def mcp_lifecycle_assessment(product_name: str, product_data: Dict) -> Dict: """Use MCP Reasoning Server for lifecycle analysis""" try: # MCP TOOL CALL: Lifecycle Impact Calculation lifecycle_result = mcp.call_mcp_tool( "reasoning", "calculate_lifecycle_impact", product_data=product_data ) return lifecycle_result except Exception as e: return {"status": "error", "error": str(e)} # ===== PHASE 3: EXECUTION WITH CONTEXT ENGINEERING ===== def execute_assessment(agent: EcoAgent, product_name: str, vision_data: Dict = None) -> Dict: """PHASE 3: EXECUTION - Agent executes assessment with RAG and MCP""" # Step 1: Retrieve knowledge (RAG) reasoning = agent.reason_about_product(product_name) rag_context = retrieve_environmental_knowledge(product_name, reasoning) agent.execution_log.append(f"Retrieved environmental knowledge (RAG)") # Step 2: Prepare product data product_data = {"name": product_name, "query": product_name} if vision_data and vision_data.get("status") == "success": analysis = vision_data.get("analysis", {}) # Only add materials if they're meaningful (not generic placeholders) materials = analysis.get("materials", []) if materials and not any("material" in str(m).lower() for m in materials): product_data["materials"] = materials packaging = analysis.get("packaging_materials", []) if packaging: product_data["packaging"] = packaging category = analysis.get("product_category", "") if category: product_data["category"] = category agent.execution_log.append(f"Vision analysis integrated") # Step 3: MCP Lifecycle Assessment lifecycle_result = mcp_lifecycle_assessment(product_name, product_data) if lifecycle_result.get("status") == "success": agent.execution_log.append(f"Lifecycle assessment completed via MCP") else: agent.execution_log.append(f"MCP lifecycle assessment skipped") # Step 4: CONTEXT ENGINEERING - Comprehensive prompt with all context engineered_prompt = f"""You are an environmental assessment expert. Assess this product: {product_name} **AGENT REASONING:** {reasoning} **RETRIEVED KNOWLEDGE (RAG):** {rag_context} **MCP LIFECYCLE ANALYSIS:** {json.dumps(lifecycle_result, indent=2) if lifecycle_result.get("status") == "success" else "Pending detailed analysis"} **VISION ANALYSIS:** {json.dumps(vision_data.get("analysis", {}), indent=2) if vision_data else "No image data"} Using ALL the above context, provide a comprehensive assessment in this EXACT JSON format: {{ "name": "{product_name}", "eco_score": 5.5, "carbon": "XX kg CO2e or XX g CO2e", "issues": "Main environmental concerns separated by commas", "alternative": "Best eco-friendly alternative with score (X.X/10) - Key benefit", "lifecycle_insights": "Key insights from lifecycle analysis", "materials_analysis": "Summary of materials and their impact" }} IMPORTANT: - name MUST be: {product_name} - eco_score: realistic number 1-10 for THIS SPECIFIC product - alternative: Recommend ONLY NEW sustainable products (NOT second-hand, vintage, used, refurbished, or pre-owned items). Focus on eco-friendly materials, sustainable manufacturing, certifications, or innovative green technologies - Be specific and accurate based on the context provided - Return ONLY valid JSON, no extra text. Examples of good alternatives: - For shoes: "Vegan leather shoes made from recycled materials with score (8.5/10) - Zero animal products and 70% lower carbon footprint" - For electronics: "Energy Star certified laptop with modular design with score (7.8/10) - 50% longer lifespan and easy repair" - For clothing: "Organic cotton t-shirt with Fair Trade certification with score (8.2/10) - Pesticide-free farming and ethical production" DO NOT recommend: second-hand, vintage, used, refurbished, pre-owned, thrift store items.""" try: message = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=2000, messages=[{"role": "user", "content": engineered_prompt}] ) response_text = message.content[0].text.strip() if "```json" in response_text: response_text = response_text.split("```json")[1].split("```")[0].strip() elif "```" in response_text: response_text = response_text.split("```")[1].split("```")[0].strip() product_assessment = json.loads(response_text) product_assessment["name"] = product_name.title() if isinstance(product_assessment.get("eco_score"), str): product_assessment["eco_score"] = float(product_assessment["eco_score"]) agent.execution_log.append(f"Final assessment generated") return { "assessment": product_assessment, "reasoning": reasoning, "rag_context": rag_context, "lifecycle": lifecycle_result, "execution_log": agent.execution_log.copy() } except Exception as e: agent.execution_log.append(f"Error: {str(e)}") return { "assessment": { "name": product_name.title(), "eco_score": 5.0, "carbon": "Data unavailable", "issues": "Assessment in progress", "alternative": "Consult environmental guidelines", "lifecycle_insights": "Analysis pending" }, "execution_log": agent.execution_log.copy() } def format_score(score) -> str: """Format eco score with emoji""" try: score_num = float(score) except (ValueError, TypeError): score_num = 5.0 if score_num >= 8: return f"{score_num}/10 Excellent" elif score_num >= 6: return f"{score_num}/10 Good" elif score_num >= 4: return f"{score_num}/10 Moderate" else: return f"{score_num}/10 Poor" # ===== GRADIO 6 FEATURE: Progress Tracking ===== def assess_product_agentic(image: str, progress=gr.Progress()) -> str: """Main assessment with Gradio 6 Progress tracking""" try: progress(0, desc="Initializing agent...") agent = EcoAgent() # Get product query query = "" source = "" vision_data = None if image is not None: progress(0.1, desc="Analyzing image...") query, vision_data = analyze_image_with_mcp(image) # Don't stop if image analysis fails, just log it if query and not query.startswith("Error"): source = f"**Detected:** {query}\n\n" if vision_data and vision_data.get("status") == "success": analysis = vision_data.get("analysis", {}) materials = analysis.get("materials", []) if materials and not any("material" in str(m).lower() for m in materials): source += f"**Materials:** {', '.join(materials)}\n\n" else: # Image analysis failed, clear query so text input can be used query = "" # Only return error if BOTH image and text are missing/invalid if not query: return "**Take a photo** or type a product name to get started!" # PHASE 1: PLANNING progress(0.3, desc="Agent planning assessment...") plan = agent.plan_assessment(query) # PHASE 2: REASONING progress(0.5, desc="Agent reasoning...") # PHASE 3: EXECUTION progress(0.7, desc="Executing assessment...") result = execute_assessment(agent, query, vision_data) progress(0.9, desc="Formatting results...") product = result["assessment"] # Get score for color coding eco_score = float(product.get('eco_score', 5.0)) # Determine score color based on value if eco_score < 5: score_color = "#ef4444" # Red score_label = "Poor" score_bg = "#fee2e2" elif eco_score < 7: score_color = "#f59e0b" # Amber score_label = "Moderate" score_bg = "#fef3c7" else: score_color = "#10b981" # Green score_label = "Good" score_bg = "#d1fae5" # Format main output with styled HTML main_output = f"""
{product.get('carbon', 'N/A')}
{product.get('issues', 'N/A')}
" + product['materials_analysis'] + "
" + product['lifecycle_insights'] + "
{product.get('alternative', 'N/A')}
Analyze products instantly and discover their environmental footprint
Upload a product image or type a product name to get started with your environmental assessment