Update app.py

app.py

@@ -1,292 +1,261 @@
-import os
-import logging
 import asyncio
-import yaml
-from typing import Dict, List, Any, Tuple, Optional
-from abc import ABC, abstractmethod
-
 import gradio as gr
-from dotenv import load_dotenv
-from langchain_community.llms import HuggingFaceHub
-from langchain.agents import initialize_agent, AgentType
-from langchain.chains import LLMChain
-from langchain.prompts import PromptTemplate
-from langchain.tools import Tool
-
-# Load environment variables
-load_dotenv()
-
-# Custom Exceptions
-class CodeFusionError(Exception):
-    """Base exception class for CodeFusion."""
-    pass
-
-class AgentInitializationError(CodeFusionError):
-    """Raised when there's an error initializing the agent."""
-    pass
-
-class ToolExecutionError(CodeFusionError):
-    """Raised when there's an error executing a tool."""
-    pass
+from langchain_community.llms import HuggingFaceEndpoint
+from langchain_core.prompts import PromptTemplate
+from langchain.agents import create_react_agent, AgentExecutor
+from langchain_core.tools import BaseTool
+from typing import List
+import yaml
+import os
+import json
 
-# Utility Functions
-def load_config() -> Dict:
-    """Load configuration from config.yaml file or use default values."""
-    config_path = 'config.yaml'
-    default_config = {
-        'model_name': "google/flan-t5-xl",
-        'api_key': "your_default_api_key_here",
-        'temperature': 0.5,
-        'verbose': True
-    }
-    
+def load_config():
+    config_path = os.path.join(os.path.dirname(__file__), "config.yaml")
     try:
-        with open(config_path, 'r') as config_file:
-            config = yaml.safe_load(config_file)
+        with open(config_path, "r") as config_file:
+            return yaml.safe_load(config_file)
     except FileNotFoundError:
-        print(f"Config file not found at {config_path}. Using default configuration.")
-        config = default_config
-    
-    # Override with environment variables if set
-    config['api_key'] = os.getenv('HUGGINGFACE_API_KEY', config['api_key'])
-    return config
-
-def setup_logging() -> logging.Logger:
-    """Set up logging configuration."""
-    logging.basicConfig(
-        level=logging.INFO,
-        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-        filename='codefusion.log'
-    )
-    return logging.getLogger(__name__)
+        print("Config file not found. Using default configuration.")
+        return {
+            "model": "mistralai/Mixtral-8x7B-Instruct-v0.1",
+            "hf_api_token": os.environ.get("HUGGINGFACEHUB_API_TOKEN", "your_default_token_here")
+        }
 
-# Load configuration and set up logging
 config = load_config()
-logger = setup_logging()
 
-# Tool Classes
-class Tool(ABC):
-    """Abstract base class for all tools used by the agent."""
-
-    def __init__(self, name: str, description: str):
-        self.name = name
-        self.description = description
-        self.llm = HuggingFaceHub(
-            repo_id=config['model_name'],
-            model_kwargs={"temperature": config['temperature']},
-            huggingfacehub_api_token=config['api_key']
-        )
+class AgentInitializationError(Exception):
+    pass
 
-    @abstractmethod
-    async def run(self, arguments: Dict[str, Any]) -> Dict[str, str]:
-        """Execute the tool's functionality."""
-        pass
+class CodeGenerationTool(BaseTool):
+    name = "CodeGeneration"
+    description = "Generates code based on a prompt"
+
+    def _run(self, prompt: str) -> str:
+        # Implement basic code generation logic
+        if "Flask app structure" in prompt:
+            return self.generate_flask_app_structure()
+        elif "binary search algorithm" in prompt:
+            return self.generate_binary_search()
+        elif "responsive navbar" in prompt:
+            return self.generate_responsive_navbar()
+        else:
+            return f"Generated code placeholder for: {prompt}"
+
+    async def _arun(self, prompt: str) -> str:
+        return self._run(prompt)
+
+    def generate_flask_app_structure(self):
+        return """
+# app.py
+from flask import Flask, render_template
+
+app = Flask(__name__)
+
+@app.route('/')
+def home():
+    return render_template('index.html')
+
+if __name__ == '__main__':
+    app.run(debug=True)
+
+# templates/index.html
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Flask App</title>
+</head>
+<body>
+    <h1>Welcome to Flask!</h1>
+</body>
+</html>
+"""
+
+    def generate_binary_search(self):
+        return """
+def binary_search(arr, target):
+    left, right = 0, len(arr) - 1
+    
+    while left <= right:
+        mid = (left + right) // 2
+        if arr[mid] == target:
+            return mid
+        elif arr[mid] < target:
+            left = mid + 1
+        else:
+            right = mid - 1
+    
+    return -1  # Target not found
+
+# Example usage
+sorted_array = [1, 3, 5, 7, 9, 11, 13, 15]
+target = 7
+result = binary_search(sorted_array, target)
+print(f"Target {target} found at index: {result}")
+"""
+
+    def generate_responsive_navbar(self):
+        return """
+<!-- HTML -->
+<nav class="navbar">
+    <div class="navbar-logo">Logo</div>
+    <ul class="navbar-links">
+        <li><a href="#home">Home</a></li>
+        <li><a href="#about">About</a></li>
+        <li><a href="#services">Services</a></li>
+        <li><a href="#contact">Contact</a></li>
+    </ul>
+    <div class="navbar-toggle">
+        <span class="bar"></span>
+        <span class="bar"></span>
+        <span class="bar"></span>
+    </div>
+</nav>
+
+<!-- CSS -->
+<style>
+.navbar {
+    display: flex;
+    justify-content: space-between;
+    align-items: center;
+    padding: 1rem 2rem;
+    background-color: #333;
+    color: white;
+}
+
+.navbar-logo {
+    font-size: 1.5rem;
+    font-weight: bold;
+}
+
+.navbar-links {
+    display: flex;
+    list-style: none;
+}
+
+.navbar-links li {
+    margin-left: 1rem;
+}
+
+.navbar-links a {
+    color: white;
+    text-decoration: none;
+}
+
+.navbar-toggle {
+    display: none;
+    flex-direction: column;
+    cursor: pointer;
+}
+
+.bar {
+    width: 25px;
+    height: 3px;
+    background-color: white;
+    margin: 3px 0;
+}
+
+@media (max-width: 768px) {
+    .navbar-links {
+        display: none;
+        flex-direction: column;
+        width: 100%;
+        position: absolute;
+        top: 60px;
+        left: 0;
+        background-color: #333;
+    }
 
-class CodeGenerationTool(Tool):
-    """Tool for generating code snippets in various languages."""
+    .navbar-links.active {
+        display: flex;
+    }
 
-    def __init__(self):
-        super().__init__("Code Generation", "Generates code snippets in various languages.")
-        self.prompt_template = PromptTemplate(
-            input_variables=["language", "code_description"],
-            template="Generate {language} code for: {code_description}"
-        )
-        self.chain = LLMChain(llm=self.llm, prompt=self.prompt_template)
-        # Add this line:
-        self.is_single_input = True
+    .navbar-links li {
+        margin: 1rem 0;
+    }
 
-    async def run(self, arguments: Dict[str, str]) -> Dict[str, str]:
-        language = arguments.get("language", "python")
-        code_description = arguments.get("code_description", "print('Hello, World!')")
-        try:
-            code = await self.chain.arun(language=language, code_description=code_description)
-            return {"output": code}
-        except Exception as e:
-            logger.error(f"Error in CodeGenerationTool: {e}")
-            raise ToolExecutionError(f"Failed to generate code: {e}")
+    .navbar-toggle {
+        display: flex;
+    }
+}
+</style>
 
-class CodeExplanationTool(Tool):
-    """Tool for explaining code snippets."""
+<!-- JavaScript -->
+<script>
+document.querySelector('.navbar-toggle').addEventListener('click', function() {
+    document.querySelector('.navbar-links').classList.toggle('active');
+});
+</script>
+"""
 
-    def __init__(self):
-        super().__init__("Code Explanation", "Explains code snippets in simple terms.")
-        self.prompt_template = PromptTemplate(
-            input_variables=["code"],
-            template="Explain the following code in simple terms:\n\n{code}"
+class Agent:
+    def __init__(self, name: str, description: str, tools: List[BaseTool]):
+        self.name = name
+        self.description = description
+        self.llm = HuggingFaceEndpoint(
+            repo_id=config["model"],
+            task="text-generation",
+            model_kwargs={"temperature": 0.7, "max_length": 1024},
+            huggingfacehub_api_token=config["hf_api_token"]
         )
-        self.chain = LLMChain(llm=self.llm, prompt=self.prompt_template)
-        # Add this line:
-        self.is_single_input = True
-
-    async def run(self, arguments: Dict[str, str]) -> Dict[str, str]:
-        code = arguments.get("code", "print('Hello, World!')")
-        try:
-            explanation = await self.chain.arun(code=code)
-            return {"output": explanation}
-        except Exception as e:
-            logger.error(f"Error in CodeExplanationTool: {e}")
-            raise ToolExecutionError(f"Failed to explain code: {e}")
-
-class DebuggingTool(Tool):
-    """Tool for debugging code snippets."""
-
-    def __init__(self):
-        super().__init__("Debugging", "Helps identify and fix issues in code snippets.")
+        
         self.prompt_template = PromptTemplate(
-            input_variables=["code", "error_message"],
-            template="Debug the following code:\n\n{code}\n\nError message: {error_message}"
+            template="You are {name}, {description}. Respond to the following: {input}",
+            input_variables=["name", "description", "input"]
         )
-        self.chain = LLMChain(llm=self.llm, prompt=self.prompt_template)
-        # Add this line:
-        self.is_single_input = True
-
-    async def run(self, arguments: Dict[str, str]) -> Dict[str, str]:
-        code = arguments.get("code", "")
-        error_message = arguments.get("error_message", "")
-        try:
-            debug_result = await self.chain.arun(code=code, error_message=error_message)
-            return {"output": debug_result}
-        except Exception as e:
-            logger.error(f"Error in DebuggingTool: {e}")
-            raise ToolExecutionError(f"Failed to debug code: {e}")
-
-# Agent Class
-class Agent:
-    """Represents an AI agent with specific tools and capabilities."""
-
-    def __init__(self, name: str, role: str, tools: List[Tool]):
-        self.name = name
-        self.role = role
-        self.tools = tools
-        self.memory: List[tuple] = []
         
         try:
-            self.llm = HuggingFaceHub(
-                repo_id=config['model_name'],
-                model_kwargs={"temperature": config['temperature']},
-                huggingfacehub_api_token=config['api_key']
-            )
-            # Convert tools to dictionaries
-            tools_dict = [
-                {"name": tool.name, "description": tool.description, "func": tool.run}
-                for tool in self.tools
-            ]
-            self.agent = initialize_agent(
-                llm=self.llm,
-                tools=tools_dict,
-                agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
-                verbose=config['verbose']
-            )
+            self.agent = create_react_agent(self.llm, tools, self.prompt_template)
+            self.agent_executor = AgentExecutor(agent=self.agent, tools=tools, verbose=True)
         except Exception as e:
-            logger.error(f"Error initializing agent: {e}")
             raise AgentInitializationError(f"Failed to initialize agent: {e}")
 
-    async def act(self, prompt: str, context: str) -> str:
-        """Perform an action based on the given prompt and context."""
-        self.memory.append((prompt, context))
+    async def run(self, input_text: str) -> str:
         try:
-            action = await self.agent.arun(prompt, context)
-            return action
+            result = await self.agent_executor.arun(input_text)
+            return result
         except Exception as e:
-            logger.error(f"Error during agent action: {e}")
-            raise
-
-    def __str__(self) -> str:
-        return f"Agent: {self.name} (Role: {self.role})"
-
-# Main application functions
-async def run(message: str, history: List[Tuple[str, str]]) -> str:
-    """Process user input and generate a response using the agent system."""
-    # Create a trio of agents for sequential app building
-    agent_structure = [
-        Agent("CodeFusion_Structure", "App Structure Designer", [CodeGenerationTool()]),
-        Agent("CodeFusion_Logic", "App Logic Implementer", [CodeGenerationTool(), CodeExplanationTool()]),
-        Agent("CodeFusion_UI", "App UI Designer", [CodeGenerationTool(), CodeExplanationTool(), DebuggingTool()])
-    ]
-    
-    # Initialize the current agent based on the user's request
-    current_agent_index = 0
-    for i, agent in enumerate(agent_structure):
-        if "structure" in message.lower():
-            current_agent_index = i
-            break
-        elif "logic" in message.lower():
-            current_agent_index = i
-            break
-        elif "ui" in message.lower():
-            current_agent_index = i
-            break
-
-    # Use the selected agent to process the message
-    current_agent = agent_structure[current_agent_index]
-    context = "\n".join([f"Human: {h[0]}\nAI: {h[1]}" for h in history])
-    try:
-        response = await current_agent.act(message, context)
-        return response
-    except Exception as e:
-        logger.error(f"Error processing request: {e}")
-        return "I apologize, but an error occurred while processing your request. Please try again."
+            return f"Error: {str(e)}"
+
+class CodeFusion:
+    def __init__(self):
+        code_gen_tool = CodeGenerationTool()
+        self.agents = [
+            Agent("CodeFusion_Structure", "App Structure Designer", [code_gen_tool]),
+            Agent("CodeFusion_Logic", "Logic Implementation Expert", [code_gen_tool]),
+            Agent("CodeFusion_UI", "User Interface Designer", [code_gen_tool])
+        ]
+
+    async def run(self, input_text: str) -> str:
+        results = []
+        for agent in self.agents:
+            result = await agent.run(input_text)
+            results.append(f"{agent.name}: {result}")
+        return "\n\n".join(results)
+
+code_fusion = CodeFusion()
+
+async def chat(message, history):
+    response = await code_fusion.run(message)
+    return response
 
 async def main():
-    """Main function to run the Gradio interface."""
-    examples = [
-        ["Can you help me structure an app to manage a to-do list?", "Sure, here's a basic structure for a to-do list app:"],
-        ["Now, implement the logic for adding and deleting tasks.", "OK, here's the logic for adding and deleting tasks:"],
-        ["Finally, design a simple UI for the app.", "Here's a simple UI design for the to-do list app:"],
-    ]
-
-    gr.ChatInterface(
-        fn=run,
-        title="CodeFusion: Your AI Coding Assistant",
-        description="Ask me about code generation, explanation, debugging, or any other coding task!",
-        examples=examples,
-        theme="default"
-    ).launch()
-
-# Simple testing framework
-def run_tests():
-    """Run basic tests for the CodeFusion components."""
-    
-    async def test_code_generation():
-        tool = CodeGenerationTool()
-        result = await tool.run({"language": "python", "code_description": "function to add two numbers"})
-        assert "def" in result["output"], "Code generation failed to produce a function"
-        print("Code Generation Test: Passed")
-
-    async def test_code_explanation():
-        tool = CodeExplanationTool()
-        result = await tool.run({"code": "def factorial(n):\n    return 1 if n == 0 else n * factorial(n-1)"})
-        assert "recursive" in result["output"].lower(), "Code explanation failed to mention recursion"
-        print("Code Explanation Test: Passed")
-
-    async def test_debugging():
-        tool = DebuggingTool()
-        result = await tool.run({"code": "def divide(a, b):\n    return a / b", "error_message": "ZeroDivisionError"})
-        assert "zero" in result["output"].lower(), "Debugging failed to address division by zero"
-        print("Debugging Test: Passed")
-
-    async def test_agent():
-        agent = Agent("TestAgent", "Tester", [CodeGenerationTool(), CodeExplanationTool(), DebuggingTool()])
-        result = await agent.act("Generate a Python function to calculate the square of a number", "")
-        assert "def" in result and "return" in result, "Agent failed to generate a proper function"
-        print("Agent Test: Passed")
-
-    async def run_all_tests():
-        await test_code_generation()
-        await test_code_explanation()
-        await test_debugging()
-        await test_agent()
-
-    asyncio.run(run_all_tests())
+    iface = gr.ChatInterface(
+        fn=chat,
+        title="CodeFusion AI",
+        description="Your AI-powered coding assistant",
+        examples=[
+            "Create a basic Flask app structure",
+            "Implement a binary search algorithm in Python",
+            "Design a responsive navbar using HTML and CSS"
+        ],
+        retry_btn=None,
+        undo_btn="Delete Previous",
+        clear_btn="Clear",
+    )
 
-import asyncio
+    await iface.launch()
 
 if __name__ == "__main__":
-    import sys
-    if len(sys.argv) > 1 and sys.argv[1] == "--test":
-        run_tests()
-    else:
-        asyncio.run(main())
-        interface = gradio_interface()
-        interface.queue()
-        interface.launch()
+    asyncio.run(main())