CakeCN/hsa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
CakeCN
2024-12-03 16:21:19 +08:00
commit 4198ca63b1
975 changed files with 333413 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

291
AlgoriAgent/examples/conversation_with_swe-agent/main.ipynb Normal file
View File

@@ -0,0 +1,291 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Conversation with Software Engineering Agent\n",
"\n",
"SWE-agent(SoftWare Engineering Agent) is an agent designed for solving real world software engineering problems, such as fixing github issues.\n",
"More details can be found in the project's [homepage](https://swe-agent.com/) and related [github repo](https://swe-agent.com/).\n",
"\n",
"In the example here, we partially implement the SWE-agent, and provide a simple example of how to use the implemented SWE-agent to fix a bug in a python file.\n",
"You should note that currently how to enable agents with stronger programming capabilities remains an open challenge, and the performance of the paritially implemented SWE-agent is not guaranteed.\n",
"\n",
"## Prerequisites\n",
"\n",
"- Follow [READMD.md](https://github.com/modelscope/agentscope) to install AgentScope. We require the lastest version, so you should build from source by running `pip install -e .` instead of intalling from pypi. \n",
"- Prepare a model configuration. AgentScope supports both local deployed model services (CPU or GPU) and third-party services. More details and example model configurations please refer to our [tutorial](https://modelscope.github.io/agentscope/en/tutorial/203-model.html).\n",
"- Understand the ServiceToolkit module and how to use it to pre-process the tool functions for LLMs. You can refer to the [ReAct agent example](../conversation_with_react_agent/main.ipynb) and you should also refer to the [tutorial](https://modelscope.github.io/agentscope/en/tutorial/204-service.html) for service functions.\n",
"\n",
"\n",
"## Note\n",
"\n",
"- The example is tested with the following models. For other models, you may need to adjust the prompt.\n",
" - gpt-4\n",
" - dashscope_chat (qwen-max)\n",
" - gemini_chat (gemini-pro)\n",
" - ollama_chat (llama3_8b)\n",
"- How to enable agents with stronger programming capabilities remains an open challenge, and the current implementations are not perfect. Please feel free to explore it yourself."
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [],
"source": [
"YOUR_MODEL_CONFIGURATION_NAME = \"{YOUR_MODEL_CONFIGURATION_NAME}\"\n",
"\n",
"YOUR_MODEL_CONFIGURATION = {\n",
" \"model_type\": \"xxx\", \n",
" \"config_name\": YOUR_MODEL_CONFIGURATION_NAME\n",
" \n",
" # ...\n",
"}"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 1: Initialize the AgentScope environment and SWE-agent\n",
"\n",
"Here we init the agentscope environment and load the SWE-agent.\n",
"\n",
"The code of SWE-agent is in `swe_agent.py`, and the related prompts are in `swe_agent_prompts.py`.\n",
"\n",
"If you are interested in the details, please refer to the code and the origianl SWE-agent repo [here](https://github.com/princeton-nlp/SWE-agent)."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from swe_agent import SWEAgent\n",
"\n",
"import agentscope\n",
"\n",
"agentscope.init(\n",
" model_configs=YOUR_MODEL_CONFIGURATION,\n",
" project=\"Conversation with SWE-agent\",\n",
")\n",
"\n",
"agent = SWEAgent(\n",
" name=\"assistant\",\n",
" model_config_name=YOUR_MODEL_CONFIGURATION_NAME,\n",
")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 2: Create the code to be processed by the SWE-agent\n",
"\n",
"Here we use the `write_file` function to write the following code into `gcd.py`.\n",
"The code here is a wrong implementation of the [Greatest Common Divisor (GCD) algorithm](https://en.wikipedia.org/wiki/Euclidean_algorithm).\n",
"We will ask the SWE-agent to correct it in our next step."
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"{'status': <ServiceExecStatus.SUCCESS: 1>,\n",
" 'content': 'WRITE OPERATION:\\nYou have written to \"gcd.py\" on these lines: 0:-1.\\ndef gcd(a, b):\\n if a == 0:\\n return b\\n while a != 0:\\n a, b = b, a\\n return b\\n\\ndef lcm(a, b):\\n return (a * b) // gcd(a, b)\\n\\n# testing on GCD and LCM functions\\nprint(\"GCD of 12 and 18 is:\", gcd(12, 18))\\nprint(\"LCM of 12 and 18 is:\", lcm(12, 18))\\n\\n'}"
]
},
"execution_count": 3,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"from swe_agent_service_func import write_file\n",
"\n",
"# here the gcd code is written in a wrong way\n",
"wrong_gcd_text = \"\"\"\n",
"def gcd(a, b):\n",
" if a == 0:\n",
" return b\n",
" while a != 0:\n",
" a, b = b, a\n",
" return b\n",
"\n",
"def lcm(a, b):\n",
" return (a * b) // gcd(a, b)\n",
"\n",
"# testing on GCD and LCM functions\n",
"print(\"GCD of 12 and 18 is:\", gcd(12, 18))\n",
"print(\"LCM of 12 and 18 is:\", lcm(12, 18))\n",
"\"\"\"\n",
"\n",
"write_file(\"gcd.py\", wrong_gcd_text)\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Step 3: Ask the SWE-agent to correct the code\n",
"\n",
"Here we simple ask the SWE-agent to correct the code in `gcd.py`."
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"user: The file in gcd.py is not returning the correct answer, can you fix the file for me?\n",
"assistant: {\n",
" \"thought\": \"First thing I'll do is read the content of gcd.py file to understand the existing logic.\",\n",
" \"action\": {\n",
" \"name\": \"read_file\",\n",
" \"arguments\": {\n",
" \"file_path\": \"gcd.py\"\n",
" }\n",
" }\n",
"}\n",
"assistant: \n",
"====Observation====\n",
"\n",
"def gcd(a, b):\n",
" if a == 0:\n",
" return b\n",
" while a != 0:\n",
" a, b = b, a\n",
" return b\n",
"\n",
"def lcm(a, b):\n",
" return (a * b) // gcd(a, b)\n",
"\n",
"# testing on GCD and LCM functions\n",
"print(\"GCD of 12 and 18 is:\", gcd(12, 18))\n",
"print(\"LCM of 12 and 18 is:\", lcm(12, 18))\n",
"\n",
"\n",
"assistant: {\n",
" \"thought\": \"I can see the error in the gcd calculation. The gcd calculation in the file is incorrect. The correct way to find the gcd of two numbers using the Euclidean algorithm is to continually reduce a and b by finding the remainder of a % b until b is 0, at which point a will be the gcd of the original a and b values. In the given code, instead of finding the remainder, it is incorrectly swapping the values of a and b. I'll correct this error by modifying the gcd function using the write_file command, reflecting the correct Euclidean algorithm.\",\n",
" \"action\": {\n",
" \"name\": \"write_file\",\n",
" \"arguments\": {\n",
" \"start_line\": 0,\n",
" \"end_line\": 15,\n",
" \"content\": \"def gcd(a, b):\\n while b != 0:\\n a, b = b, a % b\\n return a\\n\\n\\ndef lcm(a, b):\\n return (a * b) // gcd(a, b)\\n\\n\\n# testing on GCD and LCM functions\\nprint(\\\"GCD of 12 and 18 is:\\\", gcd(12, 18))\\nprint(\\\"LCM of 12 and 18 is:\\\", lcm(12, 18))\\n\",\n",
" \"file_path\": \"gcd.py\"\n",
" }\n",
" }\n",
"}\n",
"assistant: \n",
"====Observation====\n",
"WRITE OPERATION:\n",
"You have written to \"gcd.py\" on these lines: 0:15.def gcd(a, b):\n",
" while b != 0:\n",
" a, b = b, a % b\n",
" return a\n",
"\n",
"\n",
"def lcm(a, b):\n",
" return (a * b) // gcd(a, b)\n",
"\n",
"\n",
"# testing on GCD and LCM functions\n",
"print(\"GCD of 12 and 18 is:\", gcd(12, 18))\n",
"print(\"LCM of 12 and 18 is:\", lcm(12, 18))\n",
"\n",
"\n",
"assistant: {\n",
" \"thought\": \"I should run the python file to see if the changes I made have fixed the gcd function.\",\n",
" \"action\": {\n",
" \"name\": \"execute_shell_command\",\n",
" \"arguments\": {\n",
" \"command\": \"python3 gcd.py\"\n",
" }\n",
" }\n",
"}\n",
"assistant: \n",
"====Observation====\n",
"GCD of 12 and 18 is: 6\n",
"LCM of 12 and 18 is: 36\n",
"assistant: {\n",
" \"thought\": \"The gcd.py file has been fixed and tested successfully. Therefore, no further actions are needed.\",\n",
" \"action\": {\n",
" \"name\": \"exit\",\n",
" \"arguments\": {}\n",
" }\n",
"}\n",
"assistant: \n",
"====Observation====\n",
"Current task finished, exitting.\n"
]
}
],
"source": [
"from loguru import logger\n",
"from agentscope.message import Msg\n",
"\n",
"mss = Msg(\n",
" name=\"user\", \n",
" content=\"The file in gcd.py is not returning the correct answer, can you fix the file for me?\", \n",
" role=\"user\"\n",
")\n",
"logger.chat(mss)\n",
"answer_mss = agent(mss)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Conlusion\n",
"\n",
"After a few iterations, the SWE-agent assistant finish the job successfully, and the code is now working fine."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Above we shown a example of how to use the SWE-agent to fix code errors.\n",
"Although the design of the SWE-agent is primarily aimed at addressing GitHub issues, with modifications, it can also be utilized for more general programming tasks.\n",
"\n",
"Currently, how to enable agent with general programming ablities remains a challenging open question, with the efficacy of agent programming potentially influenced by factors such as prompt construction, model capabilities, and the complexity of the task at hand. Here we just provide an interesting toy example. \n",
"\n",
"We encourage users to experiment by altering the prompts within this example or by assigning different tasks to the agent, among other methods of exploration. Please feel free to experiment and explore on your own. The AgentScope team will continue to provide updates, enhancing the capabilities of the Programming Agents in the future!"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "datajuicer",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.18"
}
},
"nbformat": 4,
"nbformat_minor": 2
}

274
AlgoriAgent/examples/conversation_with_swe-agent/swe_agent.py Normal file
View File

@@ -0,0 +1,274 @@
# -*- coding: utf-8 -*-
"""An agent class that partially implements the SWE-agent.
SWE-agent is an agent designed for solving github issues.
More details can be found in https://swe-agent.com/.
Here we partially implement and modified the SWE-agent,
try to make it work with wider range of tasks then just fixing github issues.
"""
from agentscope.agents import AgentBase
from agentscope.message import Msg
from agentscope.exception import ResponseParsingError
from agentscope.parsers import MarkdownJsonDictParser
from typing import List, Callable, Optional, Union, Sequence
import json
from agentscope.service import (
ServiceFactory,
execute_shell_command,
)
from swe_agent_service_func import (
exec_py_linting,
write_file,
read_file,
)
from swe_agent_prompts import (
get_system_prompt,
get_context_prompt,
get_step_prompt,
)
def prepare_func_prompt(function: Callable) -> str:
func, desc = ServiceFactory.get(function)
func_name = desc["function"]["name"]
func_desc = desc["function"]["description"]
args_desc = desc["function"]["parameters"]["properties"]
args_list = [f"{func_name}: {func_desc}"]
for args_name, args_info in args_desc.items():
if "type" in args_info:
args_line = (
f'\t{args_name} ({args_info["type"]}): '
f'{args_info.get("description", "")}'
)
else:
args_line = f'\t{args_name}: {args_info.get("description", "")}'
args_list.append(args_line)
func_prompt = "\n".join(args_list)
return func_prompt
COMMANDS_DISCRIPTION_DICT = {
"exit": "exit: Executed when the current task is complete, takes no arguments", # noqa
"scroll_up": "scroll_up: Scrolls up the current open file, will scroll up and show you the 100 lines above your current lines, takes no arguments", # noqa
"scroll_down": "scroll_down: Scrolls down the current open file, will scroll down and show you the 100 lines below your current lines'takes no arguments", # noqa
"goto": "goto: This will take you directly to the line <line_num> and show you the 100 lines below it. \n line_num (int): The line number to go to.", # noqa
}
COMMANDS_DISCRIPTION_DICT["write_file"] = prepare_func_prompt(write_file)
COMMANDS_DISCRIPTION_DICT["read_file"] = prepare_func_prompt(read_file)
COMMANDS_DISCRIPTION_DICT["execute_shell_command"] = prepare_func_prompt(
execute_shell_command,
)
COMMANDS_DISCRIPTION_DICT["exec_py_linting"] = prepare_func_prompt(
exec_py_linting,
)
ERROR_INFO_PROMPT = """Your response is not a JSON object, and cannot be parsed by `json.loads` in parse function:
## Your Response:
[YOUR RESPONSE BEGIN]
{response}
[YOUR RESPONSE END]
## Error Information:
{error_info}
Analyze the reason, and re-correct your response in the correct format.""" # pylint: disable=all # noqa
def count_file_lines(file_path: str) -> int:
with open(file_path, "r") as file:
lines = file.readlines()
return len(lines)
class SWEAgent(AgentBase):
"""
The SWE-agent
"""
def __init__(
self,
name: str,
model_config_name: str,
) -> None:
""" """
super().__init__(
name=name,
model_config_name=model_config_name,
)
self.memory_window = 6
self.max_retries = 2
self.running_memory: List[str] = []
self.cur_file: str = ""
self.cur_line: int = 0
self.cur_file_content: str = ""
self.main_goal = ""
self.commands_prompt = ""
self.parser = MarkdownJsonDictParser()
self.get_commands_prompt()
def get_current_file_content(self) -> None:
"""
Get the current file content.
"""
if self.cur_file == "":
return
start_line = self.cur_line - 50
if start_line < 0:
start_line = 0
end_line = self.cur_line + 50
if end_line > count_file_lines(self.cur_file):
end_line = -1
read_res = read_file(self.cur_file, start_line, end_line)
self.cur_file_content = read_res.content
def step(self) -> Msg:
"""
Step the SWE-agent.
"""
message_list = []
# construct system prompt
system_prompt = get_system_prompt(self.commands_prompt)
message_list.append(Msg("user", system_prompt, role="system"))
# construct context prompt, i.e. previous actions
context_prompt = get_context_prompt(
self.running_memory,
self.memory_window,
)
message_list.append(Msg("user", context_prompt, role="user"))
# construct step prompt for this instance
self.get_current_file_content()
step_prompt = get_step_prompt(
self.main_goal,
self.cur_file,
self.cur_line,
self.cur_file_content,
)
message_list.append(Msg("user", step_prompt, role="user"))
# get response from agent
try:
in_prompt = self.model.format(message_list)
res = self.model(
in_prompt,
parse_func=self.parser.parse,
max_retries=1,
)
except ResponseParsingError as e:
response_msg = Msg(self.name, e.raw_response, "assistant")
self.speak(response_msg)
# Re-correct by model itself
error_msg = Msg(
name="system",
content={
"action": {"name": "error"},
"error_msg": ERROR_INFO_PROMPT.format(
parse_func=self.parser.parse,
error_info=e.message,
response=e.raw_response,
),
},
role="system",
)
self.speak(error_msg)
# continue
self.running_memory.append(error_msg)
return error_msg
msg_res = Msg(self.name, res.parsed, role="assistant")
self.speak(
Msg(self.name, json.dumps(res.parsed, indent=4), role="assistant"),
)
# parse and execute action
action = res.parsed.get("action")
obs = self.prase_command(res.parsed["action"])
self.speak(
Msg(self.name, "\n====Observation====\n" + obs, role="assistant"),
)
# add msg to context windows
self.running_memory.append(str(action) + str(obs))
return msg_res
def reply(self, x: Optional[Union[Msg, Sequence[Msg]]] = None) -> Msg:
action_name = None
self.main_goal = x.content
while not action_name == "exit":
msg = self.step()
action_name = msg.content["action"]["name"]
return msg
def prase_command(self, command_call: dict) -> str:
command_name = command_call["name"]
command_args = command_call["arguments"]
if command_name == "exit":
return "Current task finished, exitting."
if command_name in ["goto", "scroll_up", "scroll_down"]:
if command_name == "goto":
line = command_call["arguments"]["line_num"]
command_str = f"Going to {self.cur_file} line \
{command_args['line_mum']}."
command_failed_str = f"Failed to go to {self.cur_file} \
line {command_args['line_num']}"
if command_name == "scroll_up":
line = self.cur_line - 100
if line < 0:
line = 0
command_str = (
f"Scrolling up from file {self.cur_file} to line {line}."
)
command_failed_str = (
f"Failed to scroll up {self.cur_file} to line {line}"
)
if command_name == "scroll_down":
line = self.cur_line + 100
if line > count_file_lines(self.cur_file):
line = count_file_lines(self.cur_file)
command_str = (
f"Scrolling down from file {self.cur_file} to line {line}."
)
command_failed_str = (
f"Failed to scrool down {self.cur_file} to line {line}"
)
read_status = read_file(self.cur_file, line, line + 100)
if read_status.status == "success":
self.cur_line = line
obs = read_status.content
return f"{command_str}. Observe file content: {obs}"
else:
return command_failed_str
if command_name == "execute_shell_command":
return execute_shell_command(**command_args).content
if command_name == "write_file":
self.cur_file = command_args["file_path"]
self.cur_line = command_args.get("start_line", 0)
write_status = write_file(**command_args)
return write_status.content
if command_name == "read_file":
self.cur_file = command_args["file_path"]
self.cur_line = command_args.get("start_line", 0)
read_status = read_file(**command_args)
return read_status.content
if command_name == "exec_py_linting":
return exec_py_linting(**command_args).content
return "No such command"
def get_commands_prompt(self) -> None:
for name, desc in COMMANDS_DISCRIPTION_DICT.items():
self.commands_prompt += f"{name}: {desc}\n"

123
AlgoriAgent/examples/conversation_with_swe-agent/swe_agent_prompts.py Normal file
View File

@@ -0,0 +1,123 @@
# -*- coding: utf-8 -*-
# pylint: disable=C0301
"""The SWE-agent relay heavily on it's prompts.
This file contains the neccessary prompts for the SWE-agent.
Some prompts are taken and modified from the original SWE-agent repo
or the SWE-agent implementation from Open-Devin.
"""
WINDOW = 100
def get_system_prompt(command_prompt: str) -> str:
"""
Get the system prompt for SWE-agent.
"""
return f"""
SETTING:
You are an autonomous coding agent, here to perform codding tasks given the instruction.
You have been designed with a wide range of programming tasks, from code editing and debugging to testing and deployment.
You have access to a variety of tools and commands that you can use to help you solve problems efficiently.
You're working directly in the command line with a special interface.
The special interface consists of a file editor that shows you {WINDOW} lines of a file at a time.
In addition to typical bash commands, you can also use the following commands to help you navigate and edit files.
COMMANDS:
{command_prompt}
Please note that THE WRITE COMMAND REQUIRES PROPER INDENTATION.
If you'd like to add the line ' print(x)' you must fully write that out, with all those spaces before the code!
Indentation is important and code that is not indented correctly will fail and require fixing before it can be run.
If you'd like to issue two commands at once, PLEASE DO NOT DO THAT! Please instead first submit just the first command, and then after receiving a response you'll be able to issue the second command.
You're free to use any other bash commands you want (e.g. find, grep, cat, ls) in addition to the special commands listed above.
However, the environment does NOT support interactive session commands (e.g. vim, python), so please do not invoke them.
{RESPONSE_FORMAT_PROMPT}
""" # noqa
RESPONSE_FORMAT_PROMPT = """
## Response Format:
You should respond with a JSON object in the following format.
```json
{
"thought": "what you thought",
"action": {"name": "{command name}", "arguments": {"{argument1 name}": xxx, "{argument2 name}": xxx}}
}
```
For Example:
```json
{
"thought": "First I'll start by using ls to see what files are in the current directory. Then maybe we can look at some relevant files to see what they look like.",
"action": {"name": "execute_shell_command", "arguments": {"command": "ls -a"}}
}
```
OUTPUT the JSON format and ONLY OUTPUT the JSON format.
Your Response should always be a valid JSON string that can be parsed.
""" # noqa
def get_step_prompt(
task: str,
file: str,
line: int,
current_file_content: str,
) -> str:
"""
Get the step prompt for SWE-agent.
"""
return f"""
We're currently perform the following coding task. Here's the original task description from the user.
{task}
CURRENT
Open File: {file} on line {line}
Current File Content:
{current_file_content}
You can use these commands with the current file:
Navigation: `scroll_up`, `scroll_down`, and `goto <line>`
INSTRUCTIONS:
1. If you run a command and it doesn't work, try running a different command. A command that did not work once will not work the second time unless you modify it!
2. If you open a file and need to get to an area around a specific line that is not in the first 100 lines, say line 583, don't just use the scroll_down command multiple times. Instead, use the goto 583 command. It's much quicker.
3. Always make sure to look at the currently open file and the current working directory (which appears right after the currently open file). The currently open file might be in a different directory! Note that some commands, such as 'write_file' and 'read_file', open files, so they might change the current open file.
4. When editing files, it is easy to accidentally specify a wrong line number or to write code with incorrect indentation. Always check the code after you issue an edit to make sure that it reflects what you wanted to accomplish. If it didn't, issue another command to fix it.
5. After modifying python files, you can run `exec_py_linting` to check for errors. If there are errors, fix them and repeat the previous step.
NOTE THAT THIS ENVIRONMENT DOES NOT SUPPORT INTERACTIVE SESSION COMMANDS, such as "vim" or "python", or "python3". So DONOT execute them by running `execute_shell_command` with `python` command or `python3` command if the code need additional inputs.
If you want to check whether a python file is valid, you can use `exec_py_linting` to check for errors.
You should always notice your response format and respond with a JSON object in the following format.
{RESPONSE_FORMAT_PROMPT}
""" # noqa
def get_context_prompt(memory: list, window: int) -> str:
"""
Get the context prompt for the given memory and window.
"""
res = f"These are your past {window} actions:\n"
window_size = window if len(memory) > window else len(memory)
cur_mems = memory[-window_size:]
res += "===== Previous Actions =====\n"
for idx, mem in enumerate(cur_mems):
res += f"\nMemory {idx}:\n{mem}\n"
res += "======= End Actions =======\n"
res += "Use these memories to provide additional context to \
the problem you are solving.\nRemember that you have already \
completed these steps so you do not need to perform them again."
return res

147
AlgoriAgent/examples/conversation_with_swe-agent/swe_agent_service_func.py Normal file
View File

@@ -0,0 +1,147 @@
# -*- coding: utf-8 -*-
# pylint: disable=C0301
"""
Tools for swe-agent, such as checking files with linting and formatting,
writing and reading files by lines, etc.
"""
import subprocess
import os
from agentscope.service.service_response import ServiceResponse
from agentscope.service.service_status import ServiceExecStatus
def exec_py_linting(file_path: str) -> ServiceResponse:
"""
Executes flake8 linting on the given .py file with specified checks and
returns the linting result.
Args:
file_path (`str`): The path to the Python file to lint.
Returns:
ServiceResponse: Contains either the output from the flake8 command as
a string if successful, or an error message including the error type.
"""
command = f"flake8 --isolated --select=F821,F822,F831,\
E111,E112,E113,E999,E902 {file_path}"
try:
result = subprocess.run(
command,
shell=True,
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
return ServiceResponse(
status=ServiceExecStatus.SUCCESS,
content=result.stdout.strip()
if result.stdout
else "No lint errors found.",
)
except subprocess.CalledProcessError as e:
error_message = (
e.stderr.strip()
if e.stderr
else "An error occurred while linting the file."
)
return ServiceResponse(
status=ServiceExecStatus.ERROR,
content=error_message,
)
except Exception as e:
return ServiceResponse(
status=ServiceExecStatus.ERROR,
content=str(e),
)
def write_file(
file_path: str,
content: str,
start_line: int = 0,
end_line: int = -1,
) -> ServiceResponse:
"""
Write content to a file by replacing the current lines between <start_line> and <end_line> with <content>. Default start_line = 0 and end_line = -1. Calling this with no <start_line> <end_line> args will replace the whole file, so besure to use this with caution when writing to a file that already exists.
Args:
file_path (`str`): The path to the file to write to.
content (`str`): The content to write to the file.
start_line (`Optional[int]`, defaults to `0`): The start line of the file to be replace with <content>.
end_line (`Optional[int]`, defaults to `-1`): The end line of the file to be replace with <content>. end_line = -1 means the end of the file, otherwise it should be a positive integer indicating the line number.
""" # noqa
try:
mode = "w" if not os.path.exists(file_path) else "r+"
insert = content.split("\n")
with open(file_path, mode, encoding="utf-8") as file:
if mode != "w":
all_lines = file.readlines()
new_file = [""] if start_line == 0 else all_lines[:start_line]
new_file += [i + "\n" for i in insert]
last_line = end_line + 1
new_file += [""] if end_line == -1 else all_lines[last_line:]
else:
new_file = insert
file.seek(0)
file.writelines(new_file)
file.truncate()
obs = f'WRITE OPERATION:\nYou have written to "{file_path}" \
on these lines: {start_line}:{end_line}.'
return ServiceResponse(
status=ServiceExecStatus.SUCCESS,
content=obs + "".join(new_file),
)
except Exception as e:
error_message = f"{e.__class__.__name__}: {e}"
return ServiceResponse(
status=ServiceExecStatus.ERROR,
content=error_message,
)
def read_file(
file_path: str,
start_line: int = 0,
end_line: int = -1,
) -> ServiceResponse:
"""
Shows a given file's contents starting from <start_line> up to <end_line>. Default: start_line = 0, end_line = -1. By default the whole file will be read.
Args:
file_path (`str`): The path to the file to read.
start_line (`Optional[int]`, defaults to `0`): The start line of the file to be read.
end_line (`Optional[int]`, defaults to `-1`): The end line of the file to be read.
""" # noqa
start_line = max(start_line, 0)
try:
with open(file_path, "r", encoding="utf-8") as file:
if end_line == -1:
if start_line == 0:
code_view = file.read()
else:
all_lines = file.readlines()
code_slice = all_lines[start_line:]
code_view = "".join(code_slice)
else:
all_lines = file.readlines()
num_lines = len(all_lines)
begin = max(0, min(start_line, num_lines - 2))
end_line = (
-1 if end_line > num_lines else max(begin + 1, end_line)
)
code_slice = all_lines[begin:end_line]
code_view = "".join(code_slice)
return ServiceResponse(
status=ServiceExecStatus.SUCCESS,
content=f"{code_view}",
)
except Exception as e:
error_message = f"{e.__class__.__name__}: {e}"
return ServiceResponse(
status=ServiceExecStatus.ERROR,
content=error_message,
)