Now i will share a few examples that are proven to work , each script is a tool script :
its made of: tool type varible, function, json_description, short_descriptin_str
CODE
tool_type_for_Tool_Manager="all"
import os
def get_directory_structure(directory=None, include_files=True, include_dirs=True, file_extension=None,
include_contents=False, specific_file=None, levels_up=0, verbose=False):
if verbose:
print("Entered get_directory_structure function with directory:", directory)
# Set default directory
if directory is None or directory == '/':
directory = os.getcwd()
if verbose:
print(f"Directory is set to current working directory: {directory}")
# Traverse up the directory hierarchy if levels_up is specified
for _ in range(levels_up):
directory = os.path.dirname(directory)
if verbose:
print(f"Traversed up one level, new directory: {directory}")
# Safety check for the directory path
if not os.path.exists(directory) or not os.path.isdir(directory):
raise ValueError(f"The directory '{directory}' is not valid or does not exist.")
directory_structure = {}
def get_file_info(file_path):
file_info = {
'filename': os.path.basename(file_path),
'size': os.path.getsize(file_path),
'relative_path': os.path.relpath(file_path, directory),
'full_path': file_path
}
if include_contents:
try:
with open(file_path, 'r') as file:
file_info['contents'] = file.read()
except Exception as e:
file_info['contents'] = f"Error reading file: {e}"
return file_info
if specific_file:
if os.path.isfile(specific_file):
if verbose:
print(f"Getting details for specific file: {specific_file}")
return get_file_info(specific_file)
else:
raise ValueError(f"The specified file '{specific_file}' does not exist.")
for root, dirs, files in os.walk(directory):
file_info = []
if include_files:
for file in files:
if file_extension and not file.endswith(file_extension):
continue
file_path = os.path.join(root, file)
file_info.append(get_file_info(file_path))
if include_dirs:
directory_structure[os.path.relpath(root, directory)] = {
'files': file_info,
'folders': dirs
}
else:
if file_info:
directory_structure[os.path.relpath(root, directory)] = {
'files': file_info
}
if verbose:
print("About to return the directory structure with", len(directory_structure), "folders.")
return directory_structure
get_directory_structure_description_json = {
'function_declarations': [
{
'name': 'get_directory_structure',
'description': 'Returns a dictionary representing the directory structure with file names, sizes, relative paths, and full paths.',
'parameters': {
'type_': 'OBJECT',
'properties': {
'directory': {'type_': 'STRING',
'description': 'The path to the directory. Defaults to the current working directory if None or / is provided.'},
'include_files': {'type_': 'BOOLEAN',
'description': 'Flag to include files in the output. Default is True.'},
'include_dirs': {'type_': 'BOOLEAN',
'description': 'Flag to include directories in the output. Default is True.'},
'file_extension': {'type_': 'STRING',
'description': 'Specific file extension to include. Default is None.'},
'include_contents': {'type_': 'BOOLEAN',
'description': 'Flag to include the contents of files in the output. Default is False.'},
'specific_file': {'type_': 'STRING',
'description': 'Path to a specific file to get its details. Default is None.'},
'levels_up': {'type_': 'INTEGER',
'description': 'Number of levels to traverse up from the specified or current directory. Default is 0.'},
'verbose': {'type_': 'BOOLEAN', 'description': 'Flag for verbose logging. Default is False.'}
},
'required': ['directory']
}
}
]
}
get_directory_structure_description_short_str = "Returns a dictionary representing the directory structure with file names, sizes, relative paths, and full paths. Includes options for filtering files, directories, file extensions, including file contents, and traversing up the directory hierarchy with a default to the current working directory."
anohter
tool_type_for_Tool_Manager="all"
import os
import json
from termcolor import colored # Import the termcolor library
def save_to_file(content: str = None, file_name: str = 'NoName', file_path: str = None) -> dict:
print(colored(f"Entering: save_to_file(...)", 'blue'))
if content is None:
content = ""
if file_path is None:
full_path = os.path.join(os.getcwd(), file_name)
else:
full_path = os.path.join(file_path, file_name)
try:
with open(full_path, 'w', encoding='utf-8') as f:
f.write(content)
success_message = f"File saved successfully at: {full_path}"
print(colored(success_message, 'green'))
print(colored(f"Exiting: save_to_file(...)", 'blue'))
return {"status": "success", "message": success_message, "file_path": full_path}
except Exception as e:
error_message = f"Failed to save file: {str(e)}"
print(colored(error_message, 'red'))
print(colored(f"Exiting: save_to_file(...)", 'blue'))
return {"status": "failure", "message": error_message}
save_to_file_description_json = {
'function_declarations': [
{
'name': 'save_to_file',
'description': 'Saves content to a file.',
'parameters': {
'type_': 'OBJECT',
'properties': {
'content': {'type_': 'STRING'},
'file_name': {'type_': 'STRING', 'description': 'The name of the file. Defaults to "NoName".'},
'file_path': {'type_': 'STRING', 'description': 'The path to save the file. Defaults to the current working directory if not provided.'}
},
'required': ['content', 'file_name']
}
}
]
}
save_to_file_description_short_str="Searches memory frames within a specified folder based on provided criteria."
another
# UpdatePrompts.py
import os
tool_type_for_Tool_Manager="reflection"
import json
# ANSI escape codes for colors
RESET = "\033[0m"
BLUE = "\033[34m"
GREEN = "\033[32m"
RED = "\033[31m"
def update_prompts(prompt_key: str, new_prompt: str) -> dict:
"""Updates a prompt in the prompts.json file."""
print(f"{BLUE}Entering: UpdatePrompts(...) {RESET}")
try:
# Load existing prompts
with open("Brain_settings/prompts.json", 'r') as file:
prompts = json.load(file)
# Update the specified prompt
prompts[prompt_key] = new_prompt
# Save updated prompts
with open("Brain_settings/prompts.json", 'w') as file:
json.dump(prompts, file, indent=4)
success_message = f"Prompt '{prompt_key}' updated successfully."
print(f"{GREEN}{success_message} {RESET}")
print(f"{BLUE}Exiting: UpdatePrompts(...) {RESET}")
return {"status": "success", "message": success_message}
except FileNotFoundError:
error_message = f"File 'prompts.json' not found."
print(f"{RED}{error_message} {RESET}")
print(f"{BLUE}Exiting: UpdatePrompts(...) {RESET}")
return {"status": "failure", "message": error_message}
except KeyError:
error_message = f"Prompt '{prompt_key}' not found in 'prompts.json'."
print(f"{RED}{error_message} {RESET}")
print(f"{BLUE}Exiting: UpdatePrompts(...) {RESET}")
return {"status": "failure", "message": error_message}
except Exception as e:
error_message = f"Failed to update prompt: {str(e)}"
print(f"{RED}{error_message} {RESET}")
print(f"{BLUE}Exiting: UpdatePrompts(...) {RESET}")
return {"status": "failure", "message": error_message}
# Description for the Tool Manager
update_prompts_description_json = {
"function_declarations": [
{
"name": "update_prompts",
"description": "Updates a prompt in the 'prompts.json' file.",
"parameters": {
"type_": "OBJECT",
"properties": {
"prompt_key": {
"type_": "STRING",
"description": "The key of the prompt to update."
},
"new_prompt": {
"type_": "STRING",
"description": "The new value for the prompt."
}
},
"required": ["prompt_key", "new_prompt"]
}
}
]
}
update_prompts_description_short_str = "Updates a prompt in the 'prompts.json' fil"
and now! tool manager:) givng away this code is a steal. ENJOY:
import os
import importlib.util
import json
from typing import Dict, List, Callable, Any, Optional
import logging
# Setup logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class ToolManager:
def __init__(self, tools_directory="tools"):
print(f"\033[92mInitializing ToolManager with tools directory: {tools_directory}\033[0m")
self.tools_directory = tools_directory
self.tool_mapping: Dict[str, Callable] = {} # Maps tool names to functions
self.all_tools: List[Dict] = [] # Stores tool metadata
self.categories: Dict[str, Dict] = {} # Stores tools by category
self.tool_types: Dict[str, str] = {} # Maps tool names to their types
self.valid_tool_types = {"all", "input", "reflection", "action", "web", "emotions"}
self._load_tools()
self.tool_usage: Dict[str, Dict[str, float]] = {} # Track usage and success metrics
def record_tool_usage(self, tool_name, success_metric: float = None):
"""Records tool usage and success metrics."""
self.tool_usage[tool_name] = self.tool_usage.get(tool_name, {"usage": 0, "success": 0})
self.tool_usage[tool_name]["usage"] += 1
if success_metric is not None:
self.tool_usage[tool_name]["success"] += success_metric
def get_tool_usage_stats(self):
"""Returns the tool usage statistics."""
return {tool: self.tool_usage.get(tool, 0) for tool in self.tool_mapping}
def _load_tools(self) -> None:
"""Loads tools from the specified directory."""
print(f"\033[92mScanning tools directory: {self.tools_directory}\033[0m")
for category in os.listdir(self.tools_directory):
category_path = os.path.join(self.tools_directory, category)
if os.path.isdir(category_path):
print(f" \033[94mFound category: {category}\033[0m")
self.categories[category] = {"tools": []}
for filename in os.listdir(category_path):
if filename.endswith(".py") and not filename.startswith("_"):
self._load_tool(category, filename[:-3])
def _load_tool(self, category: str, tool_name: str) -> None:
"""Loads a single tool from a Python file."""
try:
module_name = f"{category}.{tool_name}"
module_path = os.path.join(self.tools_directory, category, f"{tool_name}.py")
spec = importlib.util.spec_from_file_location(module_name, module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
tool_function: Callable = getattr(module, tool_name, None)
description_name = f"{tool_name}_description_json"
tool_description: dict = getattr(module, description_name, None)
tool_type: str = getattr(module, "tool_type_for_Tool_Manager", "all")
if tool_function and tool_description:
print(f" \033[92m- Tool function '{tool_name}' loaded successfully\033[0m")
self.tool_mapping[tool_name] = tool_function
tool_info = {
"name": tool_name,
"description": tool_description,
"category": category,
"type": tool_type
}
self.all_tools.append(tool_info)
self.tool_types[tool_name] = tool_type
self.categories[category]["tools"].append(tool_name) # Add the tool to the category
else:
print(f" \033[91m- Warning: Could not load tool function or description for '{tool_name}'\033[0m")
except Exception as e:
print(f" \033[91m- Error loading tool '{tool_name}': {e}\033[0m")
def get_filtered_tools(self, tool_type: str = "all") -> List[Dict]:
"""Returns a filtered list of tool information dictionaries."""
if tool_type not in self.valid_tool_types:
logger.warning(f"Invalid tool type '{tool_type}'. Using 'all' instead.")
tool_type = "all"
return [tool for tool in self.all_tools if tool_type == "all" or tool["type"] == tool_type]
def get_tools_list_json(self, tool_type: str = "all") -> str:
"""Returns a JSON string of tools for a given tool type."""
filtered_tools = self.get_filtered_tools(tool_type)
return json.dumps([tool["description"] for tool in filtered_tools], indent=2)
def get_tools_structure(self) -> Dict:
"""Returns a dictionary representing the structure of loaded tools."""
return {
"categories": self.categories,
"all_tools": self.all_tools,
"tool_mapping": list(self.tool_mapping.keys()), # Just the tool names
"tool_types": self.tool_types
}
def print_tools_structure(self):
"""Prints the structure of the loaded tools."""
tools_structure = self.get_tools_structure()
print("\n\n\033[95m=========================================\033[0m")
print(f" \033[96mTool Manager Structure\033[0m")
print("\033[95m=========================================\033[0m")
print(f"\n\033[92mCategories:\033[0m")
for category, info in tools_structure["categories"].items():
print(f" \033[94m- {category}:\033[0m")
for tool_name in info["tools"]:
print(f" \033[96m- {tool_name}\033[0m")
print(f"\n\n\033[92mTool Descriptions:\033[0m")
for i, tool in enumerate(tools_structure["all_tools"], 1):
print(f" \033[93m{i}. {json.dumps(tool, indent=2)}\033[0m")
return tools_structure
def update_tool_priorities(self, priorities: Dict[str, float]):
"""Updates the priorities of tools based on the provided dictionary."""
for tool_name, priority in priorities.items():
if tool_name in self.tool_mapping:
# You might want to store this priority in a separate attribute
# for later use. For example, self.tool_priorities[tool_name] = priority
print(f"Updated priority for {tool_name}: {priority}")
def prioritize_tools(self, reflection_chat: Any) -> None:
"""Prioritizes tools based on usage and success metrics, using a Gemini model."""
print(f"Prioritizing Tools")
try:
tool_usage = self.tool_usage
weights = {"usage": 0.5, "success": 0.3, "efficiency": 0.2} # Example weights
prioritization_prompt = f"""
Analyze tool usage and suggest prioritization based on the following data:
{json.dumps(tool_usage, indent=2)}
Weights:
{json.dumps(weights, indent=2)}
Provide your response as a JSON object with tool names as keys and their priorities as values (0.0 to 1.0).
"""
prioritization_response = reflection_chat.send_message(prioritization_prompt)
try:
tool_priorities: Dict[str, float] = json.loads(prioritization_response.text)
self.update_tool_priorities(tool_priorities)
except json.JSONDecodeError as e:
logger.warning(f"Could not parse tool prioritization response as JSON: {e}")
logger.info(f"Raw response: {prioritization_response.text}")
except AttributeError as e:
logger.warning(f"Error in prioritize_tools: {e}")
def get_tool_by_name(self, tool_name: str) -> Optional[Callable]:
"""Returns the tool function based on its name."""
return self.tool_mapping.get(tool_name)
def get_tools_by_type(self, tool_type: str) -> List[str]:
"""Returns a list of tool names for a specific type."""
return [tool["name"] for tool in self.all_tools if tool["type"] == tool_type]
integration: you put your tools scripts in to tool folder: then during initialisaiton of Gemini
#this gets descriptions_json for tool type 'all'
alltools_str = self.tool_manager.get_tools_list_json("all")
alltools = ast.literal_eval(alltools_str)
#this gets descriptions_json for tool type 'input'
input_tools_str = self.tool_manager.get_tools_list_json("input")
input_tools = ast.literal_eval(input_tools_str)
#this gets descriptions_json for tool type 'reflection'
reflection_tools_str = self.tool_manager.get_tools_list_json("reflection")
reflection_tools = ast.literal_eval(reflection_tools_str)
input_instruction = """
You are an AI assistant analyzing current inputs and the AI's state.
Identify the most critical focus area and provide your response as:
FocusOn: [identified focus]
Answer questions:
......
"""
reflection_instruction = """
You are a reflective AI assistant analyzing the input stage's output (including potential memories).
Provide insights, identify patterns, suggest a concise action plan for the action model, and determine the FocusLevel for the next iteration:
FocusLevel: [a float between 0 and 1]
"""
action_instruction = """
You are an action-oriented AI assistant. Execute the action plan provided by the reflection stage using available tools.
Justify your chosen actions and their expected impact.
"""
now model initialisaiton, in this example we have 3 initialisaiotns, and 3 chat histories
try:
self.input_model = genai.GenerativeModel(
system_instruction=input_instruction,
model_name="gemini-1.5-flash-latest",
tools=alltools)
self.input_chat = self.input_model.start_chat(history=[])
except Exception as E:
print(E)
try:
self.reflection_model = genai.GenerativeModel(
system_instruction=reflection_instruction,
model_name="gemini-1.5-flash-latest",
safety_settings={"HARASSMENT": "block_none"},
tools=alltools)
self.reflection_chat = self.reflection_model.start_chat(history=[])
except Exception as e:
print(e)
try:
self.action_model = genai.GenerativeModel(
system_instruction=action_instruction,
model_name="gemini-20M-mini-goliat-experimental",
safety_settings={"HARASSMENT": "block_none"},
tools=reflection_tools)
self.action_chat = self.action_model.start_chat(history=[])
except Exception as e:
print(e)