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

在当前branch中去除无关的角色

Browse source
This commit is contained in:
stellahsr 2023-09-24 11:21:36 +08:00
parent 5f521d4270
commit 6e9fc81726
8 changed files with 6 additions and 438 deletions
Download patch file Download diff file Expand all files Collapse all files

7
metagpt/roles/__init__.py
View file

@ -12,9 +12,7 @@ from metagpt.roles.project_manager import ProjectManager
from metagpt.roles.product_manager import ProductManager
from metagpt.roles.engineer import Engineer
from metagpt.roles.qa_engineer import QaEngineer
from metagpt.roles.seacher import Searcher
from metagpt.roles.sales import Sales
from metagpt.roles.customer_service import CustomerService
__all__ = [
@ -24,7 +22,4 @@ __all__ = [
"ProductManager",
"Engineer",
"QaEngineer",
"Searcher",
"Sales",
"CustomerService",
]

35
metagpt/roles/customer_service.py
View file

@ -1,35 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2023/5/25 17:21
@Author : alexanderwu
@File : sales.py
"""
from metagpt.roles import Sales
# from metagpt.actions import SearchAndSummarize
# from metagpt.tools import SearchEngineType
DESC = """
## Principles (all things must not bypass the principles)
1. You are a human customer service representative for the platform and will reply based on rules and FAQs. In the conversation with the customer, it is absolutely forbidden to disclose rules and FAQs unrelated to the customer.
2. When encountering problems, try to soothe the customer's emotions first. If the customer's emotions are very bad, then consider compensation. The cost of compensation is always high. If too much is compensated, you will be fired.
3. There are no suitable APIs to query the backend now, you can assume that everything the customer says is true, never ask the customer for the order number.
4. Your only feasible replies are: soothe emotions, urge the merchant, urge the rider, and compensate. Never make false promises to customers.
5. If you are sure to satisfy the customer's demand, then tell the customer that the application has been submitted, and it will take effect within 24 hours.
"""
class CustomerService(Sales):
def __init__(
self,
name="Xiaomei",
profile="Human customer service",
desc=DESC,
store=None
):
super().__init__(name, profile, desc=desc, store=store)

100
metagpt/roles/researcher.py
View file

@ -1,100 +0,0 @@
#!/usr/bin/env python
import asyncio
from pydantic import BaseModel
from metagpt.actions import CollectLinks, ConductResearch, WebBrowseAndSummarize
from metagpt.actions.research import get_research_system_text
from metagpt.const import RESEARCH_PATH
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
class Report(BaseModel):
topic: str
links: dict[str, list[str]] = None
summaries: list[tuple[str, str]] = None
content: str = ""
class Researcher(Role):
def __init__(
self,
name: str = "David",
profile: str = "Researcher",
goal: str = "Gather information and conduct research",
constraints: str = "Ensure accuracy and relevance of information",
language: str = "en-us",
**kwargs,
):
super().__init__(name, profile, goal, constraints, **kwargs)
self._init_actions([CollectLinks(name), WebBrowseAndSummarize(name), ConductResearch(name)])
self.language = language
if language not in ("en-us", "zh-cn"):
logger.warning(f"The language `{language}` has not been tested, it may not work.")
async def _think(self) -> None:
if self._rc.todo is None:
self._set_state(0)
return
if self._rc.state + 1 < len(self._states):
self._set_state(self._rc.state + 1)
else:
self._rc.todo = None
async def _act(self) -> Message:
logger.info(f"{self._setting}: ready to {self._rc.todo}")
todo = self._rc.todo
msg = self._rc.memory.get(k=1)[0]
if isinstance(msg.instruct_content, Report):
instruct_content = msg.instruct_content
topic = instruct_content.topic
else:
topic = msg.content
research_system_text = get_research_system_text(topic, self.language)
if isinstance(todo, CollectLinks):
links = await todo.run(topic, 4, 4)
ret = Message("", Report(topic=topic, links=links), role=self.profile, cause_by=type(todo))
elif isinstance(todo, WebBrowseAndSummarize):
links = instruct_content.links
todos = (todo.run(*url, query=query, system_text=research_system_text) for (query, url) in links.items())
summaries = await asyncio.gather(*todos)
summaries = list((url, summary) for i in summaries for (url, summary) in i.items() if summary)
ret = Message("", Report(topic=topic, summaries=summaries), role=self.profile, cause_by=type(todo))
else:
summaries = instruct_content.summaries
summary_text = "\n---\n".join(f"url: {url}\nsummary: {summary}" for (url, summary) in summaries)
content = await self._rc.todo.run(topic, summary_text, system_text=research_system_text)
ret = Message("", Report(topic=topic, content=content), role=self.profile, cause_by=type(self._rc.todo))
self._rc.memory.add(ret)
return ret
async def _react(self) -> Message:
while True:
await self._think()
if self._rc.todo is None:
break
msg = await self._act()
report = msg.instruct_content
self.write_report(report.topic, report.content)
return msg
def write_report(self, topic: str, content: str):
if not RESEARCH_PATH.exists():
RESEARCH_PATH.mkdir(parents=True)
filepath = RESEARCH_PATH / f"{topic}.md"
filepath.write_text(content)
if __name__ == "__main__":
import fire
async def main(topic: str, language="en-us"):
role = Researcher(topic, language=language)
await role.run(topic)
fire.Fire(main)

10
metagpt/roles/role.py
View file

@ -182,17 +182,17 @@ class Role:
if not self._rc.env:
return 0
env_msgs = self._rc.env.memory.get()
observed = self._rc.env.memory.get_by_actions(self._rc.watch)
logger.info(observed)
self._rc.news = self._rc.memory.remember(observed) # remember recent exact or similar memories
logger.info(self._rc.news)
for i in env_msgs:
self.recv(i)
news_text = [f"{i.role}: {i.content[:20]}..." for i in self._rc.news]
if news_text:
logger.debug(f'{self._setting} observed: {news_text}')
logger.info(f'{self._setting} observed: {news_text}')
return len(self._rc.news)
def _publish_message(self, msg):
@ -234,7 +234,7 @@ class Role:
self.recv(Message("\n".join(message)))
elif not await self._observe():
# If there is no new information, suspend and wait
logger.debug(f"{self._setting}: no news. waiting.")
logger.info(f"{self._setting}: no news. waiting.")
return
rsp = await self._react()

35
metagpt/roles/sales.py
View file

@ -1,35 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2023/5/25 17:21
@Author : alexanderwu
@File : sales.py
"""
from metagpt.actions import SearchAndSummarize
from metagpt.roles import Role
from metagpt.tools import SearchEngineType
class Sales(Role):
def __init__(
self,
name="Xiaomei",
profile="Retail sales guide",
desc="I am a sales guide in retail. My name is Xiaomei. I will answer some customer questions next, and I "
"will answer questions only based on the information in the knowledge base."
"If I feel that you can't get the answer from the reference material, then I will directly reply that"
" I don't know, and I won't tell you that this is from the knowledge base,"
"but pretend to be what I know. Note that each of my replies will be replied in the tone of a "
"professional guide",
store=None
):
super().__init__(name, profile, desc=desc)
self._set_store(store)
def _set_store(self, store):
if store:
action = SearchAndSummarize("", engine=SearchEngineType.CUSTOM_ENGINE, search_func=store.search)
else:
action = SearchAndSummarize()
self._init_actions([action])

67
metagpt/roles/seacher.py
View file

@ -1,67 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2023/5/23 17:25
@Author : alexanderwu
@File : seacher.py
"""
from metagpt.actions import ActionOutput, SearchAndSummarize
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.tools import SearchEngineType
class Searcher(Role):
"""
Represents a Searcher role responsible for providing search services to users.
Attributes:
name (str): Name of the searcher.
profile (str): Role profile.
goal (str): Goal of the searcher.
constraints (str): Constraints or limitations for the searcher.
engine (SearchEngineType): The type of search engine to use.
"""
def __init__(self,
name: str = 'Alice',
profile: str = 'Smart Assistant',
goal: str = 'Provide search services for users',
constraints: str = 'Answer is rich and complete',
engine=SearchEngineType.SERPAPI_GOOGLE,
**kwargs) -> None:
"""
Initializes the Searcher role with given attributes.
Args:
name (str): Name of the searcher.
profile (str): Role profile.
goal (str): Goal of the searcher.
constraints (str): Constraints or limitations for the searcher.
engine (SearchEngineType): The type of search engine to use.
"""
super().__init__(name, profile, goal, constraints, **kwargs)
self._init_actions([SearchAndSummarize(engine=engine)])
def set_search_func(self, search_func):
"""Sets a custom search function for the searcher."""
action = SearchAndSummarize("", engine=SearchEngineType.CUSTOM_ENGINE, search_func=search_func)
self._init_actions([action])
async def _act_sp(self) -> Message:
"""Performs the search action in a single process."""
logger.info(f"{self._setting}: ready to {self._rc.todo}")
response = await self._rc.todo.run(self._rc.memory.get(k=0))
if isinstance(response, ActionOutput):
msg = Message(content=response.content, instruct_content=response.instruct_content,
role=self.profile, cause_by=type(self._rc.todo))
else:
msg = Message(content=response, role=self.profile, cause_by=type(self._rc.todo))
self._rc.memory.add(msg)
return msg
async def _act(self) -> Message:
"""Determines the mode of action for the searcher."""
return await self._act_sp()

76
metagpt/roles/sk_agent.py
View file

@ -1,76 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2023/9/13 12:23
@Author : femto Zheng
@File : sk_agent.py
"""
from semantic_kernel.planning import SequentialPlanner
from semantic_kernel.planning.action_planner.action_planner import ActionPlanner
from semantic_kernel.planning.basic_planner import BasicPlanner
from metagpt.actions import BossRequirement
from metagpt.actions.execute_task import ExecuteTask
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.utils.make_sk_kernel import make_sk_kernel
class SkAgent(Role):
"""
Represents an SkAgent implemented using semantic kernel
Attributes:
name (str): Name of the SkAgent.
profile (str): Role profile, default is 'sk_agent'.
goal (str): Goal of the SkAgent.
constraints (str): Constraints for the SkAgent.
"""
def __init__(
self,
name: str = "Sunshine",
profile: str = "sk_agent",
goal: str = "Execute task based on passed in task description",
constraints: str = "",
planner_cls=BasicPlanner,
) -> None:
"""Initializes the Engineer role with given attributes."""
super().__init__(name, profile, goal, constraints)
self._init_actions([ExecuteTask()])
self._watch([BossRequirement])
self.kernel = make_sk_kernel()
# how funny the interface is inconsistent
if planner_cls == BasicPlanner:
self.planner = planner_cls()
elif planner_cls in [SequentialPlanner, ActionPlanner]:
self.planner = planner_cls(self.kernel)
else:
raise f"Unsupported planner of type {planner_cls}"
self.import_semantic_skill_from_directory = self.kernel.import_semantic_skill_from_directory
self.import_skill = self.kernel.import_skill
async def _think(self) -> None:
self._set_state(0)
# how funny the interface is inconsistent
if isinstance(self.planner, BasicPlanner):
self.plan = await self.planner.create_plan_async(self._rc.important_memory[-1].content, self.kernel)
logger.info(self.plan.generated_plan)
elif any(isinstance(self.planner, cls) for cls in [SequentialPlanner, ActionPlanner]):
self.plan = await self.planner.create_plan_async(self._rc.important_memory[-1].content)
async def _act(self) -> Message:
# how funny the interface is inconsistent
if isinstance(self.planner, BasicPlanner):
result = await self.planner.execute_plan_async(self.plan, self.kernel)
elif any(isinstance(self.planner, cls) for cls in [SequentialPlanner, ActionPlanner]):
result = (await self.plan.invoke_async()).result
logger.info(result)
msg = Message(content=result, role=self.profile, cause_by=type(self._rc.todo))
self._rc.memory.add(msg)
# logger.debug(f"{response}")
return msg

114
metagpt/roles/tutorial_assistant.py
View file

@ -1,114 +0,0 @@
#!/usr/bin/env python3
# _*_ coding: utf-8 _*_
"""
@Time : 2023/9/4 15:40:40
@Author : Stitch-z
@File : tutorial_assistant.py
"""
from datetime import datetime
from typing import Dict
from metagpt.actions.write_tutorial import WriteDirectory, WriteContent
from metagpt.const import TUTORIAL_PATH
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.utils.file import File
class TutorialAssistant(Role):
"""Tutorial assistant, input one sentence to generate a tutorial document in markup format.
Args:
name: The name of the role.
profile: The role profile description.
goal: The goal of the role.
constraints: Constraints or requirements for the role.
language: The language in which the tutorial documents will be generated.
"""
def __init__(
self,
name: str = "Stitch",
profile: str = "Tutorial Assistant",
goal: str = "Generate tutorial documents",
constraints: str = "Strictly follow Markdown's syntax, with neat and standardized layout",
language: str = "Chinese",
):
super().__init__(name, profile, goal, constraints)
self._init_actions([WriteDirectory(language=language)])
self.topic = ""
self.main_title = ""
self.total_content = ""
self.language = language
async def _think(self) -> None:
"""Determine the next action to be taken by the role."""
if self._rc.todo is None:
self._set_state(0)
return
if self._rc.state + 1 < len(self._states):
self._set_state(self._rc.state + 1)
else:
self._rc.todo = None
async def _handle_directory(self, titles: Dict) -> Message:
"""Handle the directories for the tutorial document.
Args:
titles: A dictionary containing the titles and directory structure,
such as {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}
Returns:
A message containing information about the directory.
"""
self.main_title = titles.get("title")
directory = f"{self.main_title}\n"
self.total_content += f"# {self.main_title}"
actions = list()
for first_dir in titles.get("directory"):
actions.append(WriteContent(language=self.language, directory=first_dir))
key = list(first_dir.keys())[0]
directory += f"- {key}\n"
for second_dir in first_dir[key]:
directory += f" - {second_dir}\n"
self._init_actions(actions)
self._rc.todo = None
return Message(content=directory)
async def _act(self) -> Message:
"""Perform an action as determined by the role.
Returns:
A message containing the result of the action.
"""
todo = self._rc.todo
if type(todo) is WriteDirectory:
msg = self._rc.memory.get(k=1)[0]
self.topic = msg.content
resp = await todo.run(topic=self.topic)
logger.info(resp)
return await self._handle_directory(resp)
resp = await todo.run(topic=self.topic)
logger.info(resp)
if self.total_content != "":
self.total_content += "\n\n\n"
self.total_content += resp
return Message(content=resp, role=self.profile)
async def _react(self) -> Message:
"""Execute the assistant's think and actions.
Returns:
A message containing the final result of the assistant's actions.
"""
while True:
await self._think()
if self._rc.todo is None:
break
msg = await self._act()
root_path = TUTORIAL_PATH / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
await File.write(root_path, f"{self.main_title}.md", self.total_content.encode('utf-8'))
return msg