github/AutoGPT
1
1
Fork 0
mirror of https://github.com/Significant-Gravitas/AutoGPT.git synced 2026-04-08 03:00:28 -04:00
Code Issues Packages Projects Releases Wiki Activity

Refactor/move functions in app to agent (#4957)

Browse Source 
* Add links to github issues in the README and clarify run instructions

* Move things only used by the agent out of app.py and into the agent module

* Fix busted dynamic import
This commit is contained in:
James Collins
2023-07-12 19:36:00 -07:00
committed by GitHub
parent 077e143cc2
commit 4177c37b51
5 changed files with 111 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
autogpt/agent/agent.py
View File

@@ -9,6 +9,7 @@ from colorama import Fore, Style
from autogpt.config import Config
from autogpt.config.ai_config import AIConfig
from autogpt.json_utils.utilities import extract_json_from_response, validate_json
from autogpt.llm import ChatModelResponse
from autogpt.llm.chat import chat_with_ai
from autogpt.llm.providers.openai import OPEN_AI_CHAT_MODELS
from autogpt.llm.utils import count_string_tokens
@@ -86,9 +87,6 @@ class Agent:
self.smart_token_limit = OPEN_AI_CHAT_MODELS.get(config.smart_llm).max_tokens
def start_interaction_loop(self):
# Avoid circular imports
from autogpt.app import execute_command, extract_command
# Interaction Loop
self.cycle_count = 0
command_name = None
@@ -307,3 +305,94 @@ class Agent:
logger.typewriter_log(
"SYSTEM: ", Fore.YELLOW, "Unable to execute command"
)
def extract_command(
assistant_reply_json: dict, assistant_reply: ChatModelResponse, config: Config
):
"""Parse the response and return the command name and arguments
Args:
assistant_reply_json (dict): The response object from the AI
assistant_reply (ChatModelResponse): The model response from the AI
config (Config): The config object
Returns:
tuple: The command name and arguments
Raises:
json.decoder.JSONDecodeError: If the response is not valid JSON
Exception: If any other error occurs
"""
if config.openai_functions:
if assistant_reply.function_call is None:
return "Error:", "No 'function_call' in assistant reply"
assistant_reply_json["command"] = {
"name": assistant_reply.function_call.name,
"args": json.loads(assistant_reply.function_call.arguments),
}
try:
if "command" not in assistant_reply_json:
return "Error:", "Missing 'command' object in JSON"
if not isinstance(assistant_reply_json, dict):
return (
"Error:",
f"The previous message sent was not a dictionary {assistant_reply_json}",
)
command = assistant_reply_json["command"]
if not isinstance(command, dict):
return "Error:", "'command' object is not a dictionary"
if "name" not in command:
return "Error:", "Missing 'name' field in 'command' object"
command_name = command["name"]
# Use an empty dictionary if 'args' field is not present in 'command' object
arguments = command.get("args", {})
return command_name, arguments
except json.decoder.JSONDecodeError:
return "Error:", "Invalid JSON"
# All other errors, return "Error: + error message"
except Exception as e:
return "Error:", str(e)
def execute_command(
command_name: str,
arguments: dict[str, str],
agent: Agent,
):
"""Execute the command and return the result
Args:
command_name (str): The name of the command to execute
arguments (dict): The arguments for the command
agent (Agent): The agent that is executing the command
Returns:
str: The result of the command
"""
try:
# Execute a native command with the same name or alias, if it exists
if command := agent.command_registry.get_command(command_name):
return command(**arguments, agent=agent)
# Handle non-native commands (e.g. from plugins)
for command in agent.ai_config.prompt_generator.commands:
if (
command_name == command["label"].lower()
or command_name == command["name"].lower()
):
return command["function"](**arguments)
raise RuntimeError(
f"Cannot execute '{command_name}': unknown command."
" Do not try to use this command again."
)
except Exception as e:
return f"Error: {str(e)}"

114
autogpt/app.py
View File

@@ -1,114 +0,0 @@
""" Command and Control """
import json
from typing import Dict
from autogpt.agent.agent import Agent
from autogpt.config import Config
from autogpt.llm import ChatModelResponse
def is_valid_int(value: str) -> bool:
"""Check if the value is a valid integer
Args:
value (str): The value to check
Returns:
bool: True if the value is a valid integer, False otherwise
"""
try:
int(value)
return True
except ValueError:
return False
def extract_command(
assistant_reply_json: Dict, assistant_reply: ChatModelResponse, config: Config
):
"""Parse the response and return the command name and arguments
Args:
assistant_reply_json (dict): The response object from the AI
assistant_reply (ChatModelResponse): The model response from the AI
config (Config): The config object
Returns:
tuple: The command name and arguments
Raises:
json.decoder.JSONDecodeError: If the response is not valid JSON
Exception: If any other error occurs
"""
if config.openai_functions:
if assistant_reply.function_call is None:
return "Error:", "No 'function_call' in assistant reply"
assistant_reply_json["command"] = {
"name": assistant_reply.function_call.name,
"args": json.loads(assistant_reply.function_call.arguments),
}
try:
if "command" not in assistant_reply_json:
return "Error:", "Missing 'command' object in JSON"
if not isinstance(assistant_reply_json, dict):
return (
"Error:",
f"The previous message sent was not a dictionary {assistant_reply_json}",
)
command = assistant_reply_json["command"]
if not isinstance(command, dict):
return "Error:", "'command' object is not a dictionary"
if "name" not in command:
return "Error:", "Missing 'name' field in 'command' object"
command_name = command["name"]
# Use an empty dictionary if 'args' field is not present in 'command' object
arguments = command.get("args", {})
return command_name, arguments
except json.decoder.JSONDecodeError:
return "Error:", "Invalid JSON"
# All other errors, return "Error: + error message"
except Exception as e:
return "Error:", str(e)
def execute_command(
command_name: str,
arguments: dict[str, str],
agent: Agent,
):
"""Execute the command and return the result
Args:
command_name (str): The name of the command to execute
arguments (dict): The arguments for the command
agent (Agent): The agent that is executing the command
Returns:
str: The result of the command
"""
try:
# Execute a native command with the same name or alias, if it exists
if command := agent.command_registry.get_command(command_name):
return command(**arguments, agent=agent)
# Handle non-native commands (e.g. from plugins)
for command in agent.ai_config.prompt_generator.commands:
if (
command_name == command["label"].lower()
or command_name == command["name"].lower()
):
return command["function"](**arguments)
raise RuntimeError(
f"Cannot execute '{command_name}': unknown command."
" Do not try to use this command again."
)
except Exception as e:
return f"Error: {str(e)}"

1
autogpt/main.py
View File

@@ -28,7 +28,6 @@ COMMAND_CATEGORIES = [
"autogpt.commands.file_operations",
"autogpt.commands.web_search",
"autogpt.commands.web_selenium",
"autogpt.app",
"autogpt.commands.task_statuses",
]

57
tests/unit/test_agent.py
View File

@@ -1,46 +1,27 @@
from unittest.mock import MagicMock
import pytest
from autogpt.agent import Agent
from autogpt.config import AIConfig
from autogpt.config.config import Config
@pytest.fixture
def agent(config: Config):
ai_name = "Test AI"
memory = MagicMock()
next_action_count = 0
command_registry = MagicMock()
ai_config = AIConfig(ai_name=ai_name)
system_prompt = "System prompt"
triggering_prompt = "Triggering prompt"
workspace_directory = "workspace_directory"
agent = Agent(
ai_name=ai_name,
memory=memory,
next_action_count=next_action_count,
command_registry=command_registry,
ai_config=ai_config,
config=config,
system_prompt=system_prompt,
triggering_prompt=triggering_prompt,
workspace_directory=workspace_directory,
)
return agent
from autogpt.agent.agent import Agent, execute_command
def test_agent_initialization(agent: Agent):
assert agent.ai_name == "Test AI"
assert agent.memory == agent.memory
assert agent.ai_name == "Base"
assert agent.history.messages == []
assert agent.next_action_count == 0
assert agent.command_registry == agent.command_registry
assert agent.ai_config == agent.ai_config
assert agent.system_prompt == "System prompt"
assert agent.triggering_prompt == "Triggering prompt"
def test_execute_command_plugin(agent: Agent):
"""Test that executing a command that came from a plugin works as expected"""
command_name = "check_plan"
agent.ai_config.prompt_generator.add_command(
command_name,
"Read the plan.md with the next goals to achieve",
{},
lambda: "hi",
)
command_result = execute_command(
command_name=command_name,
arguments={},
agent=agent,
)
assert command_result == "hi"
# More test methods can be added for specific agent interactions

23
tests/unit/test_execute_command.py
View File

@@ -1,23 +0,0 @@
from autogpt.agent import Agent
from autogpt.app import execute_command
def check_plan():
return "hi"
def test_execute_command_plugin(agent: Agent):
"""Test that executing a command that came from a plugin works as expected"""
command_name = "check_plan"
agent.ai_config.prompt_generator.add_command(
command_name,
"Read the plan.md with the next goals to achieve",
{},
check_plan,
)
command_result = execute_command(
command_name=command_name,
arguments={},
agent=agent,
)
assert command_result == "hi"