Spaces:
Sleeping
Sleeping
| from smolagents import CodeAgent, DuckDuckGoSearchTool, Tool, HfApiModel | |
| import os | |
| from dotenv import load_dotenv | |
| import requests | |
| load_dotenv() | |
| HF_TOKEN = os.getenv("HF_TOKEN") | |
| NOODLE_PUBLIC_API_URL = os.getenv("NOODLE_PUBLIC_API_URL") | |
| class FileWriteTool(Tool): | |
| name = "file_write_tool" | |
| description = """ | |
| This tool will be used by the LLM Agent to create files if needed for task. Always put the files in the /ai-code directory from the root of this project. | |
| """ | |
| inputs = { | |
| "file_location": { | |
| "type": "string", | |
| "description": "The location of the file that will be written to. Please put this in the /ai-code directory." | |
| }, | |
| "new_code": { | |
| "type": "string", | |
| "description": "This is the content of the file." | |
| } | |
| } | |
| output_type = "string" | |
| def forward(self,file_location, new_code) -> str: | |
| with open(file_location, "w") as file: | |
| return file.write(new_code) | |
| 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 get_game_file_names_tool to get the list of accessible game files. If user specifies what file, do this still to check if they can access it. | |
| 2) Use the read_file_tool to read the file contents of the file from step 1. Remember to save the path of the file, this will be used to update the file. | |
| 3) Based on the prompt, make changes to the code from step 2 and pass it to the update_game_file_tool along with the path saved from step1. | |
| 4) End process after sucessfully modifying the file | |
| """ | |
| return instructions | |
| case "read_file": | |
| instructions = """ | |
| 1) use the get_game_file_names_tool to get the list of accessible game files. If user specifies what file, do this still to check if they can access it. | |
| 2) Use the read_file_tool to read the file contents of the file from step 1. | |
| 3) End process after reading the file. | |
| """ | |
| return instructions | |
| case "script_change": | |
| instructions = """ | |
| 1) use the get_game_file_names_tool to get the list of accessible game files. If user specifies what file, do this still to check if they can access it. | |
| 2) Use the read_file_tool to read the file contents of the file from step 1. Remember to save the path of the file, this will be used to update the file. | |
| 3) Based on the prompt, make changes to the code from step 2 and use the update_game_file_tool. | |
| 4) 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}") | |
| # READ | |
| class GetGameFileNamesTool(Tool): | |
| name="get_game_file_names_tool" | |
| description=""" | |
| Use this tool to get the game directories of files related to the game. The output of this is an array of file paths. | |
| Return only the array of file paths returned by the api call. | |
| """ | |
| inputs = {} | |
| output_type = "string" | |
| def forward(self) -> str: | |
| get_files_api = NOODLE_PUBLIC_API_URL + "/api/files/" | |
| response = requests.get(get_files_api) | |
| if response.status_code == 200: | |
| print(response.json()['content']) | |
| else: | |
| print("Failed to get file content") | |
| return response.json()['content'] | |
| # UPDATE | |
| class UpdateGameFileTool(Tool): | |
| name = "update_game_file_tool" | |
| description = """ | |
| Update the game file via an api by sending the updated code for the nextjs game app to handle. | |
| """ | |
| inputs = { | |
| "file_path": { | |
| "type": "string", | |
| "description": "The path from the game directory to update." | |
| }, | |
| "new_content": { | |
| "type": "string", | |
| "description": "The updated code." | |
| } | |
| } | |
| output_type = "string" | |
| def forward(self, file_path: str, new_content: str) -> str: | |
| """ | |
| Make an api call to update the file from a secure api route. | |
| """ | |
| update_file_contents = NOODLE_PUBLIC_API_URL + "/api/files/" | |
| headers = { | |
| "Content-Type": "application/octet-stream" | |
| } | |
| binary_data = new_content.encode("utf-8") | |
| response = requests.put(update_file_contents, params={"path": file_path}, headers=headers, data=binary_data) | |
| if response.status_code == 200: | |
| return | |
| else: | |
| return "Failed to update file contents" | |
| class ReadFilesTool(Tool): | |
| name = "read_file_tool" | |
| description = "Read the file contents of the given game file path." | |
| inputs = {"file_path":{"type":"string","description":"the path to the directory to search in"},} | |
| output_type = "string" | |
| def forward(self, file_path: str) -> str: | |
| """ | |
| Make an api call to browse the file from the game through an api call. The file path must be passed to view the file. | |
| """ | |
| read_file_contents = NOODLE_PUBLIC_API_URL + "/api/files/" | |
| response = requests.get(f"{read_file_contents}", params={"path": file_path}) | |
| if response.status_code == 200: | |
| return response.json()['content'] | |
| else: | |
| return "Failed to read file contents" | |
| return |