Sde team remove langchain

agentverse/environments/__init__.py

@@ -6,3 +6,5 @@
 from .base import BaseEnvironment
 from .basic import BasicEnvironment
 from .pokemon import PokemonEnvironment
+from .sde_team import SdeTeamEnvironment
+from .sde_team_given_tests import SdeTeamGivenTestsEnvironment

agentverse/environments/rules/order/__init__.py

@@ -7,3 +7,5 @@
 from .concurrent import ConcurrentOrder
 from .classroom import ClassroomOrder
 from .prisoner import PrisonerOrder
+from .sde_team import SdeTeamOrder
+from .sde_team_given_tests import SdeTeamGivenTestsOrder

agentverse/environments/rules/order/sde_team.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import logging
+import re
+import random
+from typing import TYPE_CHECKING, Any, List, Optional
+
+from . import order_registry as OrderRegistry
+from .base import BaseOrder
+
+if TYPE_CHECKING:
+    from agentverse.environments import BaseEnvironment
+
+
+@OrderRegistry.register("sde_team")
+class SdeTeamOrder(BaseOrder):
+    """The order for a code problem solving
+    """
+    next_agent_idx: int = 2
+
+    def get_next_agent_idx(self, environment: BaseEnvironment) -> List[int]:
+        if self.next_agent_idx == 2:
+            self.next_agent_idx = 0
+            return [2] * 5 # TODO set the number in yaml
+        elif self.next_agent_idx == 0:
+            self.next_agent_idx = 1
+            return [0]
+        elif self.next_agent_idx == 1:
+            self.next_agent_idx = 0
+            return [1]

agentverse/environments/rules/order/sde_team_given_tests.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import logging
+import re
+import random
+from typing import TYPE_CHECKING, Any, List, Optional
+
+from . import order_registry as OrderRegistry
+from .base import BaseOrder
+
+if TYPE_CHECKING:
+    from agentverse.environments import BaseEnvironment
+
+
+@OrderRegistry.register("sde_team_given_tests")
+class SdeTeamGivenTestsOrder(BaseOrder):
+    """The order for a code problem solving given unit tests
+    """
+    next_agent_idx: int = 0
+
+    def get_next_agent_idx(self, environment: BaseEnvironment) -> List[int]:
+        if self.next_agent_idx == 0:
+            self.next_agent_idx = 1
+            return [0]
+        elif self.next_agent_idx == 1:
+            self.next_agent_idx = 0
+            return [1]

agentverse/environments/rules/selector/__init__.py

@@ -5,3 +5,5 @@
 from .base import BaseSelector
 from .basic import BasicSelector
 from .classroom import ClassroomSelector
+from .sde_team import SdeTeamSelector
+from .sde_team_given_tests import SdeTeamGivenTestsSelector

agentverse/environments/rules/selector/code_api.py

@@ -0,0 +1,97 @@
+import io
+import sys
+import ast
+import json
+import astunparse
+import concurrent.futures
+import traceback
+
+
+def get_call_str(assert_statement: str) -> str:
+    call_str = ast.parse(assert_statement).body[0].test.left # type: ignore
+    return astunparse.unparse(call_str).strip()
+
+def get_output(func: str, assert_statement: str) -> str:
+    try:
+        func_call = get_call_str(assert_statement)
+        try:
+            exec(func, globals())
+            output = eval(func_call)
+            return output
+        except Exception as e:
+            return str(e)
+    except:
+        return "get_call_str error"
+
+def worker(code, globals=None, locals=None):
+    old_stdout = sys.stdout
+    redirected_output = sys.stdout = io.StringIO()
+    if locals is None:
+        locals = {}
+    try:
+        # TODO: exec(code, globals, locals) could be buggy
+        # In cases where both import statement and function exits in the code, if the locals are given, 
+        # the code will not find the imported package. 
+        # For example,
+        # code = "import math\ndef f(x):\n\treturn math.pow(x, 2)\nassert f(2) == 4"
+        # It will return "NameError: name 'math' is not defined"
+        exec(code, locals, locals)
+        stdout = redirected_output.getvalue()
+        return stdout, globals, locals
+    except Exception as e:
+        trace_str = traceback.format_exc()
+        return f"Error: {trace_str}", globals, locals
+    finally:
+        sys.stdout = old_stdout  # restore the original stdout
+
+def execute_code(code: str) -> str:
+    """Execute a snippet of python code and return the output or the error message.
+    """
+    timeout = 5
+    try:
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            future = executor.submit(worker, code)
+            result, _, _ = future.result(timeout)
+            return result
+    except concurrent.futures.TimeoutError:
+        return "Timeout"
+
+def execute_unit_tests(func_impl: str, tests: str) -> str:
+    """Run a python function on a bunch of unit tests tests and return detailed feedback.
+    """
+    # tests = eval(tests)
+    # assert type(tests) == list
+
+    # Combine function code and assert statement
+    func_test_list = [f'{func_impl}\n{test}' for test in tests]
+
+    # Run the tests and collect the results
+    success_tests = []
+    failed_tests = []
+    is_passing = True
+    num_tests = len(func_test_list)
+    for i in range(num_tests):
+        output = execute_code(func_test_list[i])
+        if output == "Timeout":
+            failed_tests += [f"{tests[i]} # output: Timeout"]
+            is_passing = False
+        elif output.startswith("Error: "):
+            # print(output)
+            func_output = get_output(func_impl, tests[i])
+            if func_output == "get_call_str error":
+                func_output = output
+            failed_tests += [f"{tests[i]} # output: {func_output}"]
+            is_passing = False
+        else:
+            success_tests += [tests[i]]
+
+    feedback = "Tested passed:\n\n"
+    for test in success_tests:
+        feedback += f"{test}\n\n"
+    feedback += "Tests failed:\n\n"
+    for test in failed_tests:
+        feedback += f"{test}\n\n"
+
+    return json.dumps({"is_passing": is_passing, 
+            "feedback": feedback})
+

agentverse/environments/rules/selector/sde_team.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, List
+
+from agentverse.message import Message
+
+from . import selector_registry as SelectorRegistry
+from .base import BaseSelector
+
+import json
+import re
+
+if TYPE_CHECKING:
+    from agentverse.environments import BaseEnvironment
+
+def extract(content: str, keyword: str):
+    result = ""
+    flag = False
+    for line in content.split('\n'):
+        if line.strip().startswith(keyword):
+            flag = True
+            continue
+        if flag:
+            result += line
+            result += "\n"
+    return result
+
+
+@SelectorRegistry.register("sde_team")
+class SdeTeamSelector(BaseSelector):
+    def select_message(self, environment: BaseEnvironment, messages: List[Message]) -> List[Message]:
+        last_sender = environment.last_messages[0].sender
+        selected = messages
+
+        if last_sender == "unit_test_generator":
+            unit_tests = set()
+            for message in selected:
+                unit_test = extract(message.content, "<unit test>:")
+                if unit_test not in unit_tests:
+                    unit_tests.add(extract(message.content, "<unit test>:"))
+            unit_tests = list(unit_tests)
+            environment.rule_params["unit_tests"] = str(unit_tests)
+            new_message = Message(
+                            content="",
+                            sender="unit_test_generator",
+                            receiver=[],
+                        )   # TODO: set the content of the message
+            selected = [new_message]
+
+        elif last_sender == "code_writer":
+            cur_code = extract(selected[0].content, "<code>:")
+            environment.rule_params["code"] = cur_code
+
+            from .code_api import execute_unit_tests
+            feedback = execute_unit_tests(environment.rule_params["code"], eval(environment.rule_params["unit_tests"]))
+
+            environment.rule_params["feedback"] = feedback
+            selected[0].content = f"<current code>:\n\n{cur_code}\n\n<unit test feedback>:\n{feedback}"
+            f_dict = json.loads(feedback)
+            if f_dict["is_passing"]:
+                environment.rule_params["end_flag"] = True
+
+        elif last_sender == "code_reviewer":
+            code_review = selected[0].content
+            cur_code = environment.rule_params["code"]
+            selected[0].content = f"<current code>:\n\n{cur_code}\n\n{code_review}"
+            feedback = environment.rule_params["feedback"]
+            f_dict = json.loads(feedback)
+            if f_dict["is_passing"]:
+                environment.rule_params["end_flag"] = True
+
+        return selected

agentverse/environments/rules/selector/sde_team_given_tests.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, List
+
+from agentverse.message import Message
+
+from . import selector_registry as SelectorRegistry
+from .base import BaseSelector
+
+import json
+import re
+
+if TYPE_CHECKING:
+    from agentverse.environments import BaseEnvironment
+
+def extract(content: str, keyword: str):
+    result = ""
+    flag = False
+    for line in content.split('\n'):
+        if line.strip().startswith(keyword):
+            flag = True
+            continue
+        if flag:
+            result += line
+            result += "\n"
+    return result
+
+
+@SelectorRegistry.register("sde_team_given_tests")
+class SdeTeamGivenTestsSelector(BaseSelector):
+    def select_message(self, environment: BaseEnvironment, messages: List[Message]) -> List[Message]:
+        last_sender = environment.last_messages[0].sender
+        selected = messages
+
+        if last_sender == "code_writer":
+            cur_code = extract(selected[0].content, "<code>:")
+            environment.rule_params["code"] = cur_code
+
+            from .code_api import execute_unit_tests
+            feedback = execute_unit_tests(environment.rule_params["code"], eval(environment.unit_tests))
+
+            environment.rule_params["feedback"] = feedback
+            selected[0].content = f"<current code>:\n\n{cur_code}\n\n<unit test feedback>:\n{feedback}"
+            f_dict = json.loads(feedback)
+            if f_dict["is_passing"]:
+                environment.rule_params["end_flag"] = True
+
+        elif last_sender == "code_reviewer":
+            code_review = selected[0].content
+            cur_code = environment.rule_params["code"]
+            selected[0].content = f"<current code>:\n\n{cur_code}\n\n{code_review}"
+            feedback = environment.rule_params["feedback"]
+            f_dict = json.loads(feedback)
+            if f_dict["is_passing"]:
+                environment.rule_params["end_flag"] = True
+
+        return selected

agentverse/environments/rules/updater/__init__.py

@@ -5,3 +5,4 @@
 from .base import BaseUpdater
 from .basic import BasicUpdater
 from .classroom import ClassroomUpdater
+from .sde_team import SdeTeamUpdater

agentverse/environments/rules/updater/sde_team.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, List, Tuple
+
+from . import updater_registry as UpdaterRegistry
+from .base import BaseUpdater
+from agentverse.message import Message
+
+if TYPE_CHECKING:
+    from agentverse.environments import BaseEnvironment
+    from agentverse.agents import BaseAgent
+
+
+@UpdaterRegistry.register("sde_team")
+class SdeTeamUpdater(BaseUpdater):
+    """
+    The basic version of updater.
+    The messages will be seen by all the receiver specified in the message.
+    """
+
+    def update_memory(self, environment: BaseEnvironment):
+        added = False
+        for message in environment.last_messages:
+            if message.content == "":
+                continue
+            added |= self.add_message_to_all_agents(environment.agents, message)
+
+    def add_message_to_all_agents(
+        self, agents: List[BaseAgent], message: Message
+    ) -> bool:
+        if "all" in message.receiver:
+            # If receiver is all, then add the message to all agents
+            for agent in agents:
+                agent.add_message_to_memory([message])
+            return True
+        else:
+            # If receiver is not all, then add the message to the specified agents
+            receiver_set = message.receiver
+            for agent in agents:
+                if agent.name in receiver_set:
+                    agent.add_message_to_memory([message])
+                    receiver_set.remove(agent.name)
+            if len(receiver_set) > 0:
+                missing_receiver = ", ".join(list(receiver_set))
+                raise ValueError(
+                    "Receiver {} not found. Message discarded".format(missing_receiver)
+                )
+            return True

agentverse/environments/rules/visibility/__init__.py

@@ -9,3 +9,4 @@
 from .classroom import ClassroomVisibility
 from .oneself import OneselfVisibility
 from .prisoner import PrisonerVisibility
+from .sde_team import SdeTeamVisibility

agentverse/environments/rules/visibility/sde_team.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import random
+from typing import TYPE_CHECKING, Any, List, Union
+
+from . import visibility_registry as VisibilityRegistry
+from .base import BaseVisibility
+
+if TYPE_CHECKING:
+    from agentverse.environments import BaseEnvironment
+
+
+@VisibilityRegistry.register("sde_team")
+class SdeTeamVisibility(BaseVisibility):
+    """
+    Visibility function for code problem. No need to change visibility.
+
+    """
+
+    def update_visible_agents(self, environment: BaseEnvironment):
+        return
+
+    def reset(self):
+        return