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

Merge branch 'da_change' into 'mgx_ops'

Browse source 
rm redundant

See merge request pub/MetaGPT!342
This commit is contained in:
林义章 2024-08-20 13:09:22 +00:00
commit 8844bced88
4 changed files with 0 additions and 444 deletions
Download patch file Download diff file Expand all files Collapse all files

141
metagpt/actions/di/detect_intent.py
View file

@ -1,141 +0,0 @@
from __future__ import annotations
import asyncio
import re
from enum import Enum
from typing import Tuple
from pydantic import BaseModel
from metagpt.actions import Action
from metagpt.schema import Message
class SOPItemDef(BaseModel):
"""
Represents an item in a Standard Operating Procedure (SOP).
Attributes:
name (str): name of the SOP item.
description (str): The description or title of the SOP.
sop (List[str]): The steps or details of the SOP.
"""
name: str
description: str
sop: list[str] = []
class SOPItem(Enum):
SOFTWARE_DEVELOPMENT = SOPItemDef(
name="software development",
description="Software development intention including developing or building software, games, app, websites, etc. EXCLUDING bug fixes, report any issues, environment setup, terminal operations, and pip install.",
sop=[
"Writes a PRD based on software requirements.",
"Writes a system design to the project repository, based on the PRD of the project. Write high-level system design instead of the actual code.",
"Writes a project plan to the project repository, based on the design of the project.",
"Writes code to implement designed features according to the project plan and adds them to the project repository.",
"Stage and commit changes for the project repository using Git.",
],
)
# FIX_BUGS = SOPItemDef(
# name="fix bugs",
# description="Fix bugs in a given project.",
# sop=[
# "Fix bugs in the project repository.",
# "Stage and commit changes for the project repository using Git.",
# ],
# )
# FORMAT_REPO = SOPItemDef(
# name="format repo",
# description="download repository from git and format the project to MetaGPT project",
# sop=[
# "Imports a project from a Git website and formats it to MetaGPT project format to enable incremental appending requirements.",
# "Stage and commit changes for the project repository using Git.",
# ],
# )
WEB_OPERATION = SOPItemDef(
name="web operation",
description="web browsing, scraping, imitation and other interaction with the web",
)
OTHER = SOPItemDef(
name="other",
description="Other intentions that do not fall into the above categories, including data science, data analysis, machine learning, deep learning and text-to-image etc.",
)
@property
def type_name(self):
return self.value.name
@classmethod
def get_type(cls, type_name):
for member in cls:
if member.type_name == type_name:
return member.value
return None
DETECT_PROMPT = """
# User Requirement
{user_requirement}
# Intentions
{intentions}
# Task
Classify user requirement into one type of the above intentions, output the index of the intention directly.
Intention index:
"""
REQ_WITH_SOP = """
{user_requirement}
## Knowledge
To meet user requirements, the following standard operating procedure(SOP) must be used:
{sop}
"""
class DetectIntent(Action):
async def run(self, with_message: Message, **kwargs) -> Tuple[str, str]:
user_requirement = with_message.content
mappings = {i + 1: si for i, si in enumerate(SOPItem)}
intentions = "\n".join([f"{i + 1}. {si.type_name}: {si.value.description}" for i, si in enumerate(SOPItem)])
prompt = DETECT_PROMPT.format(user_requirement=user_requirement, intentions=intentions)
rsp = await self._aask(prompt)
match = re.search(r"\d+", rsp)
index = len(SOPItem) + 1 # 1-based
if match:
index = int(match.group()) # 1-based
sop = mappings[index].value.sop if index in mappings else None
sop_type = mappings[index].type_name if index in mappings else SOPItem.OTHER.type_name
req_with_sop = (
REQ_WITH_SOP.format(
user_requirement=user_requirement, sop="\n".join([f"{i + 1}. {v}" for i, v in enumerate(sop)])
)
if sop
else user_requirement
)
return req_with_sop, sop_type
async def main():
# Example usage of the DetectIntent action
user_requirements = [
"Develop a 2048 game.",
"Run data analysis on sklearn wine dataset",
"帮我把pip的源设置成：https://pypi.tuna.tsinghua.edu.cn/simple",
"This is a website url does not require login: https://demosc.chinaz.net/Files/DownLoad//moban/202404/moban7767 please write a similar web page,developed in vue language, The package.json dependency must be generated",
"I would like to imitate the website available at https://demosc.chinaz.net/Files/DownLoad//moban/202404/moban7767. Could you please browse through it?",
]
detect_intent = DetectIntent()
for user_requirement in user_requirements:
req_with_sop, sop_type = await detect_intent.run(Message(role="user", content=user_requirement))
print(req_with_sop)
print(f"Detected SOP Type: {sop_type}")
if __name__ == "__main__":
asyncio.run(main())

68
metagpt/roles/di/mgx.py
View file

@ -1,68 +0,0 @@
# -*- coding: utf-8 -*-
# @Author : stellahong (stellahong@fuzhi.ai)
# @Desc :
import asyncio
from typing import Dict
from metagpt.actions.di.detect_intent import DetectIntent, SOPItem
from metagpt.logs import logger
from metagpt.roles.di.data_interpreter import DataInterpreter
from metagpt.schema import Message
from metagpt.tools.tool_recommend import BM25ToolRecommender
class MGX(DataInterpreter):
use_intent: bool = True
intents: Dict = {}
async def _detect_intent(self, user_msg: Message) -> str:
todo = DetectIntent(context=self.context)
request_with_sop, sop_type = await todo.run(user_msg)
logger.info(f"{sop_type} {request_with_sop}")
if sop_type == SOPItem.SOFTWARE_DEVELOPMENT.type_name:
self.tool_recommender = BM25ToolRecommender(tools=["software development"])
else:
self.tool_recommender = BM25ToolRecommender(tools=["<all>"])
return request_with_sop
async def _plan_and_act(self) -> Message:
"""first plan, then execute an action sequence, i.e. _think (of a plan) -> _act -> _act -> ... Use llm to come up with the plan dynamically."""
# create initial plan and update it until confirmation
goal = self.rc.memory.get()[-1].content # retrieve latest user requirement
if self.use_intent: # add mode
user_message = Message(content=goal, role="user")
goal = await self._detect_intent(user_message)
logger.info(f"Goal is {goal}")
await self.planner.update_plan(goal=goal)
# take on tasks until all finished
while self.planner.current_task:
task = self.planner.current_task
logger.info(f"ready to take on task {task}")
# take on current task
task_result = await self._act_on_task(task)
# process the result, such as reviewing, confirming, plan updating
await self.planner.process_task_result(task_result)
rsp = self.planner.get_useful_memories()[0] # return the completed plan as a response
self.rc.memory.add(rsp) # add to persistent memory
return rsp
if __name__ == "__main__":
test_requirements = [
"design a game using Gym (an open source Python library), including a graphical interface and interactive gameplay"
# "Run data analysis on sklearn Wine recognition dataset, include a plot, and train a model to predict wine class (20% as validation), and show validation accuracy"
# "获取https://www.stats.gov.cn/sj/sjjd/202307/t20230718_1941322.html的内容，并返回上半年cpi的增长或下降幅度"
]
for requirement in test_requirements:
mgx = MGX()
rsp = asyncio.run(mgx.run(requirement))