mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-06 22:58:30 +00:00
Compare commits
2 Commits
devin/1765
...
devin/1753
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
115af9b495 | ||
|
|
f2fa1755ae |
@@ -1,7 +1,9 @@
|
||||
from .base_tool import BaseTool, tool, EnvVar
|
||||
from .agent_tools.edit_file_tool import EditFileTool
|
||||
|
||||
__all__ = [
|
||||
"BaseTool",
|
||||
"tool",
|
||||
"tool",
|
||||
"EnvVar",
|
||||
]
|
||||
"EditFileTool",
|
||||
]
|
||||
|
||||
@@ -1 +1,7 @@
|
||||
"""Agent tools for crewAI."""
|
||||
|
||||
from .edit_file_tool import EditFileTool
|
||||
|
||||
__all__ = [
|
||||
"EditFileTool"
|
||||
]
|
||||
|
||||
126
src/crewai/tools/agent_tools/edit_file_tool.py
Normal file
126
src/crewai/tools/agent_tools/edit_file_tool.py
Normal file
@@ -0,0 +1,126 @@
|
||||
from crewai.tools import BaseTool
|
||||
from crewai.llm import LLM
|
||||
from typing import Type, Optional
|
||||
from pydantic import BaseModel, Field
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class EditFileToolInput(BaseModel):
|
||||
"""Input schema for EditFileTool."""
|
||||
file_path: str = Field(..., description="Path to the file to edit")
|
||||
edit_instructions: str = Field(..., description="Clear instructions for what changes to make to the file")
|
||||
context: Optional[str] = Field(None, description="Additional context about the changes needed")
|
||||
|
||||
|
||||
class EditFileTool(BaseTool):
|
||||
name: str = "edit_file"
|
||||
description: str = (
|
||||
"Edit files using Fast Apply model approach. Performs full-file rewrites instead of "
|
||||
"brittle search-and-replace operations. Provide clear edit instructions and the tool "
|
||||
"will generate an accurate, complete rewrite of the file with your changes applied."
|
||||
)
|
||||
args_schema: Type[BaseModel] = EditFileToolInput
|
||||
|
||||
def __init__(self, llm: Optional[LLM] = None, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self.llm = llm or LLM(model="gpt-4o-mini", temperature=0.1)
|
||||
|
||||
def _run(self, file_path: str, edit_instructions: str, context: Optional[str] = None) -> str:
|
||||
"""
|
||||
Execute file editing using Fast Apply approach.
|
||||
|
||||
Args:
|
||||
file_path: Path to the file to edit
|
||||
edit_instructions: Instructions for what changes to make
|
||||
context: Optional additional context
|
||||
|
||||
Returns:
|
||||
Success message with details of the edit
|
||||
"""
|
||||
try:
|
||||
path = Path(file_path)
|
||||
if not path.exists():
|
||||
return f"Error: File {file_path} does not exist"
|
||||
|
||||
if not path.is_file():
|
||||
return f"Error: {file_path} is not a file"
|
||||
|
||||
try:
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
current_content = f.read()
|
||||
except UnicodeDecodeError:
|
||||
return f"Error: Cannot read {file_path} - file appears to be binary or uses unsupported encoding"
|
||||
|
||||
prompt = self._build_fast_apply_prompt(
|
||||
current_content=current_content,
|
||||
edit_instructions=edit_instructions,
|
||||
file_path=file_path,
|
||||
context=context
|
||||
)
|
||||
|
||||
response = self.llm.call(prompt)
|
||||
new_content = self._extract_file_content(response)
|
||||
|
||||
if new_content is None:
|
||||
return "Error: Failed to generate valid file content. LLM response was malformed."
|
||||
|
||||
backup_path = f"{file_path}.backup"
|
||||
with open(backup_path, 'w', encoding='utf-8') as f:
|
||||
f.write(current_content)
|
||||
|
||||
with open(path, 'w', encoding='utf-8') as f:
|
||||
f.write(new_content)
|
||||
|
||||
original_lines = len(current_content.splitlines())
|
||||
new_lines = len(new_content.splitlines())
|
||||
|
||||
return (
|
||||
f"Successfully edited {file_path}. "
|
||||
f"Original: {original_lines} lines, New: {new_lines} lines. "
|
||||
f"Backup saved as {backup_path}"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return f"Error editing file {file_path}: {str(e)}"
|
||||
|
||||
def _build_fast_apply_prompt(self, current_content: str, edit_instructions: str,
|
||||
file_path: str, context: Optional[str] = None) -> str:
|
||||
"""Build the Fast Apply prompt for the LLM."""
|
||||
file_extension = Path(file_path).suffix
|
||||
|
||||
prompt = f"""You are an expert code editor implementing Fast Apply file editing. Your task is to rewrite the entire file with the requested changes applied.
|
||||
|
||||
IMPORTANT INSTRUCTIONS:
|
||||
1. You must output the COMPLETE rewritten file content
|
||||
2. Apply the edit instructions precisely while preserving all other functionality
|
||||
3. Maintain the original file's style, formatting, and structure
|
||||
4. Do not add explanatory comments unless they were in the original
|
||||
5. Output ONLY the file content, no explanations or markdown formatting
|
||||
|
||||
FILE TO EDIT: {file_path}
|
||||
FILE TYPE: {file_extension}
|
||||
|
||||
CURRENT FILE CONTENT:
|
||||
```
|
||||
{current_content}
|
||||
```
|
||||
|
||||
EDIT INSTRUCTIONS:
|
||||
{edit_instructions}"""
|
||||
|
||||
if context:
|
||||
prompt += f"\n\nADDITIONAL CONTEXT:\n{context}"
|
||||
|
||||
prompt += "\n\nOUTPUT THE COMPLETE REWRITTEN FILE CONTENT:"
|
||||
|
||||
return prompt
|
||||
|
||||
def _extract_file_content(self, llm_response: str) -> Optional[str]:
|
||||
"""Extract the file content from LLM response."""
|
||||
content = llm_response.strip()
|
||||
|
||||
if content.startswith('```') and content.endswith('```'):
|
||||
lines = content.split('\n')
|
||||
content = '\n'.join(lines[1:-1])
|
||||
|
||||
return content if content else None
|
||||
181
tests/tools/test_edit_file_tool.py
Normal file
181
tests/tools/test_edit_file_tool.py
Normal file
@@ -0,0 +1,181 @@
|
||||
import pytest
|
||||
import tempfile
|
||||
import os
|
||||
from unittest.mock import Mock, patch
|
||||
from crewai.tools.agent_tools.edit_file_tool import EditFileTool, EditFileToolInput
|
||||
|
||||
|
||||
class TestEditFileTool:
|
||||
|
||||
def setup_method(self):
|
||||
"""Set up test fixtures."""
|
||||
self.tool = EditFileTool()
|
||||
|
||||
def test_tool_initialization(self):
|
||||
"""Test that the tool initializes correctly."""
|
||||
assert self.tool.name == "edit_file"
|
||||
assert "Fast Apply" in self.tool.description
|
||||
assert self.tool.args_schema == EditFileToolInput
|
||||
|
||||
def test_input_schema_validation(self):
|
||||
"""Test input schema validation."""
|
||||
valid_input = EditFileToolInput(
|
||||
file_path="/path/to/file.py",
|
||||
edit_instructions="Add a new function",
|
||||
context="This is for testing"
|
||||
)
|
||||
assert valid_input.file_path == "/path/to/file.py"
|
||||
assert valid_input.edit_instructions == "Add a new function"
|
||||
assert valid_input.context == "This is for testing"
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
EditFileToolInput(file_path="/path/to/file.py")
|
||||
|
||||
def test_file_not_exists(self):
|
||||
"""Test handling of non-existent files."""
|
||||
result = self.tool._run(
|
||||
file_path="/non/existent/file.py",
|
||||
edit_instructions="Add a function"
|
||||
)
|
||||
assert "Error: File /non/existent/file.py does not exist" in result
|
||||
|
||||
def test_path_is_directory(self):
|
||||
"""Test handling when path points to a directory."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
result = self.tool._run(
|
||||
file_path=temp_dir,
|
||||
edit_instructions="Add a function"
|
||||
)
|
||||
assert f"Error: {temp_dir} is not a file" in result
|
||||
|
||||
def test_binary_file_handling(self):
|
||||
"""Test handling of binary files."""
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix='.bin') as temp_file:
|
||||
temp_file.write(b'\x00\x01\x02\x03')
|
||||
temp_file.flush()
|
||||
|
||||
try:
|
||||
result = self.tool._run(
|
||||
file_path=temp_file.name,
|
||||
edit_instructions="Edit this file"
|
||||
)
|
||||
assert "Cannot read" in result and "unsupported encoding" in result
|
||||
finally:
|
||||
os.unlink(temp_file.name)
|
||||
|
||||
@patch('crewai.tools.agent_tools.edit_file_tool.LLM')
|
||||
def test_successful_file_edit(self, mock_llm_class):
|
||||
"""Test successful file editing."""
|
||||
mock_llm = Mock()
|
||||
mock_llm.call.return_value = "def new_function():\n return 'edited'"
|
||||
mock_llm_class.return_value = mock_llm
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.py') as temp_file:
|
||||
temp_file.write("def old_function():\n return 'original'")
|
||||
temp_file.flush()
|
||||
|
||||
try:
|
||||
tool = EditFileTool()
|
||||
result = tool._run(
|
||||
file_path=temp_file.name,
|
||||
edit_instructions="Replace old_function with new_function"
|
||||
)
|
||||
|
||||
assert "Successfully edited" in result
|
||||
assert "Backup saved" in result
|
||||
|
||||
with open(temp_file.name, 'r') as f:
|
||||
content = f.read()
|
||||
assert "new_function" in content
|
||||
assert "old_function" not in content
|
||||
|
||||
backup_path = f"{temp_file.name}.backup"
|
||||
assert os.path.exists(backup_path)
|
||||
with open(backup_path, 'r') as f:
|
||||
backup_content = f.read()
|
||||
assert "old_function" in backup_content
|
||||
|
||||
os.unlink(backup_path)
|
||||
|
||||
finally:
|
||||
os.unlink(temp_file.name)
|
||||
|
||||
def test_build_fast_apply_prompt(self):
|
||||
"""Test Fast Apply prompt building."""
|
||||
prompt = self.tool._build_fast_apply_prompt(
|
||||
current_content="print('hello')",
|
||||
edit_instructions="Change hello to world",
|
||||
file_path="/test/file.py",
|
||||
context="Testing context"
|
||||
)
|
||||
|
||||
assert "Fast Apply file editing" in prompt
|
||||
assert "print('hello')" in prompt
|
||||
assert "Change hello to world" in prompt
|
||||
assert "/test/file.py" in prompt
|
||||
assert "Testing context" in prompt
|
||||
assert "COMPLETE rewritten file content" in prompt
|
||||
|
||||
def test_extract_file_content_plain(self):
|
||||
"""Test extracting content from plain LLM response."""
|
||||
response = "def hello():\n print('world')"
|
||||
content = self.tool._extract_file_content(response)
|
||||
assert content == "def hello():\n print('world')"
|
||||
|
||||
def test_extract_file_content_markdown(self):
|
||||
"""Test extracting content from markdown code blocks."""
|
||||
response = "```python\ndef hello():\n print('world')\n```"
|
||||
content = self.tool._extract_file_content(response)
|
||||
assert content == "def hello():\n print('world')"
|
||||
|
||||
def test_extract_file_content_empty(self):
|
||||
"""Test handling empty LLM response."""
|
||||
content = self.tool._extract_file_content("")
|
||||
assert content is None
|
||||
|
||||
content = self.tool._extract_file_content("```\n```")
|
||||
assert content == ""
|
||||
|
||||
@patch('crewai.tools.agent_tools.edit_file_tool.LLM')
|
||||
def test_llm_failure_handling(self, mock_llm_class):
|
||||
"""Test handling of LLM failures."""
|
||||
mock_llm = Mock()
|
||||
mock_llm.call.side_effect = Exception("LLM error")
|
||||
mock_llm_class.return_value = mock_llm
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.py') as temp_file:
|
||||
temp_file.write("original content")
|
||||
temp_file.flush()
|
||||
|
||||
try:
|
||||
tool = EditFileTool()
|
||||
result = tool._run(
|
||||
file_path=temp_file.name,
|
||||
edit_instructions="Edit this file"
|
||||
)
|
||||
|
||||
assert "Error editing file" in result
|
||||
assert "LLM error" in result
|
||||
|
||||
finally:
|
||||
os.unlink(temp_file.name)
|
||||
|
||||
def test_context_parameter(self):
|
||||
"""Test that context parameter is properly handled."""
|
||||
prompt_with_context = self.tool._build_fast_apply_prompt(
|
||||
current_content="test",
|
||||
edit_instructions="edit",
|
||||
file_path="/test.py",
|
||||
context="important context"
|
||||
)
|
||||
|
||||
prompt_without_context = self.tool._build_fast_apply_prompt(
|
||||
current_content="test",
|
||||
edit_instructions="edit",
|
||||
file_path="/test.py",
|
||||
context=None
|
||||
)
|
||||
|
||||
assert "important context" in prompt_with_context
|
||||
assert "ADDITIONAL CONTEXT" in prompt_with_context
|
||||
assert "ADDITIONAL CONTEXT" not in prompt_without_context
|
||||
62
tests/tools/test_edit_file_tool_integration.py
Normal file
62
tests/tools/test_edit_file_tool_integration.py
Normal file
@@ -0,0 +1,62 @@
|
||||
import tempfile
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
from crewai import Agent
|
||||
from crewai.tools.agent_tools.edit_file_tool import EditFileTool
|
||||
|
||||
|
||||
class TestEditFileToolIntegration:
|
||||
|
||||
def test_agent_with_edit_file_tool(self):
|
||||
"""Test that agents can use the EditFileTool."""
|
||||
tool = EditFileTool()
|
||||
|
||||
agent = Agent(
|
||||
role="Code Editor",
|
||||
goal="Edit files as requested",
|
||||
backstory="I am an expert at editing code files",
|
||||
tools=[tool],
|
||||
verbose=True
|
||||
)
|
||||
|
||||
assert tool in agent.tools
|
||||
assert agent.tools[0].name == "edit_file"
|
||||
|
||||
def test_tool_with_different_file_types(self):
|
||||
"""Test the tool works with different file types."""
|
||||
tool = EditFileTool()
|
||||
|
||||
test_files = [
|
||||
("test.py", "print('hello')", "Change hello to world"),
|
||||
("test.js", "console.log('hello');", "Change hello to world"),
|
||||
("test.txt", "Hello world", "Change Hello to Hi"),
|
||||
("test.md", "# Title\nContent", "Change Title to New Title")
|
||||
]
|
||||
|
||||
for filename, content, instruction in test_files:
|
||||
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix=filename) as temp_file:
|
||||
temp_file.write(content)
|
||||
temp_file.flush()
|
||||
|
||||
try:
|
||||
with patch.object(tool.llm, 'call') as mock_call:
|
||||
expected_content = content.replace('hello', 'world').replace('Hello', 'Hi').replace('Title', 'New Title')
|
||||
mock_call.return_value = expected_content
|
||||
|
||||
result = tool._run(
|
||||
file_path=temp_file.name,
|
||||
edit_instructions=instruction
|
||||
)
|
||||
|
||||
assert "Successfully edited" in result
|
||||
|
||||
with open(temp_file.name, 'r') as f:
|
||||
edited_content = f.read()
|
||||
assert edited_content == expected_content
|
||||
|
||||
backup_path = f"{temp_file.name}.backup"
|
||||
if os.path.exists(backup_path):
|
||||
os.unlink(backup_path)
|
||||
|
||||
finally:
|
||||
os.unlink(temp_file.name)
|
||||
Reference in New Issue
Block a user