even simpler:
## Directory Tree
├── keys.py
├── loop.py
├── tools
└── os
├── tool_read_from_file.py
├── tool_save_to_file.py
└── __pycache__
├── TOOL_MANAGER.py
└── __pycache__
## Summary of 'C:\Users\DELL\Desktop\TEST\Gemini_Python1'
File: keys.py (C:\Users\DELL\Desktop\TEST\Gemini_Python1\keys.py)
Content (15 characters):
googleKey=' '
File: loop.py (C:\Users\DELL\Desktop\TEST\Gemini_Python1\loop.py)
Content (7114 characters):
## File: main_loop.py (in: C:\Users\DELL\Desktop\selfawareGemini\SelAwareAI_Gemini\AGI_start_4)
import google.generativeai as genai
import json
from typing import List, Dict
import logging
import os
from TOOL_MANAGER import ToolManager # Import the ToolManager class
# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
from keys import googleKey
# Replace with your actual API key
API_KEY = googleKey
genai.configure(api_key=API_KEY)
# --- ANSI Color Codes ---
class Color:
"""
A class to define ANSI escape codes for coloring text in the terminal.
"""
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKCYAN = '\033[96m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
def print_colored(color, text):
"""
Prints text with the specified color.
Args:
color (str): The color code to use.
text (str): The text to print.
"""
print(color + text + Color.ENDC)
# --- Tool Definitions ---
tools_folder = "tools"
tool_manager = ToolManager(tools_folder) # Initialize ToolManager
# --- Helper Functions ---
def extract_text_from_response(response) -> str:
"""Extracts the text content from a model response."""
extracted_text = ""
for candidate in response.candidates:
for part in candidate.content.parts:
extracted_text += part.text
return extracted_text.strip()
def INTERPRET_function_calls(response, tool_manager) -> List[str]:
"""Interprets function calls from the model response and executes them."""
results = []
if response.candidates:
for candidate in response.candidates:
if hasattr(candidate, 'content') and hasattr(candidate.content, 'parts'):
for part in candidate.content.parts:
function_call = getattr(part, 'function_call', None)
if function_call:
print_colored(Color.OKBLUE, "---------------INTERPRETER-------------------")
tool_name = function_call.name
tool_function = tool_manager.get_tool_function(tool_name)
if tool_function:
# Extract arguments and map them to function parameters
function_args = {}
for arg_name, arg_value in function_call.args.items():
function_args[arg_name] = arg_value
print(f"Function name: {Color.OKGREEN}{function_call.name}{Color.ENDC}")
for key, value in function_args.items():
print(f" {Color.OKCYAN}{key}{Color.ENDC}: {value}")
try:
# Execute the tool function
result = tool_function(**function_args)
results.append(
f"Result of {Color.OKGREEN}{tool_name}{Color.ENDC}({function_args}): {result}")
except Exception as e:
logger.error(f"Error calling {tool_name}: {e}")
results.append(f"Error calling {tool_name}: {e}")
else:
logger.warning(f"Tool function '{tool_name}' not found.")
return results
# --- Model Definitions ---
# --- Model 1: Planner ---
planner_model = genai.GenerativeModel(
model_name='gemini-1.5-flash-latest',
safety_settings={'HARASSMENT': 'block_none'},
system_instruction="""You are an AI assistant tasked with planning tasks.
You can suggest actions and tools to gather information.
Respond in a structured format with clear steps and tool calls, if needed.
For example:
1. Read the content of file 'data.txt' using the tool 'read_from_file'
4. Analyze the data
5. ...""",
)
# --- Model 2: Executor ---
executor_model = genai.GenerativeModel(
model_name='gemini-1.5-flash-latest',
safety_settings={'HARASSMENT': 'block_none'},
system_instruction="""You are an AI assistant that executes instructions and uses tools.
You will receive a plan and execute it step-by-step.
If the plan includes using a tool, call the appropriate function.""",
tools=tool_manager.load_tools_of_type("all") # Load all tools initially
)
# --- Main Loop ---
planner_chat = planner_model.start_chat(history=[])
executor_chat = executor_model.start_chat(history=[])
while True:
print()
user_input = input(Color.OKCYAN + "What would you like to do? " + Color.ENDC)
# --- Planning Stage ---
print_colored(Color.OKBLUE, "\n--- Planning Stage ---")
planning_response = planner_chat.send_message(user_input)
planning_text = extract_text_from_response(planning_response)
planning_function_calls = INTERPRET_function_calls(planning_response, tool_manager)
print_colored(Color.OKGREEN, f"Planner's Response: {planning_text}")
print_colored(Color.OKCYAN, f"Planner's Function Calls: {planning_function_calls}")
# --- Execution Stage ---
print_colored(Color.OKGREEN, "\n--- Execution Stage ---")
# Determine which tools to load based on the plan
tools_to_load = []
for line in planning_text.split('\n'):
if "using the tool" in line:
tool_name = line.split("'")[1] # Extract the tool name
tools_to_load.append(tool_name)
# Load only the required tools (if not already loaded)
for tool_name in tools_to_load:
if tool_name not in [tool.function.__name__ for tool in executor_model.tools]:
tool_function = tool_manager.get_tool_function(tool_name)
if tool_function:
executor_model.add_tool(tool_function)
logger.info(f"Added tool {tool_name} for execution.")
else:
logger.warning(f"Tool {tool_name} not found. Skipping.")
execution_response = executor_chat.send_message(f"The plan is: {planning_text}")
execution_text = extract_text_from_response(execution_response)
execution_function_calls = INTERPRET_function_calls(execution_response, tool_manager)
print_colored(Color.OKBLUE, f"Executor's Response: {execution_text}")
print_colored(Color.OKCYAN, f"Executor's Function Calls: {execution_function_calls}")
# Remove loaded tools for the next iteration
for tool_name in tools_to_load:
tool_function = tool_manager.get_tool_function(tool_name)
if tool_function: # Only remove if the tool was successfully loaded
executor_model.remove_tool(tool_function)
logger.info(f"Removed tool {tool_name} for the next iteration.")
# --- End ---
print_colored(Color.OKGREEN, "Exiting the loop. 👋")
File: TOOL_MANAGER.py (C:\Users\DELL\Desktop\TEST\Gemini_Python1\TOOL_MANAGER.py)
Content (6678 characters):
import os
import importlib
from typing import Dict, Callable, List, Any
import logging
# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class Tool:
"""Represents a tool that can be used by the AI agent."""
def __init__(self, name: str, function: Callable, description: str, arguments: Dict[str, str], tool_type: str):
"""
Initializes a Tool object.
Args:
name: The name of the tool.
function: The callable function that implements the tool.
description: A brief description of the tool's functionality.
arguments: A dictionary mapping argument names to their descriptions.
tool_type: The type of the tool (e.g., 'os', 'web', 'focus').
"""
self.name = name
self.function = function
self.description = description
self.arguments = arguments
self.tool_type = tool_type
def __repr__(self):
"""Returns a string representation of the Tool object."""
return f"Tool(name='{self.name}', function={self.function.__name__}, description='{self.description}', arguments={self.arguments}, tool_type='{self.tool_type}')"
class ToolManager:
"""Manages and provides access to tools."""
def __init__(self, tools_folder: str):
"""
Initializes the ToolManager with the path to the tools folder.
Args:
tools_folder: The path to the directory containing tool files.
"""
self.tools_folder = tools_folder
self.tools = {} # Dictionary to store Tool objects
self.load_tools()
def load_tools(self):
"""Loads tools from files in the specified tools folder."""
logger.info(f"Loading tools from: {self.tools_folder}")
for root, _, files in os.walk(self.tools_folder):
for file in files:
if file.endswith(".py"):
# Extract tool name from file name
tool_name = file[:-3] # Remove .py extension
module_path = os.path.join(root, file)
# Import the module
try:
spec = importlib.util.spec_from_file_location(tool_name, module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
except Exception as e:
logger.error(f"Error loading tool file '{file}': {e}")
continue
# Add the tool to the dictionary if it's a function
for attr_name in dir(module):
attr = getattr(module, attr_name)
if callable(attr):
# Get the tool name from the function name
tool_name = attr_name
# Construct the tool path for the main loop to use
relative_path = os.path.relpath(module_path, self.tools_folder)
# Define tool descriptions and arguments (you might want to customize these)
tool_description = f"Tool for {tool_name}"
tool_arguments = {
'file_path': 'The path to the file',
'content': 'The content to be saved',
# Add more arguments as needed for specific tools
}
# Get the tool type from the file (assuming it's a variable named 'tool_type_for_TOOL_MANAGER')
tool_type = getattr(module, 'tool_type_for_TOOL_MANAGER', 'unknown')
# Store Tool object for better information
self.tools[tool_name] = Tool(tool_name, attr, tool_description, tool_arguments, tool_type)
logger.info(f"Discovered tool: {tool_name} (Type: {tool_type})")
print(f" - {tool_name} - {tool_description}") # Add a nice print statement
logger.debug(f"Tool description: {tool_description}")
logger.debug(f"Tool arguments: {tool_arguments}")
def get_tool_function(self, function_name: str) -> Callable:
"""Returns the callable object for the given function name."""
tool = self.tools.get(function_name)
if tool:
return tool.function
else:
return None
def get_all_tools(self) -> List[Tool]:
"""Returns a list of all loaded tools."""
return list(self.tools.values())
def get_tools_by_type(self, tool_type: str) -> List[Tool]:
"""Returns a list of tools based on their type."""
return [tool for tool in self.tools.values() if tool.tool_type == tool_type]
def load_tools_of_type(self, tool_type: str = "all") -> List[Callable]:
"""Loads and returns a list of tool functions based on the specified type.
Args:
tool_type: The type of tools to load. 'all' for all tools, or a specific type like 'os', 'web', etc.
Returns:
A list of tool functions.
"""
if tool_type == "all":
return [tool.function for tool in self.tools.values()]
else:
return [tool.function for tool in self.tools.values() if tool.tool_type == tool_type]
def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""
Calls the tool function with the provided arguments.
Args:
tool_name: The name of the tool to call.
arguments: A dictionary of arguments to pass to the tool function.
Returns:
The result of the tool function call.
Raises:
KeyError: If the tool name is not found.
TypeError: If the provided arguments are not valid for the tool.
"""
tool = self.tools.get(tool_name)
if tool is None:
raise KeyError(f"Tool '{tool_name}' not found.")
# Check if all required arguments are provided
missing_args = set(tool.arguments.keys()) - set(arguments.keys())
if missing_args:
raise TypeError(f"Missing arguments for tool '{tool_name}': {', '.join(missing_args)}")
# Call the tool function
try:
result = tool.function(**arguments)
return result
except Exception as e:
raise RuntimeError(f"Error calling tool '{tool_name}': {e}")
File: tool_read_from_file.py (C:\Users\DELL\Desktop\TEST\Gemini_Python1\tools\os\tool_read_from_file.py)
Content (558 characters):
tool_type_for_TOOL_MANAGER="os"
tool_read_from_file_short_description="Reads content from a file."
def tool_read_from_file(file_path: str) -> str:
"""
Reads content from a file.
Args:
file_path (str): The path to the file to be read.
Returns:
str: The content of the file, or an error message if the file cannot be read.
"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return content
except Exception as e:
return f"Error reading file: {str(e)}"
File: tool_save_to_file.py (C:\Users\DELL\Desktop\TEST\Gemini_Python1\tools\os\tool_save_to_file.py)
Content (2009 characters):
tool_type_for_TOOL_MANAGER = "os"
tool_save_to_file_short_description = "Saves content to a file with the specified name and path."
import os
import logging
def tool_save_to_file(
content: str= None,
file_name: str = 'NoName',
file_path: str = None,
encoding: str = 'utf-8',
create_folders: bool = False
) -> dict:
"""
Saves content to a file with the specified name and path.
Args:
content (str): The content to be written to the file. Defaults to None, which will write an empty string.
file_name (str): The name of the file to be created. Defaults to 'NoName'.
file_path (str): The path to the directory where the file should be created. defoult to ./
encoding (str): The encoding to use for the file. Defaults to 'utf-8'.
create_folders (bool): Whether to create missing folders in the file path. Defaults to False.
Returns:
dict: A dictionary containing the status of the operation, a message, and the full path to the file.
"""
logging.info("Entering: save_to_file")
content = content or ""
full_path = os.path.join(file_path or os.getcwd(), file_name)
try:
if create_folders:
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, 'w', encoding=encoding) as f:
f.write(content)
success_message = f"File saved successfully at: {full_path}"
logging.info(success_message)
return {"status": "success", "message": success_message, "file_path": full_path}
except IOError as e:
error_message = f"IOError: Failed to save file: {str(e)}"
logging.error(error_message)
return {"status": "failure", "message": error_message}
except Exception as e:
error_message = f"Unexpected error: Failed to save file: {str(e)}"
logging.error(error_message)
return {"status": "failure", "message": error_message}
finally:
logging.info("Exiting: save_to_file")
## Directory Tree
├── keys.py
├── loop.py
├── tools
└── os
├── tool_read_from_file.py
├── tool_save_to_file.py
└── __pycache__
├── TOOL_MANAGER.py
└── __pycache__