MetaGPT
MetaGPT copied to clipboard
geekan
A new role cannot receive message from previous role
I create a program and have 4 roles: Researcher, PM, Developer, QA Message flow: Researcher -> PM -> Developer -> QA
When no QA roles/actions, the program works well and can generate the python code. When i added a new role QA and find it still ends at the Developer stage. And the role QA cannot received the message from Developer.
That's what i'm confused.
import fire
from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
class InterviewReview(Action):
PROMPT_TEMPLATE: str = """
## background
Suppose you are the researcher of Design team, you will need to review the interview text with customers and highlight the most important part of text.
## interview text
{instruction}
## output
please make a short summary and show your highlights.
"""
name: str = "SimpleWriteCode"
async def run(self, instruction: str):
prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
logger.info("LLM-InterviewReview-input: {}", prompt)
rsp = await self._aask(prompt)
return rsp
class Researcher(Role):
name: str = "Re-1"
profile: str = "Researcher"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([UserRequirement])
self.set_actions([InterviewReview])
class StoryArrangement(Action):
PROMPT_TEMPLATE: str = """
## background
Suppose you are the project manager of Dev team, you will need to review the summary from researcher and then create new features for backend developers.
## summary from researcher
{context}
## output
please create features, and each one has a title and a brief introduction
"""
name: str = "StoryArrangement"
async def run(self, context: str):
prompt = self.PROMPT_TEMPLATE.format(context=context)
logger.info("LLM-StoryArrangement-input: {}", prompt)
rsp = await self._aask(prompt)
return rsp
class ProjectManager(Role):
name: str = "PM-1"
profile: str = "PM"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([StoryArrangement])
self._watch([InterviewReview]) # feel free to try this too
async def _act(self) -> Message:
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
todo = self.rc.todo
# context = self.get_memories(k=1)[0].content # use the most recent memory as context
context = self.get_memories() # use all memories as context
code_text = await todo.run(context[-1]) # specify arguments
msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
return msg
class Coding(Action):
PROMPT_TEMPLATE: str = """
## background
Suppose you are the coding developer of Dev team, you will need to write simple python code for meet the requirements from ProjectManager.
## requirements from ProjectManager
{context}
## output
please only write python code
"""
name: str = "Coding"
async def run(self, context: str):
prompt = self.PROMPT_TEMPLATE.format(context=context)
logger.info("LLM-Coding-input: {}", prompt)
rsp = await self._aask(prompt)
return rsp
class Developer(Role):
name: str = "Squad"
profile: str = "Developer"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([Coding])
self._watch([StoryArrangement]) # feel free to try this too
async def _act(self) -> Message:
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
todo = self.rc.todo
# context = self.get_memories(k=1)[0].content # use the most recent memory as context
context = self.get_memories() # use all memories as context
code_text = await todo.run(context[-1]) # specify arguments
msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
# logger.info("Message: ", msg.sent_from, msg.sent_to, msg.role, msg.cause_by);
return msg
class WriteTest(Action):
PROMPT_TEMPLATE: str = """
## background
Suppose you are the QA tester of Dev team, you will need to write simple unit test for python code from developers.
## python code from developers
{context}
## output
please write python unit test
"""
name: str = "UnitTest"
async def run(self, context: str):
prompt = self.PROMPT_TEMPLATE.format(context=context)
logger.info("LLM-WriteTest-input: {}", prompt)
rsp = await self._aask(prompt)
return rsp
class QATester(Role):
name: str = "QA"
profile: str = "QATester"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([WriteTest])
self._watch([Coding]) # feel free to try this too
async def _act(self) -> Message:
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
todo = self.rc.todo
# context = self.get_memories(k=1)[0].content # use the most recent memory as context
context = self.get_memories() # use all memories as context
code_text = await todo.run(context[-1]) # specify arguments
msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
return msg
async def main(
idea: str = """
Here is a customer interview record t:
Salesperson: Hello, [Customer Name] , thank you for taking the time to meet with me today to discuss the ABC product. We are always working to improve the ABC product to better meet your needs. Today I would like to learn more about your suggestions for new features and optimizations for the ABC product.
Customer: Hello. Overall, I am very satisfied with the ABC product. It is powerful, easy to use, and has helped us improve our work efficiency. However, I also have some suggestions that I hope will further enhance the usability of the product.
Salesperson: Great, please tell me more.
Customer: First, I would like to add an interface to get the current time to ABC. This way, I can easily embed time information into my workflow.
Salesperson: That's a great suggestion. We are already planning to develop this feature and expect to release it in the next version.
Customer: That's great. In addition, I would also like to add an interface for a simple calculator for addition, subtraction, multiplication, and division. This way, I can perform simple calculations directly in the ABC product without having to switch to other software.
Salesperson: That's a very practical suggestion. We are also considering developing this feature.
Customer: Thank you. I believe that adding these two features will make the ABC product even more user-friendly.
Salesperson: Thank you for your valuable feedback. We will definitely take it seriously and implement it as soon as possible."
""",
investment: float = 3.0,
n_round: int = 1,
add_human: bool = False,
):
logger.info(idea)
team = Team()
team.hire(
[
Researcher(),
ProjectManager(),
Developer(),
# TestOne()
QATester()
]
)
team.invest(investment=investment)
team.run_project(idea)
await team.run()
if __name__ == "__main__":
fire.Fire(main)
I apologize for the delayed response.
The n_round parameter simulates the number of rounds, defaulting to 3. In your case, with only 3 rounds, the program stops after the Developer finishes coding, before the QA role receives messages.
You can resolve it by modifying your code as follows:
async def main(
...
n_round: int = 4, # Change the value of `n_round` to 4
...
):
...
await team.run(n_round=n_round) # Change here
...
- Round 1: Researcher InterviewReview
- Round 2: ProjectManager StoryArrangement
- Round 3: Developer Coding
- Round 4: QATester WriteTest
Due to the lack of updates or replies by the user for a long time, we will close it. Please reopen it if necessary.
