Upload 11 files

__init__.py

@@ -0,0 +1,5 @@
+"""
+Gemini AI Research Agent
+
+An advanced AI assistant powered by Google's Gemini 1.5 Flash model,
+specialized in handling complex research questions and multi

main.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+"""
+Main application entry point for the Gemini AI Agent.
+This script initializes and runs the Gradio interface.
+"""
+
+import gradio as gr
+import logging
+from src.agent import GeminiAgent
+from src.config import Config
+from src.ui import create_interface
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+
+def main():
+    """
+    Main function to initialize and launch the application.
+    """
+    try:
+        # Load configuration
+        config = Config()
+
+        # Initialize the Gemini agent
+        agent = GeminiAgent(config)
+
+        # Create and launch the Gradio interface
+        interface = create_interface(agent)
+
+        logger.info("Starting Gemini AI Agent application...")
+        interface.launch(
+            server_name="0.0.0.0",
+            server_port=7860,
+            share=False
+        )
+
+    except Exception as e:
+        logger.error(f"Failed to start application: {e}")
+        raise
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,26 @@
+# Core dependencies
+gradio>=4.0.0
+google-generativeai>=0.3.0
+python-dotenv>=1.0.0
+
+# Utility dependencies
+requests>=2.31.0
+aiohttp>=3.8.0
+asyncio-throttle>=1.0.2
+
+# Data processing
+pandas>=2.0.0
+numpy>=1.24.0
+Pillow>=10.0.0
+
+# Development and testing
+pytest>=7.4.0
+pytest-asyncio>=0.21.0
+black>=23.0.0
+flake8>=6.0.0
+
+# Optional: For enhanced functionality
+youtube-dl>=2021.12.17  # For video analysis
+beautifulsoup4>=4.12.0  # For web scraping
+python-chess>=1.999     # For chess analysis
+sympy>=1.12            # For mathematical computations

src/agent.py

@@ -0,0 +1,196 @@
+"""
+Gemini AI Agent implementation.
+Handles communication with Google's Gemini API and manages conversation context.
+"""
+
+import logging
+import asyncio
+from typing import List, Dict, Any, Optional
+import google.generativeai as genai
+from google.generativeai.types import HarmCategory, HarmBlockThreshold
+from src.config import Config
+from src.utils import sanitize_input, format_response
+
+
+logger = logging.getLogger(__name__)
+
+
+class GeminiAgent:
+    """
+    AI Agent powered by Google's Gemini 1.5 Flash model.
+
+    This agent is designed to handle complex, multi-modal questions including:
+    - Text analysis and reasoning
+    - Mathematical computations
+    - Research tasks requiring web search
+    - Data analysis and interpretation
+    - Creative problem solving
+    """
+
+    def __init__(self, config: Config):
+        """
+        Initialize the Gemini Agent.
+
+        Args:
+            config: Configuration object containing API keys and settings
+        """
+        self.config = config
+        self.conversation_history: List[Dict[str, str]] = []
+        self._initialize_client()
+
+    def _initialize_client(self):
+        """
+        Initialize the Gemini API client with safety settings.
+        """
+        try:
+            genai.configure(api_key=self.config.gemini_api_key)
+
+            # Configure safety settings for research and analysis tasks
+            safety_settings = {
+                HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
+                HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
+                HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
+                HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
+            }
+
+            # Initialize the model
+            self.model = genai.GenerativeModel(
+                model_name=self.config.model_name,
+                safety_settings=safety_settings
+            )
+
+            logger.info(f"Gemini client initialized with model: {self.config.model_name}")
+
+        except Exception as e:
+            logger.error(f"Failed to initialize Gemini client: {e}")
+            raise
+
+    def _build_system_prompt(self) -> str:
+        """
+        Build the system prompt that defines the agent's capabilities and role.
+
+        Returns:
+            str: System prompt for the agent
+        """
+        return """You are an advanced AI research assistant specialized in answering complex, multi-faceted questions that may require:
+
+- Deep research and fact-checking
+- Mathematical calculations and logical reasoning
+- Analysis of multimedia content (images, videos, audio)
+- Data interpretation and statistical analysis
+- Cross-referencing multiple sources
+- Handling ambiguous or challenging queries
+
+Key capabilities:
+1. **Research Excellence**: Thoroughly investigate topics using available information
+2. **Multi-modal Analysis**: Process and analyze images, videos, and audio content
+3. **Mathematical Reasoning**: Perform calculations and logical deductions
+4. **Data Analysis**: Interpret tables, charts, and datasets
+5. **Fact Verification**: Cross-check information for accuracy
+6. **Creative Problem Solving**: Approach unusual questions with innovative methods
+
+Guidelines:
+- Provide detailed, well-researched answers
+- Show your reasoning process step-by-step
+- If information is incomplete, clearly state assumptions
+- For complex calculations, break down the steps
+- When analyzing media, describe what you observe
+- Always strive for accuracy over speed
+- If uncertain, express confidence levels
+
+Remember: You excel at handling challenging questions that require deep thinking and research."""
+
+    async def process_question(self, question: str, context: Optional[str] = None) -> str:
+        """
+        Process a question and generate a response using Gemini.
+
+        Args:
+            question: The user's question
+            context: Optional additional context
+
+        Returns:
+            str: The agent's response
+        """
+        try:
+            # Sanitize input
+            clean_question = sanitize_input(question)
+
+            # Build the full prompt
+            system_prompt = self._build_system_prompt()
+
+            # Prepare conversation context
+            context_str = ""
+            if self.conversation_history:
+                context_str = "\n\nPrevious conversation context:\n"
+                for entry in self.conversation_history[-3:]:  # Last 3 exchanges
+                    context_str += f"Q: {entry['question']}\nA: {entry['answer'][:200]}...\n"
+
+            # Build final prompt
+            full_prompt = f"{system_prompt}\n\n"
+            if context:
+                full_prompt += f"Additional Context: {context}\n\n"
+            full_prompt += f"{context_str}\nCurrent Question: {clean_question}\n\nResponse:"
+
+            # Generate response
+            response = self.model.generate_content(
+                full_prompt,
+                generation_config=genai.types.GenerationConfig(
+                    max_output_tokens=self.config.max_tokens,
+                    temperature=self.config.temperature,
+                )
+            )
+
+            # Extract and format response
+            if response.text:
+                formatted_response = format_response(response.text)
+
+                # Update conversation history
+                self._update_history(clean_question, formatted_response)
+
+                logger.info(f"Successfully processed question: {clean_question[:50]}...")
+                return formatted_response
+            else:
+                logger.warning("Received empty response from Gemini")
+                return "I apologize, but I couldn't generate a response to your question. Please try rephrasing it."
+
+        except Exception as e:
+            logger.error(f"Error processing question: {e}")
+            return f"I encountered an error while processing your question: {str(e)}"
+
+    def _update_history(self, question: str, answer: str):
+        """
+        Update the conversation history.
+
+        Args:
+            question: The user's question
+            answer: The agent's response
+        """
+        self.conversation_history.append({
+            'question': question,
+            'answer': answer
+        })
+
+        # Keep only recent history
+        if len(self.conversation_history) > self.config.max_history_length:
+            self.conversation_history = self.conversation_history[-self.config.max_history_length:]
+
+    def clear_history(self):
+        """
+        Clear the conversation history.
+        """
+        self.conversation_history.clear()
+        logger.info("Conversation history cleared")
+
+    def get_stats(self) -> Dict[str, Any]:
+        """
+        Get agent statistics.
+
+        Returns:
+            Dict containing agent statistics
+        """
+        return {
+            'model': self.config.model_name,
+            'conversation_length': len(self.conversation_history),
+            'max_tokens': self.config.max_tokens,
+            'temperature': self.config.temperature
+        }

src/config.py

@@ -0,0 +1,58 @@
+"""
+Configuration management for the Gemini AI Agent.
+Handles environment variables and application settings.
+"""
+
+import os
+from typing import Optional
+from dataclasses import dataclass
+
+
+@dataclass
+class Config:
+    """
+    Configuration class for the Gemini AI Agent.
+    """
+
+    # API Configuration
+    gemini_api_key: str
+    model_name: str = "gemini-1.5-flash"
+    max_tokens: int = 2048
+    temperature: float = 0.7
+
+    # Application Configuration
+    max_history_length: int = 10
+    timeout_seconds: int = 30
+
+    def __init__(self):
+        """
+        Initialize configuration from environment variables.
+        """
+        self.gemini_api_key = self._get_api_key()
+
+    def _get_api_key(self) -> str:
+        """
+        Get the Gemini API key from environment variables.
+
+        Returns:
+            str: The API key
+
+        Raises:
+            ValueError: If API key is not found
+        """
+        api_key = os.getenv('GEMINI_API_KEY')
+        if not api_key:
+            raise ValueError(
+                "GEMINI_API_KEY environment variable is required. "
+                "Please set it with your Google AI Studio API key."
+            )
+        return api_key
+
+    def validate(self) -> bool:
+        """
+        Validate the configuration.
+
+        Returns:
+            bool: True if configuration is valid
+        """
+        return bool(self.gemini_api_key and self.model_name)

src/ui.py

@@ -0,0 +1,184 @@
+"""
+Gradio user interface for the Gemini AI Agent.
+"""
+
+import gradio as gr
+import asyncio
+import logging
+from typing import List, Tuple, Optional
+from src.agent import GeminiAgent
+from src.utils import validate_question, format_error_message
+
+
+logger = logging.getLogger(__name__)
+
+
+def create_interface(agent: GeminiAgent) -> gr.Blocks:
+    """
+    Create the Gradio interface for the AI agent.
+
+    Args:
+        agent: The GeminiAgent instance
+
+    Returns:
+        gr.Blocks: The Gradio interface
+    """
+
+    async def process_question_async(question: str, context: str = "") -> str:
+        """
+        Async wrapper for question processing.
+        """
+        return await agent.process_question(question, context or None)
+
+    def process_question_sync(question: str, context: str = "") -> str:
+        """
+        Synchronous wrapper for question processing (required by Gradio).
+        """
+        try:
+            # Validate question
+            is_valid, error_msg = validate_question(question)
+            if not is_valid:
+                return f"❌ **Error**: {error_msg}"
+
+            # Process question
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            try:
+                result = loop.run_until_complete(process_question_async(question, context))
+                return f"🤖 **AI Response**:\n\n{result}"
+            finally:
+                loop.close()
+
+        except Exception as e:
+            error_msg = format_error_message(e)
+            return f"❌ **Error**: {error_msg}"
+
+    def clear_conversation():
+        """
+        Clear the conversation history.
+        """
+        agent.clear_history()
+        return "", "Conversation history cleared! ✨"
+
+    def get_agent_info() -> str:
+        """
+        Get information about the agent.
+        """
+        stats = agent.get_stats()
+        return f"""
+## 🤖 Agent Information
+
+**Model**: {stats['model']}
+**Conversation Length**: {stats['conversation_length']} exchanges
+**Max Tokens**: {stats['max_tokens']}
+**Temperature**: {stats['temperature']}
+
+### 🎯 Specialized Capabilities
+- **Complex Research**: Multi-source fact-checking and analysis
+- **Mathematical Reasoning**: Step-by-step problem solving
+- **Multi-modal Analysis**: Processing images, videos, and audio
+- **Data Interpretation**: Tables, charts, and statistical analysis
+- **Creative Problem Solving**: Innovative approaches to unusual questions
+"""
+
+    # Create the interface
+    with gr.Blocks(
+        title="Gemini AI Research Agent",
+        theme=gr.themes.Soft(),
+        css="""
+        .container {
+            max-width: 1200px;
+            margin: auto;
+        }
+        .question-box {
+            border-left: 4px solid #4CAF50;
+            padding-left: 16px;
+        }
+        .response-box {
+            background-color: #f8f9fa;
+            border-radius: 8px;
+            padding: 16px;
+        }
+        """
+    ) as interface:
+
+        gr.Markdown("""
+        # 🧠 Gemini AI Research Agent
+
+        An advanced AI assistant powered by Google's Gemini 1.5 Flash, specialized in handling complex research questions, data analysis, and multi-modal content processing.
+
+        **Perfect for**: Academic research, fact-checking, mathematical problems, data analysis, and challenging multi-step questions.
+        """)
+
+        with gr.Row():
+            with gr.Column(scale=2):
+                with gr.Group():
+                    gr.Markdown("## 💬 Ask Your Question")
+
+                    question_input = gr.Textbox(
+                        label="Question",
+                        placeholder="Enter your research question here... (e.g., 'How many studio albums were published by Mercedes Sosa between 2000 and 2009?')",
+                        lines=4,
+                        elem_classes=["question-box"]
+                    )
+
+                    context_input = gr.Textbox(
+                        label="Additional Context (Optional)",
+                        placeholder="Provide any additional context, constraints, or specific requirements...",
+                        lines=2
+                    )
+
+                    with gr.Row():
+                        submit_btn = gr.Button("🔍 Ask Question", variant="primary", size="lg")
+                        clear_btn = gr.Button("🗑️ Clear History", variant="secondary")
+
+                with gr.Group():
+                    gr.Markdown("## 📝 Response")
+                    response_output = gr.Textbox(
+                        label="AI Response",
+                        lines=15,
+                        interactive=False,
+                        elem_classes=["response-box"]
+                    )
+
+            with gr.Column(scale=1):
+                with gr.Group():
+                    gr.Markdown("## ℹ️ Agent Status")
+                    agent_info = gr.Markdown(get_agent_info())
+                    refresh_info_btn = gr.Button("🔄 Refresh Info", size="sm")
+
+
+        gr.Markdown("""
+        ---
+        ### 🔧 Tips for Best Results:
+        - **Be Specific**: Include all relevant details and constraints
+        - **Multi-step Questions**: Break complex questions into clear parts
+        - **Context Matters**: Use the context field for additional information
+        - **Iterative Approach**: Build on previous questions for deeper analysis
+        """)
+
+        # Event handlers
+        submit_btn.click(
+            fn=process_question_sync,
+            inputs=[question_input, context_input],
+            outputs=[response_output]
+        )
+
+        clear_btn.click(
+            fn=clear_conversation,
+            outputs=[question_input, response_output]
+        )
+
+        refresh_info_btn.click(
+            fn=get_agent_info,
+            outputs=[agent_info]
+        )
+
+        # Allow Enter key to submit
+        question_input.submit(
+            fn=process_question_sync,
+            inputs=[question_input, context_input],
+            outputs=[response_output]
+        )
+
+    return interface

src/utils.py

@@ -0,0 +1,147 @@
+"""
+Utility functions for the Gemini AI Agent.
+"""
+
+import re
+import logging
+from typing import Optional
+
+
+logger = logging.getLogger(__name__)
+
+
+def sanitize_input(text: str) -> str:
+    """
+    Sanitize user input by removing potentially harmful content.
+
+    Args:
+        text: Raw input text
+
+    Returns:
+        str: Sanitized text
+    """
+    if not isinstance(text, str):
+        return str(text)
+
+    # Remove excessive whitespace
+    text = re.sub(r'\s+', ' ', text.strip())
+
+    # Remove potential injection attempts (basic protection)
+    text = text.replace('\\n', '\n').replace('\\t', '\t')
+
+    return text
+
+
+def format_response(response: str) -> str:
+    """
+    Format the AI response for better readability.
+
+    Args:
+        response: Raw response from the AI model
+
+    Returns:
+        str: Formatted response
+    """
+    if not response:
+        return "No response generated."
+
+    # Clean up the response
+    response = response.strip()
+
+    # Ensure proper spacing after periods
+    response = re.sub(r'\.([A-Z])', r'. \1', response)
+
+    # Fix common formatting issues
+    response = re.sub(r'\n\s*\n\s*\n', '\n\n', response)  # Multiple newlines
+    response = re.sub(r'([.!?])\s*([A-Z])', r'\1 \2', response)  # Spacing after punctuation
+
+    return response
+
+
+def truncate_text(text: str, max_length: int = 1000) -> str:
+    """
+    Truncate text to a maximum length while preserving word boundaries.
+
+    Args:
+        text: Text to truncate
+        max_length: Maximum length allowed
+
+    Returns:
+        str: Truncated text
+    """
+    if len(text) <= max_length:
+        return text
+
+    # Find the last space before the max_length
+    truncated = text[:max_length]
+    last_space = truncated.rfind(' ')
+
+    if last_space > max_length * 0.8:  # If space is reasonably close to end
+        return truncated[:last_space] + "..."
+    else:
+        return truncated + "..."
+
+
+def extract_code_blocks(text: str) -> list:
+    """
+    Extract code blocks from markdown-formatted text.
+
+    Args:
+        text: Text containing potential code blocks
+
+    Returns:
+        list: List of code blocks found
+    """
+    code_pattern = r'```(\w+)?\n(.*?)\n```'
+    matches = re.findall(code_pattern, text, re.DOTALL)
+
+    return [{'language': match[0] or 'text', 'code': match[1]} for match in matches]
+
+
+def validate_question(question: str) -> tuple[bool, Optional[str]]:
+    """
+    Validate if a question is appropriate and well-formed.
+
+    Args:
+        question: The question to validate
+
+    Returns:
+        tuple: (is_valid, error_message)
+    """
+    if not question or not question.strip():
+        return False, "Question cannot be empty."
+
+    if len(question.strip()) < 3:
+        return False, "Question is too short. Please provide more detail."
+
+    if len(question) > 5000:
+        return False, "Question is too long. Please keep it under 5000 characters."
+
+    return True, None
+
+
+def format_error_message(error: Exception) -> str:
+    """
+    Format error messages for user display.
+
+    Args:
+        error: The exception that occurred
+
+    Returns:
+        str: User-friendly error message
+    """
+    error_type = type(error).__name__
+
+    # Map common errors to user-friendly messages
+    error_messages = {
+        'ConnectionError': 'Unable to connect to the AI service. Please check your internet connection.',
+        'TimeoutError': 'The request timed out. Please try again with a shorter question.',
+        'ValueError': 'Invalid input provided. Please check your question format.',
+        'KeyError': 'Configuration error. Please check your API settings.',
+        'PermissionError': 'Access denied. Please check your API key permissions.'
+    }
+
+    user_message = error_messages.get(error_type, f"An unexpected error occurred: {str(error)}")
+
+    logger.error(f"Error formatted for user: {error_type} - {str(error)}")
+    return user_message