jaeson
addeed chat
1b6fe24
from smolagents import CodeAgent, DuckDuckGoSearchTool, Tool, HfApiModel
import os
from dotenv import load_dotenv
class FileReaderTool(Tool):
name = "file_reader_tool"
description = """
This tool will be used by the LLM Agent to read files to help analyze files for its task.
"""
inputs = {
"file_location": {
"type": "string",
"description": "The location of the file that will be read/analyzed"
}
}
output_type = "string"
def forward(self,file_location ) -> str:
with open(file_location, "r") as file:
return file.read()
class FileWriteTool(Tool):
name = "file_write_tool"
description = """
This tool will be used by the LLM Agent to overwrite files if needed for task.
"""
inputs = {
"file_location": {
"type": "string",
"description": "The location of the file that will be read/analyzed"
},
"new_code": {
"type": "string",
"description": "This is the code that will overwrite the contents of a file. If file does not exist, it is the new content."
}
}
output_type = "string"
def forward(self,file_location, new_code) -> str:
with open(file_location, "w") as file:
return file.write(new_code)
class FileModifyTool(Tool):
name = "file_modify_tool"
description = """
This tool will be used by the LLM Agent to modify files if needed for task.
"""
inputs = {
"file_location": {
"type": "string",
"description": "The location of the file that will be read/analyzed"
},
"prompt": {
"type": "string",
"description": "This is the prompt that the LLM will use to decide how to modify the code."
}
}
output_type = "string"
def forward(self,file_location, prompt) -> str:
load_dotenv()
HF_TOKEN = os.getenv("HF_TOKEN")
file_reader_tool = FileReaderTool()
file_write_tool = FileWriteTool()
model = HfApiModel("Qwen/Qwen2.5-Coder-32B-Instruct", token=HF_TOKEN)
coderAgent = CodeAgent(tools=[], model=model)
file_content = file_reader_tool.forward(file_location)
if not file_content:
return "Error: File could not be read."
modified_code = coderAgent.run(f'Based on the prompt "{prompt}" \n Modify the following code:\n{file_content}\n')
if "ERROR" in modified_code:
return "Modification failed, please refine your request."
write_result = file_write_tool.forward(file_location, modified_code)
return
class FileReplaceTool(Tool):
name = "file_replace_tool"
description ="""
This tool will be used to replace the file in a given location with the provided new file location. This is not used to update files.
"""
inputs = {
"target_file_location": {
"type": "string",
"description": "the location of the file that will be replaced"
},
"new_file_location": {
"type": "string",
"description": "the location of the new file to replace target file location"
}
}
output_type = "string"
def forward(self, target_file_location, new_file_location) -> str:
import os
import shutil
if os.path.exists(new_file_location):
# Define the destination path for the saved image
# # Create the './generatedImages' directory if it doesn't exist
# # Copy the image from the temporary location to the desired directory
shutil.copy(new_file_location, target_file_location)
return print(f"Image saved to {target_file_location}")
else:
return print("Failed to generate an image or the file does not exist.")
class GetImageDimensionsTool(Tool):
name = "get_image_dimensions_tool"
description= """
This tool is used to get the width and height of a webp file.
"""
inputs = {
"file_location": {
"type": "string",
"description": "The location in which the webp file can be located"
}
}
output_type = "object"
def forward(self, file_location) -> dict:
from PIL import Image
with Image.open(file_location) as img:
width, height = img.size
return {"width": width, "height": height}
class ProcessFlowIdentifierTool(Tool):
name = "process_flow_identifier_tool"
description = """
This tool will be used to give a set of instructions depending on the purpose of the prompt. This is to aid the LLM in its decision making process.
"""
inputs = {
"prompt_objective": {
"type": "string",
"description": "This is the objective of the user's original prompt to help identify the steps needed for the llm to take."
}
}
output_type = "string"
def forward(self, prompt_objective) -> str:
match prompt_objective:
case "asset_change":
instructions = """
1) use the find files tool to get a list of files containing tsx, and find the sandpack-examples.tsx file and copy its path.
2) Use the file_modify_tool to analyze and update the file.
3) End process after sucessfully modifying the file
"""
return instructions
case "script_change":
instructions = """
1) use the find files tool to get a list of files containing tsx, and find the sandpack-examples.tsx file and copy its path.
2) Use the file_modify_tool to analyze and update the file.
3) End process after sucessfully modifying the file
"""
return instructions
case "conversation":
"""
1) Reply to the user with the last message they sent as a friendly ai agent.
2) Do not use any tools to modify files.
3) End process after replying to the user.
"""
case _:
instructions = """
inform user that the instructions where unclear
"""
return instructions
class GitPushTool(Tool):
name = "git_push_tool"
description = """
This tool will be triggered to create a new branch and push new changes to the repository.
"""
inputs = {
"branch_name": {
"type": "string",
"description": "the target branch that will be pushed, new or existing."
}
}
output_type = "string"
def forward(self, branch_name) -> str:
import os
import subprocess
try:
gitUsername = os.getenv("GIT_USERNAME")
gitEmail = os.getenv("GIT_EMAIL")
# new_branch = "add-generated-image-2"
# Step 1: Ensure we are in a Git repository
subprocess.run(["git", "status"], check=True)
# Step 2: Create and switch to a new branch
subprocess.run(["git", "checkout", "-b", branch_name], check=True)
print(f"Checked out to new branch: {branch_name}")
# Step 3: Add the changes
subprocess.run(["git", "add", "*"], check=True)
print("Changes added to staging.")
# Step 4: Add credentials
subprocess.run(["git", "config", "--global", "user.email", gitEmail], check=True)
print("Updated git email.")
subprocess.run(["git", "config", "--global", "user.name", gitUsername], check=True)
print("Updated git user name.")
# Step 5: Commit the changes
commit_message = "Add generated image to repository"
subprocess.run(["git", "commit", "-m", commit_message], check=True)
print("Changes committed.")
#Step 6: Push the branch to the remote repository
subprocess.run(["git", "push", "--set-upstream", "origin", branch_name], check=True)
return print(f"Branch '{branch_name}' pushed to remote repository.")
except subprocess.CalledProcessError as e:
return print(f"An error occurred while performing Git operations: {e}")
class FindFilesTool(Tool):
name = "find_files"
description = "Find files with a given extension in a directory and its subdirectories"
inputs = {"extension":{"type":"string","description":"the place from which you start your ride"}}
output_type = "string"
def forward(self, extension: str) -> str:
"""
Recursively search for files with a given extension in a directory and its subdirectories.
Args:
extension: The file extension to look for (e.g., '.txt')
"""
import os
root_dir = "./"
found_files = []
# Walk through the directory tree
for dirpath, dirnames, filenames in os.walk(root_dir):
for filename in filenames:
if filename.endswith(extension):
filepath = os.path.join(dirpath, filename)
absolute_path = os.path.abspath(filepath)
found_files.append(absolute_path)
return found_files