app.py

import fastapi
from typing import Optional, Dict, Any
import copy
import requests
import json
import os
import sys
from io import StringIO
import ctypes
import subprocess
import logging
from pathlib import Path
from llama_cpp import Llama
from concurrent.futures import ThreadPoolExecutor, as_completed
import random
import time
import inspect
# Load model directly
from transformers import AutoTokenizer, AutoModelForCausalLM

tokenizer = AutoTokenizer.from_pretrained("meetkai/functionary-small-v3.1", trust_remote_code=True)
model_path = AutoModelForCausalLM.from_pretrained("meetkai/functionary-small-v3.1", trust_remote_code=True)
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class Tool(fastapi.FastAPI):
    def __init__(
            self,
            tool_name: str,
            description: str,
            name_for_human: Optional[str] = None,
            name_for_model: Optional[str] = None,
            description_for_human: Optional[str] = None,
            description_for_model: Optional[str] = None,
            logo_url: Optional[str] = None,
            author_github: Optional[str] = None,
            contact_email: str = "",
            legal_info_url: str = "",
            version: str = "0.1.0",
        ):
        super().__init__(
            title=tool_name,
            description=description,
            version=version,
        )

        if name_for_human is None:
            name_for_human = tool_name
        if name_for_model is None:
            name_for_model = name_for_human
        if description_for_human is None:
            description_for_human = description
        if description_for_model is None:
            description_for_model = description_for_human
        
        self.api_info = {
            "schema_version": "v1",
            "name_for_human": name_for_human,
            "name_for_model": name_for_model,
            "description_for_human": description_for_human,
            "description_for_model": description_for_model,
            "auth": {
                "type": "none",
            },
            "api": {
                "type": "openapi",
                "url": "/openapi.json",
                "is_user_authenticated": False,
            },
            "author_github": author_github,
            "logo_url": logo_url,
            "contact_email": contact_email,
            "legal_info_url": legal_info_url,
        }

        @self.get("/.well-known/ai-plugin.json", include_in_schema=False)
        def get_api_info(request: fastapi.Request):
            openapi_path = str(request.url).replace("/.well-known/ai-plugin.json", "/openapi.json")
            info = copy.deepcopy(self.api_info)
            info["api"]["url"] = str(openapi_path)
            return info

class SelfLearningTool:
    def __init__(self):
        self.tools = {}

    def add_tool(self, name: str, func: callable):
        self.tools[name] = func

    def use_tool(self, name: str, *args, **kwargs):
        if name in self.tools:
            return self.tools[name](*args, **kwargs)
        else:
            return f"Tool '{name}' not found."

    def list_tools(self):
        return list(self.tools.keys())

    def remove_tool(self, name: str):
        if name in self.tools:
            del self.tools[name]
            return f"Tool '{name}' removed successfully."
        else:
            return f"Tool '{name}' not found."

class PythonREPL:
    def __init__(self):
        self.globals = {}
        self.locals = {}
        self.output_buffer = StringIO()
        self.self_learning_tool = SelfLearningTool()

    def run(self, command: str) -> str:
        old_stdout = sys.stdout
        sys.stdout = self.output_buffer
        try:
            exec(command, self.globals, self.locals)
            output = self.output_buffer.getvalue()
        except Exception as e:
            output = f"Error: {repr(e)}"
        finally:
            sys.stdout = old_stdout
            self.output_buffer.truncate(0)
            self.output_buffer.seek(0)
        return output

    def add_tool(self, name: str, func: callable):
        self.self_learning_tool.add_tool(name, func)

    def use_tool(self, name: str, *args, **kwargs):
        return self.self_learning_tool.use_tool(name, *args, **kwargs)

    def list_tools(self):
        return self.self_learning_tool.list_tools()

    def remove_tool(self, name: str):
        return self.self_learning_tool.remove_tool(name)

    def self_reflect(self):
        reflection = "Self-reflection:\n"
        reflection += f"Number of defined variables: {len(self.locals)}\n"
        reflection += f"Number of available tools: {len(self.list_tools())}\n"
        reflection += "Available tools:\n"
        for tool in self.list_tools():
            reflection += f"- {tool}\n"
        return reflection

    def self_inspect(self):
        inspection = "Self-inspection:\n"
        for name, value in self.locals.items():
            inspection += f"{name}: {type(value)}\n"
            if callable(value):
                try:
                    signature = inspect.signature(value)
                    inspection += f"  Signature: {signature}\n"
                except ValueError:
                    inspection += "  Signature: Unable to inspect\n"
        return inspection

def initialize_llm(model_path: str, n_ctx: int, n_threads: int = 4, n_batch: int = 512) -> Llama:
    try:
        return Llama(model_path=model_path, n_ctx=n_ctx, n_threads=n_threads, n_batch=n_batch, verbose=True)
    except Exception as e:
        logger.error(f"Failed to initialize LLM: {e}")
        raise

llm = initialize_llm('../models/lil.gguf', 4096)

def build_tool(config) -> Tool:
    tool = Tool(
        "Advanced Python REPL",
        "Execute sophisticated Python commands with self-learning capabilities",
        name_for_model="Advanced Python REPL",
        description_for_model=(
            "An advanced Python shell for executing complex Python commands. "
            "Input should be a valid Python command or script. "
            "Use print(...) to see the output of expressions. "
            "Capable of handling multi-line code, advanced Python features, "
            "and self-learning tools."
        ),
        logo_url="https://your-app-url.com/.well-known/logo.png",
        contact_email="[email protected]",
        legal_info_url="[email protected]"
    )
    
    python_repl = PythonREPL()
    
    def sanitize_input(query: str) -> str:
        return query.strip().strip("```").strip()

    @tool.get("/run_python")
    def run_python(query: str):
        sanitized_query = sanitize_input(query)
        result = python_repl.run(sanitized_query)
        return {"result": result, "execution_time": time.time()}
    
    @tool.get("/add_tool")
    def add_tool(name: str, code: str):
        sanitized_code = sanitize_input(code)
        try:
            exec(f"def {name}({sanitized_code})", python_repl.globals, python_repl.locals)
            python_repl.add_tool(name, python_repl.locals[name])
            return f"Tool '{name}' added successfully."
        except Exception as e:
            return f"Error adding tool: {str(e)}"

    @tool.get("/use_tool")
    def use_tool(name: str, args: str):
        try:
            result = python_repl.use_tool(name, *eval(args))
            return {"result": result}
        except Exception as e:
            return {"error": str(e)}

    @tool.get("/list_tools")
    def list_tools():
        return {"tools": python_repl.list_tools()}

    @tool.get("/remove_tool")
    def remove_tool(name: str):
        return {"result": python_repl.remove_tool(name)}

    @tool.get("/self_reflect")
    def self_reflect():
        return {"reflection": python_repl.self_reflect()}

    @tool.get("/self_inspect")
    def self_inspect():
        return {"inspection": python_repl.self_inspect()}

    @tool.get("/write_file")
    def write_file(file_path: str, text: str) -> str:
        write_path = Path(file_path)
        try:
            write_path.parent.mkdir(exist_ok=True, parents=False)
            with write_path.open("w", encoding="utf-8") as f:
                f.write(text)
            return f"File written successfully to {file_path}."
        except Exception as e:
            return "Error: " + str(e)
        
    @tool.get("/read_file")
    def read_file(file_path: str) -> str:
        read_path = Path(file_path)
        try:
            with read_path.open("r", encoding="utf-8") as f:
                content = f.read()
            return content
        except Exception as e:
            return "Error: " + str(e)

    return tool

if __name__ == "__main__":
    config = {}  # Add any necessary configuration
    advanced_python_repl = build_tool(config)
    
    # Run the FastAPI server
    import uvicorn
    uvicorn.run(advanced_python_repl, host="0.0.0.0", port=8000)