apunkt/MetaGPT
1
0
Fork 0
mirror of https://github.com/FoundationAgents/MetaGPT.git synced 2026-06-29 15:59:42 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor: @cause_by.setter

Browse source
This commit is contained in:
莘权 马 2023-11-04 16:46:32 +08:00
parent 1febf168e7
commit d9775037b6
21 changed files with 73 additions and 123 deletions
Download patch file Download diff file Expand all files Collapse all files

3
metagpt/actions/action.py
View file

@ -17,10 +17,9 @@ from metagpt.llm import LLM
from metagpt.logs import logger
from metagpt.utils.common import OutputParser
from metagpt.utils.custom_decoder import CustomDecoder
from metagpt.utils.named import Named
class Action(ABC, Named):
class Action(ABC):
def __init__(self, name: str = "", context=None, llm: LLM = None):
self.name: str = name
if llm is None:

2
metagpt/actions/write_code.py
View file

@ -58,7 +58,7 @@ class WriteCode(Action):
if self._is_invalid(filename):
return
message_filter = {WriteDesign.get_class_name()}
message_filter = {WriteDesign}
design = [i for i in context if i.is_recipient(message_filter)][0]
ws_name = CodeParser.parse_str(block="Python package name", text=design.content)

12
metagpt/roles/engineer.py
View file

@ -102,7 +102,7 @@ class Engineer(Role):
return CodeParser.parse_str(block="Python package name", text=system_design_msg.content)
def get_workspace(self) -> Path:
msg = self._rc.memory.get_by_action(WriteDesign.get_class_name())[-1]
msg = self._rc.memory.get_by_action(WriteDesign)[-1]
if not msg:
return WORKSPACE_ROOT / "src"
workspace = self.parse_workspace(msg)
@ -130,7 +130,7 @@ class Engineer(Role):
todo_coros = []
for todo in self.todos:
todo_coro = WriteCode().run(
context=self._rc.memory.get_by_actions([WriteTasks.get_class_name(), WriteDesign.get_class_name()]),
context=self._rc.memory.get_by_actions([WriteTasks, WriteDesign]),
filename=todo,
)
todo_coros.append(todo_coro)
@ -185,7 +185,7 @@ class Engineer(Role):
TODO: The goal is not to need it. After clear task decomposition, based on the design idea, you should be able to write a single file without needing other codes. If you can't, it means you need a clearer definition. This is the key to writing longer code.
"""
context = []
msg_filters = [WriteDesign.get_class_name(), WriteTasks.get_class_name(), WriteCode.get_class_name()]
msg_filters = [WriteDesign, WriteTasks, WriteCode]
msg = self._rc.memory.get_by_actions(msg_filters)
for m in msg:
context.append(m.content)
@ -201,7 +201,7 @@ class Engineer(Role):
logger.error("code review failed!", e)
pass
file_path = self.write_file(todo, code)
msg = Message(content=code, role=self.profile, cause_by=WriteCode.get_class_name())
msg = Message(content=code, role=self.profile, cause_by=WriteCode)
self._rc.memory.add(msg)
self.publish_message(msg)
@ -231,7 +231,7 @@ class Engineer(Role):
return ret
# Parse task lists
message_filter = {WriteTasks.get_class_name()}
message_filter = {WriteTasks}
for message in self._rc.news:
if not message.is_recipient(message_filter):
continue
@ -241,7 +241,7 @@ class Engineer(Role):
async def _think(self) -> None:
# In asynchronous scenarios, first check if the required messages are ready.
filters = {WriteTasks.get_class_name()}
filters = {WriteTasks}
msgs = self._rc.memory.get_by_actions(filters)
if not msgs:
self._rc.todo = None

20
metagpt/roles/qa_engineer.py
View file

@ -50,7 +50,7 @@ class QaEngineer(Role):
return CodeParser.parse_str(block="Python package name", text=system_design_msg.content)
def get_workspace(self, return_proj_dir=True) -> Path:
msg = self._rc.memory.get_by_action(WriteDesign.get_class_name())[-1]
msg = self._rc.memory.get_by_action(WriteDesign)[-1]
if not msg:
return WORKSPACE_ROOT / "src"
workspace = self.parse_workspace(msg)
@ -99,7 +99,7 @@ class QaEngineer(Role):
msg = Message(
content=str(file_info),
role=self.profile,
cause_by=WriteTest.get_class_name(),
cause_by=WriteTest,
tx_from=self.profile,
tx_to=self.profile,
)
@ -133,9 +133,7 @@ class QaEngineer(Role):
recipient = parse_recipient(result_msg) # the recipient might be Engineer or myself
content = str(file_info) + FILENAME_CODE_SEP + result_msg
msg = Message(
content=content, role=self.profile, cause_by=RunCode.get_class_name(), tx_from=self.profile, tx_to=recipient
)
msg = Message(content=content, role=self.profile, cause_by=RunCode, tx_from=self.profile, tx_to=recipient)
self.publish_message(msg)
async def _debug_error(self, msg):
@ -147,7 +145,7 @@ class QaEngineer(Role):
msg = Message(
content=file_info,
role=self.profile,
cause_by=DebugError.get_class_name(),
cause_by=DebugError,
tx_from=self.profile,
tx_to=recipient,
)
@ -165,14 +163,14 @@ class QaEngineer(Role):
result_msg = Message(
content=f"Exceeding {self.test_round_allowed} rounds of tests, skip (writing code counts as a round, too)",
role=self.profile,
cause_by=WriteTest.get_class_name(),
cause_by=WriteTest,
tx_from=self.profile,
)
return result_msg
code_filters = {WriteCode.get_class_name(), WriteCodeReview.get_class_name()}
test_filters = {WriteTest.get_class_name(), DebugError.get_class_name()}
run_filters = {RunCode.get_class_name()}
code_filters = {WriteCode, WriteCodeReview}
test_filters = {WriteTest, DebugError}
run_filters = {RunCode}
for msg in self._rc.news:
# Decide what to do based on observed msg type, currently defined by human,
# might potentially be moved to _think, that is, let the agent decides for itself
@ -189,7 +187,7 @@ class QaEngineer(Role):
result_msg = Message(
content=f"Round {self.test_round} of tests done",
role=self.profile,
cause_by=WriteTest.get_class_name(),
cause_by=WriteTest,
tx_from=self.profile,
)
return result_msg

7
metagpt/roles/role.py
View file

@ -31,7 +31,6 @@ from metagpt.logs import logger
from metagpt.memory import LongTermMemory, Memory
from metagpt.schema import Message, MessageQueue
from metagpt.utils.common import get_class_name, get_object_name
from metagpt.utils.named import Named
PREFIX_TEMPLATE = """You are a {profile}, named {name}, your goal is {goal}, and the constraint is {constraints}. """
@ -107,7 +106,7 @@ class RoleContext(BaseModel):
return self.memory.get()
class Role(Named):
class Role:
"""Role/Agent"""
def __init__(self, name="", profile="", goal="", constraints="", desc=""):
@ -174,10 +173,10 @@ class Role(Named):
return self._rc.watch
return {
self.name,
self.get_object_name(),
get_object_name(self),
self.profile,
f"{self.name}({self.profile})",
f"{self.name}({self.get_object_name()})",
f"{self.name}({get_object_name(self)})",
}
def _get_prefix(self):

19
metagpt/schema.py
View file

@ -24,7 +24,7 @@ from metagpt.const import (
MESSAGE_ROUTE_TO,
)
from metagpt.logs import logger
from metagpt.utils.common import get_class_name, get_object_name
from metagpt.utils.common import any_to_str
class RawMessage(TypedDict):
@ -129,11 +129,12 @@ class Message(BaseModel):
if k in attribute_names:
continue
if k == MESSAGE_ROUTE_FROM:
self.set_from(v)
self.set_from(any_to_str(v))
continue
if k == MESSAGE_ROUTE_CAUSE_BY:
self.meta_info[k] = v
if k == MESSAGE_ROUTE_TO or k == MESSAGE_ROUTE_CAUSE_BY:
self.set_cause_by(v)
continue
if k == MESSAGE_ROUTE_TO:
self.add_to(v)
continue
self.meta_info[k] = v
@ -161,18 +162,14 @@ class Message(BaseModel):
if key == MESSAGE_ROUTE_CAUSE_BY:
self.set_cause_by(val)
return
if key == MESSAGE_ROUTE_FROM:
self.set_from(any_to_str(val))
super().__setattr__(key, val)
def set_cause_by(self, val):
"""Update the value of `cause_by` in the `meta_info` and `routes` attributes."""
old_value = self.get_meta(MESSAGE_ROUTE_CAUSE_BY)
new_value = None
if isinstance(val, str):
new_value = val
elif not callable(val):
new_value = get_object_name(val)
else:
new_value = get_class_name(val)
new_value = any_to_str(val)
self.set_meta(MESSAGE_ROUTE_CAUSE_BY, new_value)
self.route.replace(old_value, new_value)

7
metagpt/software_company.py
View file

@ -18,10 +18,9 @@ from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.utils.common import NoMoneyException
from metagpt.utils.named import Named
class SoftwareCompany(BaseModel, Named):
class SoftwareCompany(BaseModel):
"""
Software Company: Possesses a team, SOP (Standard Operating Procedures), and a platform for instant messaging,
dedicated to writing executable code.
@ -55,8 +54,8 @@ class SoftwareCompany(BaseModel, Named):
Message(
role="BOSS",
content=idea,
cause_by=BossRequirement.get_class_name(),
tx_from=SoftwareCompany.get_class_name(),
cause_by=BossRequirement,
tx_from=SoftwareCompany,
)
)

10
metagpt/utils/common.py
View file

@ -315,3 +315,13 @@ def get_object_name(obj) -> str:
"""Return class name of the object"""
cls = type(obj)
return f"{cls.__module__}.{cls.__name__}"
def any_to_str(val) -> str:
"""Return the class name or the class name of the object, or 'val' if it's a string type."""
if isinstance(val, str):
return val
if not callable(val):
return get_object_name(val)
return get_class_name(val)

21
metagpt/utils/named.py
View file

@ -1,21 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2023/11/1
@Author : mashenquan
@File : named.py
"""
class Named:
"""A base class with functions for converting classes to names and objects to class names."""
@classmethod
def get_class_name(cls):
"""Return class name"""
return f"{cls.__module__}.{cls.__name__}"
def get_object_name(self):
"""Return class name of the object"""
cls = type(self)
return f"{cls.__module__}.{cls.__name__}"