New Agent, Action, Observation Abstraction and with updated Controller

agenthub/codeact_agent/__init__.py

@@ -2,12 +2,21 @@
 import re
 from litellm import completion
 from termcolor import colored
-from typing import List, Dict
+from typing import List
+
+from opendevin.agent import Agent
+from opendevin.state import State
+from opendevin.action import (
+    Action,
+    CmdRunAction,
+    AgentEchoAction,
+    AgentFinishAction,
+)
+from opendevin.observation import (
+    CmdOutputObservation,
+    AgentMessageObservation,
+)
 
-from opendevin.agent import Agent, Message, Role
-from opendevin.lib.event import Event
-from opendevin.lib.command_manager import CommandManager
-from opendevin.sandbox.sandbox import DockerInteractive
 
 assert (
     "OPENAI_API_KEY" in os.environ
@@ -54,9 +63,7 @@ class CodeActAgent(Agent):
     def __init__(
         self,
         instruction: str,
-        workspace_dir: str,
-        model_name: str,
-        max_steps: int = 100
+        model_name: str
     ) -> None:
         """
         Initializes a new instance of the CodeActAgent class.
@@ -65,69 +72,67 @@ def __init__(
         - instruction (str): The instruction for the agent to execute.
         - max_steps (int): The maximum number of steps to run the agent.
         """
-        super().__init__(instruction, workspace_dir, model_name, max_steps)
-        self._history = [Message(Role.SYSTEM, SYSTEM_MESSAGE)]
-        self._history.append(Message(Role.USER, instruction))
-        self.env = DockerInteractive(workspace_dir=workspace_dir)
+        super().__init__(instruction, model_name)
+        self.messages = [
+            {"role": "system", "content": SYSTEM_MESSAGE},
+            {"role": "user", "content": instruction},
+        ]
         print(colored("===USER:===\n" + instruction, "green"))
 
-    def _history_to_messages(self) -> List[Dict]:
-        return [message.to_dict() for message in self._history]
-
-    def run(self) -> None:
-        """
-        Starts the execution of the assigned instruction. This method should
-        be implemented by subclasses to define the specific execution logic.
-        """
-        for _ in range(self.max_steps):
-            response = completion(
-                messages=self._history_to_messages(),
-                model=self.model_name,
-                stop=["</execute>"],
-                temperature=0.0,
-                seed=42,
-            )
-            action = parse_response(response)
-            self._history.append(Message(Role.ASSISTANT, action))
-            print(colored("===ASSISTANT:===\n" + action, "yellow"))
-
-            command = re.search(r"<execute>(.*)</execute>", action, re.DOTALL)
-            if command is not None:
-                # a command was found
-                command_group = command.group(1)
-                if command_group.strip() == "exit":
-                    print(colored("Exit received. Exiting...", "red"))
-                    break
-                # execute the code
-                # TODO: does exit_code get loaded into Message?
-                exit_code, observation = self.env.execute(command_group)
-                self._history.append(Message(Role.ASSISTANT, observation))
-                print(colored("===ENV OBSERVATION:===\n" + observation, "blue"))
-            else:
-                # we could provide a error message for the model to continue similar to
-                # https://github.com/xingyaoww/mint-bench/blob/main/mint/envs/general_env.py#L18-L23
-                observation = INVALID_INPUT_MESSAGE
-                self._history.append(Message(Role.ASSISTANT, observation))
-                print(colored("===ENV OBSERVATION:===\n" + observation, "blue"))
-
-        self.env.close()
-
-    def chat(self, message: str) -> None:
-        """
-        Optional method for interactive communication with the agent during its execution. Implementations
-        can use this method to modify the agent's behavior or state based on chat inputs.
-
-        Parameters:
-        - message (str): The chat message or command.
-        """
-        raise NotImplementedError
-
-    # TODO: implement these abstract methods
-    def add_event(self, event: Event) -> None:
-        raise NotImplementedError("Implement this abstract method")
+    def step(self, state: State) -> Action:
+        updated_info = state.updated_info
+
+        if updated_info:
+
+            for item in updated_info:
+                if isinstance(item, Action):
+                    assert isinstance(item, (CmdRunAction, AgentEchoAction)), "Expecting CmdRunAction or AgentEchoAction for Action"
+
+                elif isinstance(item, AgentMessageObservation):  # warning message from itself
+                    self.messages.append({"role": "user", "content": item.content})
+                    print(colored("===USER:===\n" + item.content, "green"))
+
+                elif isinstance(item, CmdOutputObservation):
+                    content = "OBSERVATION:\n" + item.content
+                    content += f"\n[Command {item.command_id} finished with exit code {item.exit_code}]]"
+                    self.messages.append({"role": "user", "content": content})
+                    print(colored("===ENV OBSERVATION:===\n" + content, "blue"))
+
+                else:
+                    raise NotImplementedError(f"Unknown observation type: {item}")
+
+        response = completion(
+            messages=self.messages,
+            model=self.model_name,
+            stop=["</execute>"],
+            temperature=0.0,
+            seed=42,
+        )
+        action = parse_response(response)
+        self.messages.append({"role": "assistant", "content": action})
+        print(colored("===ASSISTANT:===\n" + action, "yellow"))
+
+        command = re.search(r"<execute>(.*)</execute>", action, re.DOTALL)
+        if command is not None:
+            # a command was found
+            command_group = command.group(1)
+            if command_group.strip() == "exit":
+                print(colored("Exit received. Exiting...", "red"))
+                return AgentFinishAction()
+            return CmdRunAction(command = command_group)
+            # # execute the code
+            # # TODO: does exit_code get loaded into Message?
+            # exit_code, observation = self.env.execute(command_group)
+            # self._history.append(Message(Role.ASSISTANT, observation))
+            # print(colored("===ENV OBSERVATION:===\n" + observation, "blue"))
+        else:
+            # we could provide a error message for the model to continue similar to
+            # https://github.com/xingyaoww/mint-bench/blob/main/mint/envs/general_env.py#L18-L23
+            # observation = INVALID_INPUT_MESSAGE
+            # self._history.append(Message(Role.ASSISTANT, observation))
+            # print(colored("===ENV OBSERVATION:===\n" + observation, "blue"))
+            return AgentEchoAction(content=INVALID_INPUT_MESSAGE)  # warning message to itself
 
-    def step(self, cmd_mgr: CommandManager) -> Event:
-        raise NotImplementedError("Implement this abstract method")
 
     def search_memory(self, query: str) -> List[str]:
         raise NotImplementedError("Implement this abstract method")

agenthub/langchains_agent/__init__.py

@@ -1,9 +1,28 @@
-from typing import List
+from typing import List, Dict, Type
 
+import agenthub.langchains_agent.utils.llm as llm
 from opendevin.agent import Agent
+from opendevin.action import (
+    Action,
+    CmdRunAction,
+    CmdKillAction,
+    BrowseURLAction,
+    FileReadAction,
+    FileWriteAction,
+    AgentRecallAction,
+    AgentThinkAction,
+    AgentFinishAction,
+)
+from opendevin.observation import (
+    Observation,
+    CmdOutputObservation,
+    BrowserOutputObservation,
+)
+from opendevin.state import State
+
+from agenthub.langchains_agent.utils.monologue import Monologue
+from agenthub.langchains_agent.utils.memory import LongTermMemory
 
-from agenthub.langchains_agent.utils.agent import Agent as LangchainsAgentImpl
-from opendevin.lib.event import Event
 
 INITIAL_THOUGHTS = [
     "I exist!",
@@ -44,55 +63,131 @@
 ]
 
 
+MAX_OUTPUT_LENGTH = 5000
+MAX_MONOLOGUE_LENGTH = 20000
+
+
+ACTION_TYPE_TO_CLASS: Dict[str, Type[Action]] = {
+    "run": CmdRunAction,
+    "kill": CmdKillAction,
+    "browse": BrowseURLAction,
+    "read": FileReadAction,
+    "write": FileWriteAction,
+    "recall": AgentRecallAction,
+    "think": AgentThinkAction,
+    "finish": AgentFinishAction,
+}
+
+CLASS_TO_ACTION_TYPE: Dict[Type[Action], str] = {v: k for k, v in ACTION_TYPE_TO_CLASS.items()}
+
 class LangchainsAgent(Agent):
     _initialized = False
 
+    def __init__(self, instruction: str, model_name: str):
+        super().__init__(instruction, model_name)
+        self.monologue = Monologue(self.model_name)
+        self.memory = LongTermMemory()
+
+    def _add_event(self, event: dict):
+        if 'output' in event['args']:
+            event['args']['output'] = event['args']['output'][:MAX_OUTPUT_LENGTH] + "..."
+
+        self.monologue.add_event(event)
+        self.memory.add_event(event)
+        if self.monologue.get_total_length() > MAX_MONOLOGUE_LENGTH:
+            self.monologue.condense()
+
     def _initialize(self):
         if self._initialized:
             return
-        self.agent = LangchainsAgentImpl(self.instruction, self.model_name)
         next_is_output = False
         for thought in INITIAL_THOUGHTS:
             thought = thought.replace("$TASK", self.instruction)
             if next_is_output:
-                event = Event("output", {"output": thought})
+                d = {"action": "output", "args": {"output": thought}}
                 next_is_output = False
             else:
                 if thought.startswith("RUN"):
                     command = thought.split("RUN ")[1]
-                    event = Event("run", {"command": command})
+                    d = {"action": "run", "args": {"command": command}}
                     next_is_output = True
+
                 elif thought.startswith("RECALL"):
                     query = thought.split("RECALL ")[1]
-                    event = Event("recall", {"query": query})
+                    d = {"action": "recall", "args": {"query": query}}
                     next_is_output = True
+
                 elif thought.startswith("BROWSE"):
                     url = thought.split("BROWSE ")[1]
-                    event = Event("browse", {"url": url})
+                    d = {"action": "browse", "args": {"url": url}}
                     next_is_output = True
                 else:
-                    event = Event("think", {"thought": thought})
-            self.agent.add_event(event)
-        self._initialized = True
+                    d = {"action": "think", "args": {"thought": thought}}
 
-    def add_event(self, event: Event) -> None:
-        self.agent.add_event(event)
+        self._add_event(d)
+        self._initialized = True
 
-    def step(self, cmd_mgr) -> Event:
+    def step(self, state: State) -> Action:
         self._initialize()
-        return self.agent.get_next_action(cmd_mgr)
+        # TODO: make langchains agent use Action & Observation
+        # completly from ground up
 
-    def search_memory(self, query: str) -> List[str]:
-        return self.agent.memory.search(query)
+        # Translate state to action_dict
+        for info in state.updated_info:
+            if isinstance(info, Observation):
+                if isinstance(info, CmdOutputObservation):
+                    if info.error:
+                        d = {"action": "error", "args": {"output": info.content}}
+                    else:
+                        d = {"action": "output", "args": {"output": info.content}}
+                # elif isinstance(info, UserMessageObservation):
+                #     d = {"action": "output", "args": {"output": info.message}}
+                # elif isinstance(info, AgentMessageObservation):
+                #     d = {"action": "output", "args": {"output": info.message}}
+                elif isinstance(info, BrowserOutputObservation):
+                    d = {"action": "output", "args": {"output": info.content}}
+                else:
+                    raise NotImplementedError(f"Unknown observation type: {info}")
+                self._add_event(d)
+            elif isinstance(info, Action):
+                if isinstance(info, CmdRunAction):
+                    d = {"action": "run", "args": {"command": info.command}}
+                elif isinstance(info, CmdKillAction):
+                    d = {"action": "kill", "args": {"id": info.id}}
+                elif isinstance(info, BrowseURLAction):
+                    d = {"action": "browse", "args": {"url": info.url}}
+                elif isinstance(info, FileReadAction):
+                    d = {"action": "read", "args": {"file": info.path}}
+                elif isinstance(info, FileWriteAction):
+                    d = {"action": "write", "args": {"file": info.path, "content": info.contents}}
+                elif isinstance(info, AgentRecallAction):
+                    d = {"action": "recall", "args": {"query": info.query}}
+                elif isinstance(info, AgentThinkAction):
+                    d = {"action": "think", "args": {"thought": info.thought}}
+                elif isinstance(info, AgentFinishAction):
+                    d = {"action": "finish"}
+                else:
+                    raise NotImplementedError(f"Unknown action type: {info}")
+                self._add_event(d)
+
+        state.updated_info = []
+
+        action_dict = llm.request_action(
+            self.instruction,
+            self.monologue.get_thoughts(),
+            self.model_name,
+            state.background_commands_obs,
+        )
+        if action_dict is None:
+            action_dict = {"action": "think", "args": {"thought": "..."}}
 
-    def chat(self, message: str) -> None:
-        """
-        Optional method for interactive communication with the agent during its execution. Implementations
-        can use this method to modify the agent's behavior or state based on chat inputs.
+        # Translate action_dict to Action
+        action = ACTION_TYPE_TO_CLASS[action_dict["action"]](**action_dict["args"])
+        self.latest_action = action
+        return action
+
+    def search_memory(self, query: str) -> List[str]:
+        return self.memory.search(query)
 
-        Parameters:
-        - message (str): The chat message or command.
-        """
-        raise NotImplementedError
 
 Agent.register("LangchainsAgent", LangchainsAgent)