feat: complete assignment
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
from typing import Callable
|
||||
|
||||
from google.genai import types
|
||||
|
||||
from functions.get_file_content import (get_file_content,
|
||||
schema_get_file_content)
|
||||
from functions.get_files_info import get_files_info, schema_get_files_info
|
||||
from functions.run_python_file import run_python_file, schema_run_python_file
|
||||
from functions.write_file import schema_write_file, write_file
|
||||
|
||||
available_functions = types.Tool(
|
||||
function_declarations=[
|
||||
schema_get_file_content,
|
||||
schema_get_files_info,
|
||||
schema_run_python_file,
|
||||
schema_write_file,
|
||||
]
|
||||
)
|
||||
|
||||
function_map: dict[str, Callable[..., str]] = {
|
||||
"get_file_content": get_file_content,
|
||||
"get_files_info": get_files_info,
|
||||
"run_python_file": run_python_file,
|
||||
"write_file": write_file,
|
||||
}
|
||||
|
||||
|
||||
def call_function(
|
||||
function_call: types.FunctionCall, verbose: bool = False
|
||||
) -> types.Content:
|
||||
if verbose:
|
||||
print(f"Calling function: {function_call.name}({function_call.args})")
|
||||
else:
|
||||
print(f"Calling function: {function_call.name}")
|
||||
|
||||
function_name = function_call.name or ""
|
||||
function = function_map.get(function_name)
|
||||
|
||||
if function is None:
|
||||
return types.Content(
|
||||
role="tool",
|
||||
parts=[
|
||||
types.Part.from_function_response(
|
||||
name=function_name,
|
||||
response={"error": f"Unknown function: {function_name}"},
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
args = dict(function_call.args) if function_call.args else {}
|
||||
args["working_directory"] = "./calculator"
|
||||
function_result = function(**args)
|
||||
|
||||
return types.Content(
|
||||
role="tool",
|
||||
parts=[
|
||||
types.Part.from_function_response(
|
||||
name=function_name,
|
||||
response={"result": function_result},
|
||||
)
|
||||
],
|
||||
)
|
||||
@@ -0,0 +1,47 @@
|
||||
import os
|
||||
|
||||
from google.genai import types
|
||||
|
||||
|
||||
def get_file_content(working_directory: str, file_path: str) -> str:
|
||||
try:
|
||||
wdp = os.path.abspath(working_directory)
|
||||
target = os.path.normpath(os.path.join(wdp, file_path))
|
||||
valid_target_dir = os.path.commonpath([wdp, target]) == wdp
|
||||
|
||||
if not valid_target_dir:
|
||||
return f'Error: Cannot read "{file_path}" as it is outside the permitted working directory'
|
||||
|
||||
if not os.path.isfile(target):
|
||||
return f'Error: File not found or is not a regular file: "{file_path}"'
|
||||
|
||||
MAX_CHARS = 10000
|
||||
content = ""
|
||||
|
||||
with open(target, "r") as f:
|
||||
content += f.read(MAX_CHARS)
|
||||
|
||||
if f.read(1):
|
||||
content += (
|
||||
f'[...File "{file_path}" truncated at {MAX_CHARS} characters]'
|
||||
)
|
||||
|
||||
return content
|
||||
except Exception as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
|
||||
schema_get_file_content = types.FunctionDeclaration(
|
||||
name="get_file_content",
|
||||
description="",
|
||||
parameters=types.Schema(
|
||||
type=types.Type.OBJECT,
|
||||
properties={
|
||||
"file_path": types.Schema(
|
||||
type=types.Type.STRING,
|
||||
description="",
|
||||
),
|
||||
},
|
||||
required=["file_path"],
|
||||
),
|
||||
)
|
||||
@@ -0,0 +1,44 @@
|
||||
import os
|
||||
|
||||
from google.genai import types
|
||||
|
||||
|
||||
def get_files_info(working_directory, directory: str = ".") -> str:
|
||||
try:
|
||||
wdp = os.path.abspath(working_directory)
|
||||
target = os.path.normpath(os.path.join(wdp, directory))
|
||||
valid_target_dir = os.path.commonpath([wdp, target]) == wdp
|
||||
|
||||
if not valid_target_dir:
|
||||
return f'Error: Cannot list "{directory}" as it is outside the permitted directory'
|
||||
|
||||
if not os.path.isdir(target):
|
||||
return f'Error: "{directory}" is not a directory'
|
||||
|
||||
file_descriptors = []
|
||||
|
||||
for file_name in os.listdir(target):
|
||||
file_path = os.path.normpath(os.path.join(target, file_name))
|
||||
size = os.path.getsize(file_path)
|
||||
is_dir = os.path.isdir(file_path)
|
||||
file_descriptors.append(f"- {file_name}: file_size={size}, is_dir={is_dir}")
|
||||
|
||||
return "\n".join(file_descriptors)
|
||||
except Exception as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
|
||||
schema_get_files_info = types.FunctionDeclaration(
|
||||
name="get_files_info",
|
||||
description="Lists files in a specified directory relative to the working directory, providing file size and directory status",
|
||||
parameters=types.Schema(
|
||||
type=types.Type.OBJECT,
|
||||
properties={
|
||||
"directory": types.Schema(
|
||||
type=types.Type.STRING,
|
||||
description="Directory path to list files from, relative to the working directory (default is the working directory itself)",
|
||||
),
|
||||
},
|
||||
required=["directory"],
|
||||
),
|
||||
)
|
||||
@@ -0,0 +1,66 @@
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
from google.genai import types
|
||||
|
||||
|
||||
def run_python_file(
|
||||
working_directory: str, file_path: str, args: list[str] | None = None
|
||||
) -> str:
|
||||
try:
|
||||
wdp = os.path.abspath(working_directory)
|
||||
target = os.path.normpath(os.path.join(wdp, file_path))
|
||||
valid_target_dir = os.path.commonpath([wdp, target]) == wdp
|
||||
|
||||
if not valid_target_dir:
|
||||
return f'Error: Cannot execute "{file_path}" as it is outside the permitted working directory'
|
||||
|
||||
if not os.path.isfile(target):
|
||||
return f'Error: "{file_path}" does not exist or is not a regular file'
|
||||
|
||||
if not target.endswith(".py"):
|
||||
return f'Error: "{file_path}" is not a Python file'
|
||||
|
||||
command = ["python", target]
|
||||
|
||||
if args != None:
|
||||
command.extend(args)
|
||||
|
||||
result = subprocess.run(
|
||||
command,
|
||||
cwd=working_directory,
|
||||
text=True,
|
||||
timeout=30_000,
|
||||
capture_output=True,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
return f"Process exited with code {result.returncode}"
|
||||
|
||||
if len(result.stdout) == 0 and len(result.stderr) == 0:
|
||||
return "No output produced"
|
||||
|
||||
return f"STDOUT: {result.stdout}\nSTDERR: {result.stderr}"
|
||||
except Exception as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
|
||||
schema_run_python_file = types.FunctionDeclaration(
|
||||
name="run_python_file",
|
||||
description="",
|
||||
parameters=types.Schema(
|
||||
type=types.Type.OBJECT,
|
||||
properties={
|
||||
"file_path": types.Schema(
|
||||
type=types.Type.STRING,
|
||||
description="",
|
||||
),
|
||||
"args": types.Schema(
|
||||
type=types.Type.ARRAY,
|
||||
items=types.Schema(type=types.Type.STRING),
|
||||
description="",
|
||||
),
|
||||
},
|
||||
required=["file_path"],
|
||||
),
|
||||
)
|
||||
@@ -0,0 +1,47 @@
|
||||
import os
|
||||
|
||||
from google.genai import types
|
||||
|
||||
|
||||
def write_file(working_directory: str, file_path: str, content: str) -> str:
|
||||
try:
|
||||
wdp = os.path.abspath(working_directory)
|
||||
target = os.path.normpath(os.path.join(wdp, file_path))
|
||||
valid_target_dir = os.path.commonpath([wdp, target]) == wdp
|
||||
|
||||
if not valid_target_dir:
|
||||
return f'Error: Cannot write to "{file_path}" as it is outside the permitted working directory'
|
||||
|
||||
if os.path.isdir(target):
|
||||
return f'Error: Cannot write to "{file_path}" as it is a directory'
|
||||
|
||||
os.makedirs("/".join(target.split("/")[:-1]), exist_ok=True)
|
||||
|
||||
with open(target, "w") as f:
|
||||
f.write(content)
|
||||
|
||||
return (
|
||||
f'Successfully wrote to "{file_path}" ({len(content)} characters written)'
|
||||
)
|
||||
except Exception as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
|
||||
schema_write_file = types.FunctionDeclaration(
|
||||
name="write_file",
|
||||
description="",
|
||||
parameters=types.Schema(
|
||||
type=types.Type.OBJECT,
|
||||
properties={
|
||||
"file_path": types.Schema(
|
||||
type=types.Type.STRING,
|
||||
description="",
|
||||
),
|
||||
"content": types.Schema(
|
||||
type=types.Type.STRING,
|
||||
description="",
|
||||
),
|
||||
},
|
||||
required=["file_path", "content"],
|
||||
),
|
||||
)
|
||||
Reference in New Issue
Block a user