Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions app/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,10 @@ class BrowserSettings(BaseModel):
)


class OutputSettings(BaseModel):
directory: str = Field("output", description="Directory for output files")


class AppConfig(BaseModel):
llm: Dict[str, LLMSettings]
browser_config: Optional[BrowserSettings] = Field(
Expand All @@ -66,6 +70,9 @@ class AppConfig(BaseModel):
search_config: Optional[SearchSettings] = Field(
None, description="Search configuration"
)
output_config: Optional[OutputSettings] = Field(
None, description="Output configuration"
)

class Config:
arbitrary_types_allowed = True
Expand Down Expand Up @@ -163,6 +170,12 @@ def _load_initial_config(self):
if search_config:
search_settings = SearchSettings(**search_config)

# Parse output configuration
output_config = raw_config.get("output", {})
output_settings = None
if output_config:
output_settings = OutputSettings(**output_config)

config_dict = {
"llm": {
"default": default_settings,
Expand All @@ -173,6 +186,7 @@ def _load_initial_config(self):
},
"browser_config": browser_settings,
"search_config": search_settings,
"output_config": output_settings,
}

self._config = AppConfig(**config_dict)
Expand All @@ -188,6 +202,27 @@ def browser_config(self) -> Optional[BrowserSettings]:
@property
def search_config(self) -> Optional[SearchSettings]:
return self._config.search_config

@property
def output_config(self) -> Optional[OutputSettings]:
return self._config.output_config


config = Config()

# Define a function to get the output directory path
def get_output_directory() -> Path:
"""Get the configured output directory path"""
output_config = config.output_config
if output_config and output_config.directory:
output_dir = Path(output_config.directory)
# If it's a relative path, make it relative to the project root
if not output_dir.is_absolute():
output_dir = PROJECT_ROOT / output_dir
return output_dir
# Default to workspace directory if not configured
return WORKSPACE_ROOT

# Create the output directory if it doesn't exist
OUTPUT_ROOT = get_output_directory()
OUTPUT_ROOT.mkdir(parents=True, exist_ok=True)
106 changes: 78 additions & 28 deletions app/tool/code_editor.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import aiofiles
from pydantic import BaseModel, Field

from app.config import OUTPUT_ROOT
from app.tool.base import BaseTool


Expand All @@ -22,6 +23,9 @@ class FileEditor(BaseTool):

name: str = "file_editor"
description: str = """Edit or create any type of file using specialized formats for precise modifications.
By default, files are saved to the configured output directory (set in config.toml).
You can specify a relative path within the output directory, or set use_output_dir=False to save to an absolute path.

Supports multiple edit formats for different types of changes:

1. DIFF MODE (format="diff"): For targeted edits to specific parts of files
Expand Down Expand Up @@ -88,6 +92,11 @@ def new_function():
"enum": ["w", "a"],
"description": "File opening mode: 'w' for write (default), 'a' for append",
"default": "w"
},
"use_output_dir": {
"type": "boolean",
"description": "(optional) Whether to save the file in the configured output directory. Default is True.",
"default": True
}
},
"required": []
Expand All @@ -96,32 +105,41 @@ def new_function():
async def execute(self, format: str = "diff", edits: str = "",
content: Optional[str] = None,
file_path: Optional[str] = None,
mode: str = "w") -> str:
mode: str = "w",
use_output_dir: bool = True) -> str:
"""Execute file edits or creation using the specified format or direct content"""
try:
# Direct content mode (file saving)
if content is not None and file_path is not None:
try:
# Determine the full file path
if use_output_dir:
# Use the configured output directory
full_path = OUTPUT_ROOT / file_path
else:
# Use the exact path specified
full_path = Path(file_path)

# Create directory if needed
directory = os.path.dirname(os.path.abspath(file_path))
directory = os.path.dirname(str(full_path))
if directory:
os.makedirs(directory, exist_ok=True)

# Write the file using aiofiles for async I/O
async with aiofiles.open(file_path, mode, encoding="utf-8") as f:
async with aiofiles.open(full_path, mode, encoding="utf-8") as f:
await f.write(content)

return f"Content successfully saved to {file_path}"
return f"Content successfully saved to {full_path}"
except Exception as e:
return f"Error saving file: {str(e)}"

# Standard FileEditor functionality
if format == "whole":
result = await self._apply_whole_file_edits(edits)
result = await self._apply_whole_file_edits(edits, use_output_dir)
elif format == "udiff":
result = await self._apply_udiff_edits(edits)
result = await self._apply_udiff_edits(edits, use_output_dir)
else: # default to diff (search/replace blocks)
result = await self._apply_diff_edits(edits)
result = await self._apply_diff_edits(edits, use_output_dir)

if result.success:
return f"Successfully edited files: {', '.join(result.edited_files)}\n{result.message}"
Expand All @@ -130,7 +148,7 @@ async def execute(self, format: str = "diff", edits: str = "",
except Exception as e:
return f"Error: {str(e)}"

async def _apply_whole_file_edits(self, edits: str) -> EditResult:
async def _apply_whole_file_edits(self, edits: str, use_output_dir: bool = True) -> EditResult:
"""Apply whole file edits"""
edited_files = []
errors = []
Expand All @@ -147,14 +165,22 @@ async def _apply_whole_file_edits(self, edits: str) -> EditResult:
for filename, content in file_blocks:
filename = filename.strip()
try:
# Determine the full file path
if use_output_dir:
# Use the configured output directory
full_path = OUTPUT_ROOT / filename
else:
# Use the exact path specified
full_path = Path(filename)

# Create directory if it doesn't exist
os.makedirs(os.path.dirname(os.path.abspath(filename)), exist_ok=True)
os.makedirs(os.path.dirname(str(full_path)), exist_ok=True)

# Write the file
with open(filename, 'w') as f:
with open(full_path, 'w') as f:
f.write(content)

edited_files.append(filename)
edited_files.append(str(full_path))
except Exception as e:
errors.append(f"Error writing {filename}: {str(e)}")

Expand All @@ -171,7 +197,7 @@ async def _apply_whole_file_edits(self, edits: str) -> EditResult:
edited_files=edited_files
)

async def _apply_diff_edits(self, edits: str) -> EditResult:
async def _apply_diff_edits(self, edits: str, use_output_dir: bool = True) -> EditResult:
"""Apply search/replace block edits"""
edited_files = []
errors = []
Expand All @@ -187,31 +213,43 @@ async def _apply_diff_edits(self, edits: str) -> EditResult:

for filename, search_text, replace_text in blocks:
try:
# Determine the full file path
if use_output_dir:
# Use the configured output directory
full_path = OUTPUT_ROOT / filename
else:
# Use the exact path specified
full_path = Path(filename)

# Check if file exists
if not os.path.exists(filename) and not search_text.strip():
if not os.path.exists(full_path) and not search_text.strip():
# Creating a new file
os.makedirs(os.path.dirname(os.path.abspath(filename)), exist_ok=True)
with open(filename, 'w') as f:
os.makedirs(os.path.dirname(str(full_path)), exist_ok=True)
with open(full_path, 'w') as f:
f.write(replace_text)
edited_files.append(filename)
edited_files.append(str(full_path))
continue

# For existing files, we need to check the original path
# since we're modifying existing files
file_to_read = filename if os.path.exists(filename) else full_path

# Read existing file
with open(filename, 'r') as f:
with open(file_to_read, 'r') as f:
content = f.read()

# Apply the edit
new_content = self._replace_text(content, search_text, replace_text)

if new_content == content:
errors.append(f"No changes made to {filename} - search text not found")
errors.append(f"No changes made to {full_path} - search text not found")
continue

# Write the updated content
with open(filename, 'w') as f:
with open(full_path, 'w') as f:
f.write(new_content)

edited_files.append(filename)
edited_files.append(str(full_path))
except Exception as e:
errors.append(f"Error editing {filename}: {str(e)}")

Expand All @@ -228,7 +266,7 @@ async def _apply_diff_edits(self, edits: str) -> EditResult:
edited_files=edited_files
)

async def _apply_udiff_edits(self, edits: str) -> EditResult:
async def _apply_udiff_edits(self, edits: str, use_output_dir: bool = True) -> EditResult:
"""Apply unified diff edits"""
edited_files = []
errors = []
Expand All @@ -251,27 +289,39 @@ async def _apply_udiff_edits(self, edits: str) -> EditResult:
errors.append("Could not determine filename from diff")
continue

# Determine the full file path
if use_output_dir:
# Use the configured output directory
full_path = OUTPUT_ROOT / filename
else:
# Use the exact path specified
full_path = Path(filename)

# Check if file exists
if not os.path.exists(filename) and filename != '/dev/null':
if not os.path.exists(full_path) and filename != '/dev/null':
# Creating a new file
os.makedirs(os.path.dirname(os.path.abspath(filename)), exist_ok=True)
with open(filename, 'w') as f:
os.makedirs(os.path.dirname(str(full_path)), exist_ok=True)
with open(full_path, 'w') as f:
f.write('\n'.join(line[1:] for line in changes if line.startswith('+')))
edited_files.append(filename)
edited_files.append(str(full_path))
continue

# For existing files, we need to check the original path
# since we're modifying existing files
file_to_read = filename if os.path.exists(filename) else full_path

# Read existing file
with open(filename, 'r') as f:
with open(file_to_read, 'r') as f:
content = f.read().splitlines()

# Apply the diff
new_content = self._apply_diff_changes(content, changes)

# Write the updated content
with open(filename, 'w') as f:
with open(full_path, 'w') as f:
f.write('\n'.join(new_content))

edited_files.append(filename)
edited_files.append(str(full_path))
except Exception as e:
errors.append(f"Error applying diff: {str(e)}")

Expand Down
28 changes: 23 additions & 5 deletions app/tool/file_saver.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,18 @@
import os
from pathlib import Path

import aiofiles

from app.config import OUTPUT_ROOT
from app.tool.base import BaseTool


class FileSaver(BaseTool):
name: str = "file_saver"
description: str = """Save content to a local file at a specified path.
Use this tool when you need to save text, code, or generated content to a file on the local filesystem.
The tool accepts content and a file path, and saves the content to that location.
By default, files are saved to the configured output directory (set in config.toml).
You can specify a relative path within the output directory, or set use_output_dir=False to save to an absolute path.
"""
parameters: dict = {
"type": "object",
Expand All @@ -27,33 +30,48 @@ class FileSaver(BaseTool):
"description": "(optional) The file opening mode. Default is 'w' for write. Use 'a' for append.",
"enum": ["w", "a"],
"default": "w"
},
"use_output_dir": {
"type": "boolean",
"description": "(optional) Whether to save the file in the configured output directory. Default is True.",
"default": True
}
},
"required": ["content", "file_path"]
}

async def execute(self, content: str, file_path: str, mode: str = "w") -> str:
async def execute(self, content: str, file_path: str, mode: str = "w", use_output_dir: bool = True) -> str:
"""
Save content to a file at the specified path.

Args:
content (str): The content to save to the file.
file_path (str): The path where the file should be saved.
mode (str, optional): The file opening mode. Default is 'w' for write. Use 'a' for append.
use_output_dir (bool, optional): Whether to save the file in the configured output directory.
Default is True. If False, the file will be saved at the exact path specified.

Returns:
str: A message indicating the result of the operation.
"""
try:
# Determine the full file path
if use_output_dir:
# Use the configured output directory
full_path = OUTPUT_ROOT / file_path
else:
# Use the exact path specified
full_path = Path(file_path)

# Ensure the directory exists
directory = os.path.dirname(file_path)
directory = os.path.dirname(str(full_path))
if directory and not os.path.exists(directory):
os.makedirs(directory)

# Write directly to the file
async with aiofiles.open(file_path, mode, encoding="utf-8") as file:
async with aiofiles.open(full_path, mode, encoding="utf-8") as file:
await file.write(content)

return f"Content successfully saved to {file_path}"
return f"Content successfully saved to {full_path}"
except Exception as e:
return f"Error saving file: {str(e)}"
5 changes: 5 additions & 0 deletions config/config.example.toml
Original file line number Diff line number Diff line change
Expand Up @@ -50,3 +50,8 @@ temperature = 0.0
# [search]
# Search engine for agent to use. Default is "Google" can be set to "Baidu" or "DuckDuckGo".
#engine = "Google"

# Output directory configuration
# [output]
# Directory for output files, relative to project root
#directory = "output"