add st_action and chat action

examples/st_game/actions/agent_chat_sum_rel.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : summarize relationship in a agent chat
+
+from metagpt.logs import logger
+from metagpt.schema import Message
+
+from ..roles.st_role import STRole
+from ..actions.st_action import STAction
+
+
+class AgentChatSumRel(STAction):
+
+    def __init__(self, name="AgentChatSumRel", context: list[Message] = None, llm=None):
+        super().__init__(name, context, llm)
+
+    def _func_validate(self, llm_resp: str, prompt: str) -> bool:
+        resp = False
+        try:
+            _ = llm_resp.split('"')[0].strip()
+            resp = True
+        except Exception as exp:
+            pass
+        return resp
+
+    def _func_cleanup(self, llm_resp: str, prompt: str) -> str:
+        return llm_resp.split('"')[0].strip()
+
+    def _func_fail_default_resp(self) -> str:
+        pass
+
+    async def run(self, init_role: STRole, target_role: STRole, statements: str) -> str:
+        def create_prompt_input(init_role: STRole, target_role: STRole, statements: str) -> str:
+            prompt_input = [statements, init_role.name, target_role.name]
+            return prompt_input
+
+        prompt_input = create_prompt_input(init_role, target_role, statements)
+        prompt = self.generate_prompt_with_tmpl_filename(prompt_input,
+                                                         "summarize_chat_relationship_v2.txt")
+
+        example_output = "Jane Doe is working on a project"
+        special_instruction = "The output should be a string that responds to the question."
+        output = await self._run_v2(prompt,
+                                    example_output,
+                                    special_instruction)
+        return output[0]

examples/st_game/actions/decide_to_talk.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : device to talk to another role, return yes or no
+
+from metagpt.logs import logger
+from metagpt.schema import Message
+
+from ..roles.st_role import STRole
+from ..actions.st_action import STAction
+
+
+class DecideToTalk(STAction):
+
+    def __init__(self, name="DecideToTalk", context: list[Message] = None, llm=None):
+        super().__init__(name, context, llm)
+
+    def _func_validate(self, llm_resp: str, prompt: str) -> bool:
+        resp = False
+        try:
+            if llm_resp.split("Answer in yes or no:")[-1].strip().lower() in ["yes", "no"]:
+                resp = True
+        except ValueError as exp:
+            pass
+        return resp
+
+    def _func_cleanup(self, llm_resp: str, prompt: str) -> str:
+        return llm_resp.split("Answer in yes or no:")[-1].strip().lower()
+
+    def _func_fail_default_resp(self) -> str:
+        return "yes"
+
+    async def run(self, init_role: STRole, target_role: STRole, retrieved: dict, *args, **kwargs) -> bool:
+        """Run action"""
+        def create_prompt_input(init_role: STRole, target_role: STRole, retrieved: dict) -> str:
+            scratch = init_role._rc.scratch
+            target_scratch = target_role._rc.scratch
+            last_chat = init_role._rc.memory.get_last_chat(target_role.name)
+            last_chatted_time = ""
+            last_chat_about = ""
+            if last_chat:
+                last_chatted_time = last_chat.created.strftime("%B %d, %Y, %H:%M:%S")
+                last_chat_about = last_chat.description
+
+            context = ""
+            for c_node in retrieved["events"]:
+                curr_desc = c_node.description.split(" ")
+                curr_desc[2:3] = ["was"]
+                curr_desc = " ".join(curr_desc)
+                context += f"{curr_desc}. "
+            context += "\n"
+            for c_node in retrieved["thoughts"]:
+                context += f"{c_node.description}. "
+
+            curr_time = scratch.curr_time.strftime("%B %d, %Y, %H:%M:%S %p")
+            init_act_desc = scratch.act_description
+            if "(" in init_act_desc:
+                init_act_desc = init_act_desc.split("(")[-1][:-1]
+
+            if len(scratch.planned_path) == 0 and "waiting" not in init_act_desc:
+                init_p_desc = f"{init_role.name} is already {init_act_desc}"
+            elif "waiting" in init_act_desc:
+                init_p_desc = f"{init_role.name} is {init_act_desc}"
+            else:
+                init_p_desc = f"{init_role.name} is on the way to {init_act_desc}"
+
+            target_act_desc = scratch.act_description
+            if "(" in target_act_desc:
+                target_act_desc = target_act_desc.split("(")[-1][:-1]
+
+            if len(target_scratch.planned_path) == 0 and "waiting" not in init_act_desc:
+                target_p_desc = f"{target_role.name} is already {target_act_desc}"
+            elif "waiting" in init_act_desc:
+                target_p_desc = f"{init_role.name} is {init_act_desc}"
+            else:
+                target_p_desc = f"{target_role.name} is on the way to {target_act_desc}"
+
+            prompt_input = []
+            prompt_input += [context]
+
+            prompt_input += [curr_time]
+
+            prompt_input += [init_role.name]
+            prompt_input += [target_role.name]
+            prompt_input += [last_chatted_time]
+            prompt_input += [last_chat_about]
+
+            prompt_input += [init_p_desc]
+            prompt_input += [target_p_desc]
+            prompt_input += [init_role.name]
+            prompt_input += [target_role.name]
+            return prompt_input
+
+        prompt_input = create_prompt_input(init_role, target_role, retrieved)
+        prompt = self.generate_prompt_with_tmpl_filename(prompt_input=prompt_input,
+                                                         tmpl_filename="decide_to_talk_v2.txt")
+        output = await self._run_v1(prompt)  # yes or no
+        result = True if output == "yes" else False
+        logger.info(f"Run action: {self.__class__.__name__} with result: {result}")
+        return result

examples/st_game/actions/st_action.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# @Desc   : StanfordTown Action
+
+from typing import Union
+from abc import abstractmethod
+import json
+
+from metagpt.actions.action import Action
+from metagpt.schema import Message
+
+from ..utils.const import PROMPTS_DIR
+
+
+class STAction(Action):
+
+    def __init__(self, name="STAction", context: list[Message] = None, llm=None):
+        super().__init__(name, context, llm)
+        self.prompt_dir = PROMPTS_DIR
+
+    @abstractmethod
+    def _func_validate(self, llm_resp: str, prompt: str):
+        raise NotImplementedError
+
+    @abstractmethod
+    def _func_cleanup(self, llm_resp: str, prompt: str):
+        raise NotImplementedError
+
+    @abstractmethod
+    def _func_fail_default_resp(self):
+        raise NotImplementedError
+
+    def generate_prompt_with_tmpl_filename(self, prompt_input: Union[str, list], tmpl_filename) -> str:
+        """
+        same with `generate_prompt`
+        Args:
+            prompt_input: the input we want to feed in (IF THERE ARE MORE THAN ONE INPUT, THIS CAN BE A LIST.)
+            tmpl_filename: prompt template filename
+        Returns:
+            a str prompt that will be sent to LLM server.
+        """
+        if isinstance(prompt_input, str):
+            prompt_input = [prompt_input]
+        prompt_input = [str(i) for i in prompt_input]
+
+        f = open(str(self.prompt_dir.joinpath(tmpl_filename)), "r")
+        prompt = f.read()
+        f.close()
+        for count, i in enumerate(prompt_input):
+            prompt = prompt.replace(f"!<INPUT {count}>!", i)
+        if "<commentblockmarker>###</commentblockmarker>" in prompt:
+            prompt = prompt.split("<commentblockmarker>###</commentblockmarker>")[1]
+        return prompt.strip()
+
+    async def _run_v1(self, prompt: str, retry: int = 3) -> str:
+        """
+            same with `gpt_structure.generate_prompt`
+            default post-preprocess operations of LLM response
+        """
+        for idx in range(retry):
+            llm_resp = await self._aask(prompt)
+            if self._func_validate(llm_resp, prompt):
+                return self._func_cleanup(llm_resp, prompt)
+        return self._func_fail_default_resp()
+
+    async def _run_v2(self,
+                      prompt: str,
+                      example_output: str,
+                      special_instruction: str,
+                      retry: int = 3):
+        """ same with `gpt_structure.ChatGPT_safe_generate_response` """
+        prompt = '"""\n' + prompt + '\n"""\n'
+        prompt += f"Output the response to the prompt above in json. {special_instruction}\n"
+        prompt += "Example output json:\n"
+        prompt += '{"output": "' + str(example_output) + '"}'
+
+        for idx in range(retry):
+            try:
+                llm_resp = await self._aask(prompt)
+                end_idx = llm_resp.strip().rfind("}") + 1
+                llm_resp = llm_resp[:end_idx]
+                llm_resp = json.loads(llm_resp)["output"]
+
+                if self._func_validate(llm_resp, prompt):
+                    return self._func_cleanup(llm_resp, prompt)
+            except Exception as exp:
+                pass
+        return False
+
+    async def run(self, *args, **kwargs):
+        """Run action"""
+        raise NotImplementedError("The run method should be implemented in a subclass.")