diff --git a/examples/st_game/actions/agent_chat_sum_rel.py b/examples/st_game/actions/agent_chat_sum_rel.py new file mode 100644 index 000000000..de005f8f4 --- /dev/null +++ b/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] diff --git a/examples/st_game/actions/decide_to_talk.py b/examples/st_game/actions/decide_to_talk.py new file mode 100644 index 000000000..fb1c625b1 --- /dev/null +++ b/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 diff --git a/examples/st_game/actions/st_action.py b/examples/st_game/actions/st_action.py new file mode 100644 index 000000000..9d319c80f --- /dev/null +++ b/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"!!", i) + if "###" in prompt: + prompt = prompt.split("###")[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.")