add role steps

examples/st_game/actions/dummy_action.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : dummy action to make every STRole can deal DummyMessage which is caused by DummyAction
+
+from dataclasses import dataclass
+
+from metagpt.actions import Action
+from metagpt.schema import Message
+
+
+class DummyAction(Action):
+
+    async def run(self, *args, **kwargs):
+        raise NotImplementedError
+
+
+@dataclass
+class DummyMessage(Message):
+    """
+    dummy message to pass to role and make them to have a execution every round
+    """
+
+    def __init__(self, content: str = "dummy", cause_by=DummyAction):
+        super(DummyMessage, self).__init__(content=content,
+                                           cause_by=cause_by)

examples/st_game/roles/st_role.py

@@ -15,12 +15,16 @@ from pydantic import Field
 from pathlib import Path
 
 from metagpt.roles.role import Role, RoleContext
+from metagpt.schema import Message
 
 from ..memory.associative_memory import AssociativeMemory
+from ..actions.dummy_action import DummyAction
+from ..actions.user_requirement import UserRequirement
+from ..maze_environment import MazeEnvironment
 
 
 class STRoleContext(RoleContext):
-
+    env: 'MazeEnvironment' = Field(default=None)
     memory: AssociativeMemory = Field(default=AssociativeMemory)
 
 
@@ -28,8 +32,22 @@ class STRole(Role):
 
     # add a role's property structure to store role's age and so on like GA's Scratch.
 
-    def __init__(self, name="", profile=""):
+    def __init__(self,
+                 name: str = "Klaus Mueller",
+                 profile: str = "STMember",
+                 has_inner_voice: bool = False):
+
         self._rc = STRoleContext()
+        super(STRole, self).__init__(name=name,
+                                     profile=profile)
+
+        self._init_actions([])
+
+        if has_inner_voice:
+            # TODO add communication action
+            self._watch([UserRequirement, DummyAction])
+        else:
+            self._watch([DummyAction])
 
     def load_from(self, folder: Path):
         """
@@ -42,3 +60,36 @@ class STRole(Role):
         save role data from `storage/{simulation_name}/personas/{role_name}
         """
         pass
+
+    async def observe(self):
+        # TODO observe info from maze_env
+        pass
+
+    async def plan(self):
+        # TODO make a plan
+
+        # TODO judge if start a conversation
+
+        # TODO update plan
+
+        # TODO re-add result into memory
+        pass
+
+    async def reflect(self):
+        # TODO reflection if meet reflect condition
+
+        # TODO re-add result to memory
+        pass
+
+    async def _react(self) -> Message:
+        maze_env = self._rc.env
+        # TODO observe
+        # get maze_env from self._rc.env, and observe env info
+
+        # TODO retrieve, use self._rc.memory 's retrieve functions
+
+        # TODO plan
+
+        # TODO reflect
+
+        # TODO execute(feed-back into maze_env)

examples/st_game/run_st_game.py

@@ -7,15 +7,26 @@ import fire
 
 from stanford_town import StanfordTown
 from roles.st_role import STRole
+from utils.mg_ga_transform import get_reverie_meta
+from utils.const import STORAGE_PATH
 
 
 async def startup(idea: str,
+                  fork_sim_code: str,
+                  sim_code: str,
                   investment: float = 30.0,
                   n_round: int = 500):
 
     # get role names from `storage/{simulation_name}/reverie/meta.json` and then init roles
+    reverie_meta = get_reverie_meta(fork_sim_code)
     roles = []
     # TODO
+    for idx, role_name in enumerate(reverie_meta["persona_names"]):
+        role_stg_path = STORAGE_PATH.joinpath(fork_sim_code).joinpath(f"personas/{role_name}")
+        has_inner_voice = True if idx == 0 else False
+        role = STRole(name=role_name, has_inner_voice=has_inner_voice)
+        role.load_from(role_stg_path)
+        roles.append(role)
 
     town = StanfordTown()
     town.wakeup_roles(roles)
@@ -27,13 +38,22 @@ async def startup(idea: str,
 
 
 def main(idea: str,
+         fork_sim_code: str,
+         sim_code: str,
          investment: float = 30.0,
          n_round: int = 500):
     """
-    idea works as an `inner voice` to the first agent.
+    Args:
+        idea: idea works as an `inner voice` to the first agent.
+        fork_sim_code: old simulation name to start with
+        sim_code: new simulation name to save simulation result
+        investment: the investment of running agents
+        n_round: rounds to run agents
     """
 
     asyncio.run(startup(idea=idea,
+                        fork_sim_code=fork_sim_code,
+                        sim_code=sim_code,
                         investment=investment,
                         n_round=n_round))
 

examples/st_game/utils/mg_ga_transform.py

@@ -1,3 +1,12 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 # @Desc   : data transform of mg <-> ga under storage
+
+from .const import STORAGE_PATH
+from .utils import read_json_file, write_json_file
+
+
+def get_reverie_meta(sim_code: str) -> dict:
+    meta_file_path = STORAGE_PATH.joinpath(sim_code).joinpath("reverie/meta.json")
+    reverie_meta = read_json_file(meta_file_path)
+    return reverie_meta

examples/st_game/utils/utils.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : utils
+
+from typing import Any
+import json
+from pathlib import Path
+
+
+def read_json_file(json_file: str, encoding=None) -> list[Any]:
+    if not Path(json_file).exists():
+        raise FileNotFoundError(f"json_file: {json_file} not exist, return []")
+
+    with open(json_file, "r", encoding=encoding) as fin:
+        try:
+            data = json.load(fin)
+        except Exception as exp:
+            raise ValueError(f"read json file: {json_file} failed")
+    return data
+
+
+def write_json_file(json_file: str, data: list, encoding=None):
+    with open(json_file, "w", encoding=encoding) as fout:
+        json.dump(data, fout, ensure_ascii=False, indent=4)