apunkt/MetaGPT
1
0
Fork 0
mirror of https://github.com/FoundationAgents/MetaGPT.git synced 2026-07-17 16:41:05 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

add data analyst, explore agent controlling own plan by thinking

Browse source
This commit is contained in:
yzlin 2024-04-29 18:47:29 +08:00
parent f134126a90
commit 5447399ecb
7 changed files with 220 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

40
metagpt/strategy/thinking_command.py
View file

@ -2,6 +2,10 @@ from enum import Enum
from pydantic import BaseModel
from metagpt.environment.mgx.mgx_env import MGXEnv
from metagpt.roles import Role
from metagpt.schema import Message, Task
class CommandDef(BaseModel):
name: str
@ -66,3 +70,39 @@ def prepare_command_prompt(commands: list[Command]) -> str:
for i, command in enumerate(commands):
command_prompt += f"{i+1}. {command.value.signature}:\n{command.value.desc}\n\n"
return command_prompt
async def run_env_command(role: Role, cmd):
assert isinstance(role.rc.env, MGXEnv), "TeamLeader should only be used in an MGXEnv"
if cmd["command_name"] == Command.PUBLISH_MESSAGE.cmd_name:
role.publish_message(Message(**cmd["args"]))
if cmd["command_name"] == Command.ASK_HUMAN.cmd_name:
role.rc.working_memory.add(Message(content=cmd["args"]["question"], role="assistant"))
human_rsp = await role.rc.env.ask_human(sent_from=role, **cmd["args"])
role.rc.working_memory.add(Message(content=human_rsp, role="user"))
elif cmd["command_name"] == Command.REPLY_TO_HUMAN.cmd_name:
# TODO: consider if the message should go into memory
await role.rc.env.reply_to_human(sent_from=role, **cmd["args"])
def run_internal_command(role: Role, cmd):
if cmd["command_name"] == Command.APPEND_TASK.cmd_name:
role.planner.plan.append_task(Task(**cmd["args"]))
elif cmd["command_name"] == Command.RESET_TASK.cmd_name:
role.planner.plan.reset_task(**cmd["args"])
elif cmd["command_name"] == Command.REPLACE_TASK.cmd_name:
role.planner.plan.replace_task(Task(**cmd["args"]))
elif cmd["command_name"] == Command.FINISH_CURRENT_TASK.cmd_name:
role.planner.plan.current_task.update_task_result(task_result=role.task_result)
role.planner.plan.finish_current_task()
role.rc.working_memory.clear()
async def run_commands(role: Role, cmds):
print(*cmds, sep="\n")
for cmd in cmds:
await run_env_command(role, cmd)
run_internal_command(role, cmd)
if role.planner.plan.is_plan_finished():
role._set_state(-1)