mirror of
https://github.com/aljazceru/Auto-GPT.git
synced 2026-01-10 09:44:26 +01:00
Prompt improvements in command definitions and context template
This commit is contained in:
@@ -1,12 +1,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from autogpt.llm.base import ChatSequence
|
||||
from autogpt.models.context_item import ContextItem
|
||||
|
||||
from ..base import BaseAgent
|
||||
|
||||
from autogpt.llm.base import Message
|
||||
from autogpt.models.context_item import ContextItem
|
||||
|
||||
|
||||
class AgentContext:
|
||||
@@ -18,7 +20,7 @@ class AgentContext:
|
||||
def __bool__(self) -> bool:
|
||||
return len(self.items) > 0
|
||||
|
||||
def __contains__(self, item: ContextItem):
|
||||
def __contains__(self, item: ContextItem) -> bool:
|
||||
return any([i.source == item.source for i in self.items])
|
||||
|
||||
def add(self, item: ContextItem) -> None:
|
||||
@@ -39,18 +41,27 @@ class ContextMixin:
|
||||
|
||||
context: AgentContext
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
def __init__(self, **kwargs: Any):
|
||||
self.context = AgentContext()
|
||||
|
||||
super(ContextMixin, self).__init__(**kwargs)
|
||||
|
||||
def construct_base_prompt(self, *args, **kwargs):
|
||||
def construct_base_prompt(self, *args: Any, **kwargs: Any) -> ChatSequence:
|
||||
if kwargs.get("append_messages") is None:
|
||||
kwargs["append_messages"] = []
|
||||
|
||||
# Add context section to prompt
|
||||
if self.context:
|
||||
kwargs["append_messages"].insert(
|
||||
0, Message("system", "# Context\n" + self.context.format_numbered())
|
||||
0,
|
||||
Message(
|
||||
"system",
|
||||
"## Context\n"
|
||||
+ self.context.format_numbered()
|
||||
+ "\n\nWhen a context item is no longer needed and you are not done yet,"
|
||||
" you can hide the item by specifying its number in the list above"
|
||||
" to `hide_context_item`.",
|
||||
),
|
||||
)
|
||||
|
||||
return super(ContextMixin, self).construct_base_prompt(*args, **kwargs) # type: ignore
|
||||
|
||||
@@ -72,7 +72,7 @@ def open_file(file_path: Path, agent: Agent) -> tuple[str, FileContextItem]:
|
||||
raise DuplicateOperationError(f"The file {file_path} is already open")
|
||||
|
||||
return (
|
||||
f"File {file_path}{' created,' if created else ''} opened and added to context ✅",
|
||||
f"File {file_path}{' created,' if created else ''} has been opened and added to the context ✅",
|
||||
file,
|
||||
)
|
||||
|
||||
@@ -118,4 +118,4 @@ def open_folder(path: Path, agent: Agent) -> tuple[str, FolderContextItem]:
|
||||
if folder in agent_context:
|
||||
raise DuplicateOperationError(f"The folder {path} is already open")
|
||||
|
||||
return f"Folder {path} opened and added to context ✅", folder
|
||||
return f"Folder {path} has been opened and added to the context ✅", folder
|
||||
|
||||
@@ -190,41 +190,41 @@ def ingest_file(
|
||||
|
||||
|
||||
@command(
|
||||
"write_to_file",
|
||||
"Writes to a file",
|
||||
"write_file",
|
||||
"Write a file, creating it if necessary. If the file exists, it is overwritten.",
|
||||
{
|
||||
"filename": {
|
||||
"type": "string",
|
||||
"description": "The name of the file to write to",
|
||||
"required": True,
|
||||
},
|
||||
"text": {
|
||||
"contents": {
|
||||
"type": "string",
|
||||
"description": "The text to write to the file",
|
||||
"description": "The contents to write to the file",
|
||||
"required": True,
|
||||
},
|
||||
},
|
||||
aliases=["write_file", "create_file"],
|
||||
)
|
||||
@sanitize_path_arg("filename")
|
||||
def write_to_file(filename: Path, text: str, agent: Agent) -> str:
|
||||
"""Write text to a file
|
||||
def write_to_file(filename: Path, contents: str, agent: Agent) -> str:
|
||||
"""Write contents to a file
|
||||
|
||||
Args:
|
||||
filename (Path): The name of the file to write to
|
||||
text (str): The text to write to the file
|
||||
contents (str): The contents to write to the file
|
||||
|
||||
Returns:
|
||||
str: A message indicating success or failure
|
||||
"""
|
||||
checksum = text_checksum(text)
|
||||
checksum = text_checksum(contents)
|
||||
if is_duplicate_operation("write", filename, agent, checksum):
|
||||
raise DuplicateOperationError("File has already been updated.")
|
||||
raise DuplicateOperationError(f"File {filename.name} has already been updated.")
|
||||
|
||||
directory = os.path.dirname(filename)
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
with open(filename, "w", encoding="utf-8") as f:
|
||||
f.write(text)
|
||||
f.write(contents)
|
||||
log_operation("write", filename, agent, checksum)
|
||||
return f"File {filename.name} has been written successfully."
|
||||
|
||||
@@ -232,16 +232,13 @@ def write_to_file(filename: Path, text: str, agent: Agent) -> str:
|
||||
@sanitize_path_arg("filename")
|
||||
def append_to_file(
|
||||
filename: Path, text: str, agent: Agent, should_log: bool = True
|
||||
) -> str:
|
||||
) -> None:
|
||||
"""Append text to a file
|
||||
|
||||
Args:
|
||||
filename (Path): The name of the file to append to
|
||||
text (str): The text to append to the file
|
||||
should_log (bool): Should log output
|
||||
|
||||
Returns:
|
||||
str: A message indicating success or failure
|
||||
"""
|
||||
directory = os.path.dirname(filename)
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
@@ -253,33 +250,31 @@ def append_to_file(
|
||||
checksum = text_checksum(f.read())
|
||||
log_operation("append", filename, agent, checksum=checksum)
|
||||
|
||||
return "Text appended successfully."
|
||||
|
||||
|
||||
@command(
|
||||
"list_files",
|
||||
"Lists Files in a Directory",
|
||||
"list_folder",
|
||||
"List the items in a folder",
|
||||
{
|
||||
"directory": {
|
||||
"folder": {
|
||||
"type": "string",
|
||||
"description": "The directory to list files in",
|
||||
"description": "The folder to list files in",
|
||||
"required": True,
|
||||
}
|
||||
},
|
||||
)
|
||||
@sanitize_path_arg("directory")
|
||||
def list_files(directory: Path, agent: Agent) -> list[str]:
|
||||
"""lists files in a directory recursively
|
||||
@sanitize_path_arg("folder")
|
||||
def list_folder(folder: Path, agent: Agent) -> list[str]:
|
||||
"""Lists files in a folder recursively
|
||||
|
||||
Args:
|
||||
directory (Path): The directory to search in
|
||||
folder (Path): The folder to search in
|
||||
|
||||
Returns:
|
||||
list[str]: A list of files found in the directory
|
||||
list[str]: A list of files found in the folder
|
||||
"""
|
||||
found_files = []
|
||||
|
||||
for root, _, files in os.walk(directory):
|
||||
for root, _, files in os.walk(folder):
|
||||
for file in files:
|
||||
if file.startswith("."):
|
||||
continue
|
||||
|
||||
@@ -19,8 +19,10 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@command(
|
||||
"goals_accomplished",
|
||||
"Goals are accomplished and there is nothing left to do",
|
||||
"finish",
|
||||
"Use this to shut down once you have accomplished all of your goals,"
|
||||
" or when there are insurmountable problems that make it impossible"
|
||||
" for you to finish your task.",
|
||||
{
|
||||
"reason": {
|
||||
"type": "string",
|
||||
@@ -29,7 +31,7 @@ logger = logging.getLogger(__name__)
|
||||
}
|
||||
},
|
||||
)
|
||||
def task_complete(reason: str, agent: Agent) -> None:
|
||||
def finish(reason: str, agent: Agent) -> None:
|
||||
"""
|
||||
A function that takes in a string and exits the program
|
||||
|
||||
@@ -44,22 +46,22 @@ def task_complete(reason: str, agent: Agent) -> None:
|
||||
|
||||
|
||||
@command(
|
||||
"close_context_item",
|
||||
"Close an open file, folder or other context item",
|
||||
"hide_context_item",
|
||||
"Hide an open file, folder or other context item, to save memory.",
|
||||
{
|
||||
"index": {
|
||||
"number": {
|
||||
"type": "integer",
|
||||
"description": "The 1-based index of the context item to close",
|
||||
"description": "The 1-based index of the context item to hide",
|
||||
"required": True,
|
||||
}
|
||||
},
|
||||
available=lambda a: bool(get_agent_context(a)),
|
||||
)
|
||||
def close_context_item(index: int, agent: Agent) -> str:
|
||||
def close_context_item(number: int, agent: Agent) -> str:
|
||||
assert (context := get_agent_context(agent)) is not None
|
||||
|
||||
if index > len(context.items) or index == 0:
|
||||
raise InvalidArgumentError(f"Index {index} out of range")
|
||||
if number > len(context.items) or number == 0:
|
||||
raise InvalidArgumentError(f"Index {number} out of range")
|
||||
|
||||
context.close(index)
|
||||
return f"Context item {index} closed ✅"
|
||||
context.close(number)
|
||||
return f"Context item {number} hidden ✅"
|
||||
|
||||
@@ -57,7 +57,9 @@ class BrowsingError(CommandExecutionError):
|
||||
|
||||
@command(
|
||||
"read_webpage",
|
||||
"Read a webpage, and extract specific information from it if a question is specified.",
|
||||
"Read a webpage, and extract specific information from it if a question is specified."
|
||||
" If you are looking to extract specific information from the webpage, you should"
|
||||
" specify a question.",
|
||||
{
|
||||
"url": {"type": "string", "description": "The URL to visit", "required": True},
|
||||
"question": {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
DEFAULT_TRIGGERING_PROMPT = (
|
||||
"Determine exactly one command to use based on the given goals "
|
||||
"Determine exactly one command to use next based on the given goals "
|
||||
"and the progress you have made so far, "
|
||||
"and respond using the JSON schema specified previously:"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user