mirror of
https://github.com/FoundationAgents/MetaGPT.git
synced 2026-06-17 15:35:21 +02:00
add action detail log and more completion
This commit is contained in:
better629
parent
a37be1bd33
commit
6c947945d7
13 changed files with 348 additions and 316 deletions
|
|
@ -39,7 +39,8 @@ class AgentChatSumRel(STAction):
|
|||
|
||||
example_output = "Jane Doe is working on a project"
|
||||
special_instruction = "The output should be a string that responds to the question."
|
||||
output = self._run_v2(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
return output[0]
|
||||
output = self._run_gpt35(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
logger.info(f"Role: {init_role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
|
|
|||
|
|
@ -93,7 +93,7 @@ class DecideToTalk(STAction):
|
|||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input=prompt_input,
|
||||
tmpl_filename="decide_to_talk_v2.txt")
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
output = self._run_v1(prompt) # yes or no
|
||||
output = self._run_text_davinci(prompt, max_tokens=20) # yes or no
|
||||
result = True if output == "yes" else False
|
||||
logger.info(f"Run action: {self.__class__.__name__} with result: {result}")
|
||||
logger.info(f"Role: {init_role.name} Action: {self.cls_name} output: {result}")
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ from metagpt.schema import Message
|
|||
from ..maze import Maze
|
||||
from .st_action import STAction
|
||||
|
||||
|
||||
|
||||
class GenActionSector(STAction):
|
||||
|
||||
def __init__(self, name="GenActionSector", context: list[Message] = None, llm=None):
|
||||
|
|
@ -19,55 +19,53 @@ class GenActionSector(STAction):
|
|||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str):
|
||||
cleaned_response = llm_resp.split("}")[0]
|
||||
return cleaned_response
|
||||
return cleaned_response
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
if len(llm_resp.strip()) < 1:
|
||||
if len(llm_resp.strip()) < 1:
|
||||
return False
|
||||
if "}" not in llm_resp:
|
||||
return False
|
||||
if "," in llm_resp:
|
||||
if "," in llm_resp:
|
||||
return False
|
||||
return True
|
||||
return True
|
||||
|
||||
def _func_fail_default_resp(self):
|
||||
fs = ("kitchen")
|
||||
return fs
|
||||
return fs
|
||||
|
||||
def run(self, role: "STRole", maze: Maze, act_desp: str):
|
||||
def create_prompt_input(role, maze, act_desp):
|
||||
def create_prompt_input(role, maze, act_desp):
|
||||
act_world = f"{maze.access_tile(role.scratch.curr_tile)['world']}"
|
||||
|
||||
|
||||
prompt_input = []
|
||||
|
||||
|
||||
prompt_input += [role.scratch.get_str_name()]
|
||||
prompt_input += [role.scratch.living_area.split(":")[1]]
|
||||
x = f"{act_world}:{role.scratch.living_area.split(':')[1]}"
|
||||
prompt_input += [role.s_mem.get_str_accessible_sector_arenas(x)]
|
||||
|
||||
|
||||
prompt_input += [role.scratch.get_str_name()]
|
||||
prompt_input += [f"{maze.access_tile(role.scratch.curr_tile)['sector']}"]
|
||||
x = f"{act_world}:{maze.access_tile(role.scratch.curr_tile)['sector']}"
|
||||
prompt_input += [role.s_mem.get_str_accessible_sector_arenas(x)]
|
||||
|
||||
if role.scratch.get_str_daily_plan_req() != "":
|
||||
if role.scratch.get_str_daily_plan_req() != "":
|
||||
prompt_input += [f"\n{role.scratch.get_str_daily_plan_req()}"]
|
||||
else:
|
||||
else:
|
||||
prompt_input += [""]
|
||||
|
||||
|
||||
# MAR 11 TEMP
|
||||
prompt_input = []
|
||||
act_world = maze.access_tile(role.scratch.curr_tile)["world"]
|
||||
accessible_sector_str = role.s_mem.get_str_accessible_sectors(act_world)
|
||||
curr = accessible_sector_str.split(", ")
|
||||
fin_accessible_sectors = []
|
||||
for i in curr:
|
||||
if "'s house" in i:
|
||||
if role.scratch.last_name in i:
|
||||
for i in curr:
|
||||
if "'s house" in i:
|
||||
if role.scratch.last_name in i:
|
||||
fin_accessible_sectors += [i]
|
||||
else:
|
||||
else:
|
||||
fin_accessible_sectors += [i]
|
||||
accessible_sector_str = ", ".join(fin_accessible_sectors)
|
||||
# END MAR 11 TEMP
|
||||
|
|
@ -76,7 +74,7 @@ class GenActionSector(STAction):
|
|||
|
||||
act_desp_1 = act_desp
|
||||
act_desp_2 = act_desp
|
||||
if "(" in act_desp:
|
||||
if "(" in act_desp:
|
||||
act_desp_1 = act_desp.split("(")[0].strip()
|
||||
act_desp_2 = act_desp.split("(")[-1][:-1]
|
||||
prompt_input += [role.scratch.get_str_name()]
|
||||
|
|
@ -85,18 +83,19 @@ class GenActionSector(STAction):
|
|||
prompt_input += [act_desp_2]
|
||||
prompt_input += [role.scratch.get_str_name()]
|
||||
return prompt_input
|
||||
|
||||
|
||||
prompt_template = "action_location_sector_v1.txt"
|
||||
prompt_input = create_prompt_input(role, maze, act_desp)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, prompt_template)
|
||||
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
output = self._run_v1(prompt)
|
||||
output = self._run_text_davinci(prompt, max_tokens=15)
|
||||
y = f"{maze.access_tile(role.scratch.curr_tile)['world']}"
|
||||
x = [i.strip() for i in role.s_mem.get_str_accessible_sectors(y).split(",")]
|
||||
if output not in x:
|
||||
if output not in x:
|
||||
# output = random.choice(x)
|
||||
output = role.scratch.living_area.split(":")[1]
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
|
|
@ -108,21 +107,21 @@ class GenActionArena(STAction):
|
|||
cleaned_response = llm_resp.split("}")[0]
|
||||
return cleaned_response
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
if len(llm_resp.strip()) < 1:
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
if len(llm_resp.strip()) < 1:
|
||||
return False
|
||||
if "}" not in llm_resp:
|
||||
return False
|
||||
if "," in llm_resp:
|
||||
if "," in llm_resp:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _func_fail_default_resp(self):
|
||||
fs = ("kitchen")
|
||||
return fs
|
||||
|
||||
|
||||
def run(self, role: "STRole", maze: Maze, act_desp: str, act_world: str, act_sector: str):
|
||||
def create_prompt_input(role, maze, act_desp, act_world, act_sector):
|
||||
def create_prompt_input(role, maze, act_desp, act_world, act_sector):
|
||||
prompt_input = []
|
||||
# prompt_input += [role.scratch.get_str_name()]
|
||||
# prompt_input += [maze.access_tile(role.scratch.curr_tile)["arena"]]
|
||||
|
|
@ -135,18 +134,18 @@ class GenActionArena(STAction):
|
|||
accessible_arena_str = role.s_mem.get_str_accessible_sector_arenas(x)
|
||||
curr = accessible_arena_str.split(", ")
|
||||
fin_accessible_arenas = []
|
||||
for i in curr:
|
||||
if "'s room" in i:
|
||||
if role.scratch.last_name in i:
|
||||
for i in curr:
|
||||
if "'s room" in i:
|
||||
if role.scratch.last_name in i:
|
||||
fin_accessible_arenas += [i]
|
||||
else:
|
||||
else:
|
||||
fin_accessible_arenas += [i]
|
||||
accessible_arena_str = ", ".join(fin_accessible_arenas)
|
||||
# END MAR 11 TEMP
|
||||
prompt_input += [accessible_arena_str]
|
||||
act_desp_1 = act_desp
|
||||
act_desp_2 = act_desp
|
||||
if "(" in act_desp:
|
||||
if "(" in act_desp:
|
||||
act_desp_1 = act_desp.split("(")[0].strip()
|
||||
act_desp_2 = act_desp.split("(")[-1][:-1]
|
||||
prompt_input += [role.scratch.get_str_name()]
|
||||
|
|
@ -163,16 +162,17 @@ class GenActionArena(STAction):
|
|||
prompt_input = create_prompt_input(role, maze, act_desp, act_world, act_sector)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, prompt_template)
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
output = self._run_v1(prompt)
|
||||
output = self._run_text_davinci(prompt, max_tokens=15)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
class GenActionObject(STAction):
|
||||
def __init__(self, name="GenActionObject", context: list[Message] = None, llm=None):
|
||||
super().__init__(name, context, llm)
|
||||
super().__init__(name, context, llm)
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
if len(llm_resp.strip()) < 1:
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
if len(llm_resp.strip()) < 1:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
|
@ -186,30 +186,30 @@ class GenActionObject(STAction):
|
|||
|
||||
def run(self, role: "STRole", act_desp: str, temp_address: str):
|
||||
|
||||
def create_prompt_input(role, act_desp, temp_address):
|
||||
def create_prompt_input(role, act_desp, temp_address):
|
||||
prompt_input = []
|
||||
if "(" in act_desp:
|
||||
if "(" in act_desp:
|
||||
act_desp = act_desp.split("(")[-1][:-1]
|
||||
|
||||
|
||||
prompt_input += [act_desp]
|
||||
prompt_input += [role
|
||||
.s_mem.get_str_accessible_arena_game_objects(temp_address)]
|
||||
prompt_input += [role.s_mem.get_str_accessible_arena_game_objects(temp_address)]
|
||||
return prompt_input
|
||||
|
||||
|
||||
prompt_template = "action_object_v2.txt"
|
||||
prompt_input = create_prompt_input(role, act_desp, temp_address)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, prompt_template)
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
output = self._run_v1(prompt)
|
||||
output = self._run_text_davinci(prompt, max_tokens=15)
|
||||
x = [i.strip() for i in role.s_mem.get_str_accessible_arena_game_objects(temp_address).split(",")]
|
||||
if output not in x:
|
||||
if output not in x:
|
||||
output = random.choice(x)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
class GenPronunciatio(STAction):
|
||||
def __init__(self, name="GenPronunciatio", context: list[Message] = None, llm=None):
|
||||
super().__init__(name, context, llm)
|
||||
super().__init__(name, context, llm)
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str):
|
||||
cr = llm_resp.strip()
|
||||
|
|
@ -217,169 +217,174 @@ class GenPronunciatio(STAction):
|
|||
cr = cr[:3]
|
||||
return cr
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
try:
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
try:
|
||||
self._func_cleanup(llm_resp, prompt="")
|
||||
if len(llm_resp) == 0:
|
||||
if len(llm_resp) == 0:
|
||||
return False
|
||||
except:
|
||||
return False
|
||||
return True
|
||||
except Exception as exp:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _func_fail_default_resp(self):
|
||||
fs = "😋"
|
||||
return fs
|
||||
|
||||
def run(self, role: "STRole", act_desp: str):
|
||||
def create_prompt_input(act_desp):
|
||||
if "(" in act_desp:
|
||||
def create_prompt_input(act_desp):
|
||||
if "(" in act_desp:
|
||||
act_desp = act_desp.split("(")[-1].split(")")[0]
|
||||
prompt_input = [act_desp]
|
||||
return prompt_input
|
||||
|
||||
|
||||
prompt_template = "generate_pronunciatio_v1.txt"
|
||||
prompt_input = create_prompt_input(act_desp)
|
||||
prompt_template = "generate_pronunciatio_v1.txt"
|
||||
prompt_input = create_prompt_input(act_desp)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, prompt_template)
|
||||
example_output = "🛁🧖♀️"
|
||||
special_instruction = "The value for the output must ONLY contain the emojis."
|
||||
example_output = "🛁🧖♀️"
|
||||
special_instruction = "The value for the output must ONLY contain the emojis."
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
output = self._run_v2(prompt, example_output, special_instruction)
|
||||
output = self._run_gpt35(prompt, example_output, special_instruction)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
class GenEventTriple(STAction):
|
||||
def __init__(self, name="GenEventTriple", context: list[Message] = None, llm=None):
|
||||
super().__init__(name, context, llm)
|
||||
super().__init__(name, context, llm)
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str):
|
||||
cr = llm_resp.strip()
|
||||
cr = [i.strip() for i in cr.split(")")[0].split(",")]
|
||||
return cr
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
try:
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
try:
|
||||
llm_resp = self._func_cleanup(llm_resp, prompt="")
|
||||
if len(llm_resp) != 2:
|
||||
if len(llm_resp) != 2:
|
||||
return False
|
||||
except:
|
||||
return False
|
||||
return True
|
||||
except Exception as exp:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _func_fail_default_resp(self, role):
|
||||
def _func_fail_default_resp(self, role):
|
||||
fs = (role.name, "is", "idle")
|
||||
return fs
|
||||
return fs
|
||||
|
||||
def run(self, role: "STRole", act_desp: str):
|
||||
def create_prompt_input(role, act_desp):
|
||||
if "(" in act_desp:
|
||||
def create_prompt_input(role, act_desp):
|
||||
if "(" in act_desp:
|
||||
act_desp = act_desp.split("(")[-1].split(")")[0]
|
||||
prompt_input = [role.name,
|
||||
prompt_input = [role.name,
|
||||
act_desp,
|
||||
role.name]
|
||||
return prompt_input
|
||||
|
||||
|
||||
prompt_template = "generate_event_triple_v1.txt"
|
||||
prompt_input = create_prompt_input(role, act_desp)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, prompt_template)
|
||||
self.fail_default_resp = self._func_fail_default_resp(role)
|
||||
output = self._run_v1(prompt)
|
||||
output = self._run_text_davinci(prompt, max_tokens=30)
|
||||
output = (role.name, output[0], output[1])
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
class GenActObjDescription(STAction):
|
||||
def __init__(self, name="GenActObjDescription", context: list[Message] = None, llm=None):
|
||||
super().__init__(name, context, llm)
|
||||
super().__init__(name, context, llm)
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str):
|
||||
cr = llm_resp.strip()
|
||||
if cr[-1] == ".": cr = cr[:-1]
|
||||
if cr[-1] == ".":
|
||||
cr = cr[:-1]
|
||||
return cr
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
try:
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
try:
|
||||
llm_resp = self._func_cleanup(llm_resp, prompt="")
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
return True
|
||||
return True
|
||||
|
||||
def _func_fail_default_resp(self,act_game_object):
|
||||
def _func_fail_default_resp(self, act_game_object):
|
||||
fs = f"{act_game_object} is idle"
|
||||
return fs
|
||||
|
||||
def run(self, role: "STRole", act_game_object: str, act_desp: str):
|
||||
def create_prompt_input(act_game_object, act_desp, role):
|
||||
prompt_input = [act_game_object,
|
||||
|
||||
def run(self, role: "STRole", act_game_object: str, act_desp: str):
|
||||
def create_prompt_input(act_game_object, act_desp, role):
|
||||
prompt_input = [act_game_object,
|
||||
role.name,
|
||||
act_desp,
|
||||
act_game_object,
|
||||
act_game_object]
|
||||
return prompt_input
|
||||
|
||||
prompt_template = "generate_obj_event_v1.txt" ########
|
||||
prompt_input = create_prompt_input(act_game_object, act_desp, role) ########
|
||||
|
||||
prompt_template = "generate_obj_event_v1.txt"
|
||||
prompt_input = create_prompt_input(act_game_object, act_desp, role)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, prompt_template)
|
||||
example_output = "being fixed" ########
|
||||
special_instruction = "The output should ONLY contain the phrase that should go in <fill in>." ########
|
||||
self.fail_default_resp = self._func_fail_default_resp(act_game_object) ########
|
||||
output = self._run_v2(prompt, example_output, special_instruction)
|
||||
example_output = "being fixed"
|
||||
special_instruction = "The output should ONLY contain the phrase that should go in <fill in>."
|
||||
self.fail_default_resp = self._func_fail_default_resp(act_game_object)
|
||||
output = self._run_gpt35(prompt, example_output, special_instruction)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
|
||||
class GenObjEventTriple(STAction):
|
||||
def __init__(self, name="GenObjEventTriple", context: list[Message] = None, llm=None):
|
||||
super().__init__(name, context, llm)
|
||||
super().__init__(name, context, llm)
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str):
|
||||
cr = llm_resp.strip()
|
||||
cr = [i.strip() for i in cr.split(")")[0].split(",")]
|
||||
return cr
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
try:
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
try:
|
||||
llm_resp = self._func_cleanup(llm_resp, prompt="")
|
||||
if len(llm_resp) != 2:
|
||||
if len(llm_resp) != 2:
|
||||
return False
|
||||
except: return False
|
||||
return True
|
||||
except Exception as exp:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _func_fail_default_resp(self,act_game_object):
|
||||
def _func_fail_default_resp(self, act_game_object: str):
|
||||
fs = (act_game_object, "is", "idle")
|
||||
return fs
|
||||
|
||||
def run(self, role: "STRole", act_game_object, act_obj_desp):
|
||||
def create_prompt_input(act_game_object, act_obj_desp):
|
||||
prompt_input = [act_game_object,
|
||||
def create_prompt_input(act_game_object, act_obj_desp):
|
||||
prompt_input = [act_game_object,
|
||||
act_obj_desp,
|
||||
act_game_object]
|
||||
return prompt_input
|
||||
|
||||
|
||||
prompt_template = "generate_event_triple_v1.txt"
|
||||
prompt_input = create_prompt_input(act_game_object, act_obj_desp)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, prompt_template)
|
||||
self.fail_default_resp = self._func_fail_default_resp(act_game_object)
|
||||
output = self._run_v1(prompt)
|
||||
output = self._run_text_davinci(prompt, max_tokens=30)
|
||||
output = (act_game_object, output[0], output[1])
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
|
||||
class GenActionDetails(STAction):
|
||||
def __init__(self, name="GenActionDetails", context: list[Message] = None, llm=None):
|
||||
super().__init__(name, context, llm)
|
||||
super().__init__(name, context, llm)
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str) -> list:
|
||||
pass
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str) -> bool:
|
||||
# TODO -- this sometimes generates error
|
||||
try:
|
||||
# TODO -- this sometimes generates error
|
||||
try:
|
||||
self._func_cleanup(llm_resp)
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _func_fail_default_resp(self):
|
||||
|
||||
def _func_fail_default_resp(self):
|
||||
fs = {}
|
||||
return fs
|
||||
|
||||
|
|
@ -396,11 +401,10 @@ class GenActionDetails(STAction):
|
|||
new_address = f"{act_world}:{act_sector}:{act_arena}:{act_game_object}"
|
||||
act_pron = GenPronunciatio().run(role, act_desp)
|
||||
act_event = GenEventTriple().run(role, act_desp)
|
||||
# Persona's actions also influence the object states. We set those up here.
|
||||
# Persona's actions also influence the object states. We set those up here.
|
||||
act_obj_desp = GenActObjDescription().run(role, act_game_object, act_desp)
|
||||
act_obj_pron = GenPronunciatio().run(role, act_obj_desp)
|
||||
act_obj_event = GenObjEventTriple().run(role, act_game_object,
|
||||
act_obj_desp)
|
||||
act_obj_event = GenObjEventTriple().run(role, act_game_object, act_obj_desp)
|
||||
result_dict = {
|
||||
"action_address": new_address,
|
||||
"action_duration": int(act_dura),
|
||||
|
|
@ -414,4 +418,5 @@ class GenActionDetails(STAction):
|
|||
"act_obj_description": act_obj_desp,
|
||||
"act_obj_pronunciatio": act_obj_pron,
|
||||
"act_obj_event": act_obj_event}
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {result_dict}")
|
||||
return result_dict
|
||||
|
|
|
|||
|
|
@ -15,32 +15,32 @@ class GenDailySchedule(STAction):
|
|||
super().__init__(name, context, llm)
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str) -> bool:
|
||||
try:
|
||||
try:
|
||||
self._func_cleanup(llm_resp, prompt="")
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str) -> list:
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str) -> list:
|
||||
cr = []
|
||||
_cr = llm_resp.split(")")
|
||||
for i in _cr:
|
||||
if i[-1].isdigit():
|
||||
for i in _cr:
|
||||
if i[-1].isdigit():
|
||||
i = i[:-1].strip()
|
||||
if i[-1] == "." or i[-1] == ",":
|
||||
if i[-1] == "." or i[-1] == ",":
|
||||
cr += [i[:-1].strip()]
|
||||
return cr
|
||||
return cr
|
||||
|
||||
def _func_fail_default_resp(self) -> int:
|
||||
fs = ['wake up and complete the morning routine at 6:00 am',
|
||||
'eat breakfast at 7:00 am',
|
||||
'read a book from 8:00 am to 12:00 pm',
|
||||
'have lunch at 12:00 pm',
|
||||
'take a nap from 1:00 pm to 4:00 pm',
|
||||
'relax and watch TV from 7:00 pm to 8:00 pm',
|
||||
'go to bed at 11:00 pm']
|
||||
fs = ['wake up and complete the morning routine at 6:00 am',
|
||||
'eat breakfast at 7:00 am',
|
||||
'read a book from 8:00 am to 12:00 pm',
|
||||
'have lunch at 12:00 pm',
|
||||
'take a nap from 1:00 pm to 4:00 pm',
|
||||
'relax and watch TV from 7:00 pm to 8:00 pm',
|
||||
'go to bed at 11:00 pm']
|
||||
return fs
|
||||
|
||||
|
||||
def run(self, role: "STRole", wake_up_hour: str):
|
||||
def create_prompt_input(role, wake_up_hour):
|
||||
prompt_input = []
|
||||
|
|
@ -50,12 +50,13 @@ class GenDailySchedule(STAction):
|
|||
prompt_input += [role.scratch.get_str_firstname()]
|
||||
prompt_input += [f"{str(wake_up_hour)}:00 am"]
|
||||
return prompt_input
|
||||
|
||||
wake_up_hour = int(wake_up_hour)
|
||||
prompt_template = "daily_planning_v6.txt"
|
||||
prompt_input = create_prompt_input(role, wake_up_hour)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, prompt_template)
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
output = self._run_v1(prompt)
|
||||
output = ([f"wake up and complete the morning routine at {wake_up_hour}:00 am"]
|
||||
+ output)
|
||||
output = self._run_text_davinci(prompt, max_tokens=500)
|
||||
output = ([f"wake up and complete the morning routine at {wake_up_hour}:00 am"] + output)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
|
|
|||
|
|
@ -8,20 +8,19 @@ import string
|
|||
|
||||
from metagpt.logs import logger
|
||||
from metagpt.schema import Message
|
||||
from metagpt.config import CONFIG
|
||||
|
||||
from .st_action import STAction
|
||||
|
||||
|
||||
def get_random_alphanumeric(i=6, j=6):
|
||||
def get_random_alphanumeric(i=6, j=6):
|
||||
"""
|
||||
Returns a random alpha numeric strength that has the length of somewhere
|
||||
between i and j.
|
||||
between i and j.
|
||||
|
||||
INPUT:
|
||||
INPUT:
|
||||
i: min_range for the length
|
||||
j: max_range for the length
|
||||
OUTPUT:
|
||||
OUTPUT:
|
||||
an alpha numeric str with the length of somewhere between i and j.
|
||||
"""
|
||||
k = random.randint(i, j)
|
||||
|
|
@ -34,12 +33,12 @@ class GenHourlySchedule(STAction):
|
|||
super().__init__(name, context, llm)
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str) -> bool:
|
||||
try:
|
||||
try:
|
||||
self._func_cleanup(llm_resp, prompt="")
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str) -> list:
|
||||
cr = llm_resp.strip()
|
||||
if cr[-1] == ".":
|
||||
|
|
@ -57,28 +56,28 @@ class GenHourlySchedule(STAction):
|
|||
p_f_ds_hourly_org,
|
||||
hour_str,
|
||||
intermission2=None):
|
||||
def create_prompt_input(persona,
|
||||
curr_hour_str,
|
||||
def create_prompt_input(persona,
|
||||
curr_hour_str,
|
||||
p_f_ds_hourly_org,
|
||||
hour_str,
|
||||
intermission2=None):
|
||||
intermission2=None):
|
||||
schedule_format = ""
|
||||
for i in hour_str:
|
||||
for i in hour_str:
|
||||
schedule_format += f"[{persona.scratch.get_str_curr_date_str()} -- {i}]"
|
||||
schedule_format += f" Activity: [Fill in]\n"
|
||||
schedule_format = schedule_format[:-1]
|
||||
|
||||
intermission_str = f"Here the originally intended hourly breakdown of"
|
||||
intermission_str += f" {persona.scratch.get_str_firstname()}'s schedule today: "
|
||||
for count, i in enumerate(persona.scratch.daily_req):
|
||||
intermission_str += f"{str(count+1)}) {i}, "
|
||||
for count, i in enumerate(persona.scratch.daily_req):
|
||||
intermission_str += f"{str(count + 1)}) {i}, "
|
||||
intermission_str = intermission_str[:-2]
|
||||
|
||||
prior_schedule = ""
|
||||
if p_f_ds_hourly_org:
|
||||
if p_f_ds_hourly_org:
|
||||
prior_schedule = "\n"
|
||||
for count, i in enumerate(p_f_ds_hourly_org):
|
||||
prior_schedule += f"[(ID:{get_random_alphanumeric()})"
|
||||
for count, i in enumerate(p_f_ds_hourly_org):
|
||||
prior_schedule += f"[(ID:{get_random_alphanumeric()})"
|
||||
prior_schedule += f" {persona.scratch.get_str_curr_date_str()} --"
|
||||
prior_schedule += f" {hour_str[count]}] Activity:"
|
||||
prior_schedule += f" {persona.scratch.get_str_firstname()}"
|
||||
|
|
@ -89,7 +88,7 @@ class GenHourlySchedule(STAction):
|
|||
prompt_ending += f" -- {curr_hour_str}] Activity:"
|
||||
prompt_ending += f" {persona.scratch.get_str_firstname()} is"
|
||||
|
||||
if intermission2:
|
||||
if intermission2:
|
||||
intermission2 = f"\n{intermission2}"
|
||||
|
||||
prompt_input = []
|
||||
|
|
@ -98,38 +97,32 @@ class GenHourlySchedule(STAction):
|
|||
|
||||
prompt_input += [prior_schedule + "\n"]
|
||||
prompt_input += [intermission_str]
|
||||
if intermission2:
|
||||
if intermission2:
|
||||
prompt_input += [intermission2]
|
||||
else:
|
||||
else:
|
||||
prompt_input += [""]
|
||||
prompt_input += [prompt_ending]
|
||||
|
||||
return prompt_input
|
||||
|
||||
prompt_template = "generate_hourly_schedule_v2.txt"
|
||||
prompt_input = create_prompt_input(role,
|
||||
curr_hour_str,
|
||||
prompt_input = create_prompt_input(role,
|
||||
curr_hour_str,
|
||||
p_f_ds_hourly_org,
|
||||
hour_str,
|
||||
intermission2)
|
||||
prompt_input_str = "\n".join(prompt_input)
|
||||
raw_max_tokens_rsp = CONFIG.max_tokens_rsp
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, prompt_template)
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
|
||||
CONFIG.max_tokens_rsp = 50
|
||||
output = self._run_v1(prompt)
|
||||
CONFIG.max_tokens_rsp = raw_max_tokens_rsp
|
||||
|
||||
logger.debug(f"max_tokens_rsp: {CONFIG.max_tokens_rsp}")
|
||||
output = self._run_text_davinci(prompt, max_tokens=50)
|
||||
logger.info(f"Role: {role.name} _generate_schedule_for_given_hour prompt_input: {prompt_input_str}, "
|
||||
f"output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
def run(self, role: "STRole", wake_up_hour: int):
|
||||
hour_str = ["00:00 AM", "01:00 AM", "02:00 AM", "03:00 AM", "04:00 AM",
|
||||
"05:00 AM", "06:00 AM", "07:00 AM", "08:00 AM", "09:00 AM",
|
||||
"10:00 AM", "11:00 AM", "12:00 PM", "01:00 PM", "02:00 PM",
|
||||
hour_str = ["00:00 AM", "01:00 AM", "02:00 AM", "03:00 AM", "04:00 AM",
|
||||
"05:00 AM", "06:00 AM", "07:00 AM", "08:00 AM", "09:00 AM",
|
||||
"10:00 AM", "11:00 AM", "12:00 PM", "01:00 PM", "02:00 PM",
|
||||
"03:00 PM", "04:00 PM", "05:00 PM", "06:00 PM", "07:00 PM",
|
||||
"08:00 PM", "09:00 PM", "10:00 PM", "11:00 PM"]
|
||||
n_m1_activity = []
|
||||
|
|
@ -137,42 +130,41 @@ class GenHourlySchedule(STAction):
|
|||
for i in range(diversity_repeat_count):
|
||||
logger.info(f"diversity_repeat_count idx: {i}")
|
||||
n_m1_activity_set = set(n_m1_activity)
|
||||
if len(n_m1_activity_set) < 5:
|
||||
if len(n_m1_activity_set) < 5:
|
||||
n_m1_activity = []
|
||||
for count, curr_hour_str in enumerate(hour_str):
|
||||
if wake_up_hour > 0:
|
||||
for count, curr_hour_str in enumerate(hour_str):
|
||||
if wake_up_hour > 0:
|
||||
n_m1_activity += ["sleeping"]
|
||||
wake_up_hour -= 1
|
||||
else:
|
||||
logger.info(f"_generate_schedule_for_given_hour idx: {count}, n_m1_activity: {n_m1_activity}")
|
||||
n_m1_activity += [self._generate_schedule_for_given_hour(
|
||||
role, curr_hour_str, n_m1_activity, hour_str)]
|
||||
|
||||
# Step 1. Compressing the hourly schedule to the following format:
|
||||
# The integer indicates the number of hours. They should add up to 24.
|
||||
# [['sleeping', 6], ['waking up and starting her morning routine', 1],
|
||||
# ['eating breakfast', 1], ['getting ready for the day', 1],
|
||||
# ['working on her painting', 2], ['taking a break', 1],
|
||||
# ['having lunch', 1], ['working on her painting', 3],
|
||||
# ['taking a break', 2], ['working on her painting', 2],
|
||||
role, curr_hour_str, n_m1_activity, hour_str)]
|
||||
|
||||
# Step 1. Compressing the hourly schedule to the following format:
|
||||
# The integer indicates the number of hours. They should add up to 24.
|
||||
# [['sleeping', 6], ['waking up and starting her morning routine', 1],
|
||||
# ['eating breakfast', 1], ['getting ready for the day', 1],
|
||||
# ['working on her painting', 2], ['taking a break', 1],
|
||||
# ['having lunch', 1], ['working on her painting', 3],
|
||||
# ['taking a break', 2], ['working on her painting', 2],
|
||||
# ['relaxing and watching TV', 1], ['going to bed', 1], ['sleeping', 2]]
|
||||
_n_m1_hourly_compressed = []
|
||||
prev = None
|
||||
prev = None
|
||||
prev_count = 0
|
||||
for i in n_m1_activity:
|
||||
for i in n_m1_activity:
|
||||
if i != prev:
|
||||
prev_count = 1
|
||||
prev_count = 1
|
||||
_n_m1_hourly_compressed += [[i, prev_count]]
|
||||
prev = i
|
||||
elif _n_m1_hourly_compressed:
|
||||
elif _n_m1_hourly_compressed:
|
||||
_n_m1_hourly_compressed[-1][1] += 1
|
||||
|
||||
# Step 2. Expand to min scale (from hour scale)
|
||||
# [['sleeping', 360], ['waking up and starting her morning routine', 60],
|
||||
# [['sleeping', 360], ['waking up and starting her morning routine', 60],
|
||||
# ['eating breakfast', 60],..
|
||||
n_m1_hourly_compressed = []
|
||||
for task, duration in _n_m1_hourly_compressed:
|
||||
n_m1_hourly_compressed += [[task, duration*60]]
|
||||
|
||||
for task, duration in _n_m1_hourly_compressed:
|
||||
n_m1_hourly_compressed += [[task, duration * 60]]
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {n_m1_hourly_compressed}")
|
||||
return n_m1_hourly_compressed
|
||||
|
||||
|
|
|
|||
|
|
@ -96,5 +96,6 @@ class GenIterChatUTT(STAction):
|
|||
"iterative_convo_v1.txt")
|
||||
# original using `ChatGPT_safe_generate_response_OLD`
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
output = self._run_v1(prompt)
|
||||
output = self._run_gpt35_wo_extra_prompt(prompt)
|
||||
logger.info(f"Role: {init_role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
|
|
|||
|
|
@ -31,6 +31,6 @@ class AgentWhisperThoughtAction(STAction):
|
|||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input,
|
||||
"whisper_inner_thought_v1.txt")
|
||||
|
||||
output = self._run_v1(prompt)
|
||||
logger.info(f"Run action: {self.__class__.__name__} with result: {output}")
|
||||
output = self._run_text_davinci(prompt, max_tokens=50)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
|
|
|||
|
|
@ -108,16 +108,16 @@ class NewDecompSchedule(STAction):
|
|||
original_plan = ""
|
||||
for_time = start_time_hour
|
||||
for i in main_act_dur:
|
||||
original_plan += f'{for_time.strftime("%H:%M")} ~ {(for_time + datetime.timedelta(minutes=int(i[1]))).strftime("%H:%M")} -- ' + \
|
||||
i[0]
|
||||
original_plan += f'{for_time.strftime("%H:%M")} ~ ' \
|
||||
f'{(for_time + datetime.timedelta(minutes=int(i[1]))).strftime("%H:%M")} -- ' + i[0]
|
||||
original_plan += "\n"
|
||||
for_time += datetime.timedelta(minutes=int(i[1]))
|
||||
|
||||
new_plan_init = ""
|
||||
for_time = start_time_hour
|
||||
for count, i in enumerate(truncated_act_dur):
|
||||
new_plan_init += f'{for_time.strftime("%H:%M")} ~ {(for_time + datetime.timedelta(minutes=int(i[1]))).strftime("%H:%M")} -- ' + \
|
||||
i[0]
|
||||
new_plan_init += f'{for_time.strftime("%H:%M")} ~ ' \
|
||||
f'{(for_time + datetime.timedelta(minutes=int(i[1]))).strftime("%H:%M")} -- ' + i[0]
|
||||
new_plan_init += "\n"
|
||||
if count < len(truncated_act_dur) - 1:
|
||||
for_time += datetime.timedelta(minutes=int(i[1]))
|
||||
|
|
@ -148,5 +148,6 @@ class NewDecompSchedule(STAction):
|
|||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input,
|
||||
"new_decomp_schedule_v1.txt")
|
||||
self.fail_default_resp = self._func_fail_default_resp(main_act_dur, truncated_act_dur)
|
||||
output = self._run_v1(prompt)
|
||||
output = self._run_text_davinci(prompt, max_tokens=1000)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ class AgentFocusPt(STAction):
|
|||
try:
|
||||
self._func_cleanup(llm_resp, prompt)
|
||||
return True
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str = "") -> str:
|
||||
|
|
@ -30,7 +30,7 @@ class AgentFocusPt(STAction):
|
|||
"""
|
||||
return llm_resp
|
||||
except Exception as exp:
|
||||
logger.error(f"{self.__class__.__name__} with error {exp}")
|
||||
logger.error(f"{self.cls_name} with error {exp}")
|
||||
|
||||
def _func_fail_default_resp(self) -> str:
|
||||
pass
|
||||
|
|
@ -46,10 +46,10 @@ class AgentFocusPt(STAction):
|
|||
|
||||
example_output = '["What should Jane do for lunch", "Does Jane like strawberry", "Who is Jane"]'
|
||||
special_instruction = "Output must be a list of str."
|
||||
output = self._run_v2(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
logger.info(f"Run action: {self.__class__.__name__} with result: {output}")
|
||||
output = self._run_gpt35(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
|
|
@ -63,7 +63,7 @@ class AgentInsightAndGuidance(STAction):
|
|||
try:
|
||||
self._func_cleanup(llm_resp, prompt)
|
||||
return True
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str = "") -> dict:
|
||||
|
|
@ -79,7 +79,7 @@ class AgentInsightAndGuidance(STAction):
|
|||
ret[thought] = evi_raw
|
||||
return ret
|
||||
except Exception as exp:
|
||||
logger.error(f"{self.__class__.__name__} with error {exp}")
|
||||
logger.error(f"{self.cls_name} with error {exp}")
|
||||
|
||||
def _func_fail_default_resp(self) -> str:
|
||||
pass
|
||||
|
|
@ -93,8 +93,8 @@ class AgentInsightAndGuidance(STAction):
|
|||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input,
|
||||
"insight_and_evidence_v1.txt")
|
||||
|
||||
output = self._run_v1(prompt)
|
||||
logger.info(f"Run action: {self.__class__.__name__} with result: {output}")
|
||||
output = self._run_text_davinci(prompt, max_tokens=150)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
|
|
@ -108,7 +108,7 @@ class AgentEventTriple(STAction):
|
|||
llm_resp = self._func_cleanup(llm_resp, prompt="")
|
||||
if len(llm_resp) != 2:
|
||||
return False
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
|
@ -120,7 +120,7 @@ class AgentEventTriple(STAction):
|
|||
return cr[-2:]
|
||||
return cr
|
||||
except Exception as exp:
|
||||
logger.error(f"{self.__class__.__name__} with error {exp}")
|
||||
logger.error(f"{self.cls_name} with error {exp}")
|
||||
|
||||
def _func_fail_default_resp(self) -> str:
|
||||
pass
|
||||
|
|
@ -138,10 +138,9 @@ class AgentEventTriple(STAction):
|
|||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input,
|
||||
"generate_event_triple_v1.txt")
|
||||
|
||||
output = self._run_v1(prompt)
|
||||
output = self._run_text_davinci(prompt, max_tokens=30)
|
||||
output = (role.scratch.name, output[0], output[1])
|
||||
logger.info(f"Role: {role.name} Run action: {self.__class__.__name__} with result: {output}")
|
||||
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
|
|
@ -154,7 +153,7 @@ class AgentEventPoignancy(STAction):
|
|||
try:
|
||||
self._func_cleanup(llm_resp, prompt)
|
||||
return True
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str = "") -> int:
|
||||
|
|
@ -162,7 +161,7 @@ class AgentEventPoignancy(STAction):
|
|||
llm_resp = int(llm_resp.strip())
|
||||
return llm_resp
|
||||
except Exception as exp:
|
||||
logger.error(f"{self.__class__.__name__} with error {exp}")
|
||||
logger.error(f"{self.cls_name} with error {exp}")
|
||||
|
||||
def _func_fail_default_resp(self) -> str:
|
||||
pass
|
||||
|
|
@ -181,11 +180,10 @@ class AgentEventPoignancy(STAction):
|
|||
|
||||
example_output = "5" # ########
|
||||
special_instruction = "The output should ONLY contain ONE integer value on the scale of 1 to 10."
|
||||
output = self._run_v2(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
logger.info(f"Run action: {self.__class__.__name__} with result: {output}")
|
||||
|
||||
output = self._run_gpt35(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
|
|
@ -198,7 +196,7 @@ class AgentChatPoignancy(STAction):
|
|||
try:
|
||||
self._func_cleanup(llm_resp, prompt)
|
||||
return True
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str = "") -> int:
|
||||
|
|
@ -206,7 +204,7 @@ class AgentChatPoignancy(STAction):
|
|||
llm_resp = int(llm_resp.strip())
|
||||
return llm_resp
|
||||
except Exception as exp:
|
||||
logger.error(f"{self.__class__.__name__} with error {exp}")
|
||||
logger.error(f"{self.cls_name} with error {exp}")
|
||||
|
||||
def _func_fail_default_resp(self) -> str:
|
||||
pass
|
||||
|
|
@ -225,11 +223,10 @@ class AgentChatPoignancy(STAction):
|
|||
|
||||
example_output = "5" # ########
|
||||
special_instruction = "The output should ONLY contain ONE integer value on the scale of 1 to 10."
|
||||
output = self._run_v2(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
logger.info(f"Run action: {self.__class__.__name__} with result: {output}")
|
||||
|
||||
output = self._run_gpt35(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
|
|
@ -242,14 +239,14 @@ class AgentPlanThoughtOnConvo(STAction):
|
|||
try:
|
||||
self._func_cleanup(llm_resp, prompt)
|
||||
return True
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str = "") -> str:
|
||||
try:
|
||||
return llm_resp.split('"')[0].strip()
|
||||
except Exception as exp:
|
||||
logger.error(f"{self.__class__.__name__} with error {exp}")
|
||||
logger.error(f"{self.cls_name} with error {exp}")
|
||||
|
||||
def _func_fail_default_resp(self) -> str:
|
||||
pass
|
||||
|
|
@ -266,9 +263,8 @@ class AgentPlanThoughtOnConvo(STAction):
|
|||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input,
|
||||
"planning_thought_on_convo_v1.txt")
|
||||
|
||||
output = self._run_v1(prompt)
|
||||
logger.info(f"Run action: {self.__class__.__name__} with result: {output}")
|
||||
|
||||
output = self._run_text_davinci(prompt, max_tokens=50)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
||||
|
||||
|
|
@ -281,14 +277,14 @@ class AgentMemoryOnConvo(STAction):
|
|||
try:
|
||||
self._func_cleanup(llm_resp, prompt)
|
||||
return True
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str = "") -> str:
|
||||
try:
|
||||
return llm_resp.split('"')[0].strip()
|
||||
except Exception as exp:
|
||||
logger.error(f"{self.__class__.__name__} with error {exp}")
|
||||
logger.error(f"{self.cls_name} with error {exp}")
|
||||
|
||||
def _func_fail_default_resp(self) -> str:
|
||||
pass
|
||||
|
|
@ -304,11 +300,11 @@ class AgentMemoryOnConvo(STAction):
|
|||
prompt_input = create_prompt_input(role, statements)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input,
|
||||
"memo_on_convo_v1.txt")
|
||||
example_output = 'Jane Doe was interesting to talk to.'
|
||||
special_instruction = 'The output should ONLY contain a string that summarizes anything interesting that the agent may have noticed'
|
||||
output = self._run_v2(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
logger.info(f"Run action: {self.__class__.__name__} with result: {output}")
|
||||
|
||||
example_output = "Jane Doe was interesting to talk to."
|
||||
special_instruction = "The output should ONLY contain a string that summarizes anything interesting " \
|
||||
"that the agent may have noticed"
|
||||
output = self._run_gpt35(prompt,
|
||||
example_output,
|
||||
special_instruction)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
|
|
|||
|
|
@ -2,13 +2,14 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# @Desc : StanfordTown Action
|
||||
|
||||
from typing import Union, Optional
|
||||
from typing import Union, Optional, Any
|
||||
from abc import abstractmethod
|
||||
import json
|
||||
|
||||
from metagpt.actions.action import Action
|
||||
from metagpt.schema import Message
|
||||
from metagpt.logs import logger
|
||||
from metagpt.config import CONFIG
|
||||
|
||||
from examples.st_game.utils.const import PROMPTS_DIR
|
||||
|
||||
|
|
@ -20,6 +21,10 @@ class STAction(Action):
|
|||
self.prompt_dir = PROMPTS_DIR
|
||||
self.fail_default_resp = None
|
||||
|
||||
@property
|
||||
def cls_name(self):
|
||||
return self.__class__.__name__
|
||||
|
||||
@abstractmethod
|
||||
def _func_validate(self, llm_resp: str, prompt: str):
|
||||
raise NotImplementedError
|
||||
|
|
@ -54,26 +59,39 @@ class STAction(Action):
|
|||
prompt = prompt.split("<commentblockmarker>###</commentblockmarker>")[1]
|
||||
return prompt.strip()
|
||||
|
||||
def _ask(self, prompt: str, system_msgs: Optional[list[str]] = None) -> str:
|
||||
def _ask(self, prompt: str) -> str:
|
||||
return self.llm.ask(prompt)
|
||||
|
||||
def _run_v1(self, prompt: str, retry: int = 3) :
|
||||
def _ask_nonchat(self, prompt: str) -> str:
|
||||
return self.llm.ask_nonchat(prompt)
|
||||
|
||||
def _run_text_davinci(self, prompt: str, model_name: str = "gpt-3.5-turbo-instruct",
|
||||
max_tokens: int = 50, retry: int = 3):
|
||||
"""
|
||||
same with `gpt_structure.safe_generate_response`
|
||||
default post-preprocess operations of LLM response
|
||||
"""
|
||||
assert model_name in ["gpt-3.5-turbo-instruct", "text-davinci-002", "text-davinci-003"]
|
||||
for idx in range(retry):
|
||||
llm_resp = self._ask(prompt)
|
||||
logger.info(f"_run_v1 llm raw resp: {llm_resp}")
|
||||
tmp_model_name = self.llm.model
|
||||
tmp_max_tokens_rsp = CONFIG.max_tokens_rsp
|
||||
CONFIG.max_tokens_rsp = max_tokens
|
||||
self.llm.model = model_name
|
||||
|
||||
llm_resp = self._ask_nonchat(prompt)
|
||||
|
||||
CONFIG.max_tokens_rsp = tmp_max_tokens_rsp
|
||||
self.llm.model = tmp_model_name
|
||||
logger.info(f"llm _run_text_davinci raw resp: {llm_resp}")
|
||||
if self._func_validate(llm_resp, prompt):
|
||||
return self._func_cleanup(llm_resp, prompt)
|
||||
return self.fail_default_resp
|
||||
|
||||
def _run_v2(self,
|
||||
prompt: str,
|
||||
example_output: str,
|
||||
special_instruction: str,
|
||||
retry: int = 3):
|
||||
def _run_gpt35(self,
|
||||
prompt: str,
|
||||
example_output: str,
|
||||
special_instruction: str,
|
||||
retry: int = 3) -> Union[bool, Any]:
|
||||
""" 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"
|
||||
|
|
@ -83,7 +101,7 @@ class STAction(Action):
|
|||
for idx in range(retry):
|
||||
try:
|
||||
llm_resp = self._ask(prompt)
|
||||
logger.info(f"_run_v2 llm raw resp: {llm_resp}")
|
||||
logger.info(f"llm _run_gpt35 raw resp: {llm_resp}")
|
||||
end_idx = llm_resp.strip().rfind("}") + 1
|
||||
llm_resp = llm_resp[:end_idx]
|
||||
llm_resp = json.loads(llm_resp)["output"]
|
||||
|
|
@ -94,6 +112,19 @@ class STAction(Action):
|
|||
pass
|
||||
return False
|
||||
|
||||
def _run_gpt35_wo_extra_prompt(self,
|
||||
prompt: str,
|
||||
retry: int = 3) -> str:
|
||||
for idx in range(retry):
|
||||
try:
|
||||
llm_resp = self._ask(prompt).strip()
|
||||
logger.info(f"llm _run_gpt35_wo_extra_prompt raw resp: {llm_resp}")
|
||||
if self._func_validate(llm_resp, prompt):
|
||||
return self._func_cleanup(llm_resp, prompt)
|
||||
except Exception as exp:
|
||||
pass
|
||||
return self.fail_default_resp
|
||||
|
||||
def run(self, *args, **kwargs):
|
||||
"""Run action"""
|
||||
raise NotImplementedError("The run method should be implemented in a subclass.")
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# @Desc : summarize the content of agents' conversation
|
||||
|
||||
from metagpt.logs import logger
|
||||
from metagpt.schema import Message
|
||||
|
||||
from examples.st_game.actions.st_action import STAction
|
||||
|
|
@ -44,5 +45,6 @@ class SummarizeConv(STAction):
|
|||
special_instruction = "The output must continue the sentence above by filling in the <fill in> tag. " \
|
||||
"Don't start with 'this is a conversation about...' Just finish the sentence " \
|
||||
"but do not miss any important details (including who are chatting)."
|
||||
output = self._run_v2(prompt, example_output, special_instruction)
|
||||
output = self._run_gpt35(prompt, example_output, special_instruction)
|
||||
logger.info(f"Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
|
|
|||
|
|
@ -20,57 +20,57 @@ class TaskDecomp(STAction):
|
|||
temp = [i.strip() for i in llm_resp.split("\n")]
|
||||
_cr = []
|
||||
cr = []
|
||||
for count, i in enumerate(temp):
|
||||
if count != 0:
|
||||
_cr += [" ".join([j.strip () for j in i.split(" ")][3:])]
|
||||
else:
|
||||
for count, i in enumerate(temp):
|
||||
if count != 0:
|
||||
_cr += [" ".join([j.strip() for j in i.split(" ")][3:])]
|
||||
else:
|
||||
_cr += [i]
|
||||
for count, i in enumerate(_cr):
|
||||
for count, i in enumerate(_cr):
|
||||
k = [j.strip() for j in i.split("(duration in minutes:")]
|
||||
task = k[0]
|
||||
if task[-1] == ".":
|
||||
if task[-1] == ".":
|
||||
task = task[:-1]
|
||||
duration = int(k[1].split(",")[0].strip())
|
||||
cr += [[task, duration]]
|
||||
|
||||
total_expected_min = int(prompt.split("(total duration in minutes")[-1]
|
||||
.split("):")[0].strip())
|
||||
|
||||
# TODO -- now, you need to make sure that this is the same as the sum of
|
||||
# the current action sequence.
|
||||
curr_min_slot = [["dummy", -1],] # (task_name, task_index)
|
||||
for count, i in enumerate(cr):
|
||||
i_task = i[0]
|
||||
.split("):")[0].strip())
|
||||
|
||||
# TODO -- now, you need to make sure that this is the same as the sum of
|
||||
# the current action sequence.
|
||||
curr_min_slot = [["dummy", -1], ] # (task_name, task_index)
|
||||
for count, i in enumerate(cr):
|
||||
i_task = i[0]
|
||||
i_duration = i[1]
|
||||
|
||||
i_duration -= (i_duration % 5)
|
||||
if i_duration > 0:
|
||||
for j in range(i_duration):
|
||||
curr_min_slot += [(i_task, count)]
|
||||
curr_min_slot = curr_min_slot[1:]
|
||||
if i_duration > 0:
|
||||
for j in range(i_duration):
|
||||
curr_min_slot += [(i_task, count)]
|
||||
curr_min_slot = curr_min_slot[1:]
|
||||
|
||||
if len(curr_min_slot) > total_expected_min:
|
||||
if len(curr_min_slot) > total_expected_min:
|
||||
last_task = curr_min_slot[60]
|
||||
for i in range(1, 6):
|
||||
for i in range(1, 6):
|
||||
curr_min_slot[-1 * i] = last_task
|
||||
elif len(curr_min_slot) < total_expected_min:
|
||||
elif len(curr_min_slot) < total_expected_min:
|
||||
last_task = curr_min_slot[-1]
|
||||
for i in range(total_expected_min - len(curr_min_slot)):
|
||||
curr_min_slot += [last_task]
|
||||
|
||||
cr_ret = [["dummy", -1],]
|
||||
for task, task_index in curr_min_slot:
|
||||
if task != cr_ret[-1][0]:
|
||||
cr_ret = [["dummy", -1], ]
|
||||
for task, task_index in curr_min_slot:
|
||||
if task != cr_ret[-1][0]:
|
||||
cr_ret += [[task, 1]]
|
||||
else:
|
||||
else:
|
||||
cr_ret[-1][1] += 1
|
||||
cr = cr_ret[1:]
|
||||
|
||||
return cr
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str) -> bool:
|
||||
# TODO -- this sometimes generates error
|
||||
try:
|
||||
# TODO -- this sometimes generates error
|
||||
try:
|
||||
self._func_cleanup(llm_resp, prompt)
|
||||
except Exception as exp:
|
||||
return False
|
||||
|
|
@ -89,45 +89,46 @@ class TaskDecomp(STAction):
|
|||
def create_prompt_input(role, task, duration):
|
||||
|
||||
"""
|
||||
Today is Saturday June 25. From 00:00 ~ 06:00am, Maeve is
|
||||
planning on sleeping, 06:00 ~ 07:00am, Maeve is
|
||||
planning on waking up and doing her morning routine,
|
||||
and from 07:00am ~08:00am, Maeve is planning on having breakfast.
|
||||
Today is Saturday June 25. From 00:00 ~ 06:00am, Maeve is
|
||||
planning on sleeping, 06:00 ~ 07:00am, Maeve is
|
||||
planning on waking up and doing her morning routine,
|
||||
and from 07:00am ~08:00am, Maeve is planning on having breakfast.
|
||||
"""
|
||||
|
||||
|
||||
curr_f_org_index = role.scratch.get_f_daily_schedule_hourly_org_index()
|
||||
all_indices = []
|
||||
# if curr_f_org_index > 0:
|
||||
# if curr_f_org_index > 0:
|
||||
# all_indices += [curr_f_org_index-1]
|
||||
all_indices += [curr_f_org_index]
|
||||
if curr_f_org_index+1 <= len(role.scratch.f_daily_schedule_hourly_org):
|
||||
all_indices += [curr_f_org_index+1]
|
||||
if curr_f_org_index+2 <= len(role.scratch.f_daily_schedule_hourly_org):
|
||||
all_indices += [curr_f_org_index+2]
|
||||
if curr_f_org_index + 1 <= len(role.scratch.f_daily_schedule_hourly_org):
|
||||
all_indices += [curr_f_org_index + 1]
|
||||
if curr_f_org_index + 2 <= len(role.scratch.f_daily_schedule_hourly_org):
|
||||
all_indices += [curr_f_org_index + 2]
|
||||
|
||||
curr_time_range = ""
|
||||
|
||||
print ("DEBUG")
|
||||
print (role.scratch.f_daily_schedule_hourly_org)
|
||||
print (all_indices)
|
||||
print("DEBUG")
|
||||
print(role.scratch.f_daily_schedule_hourly_org)
|
||||
print(all_indices)
|
||||
|
||||
summ_str = f'Today is {role.scratch.curr_time.strftime("%B %d, %Y")}. '
|
||||
summ_str += f'From '
|
||||
for index in all_indices:
|
||||
print ("index", index)
|
||||
if index < len(role.scratch.f_daily_schedule_hourly_org):
|
||||
for index in all_indices:
|
||||
print("index", index)
|
||||
if index < len(role.scratch.f_daily_schedule_hourly_org):
|
||||
start_min = 0
|
||||
for i in range(index):
|
||||
for i in range(index):
|
||||
start_min += role.scratch.f_daily_schedule_hourly_org[i][1]
|
||||
end_min = start_min + role.scratch.f_daily_schedule_hourly_org[index][1]
|
||||
start_time = (datetime.datetime.strptime("00:00:00", "%H:%M:%S")
|
||||
+ datetime.timedelta(minutes=start_min))
|
||||
end_time = (datetime.datetime.strptime("00:00:00", "%H:%M:%S")
|
||||
+ datetime.timedelta(minutes=end_min))
|
||||
start_time = (datetime.datetime.strptime("00:00:00", "%H:%M:%S")
|
||||
+ datetime.timedelta(minutes=start_min))
|
||||
end_time = (datetime.datetime.strptime("00:00:00", "%H:%M:%S")
|
||||
+ datetime.timedelta(minutes=end_min))
|
||||
start_time_str = start_time.strftime("%H:%M%p")
|
||||
end_time_str = end_time.strftime("%H:%M%p")
|
||||
summ_str += f"{start_time_str} ~ {end_time_str}, {role.name} is planning on {role.scratch.f_daily_schedule_hourly_org[index][0]}, "
|
||||
if curr_f_org_index+1 == index:
|
||||
summ_str += f"{start_time_str} ~ {end_time_str}, {role.name} is planning " \
|
||||
f"on {role.scratch.f_daily_schedule_hourly_org[index][0]}, "
|
||||
if curr_f_org_index + 1 == index:
|
||||
curr_time_range = f'{start_time_str} ~ {end_time_str}'
|
||||
summ_str = summ_str[:-2] + "."
|
||||
|
||||
|
|
@ -149,8 +150,8 @@ class TaskDecomp(STAction):
|
|||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input,
|
||||
"task_decomp_v3.txt")
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
output = self._run_v1(prompt)
|
||||
logger.info(f"Role: {role.name} {self.__class__.__name__} output: {output}")
|
||||
output = self._run_text_davinci(prompt, max_tokens=1000)
|
||||
logger.info(f"Role: {role.name} {self.cls_name} output: {output}")
|
||||
|
||||
fin_output = []
|
||||
time_sum = 0
|
||||
|
|
@ -175,5 +176,5 @@ class TaskDecomp(STAction):
|
|||
for decomp_task, duration in task_decomp:
|
||||
ret += [[f"{task_desc} ({decomp_task})", duration]]
|
||||
output = ret
|
||||
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
|
|
|||
|
|
@ -16,22 +16,22 @@ class WakeUp(STAction):
|
|||
super().__init__(name, context, llm)
|
||||
|
||||
def _func_validate(self, llm_resp: str, prompt: str) -> bool:
|
||||
try:
|
||||
try:
|
||||
self._func_cleanup(llm_resp, prompt="")
|
||||
except:
|
||||
except Exception as exp:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str) -> list:
|
||||
|
||||
def _func_cleanup(self, llm_resp: str, prompt: str) -> list:
|
||||
cr = int(llm_resp.strip().lower().split("am")[0])
|
||||
return cr
|
||||
return cr
|
||||
|
||||
def _func_fail_default_resp(self) -> int:
|
||||
fs = 8
|
||||
return fs
|
||||
|
||||
|
||||
def run(self, role: "STRole"):
|
||||
def create_prompt_input(role):
|
||||
def create_prompt_input(role):
|
||||
prompt_input = [role.scratch.get_str_iss(),
|
||||
role.scratch.get_str_lifestyle(),
|
||||
role.scratch.get_str_firstname()]
|
||||
|
|
@ -40,5 +40,6 @@ class WakeUp(STAction):
|
|||
prompt_input = create_prompt_input(role)
|
||||
prompt = self.generate_prompt_with_tmpl_filename(prompt_input, "wake_up_hour_v1.txt")
|
||||
self.fail_default_resp = self._func_fail_default_resp()
|
||||
output = self._run_v1(prompt)
|
||||
return output
|
||||
output = self._run_text_davinci(prompt, max_tokens=5)
|
||||
logger.info(f"Role: {role.name} Action: {self.cls_name} output: {output}")
|
||||
return output
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue