app.py

import gradio as gr
import os
import asyncio
import nest_asyncio
from datetime import datetime
from typing import Optional, Dict, Any

from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.teams import SelectorGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.agents.web_surfer import MultimodalWebSurfer

# Enable nested event loops for Jupyter compatibility
nest_asyncio.apply()

class AIShoppingAnalyzer:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.model_client = OpenAIChatCompletionClient(model="gpt-4o")
        self.termination = MaxMessageTermination(max_messages=20) | TextMentionTermination("TERMINATE")
        
    def create_websurfer(self) -> MultimodalWebSurfer:
        """Initialize the web surfer agent for e-commerce research"""
        return MultimodalWebSurfer(
            name="websurfer_agent",
            description="""E-commerce research specialist that:
            1. Searches multiple retailers for product options
            2. Compares prices and reviews
            3. Checks product specifications and availability
            4. Analyzes website structure and findability""",
            model_client=self.model_client,
            headless=True
        )

    def create_assistant(self) -> AssistantAgent:
        """Initialize the shopping assistant agent"""
        return AssistantAgent(
            name="assistant_agent",
            description="E-commerce shopping advisor and website analyzer",
            system_message="""You are an expert shopping assistant and e-commerce analyst. Your role is to:
            1. Help find products based on user needs
            2. Compare prices and features across different sites
            3. Analyze website usability and product findability
            4. Evaluate product presentation and information quality
            5. Assess the overall e-commerce experience

            When working with the websurfer_agent:
            - Guide their research effectively
            - Verify the information they find
            - Analyze how easy it was to find products
            - Evaluate product page quality
            - Say 'keep going' if more research is needed
            - Say 'TERMINATE' only when you have a complete analysis""",
            model_client=self.model_client
        )

    def create_team(self, websurfer_agent: MultimodalWebSurfer, assistant_agent: AssistantAgent) -> SelectorGroupChat:
        """Set up the team of agents"""
        user_proxy = UserProxyAgent(
            name="user_proxy",
            description="An e-commerce site owner looking for AI shopping analysis"
        )

        return SelectorGroupChat(
            participants=[websurfer_agent, assistant_agent, user_proxy],
            selector_prompt="""You are coordinating an e-commerce analysis system. The following roles are available:
            {roles}

            Given the conversation history {history}, select the next role from {participants}.
            - The websurfer_agent searches products and analyzes website structure
            - The assistant_agent evaluates findings and makes recommendations
            - The user_proxy provides input when needed

            Return only the role name.""",
            model_client=self.model_client,
            termination_condition=self.termination
        )

    async def analyze_site(self, 
                         website_url: str, 
                         product_category: str, 
                         specific_product: Optional[str] = None) -> str:
        """Run the analysis with proper cleanup"""
        websurfer = None
        try:
            # Set up the analysis query
            query = f"""Analyze the e-commerce experience for {website_url} focusing on:
            1. Product findability in the {product_category} category
            2. Product information quality
            3. Navigation and search functionality
            4. Price visibility and comparison features"""
            
            if specific_product:
                query += f"\n5. Detailed analysis of this specific product: {specific_product}"

            # Initialize agents
            websurfer = self.create_websurfer()
            assistant = self.create_assistant()

            # Create team
            team = self.create_team(websurfer, assistant)

            # Run the analysis
            result = []
            async for message in team.run_stream(task=query):
                result.append(message)
            
            return "\n".join(result)

        except Exception as e:
            return f"Analysis error: {str(e)}"
        finally:
            if websurfer:
                try:
                    await websurfer.close()
                except Exception as e:
                    return f"Cleanup error: {str(e)}"

def create_gradio_interface() -> gr.Interface:
    """Create the Gradio interface for the AI Shopping Analyzer"""
    
    def validate_api_key(api_key: str) -> bool:
        """Validate the OpenAI API key format"""
        return api_key.startswith("sk-") and len(api_key) > 20

    async def run_analysis(api_key: str,
                         website_url: str,
                         product_category: str,
                         specific_product: str) -> str:
        """Handle the analysis submission"""
        if not validate_api_key(api_key):
            return "Please enter a valid OpenAI API key (should start with 'sk-')"
        
        if not website_url:
            return "Please enter a website URL"
            
        if not product_category:
            return "Please specify a product category"
            
        try:
            analyzer = AIShoppingAnalyzer(api_key)
            result = await analyzer.analyze_site(
                website_url=website_url,
                product_category=product_category,
                specific_product=specific_product if specific_product else None
            )
            return result
        except Exception as e:
            return f"Error during analysis: {str(e)}"

    # Create the interface
    return gr.Interface(
        fn=run_analysis,
        inputs=[
            gr.Textbox(label="OpenAI API Key", placeholder="sk-...", type="password"),
            gr.Textbox(label="Website URL", placeholder="https://your-store.com"),
            gr.Textbox(label="Product Category", placeholder="e.g., Electronics, Clothing, etc."),
            gr.Textbox(label="Specific Product (Optional)", placeholder="e.g., Blue Widget Model X")
        ],
        outputs=gr.Textbox(label="Analysis Results", lines=20),
        title="AI Shopping Agent Analyzer",
        description="""Analyze how your e-commerce site performs when the shopper is an AI agent.
        This tool helps you understand your site's effectiveness for AI-powered shopping assistants.""",
        theme="default",
        allow_flagging="never"
    )

if __name__ == "__main__":
    # Create and launch the interface
    iface = create_gradio_interface()
    iface.launch(share=True)