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

Merge branch 'dev_ldy' into 'dev'

Browse source 
Dev ldy

See merge request agents/data_agents_opt!10
This commit is contained in:
林义章 2023-11-30 06:43:46 +00:00
commit 725dac2525
5 changed files with 375 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

158
metagpt/actions/write_analysis_code.py
View file

@ -4,34 +4,52 @@
@Author : orange-crow
@File : write_code_v2.py
"""
import json
from typing import Dict, List, Union
from metagpt.actions import Action
from metagpt.prompts.ml_engineer import (
TOOL_RECOMMENDATION_PROMPT,
SELECT_FUNCTION_TOOLS,
CODE_GENERATOR_WITH_TOOLS,
TOO_ORGANIZATION_PROMPT,
ML_SPECIFIC_PROMPT,
ML_MODULE_MAP,
TOOL_OUTPUT_DESC,
)
from metagpt.schema import Message, Plan
from metagpt.tools.functions import registry
from metagpt.utils.common import create_func_config
class BaseWriteAnalysisCode(Action):
async def run(self, context: List[Message], plan: Plan = None, task_guide: str = "") -> str:
async def run(
self, context: List[Message], plan: Plan = None, task_guide: str = ""
) -> str:
"""Run of a code writing action, used in data analysis or modeling
Args:
context (List[Message]): Action output history, source action denoted by Message.cause_by
plan (Plan, optional): Overall plan. Defaults to None.
task_guide (str, optional): suggested step breakdown for the current task. Defaults to "".
Returns:
str: The code string.
"""
class WriteCodeByGenerate(BaseWriteAnalysisCode):
"""Write code fully by generation"""
def __init__(self, name: str = "", context=None, llm=None) -> str:
super().__init__(name, context, llm)
def process_msg(self, prompt: Union[str, List[Dict], Message, List[Message]], system_msg: str = None):
# Reference: https://github.com/KillianLucas/open-interpreter/blob/v0.1.4/interpreter/system_message.txt
default_system_msg = """You are Code Interpreter, a world-class programmer that can complete any goal by executing code. Strictly follow the plan and generate code step by step. Each step of the code will be executed on the user's machine, and the user will provide the code execution results to you.**Notice: The code for the next step depends on the code for the previous step. Reuse existing code directly. Use !pip install to install missing packages.**"""
def process_msg(
self,
prompt: Union[str, List[Dict], Message, List[Message]],
system_msg: str = None,
):
default_system_msg = """You are Open Interpreter, a world-class programmer that can complete any goal by executing code. Strictly follow the plan and generate code step by step. Each step of the code will be executed on the user's machine, and the user will provide the code execution results to you.**Notice: The code for the next step depends on the code for the previous step.**"""
# 全部转成list
if not isinstance(prompt, list):
prompt = [prompt]
@ -40,24 +58,38 @@ class WriteCodeByGenerate(BaseWriteAnalysisCode):
messages = []
for p in prompt:
if isinstance(p, str):
messages.append({'role': 'user', 'content': p})
messages.append({"role": "user", "content": p})
elif isinstance(p, dict):
messages.append(p)
elif isinstance(p, Message):
if isinstance(p.content, str):
messages.append(p.to_dict())
elif isinstance(p.content, dict) and 'code' in p.content:
messages.append(p.content['code'])
elif isinstance(p.content, dict) and "code" in p.content:
messages.append(p.content["code"])
# 添加默认的提示词
if default_system_msg not in messages[0]['content'] and messages[0]['role'] != 'system':
messages.insert(0, {'role': 'system', 'content': default_system_msg})
elif default_system_msg not in messages[0]['content'] and messages[0]['role'] == 'system':
messages[0] = {'role': 'system', 'content': messages[0]['content']+default_system_msg}
if (
default_system_msg not in messages[0]["content"]
and messages[0]["role"] != "system"
):
messages.insert(0, {"role": "system", "content": default_system_msg})
elif (
default_system_msg not in messages[0]["content"]
and messages[0]["role"] == "system"
):
messages[0] = {
"role": "system",
"content": messages[0]["content"] + default_system_msg,
}
return messages
async def run(
self, context: [List[Message]], plan: Plan = None, task_guide: str = "", system_msg: str = None, **kwargs
self,
context: [List[Message]],
plan: Plan = None,
task_guide: str = "",
system_msg: str = None,
**kwargs,
) -> str:
prompt = self.process_msg(context, system_msg)
code_content = await self.llm.aask_code(prompt, **kwargs)
@ -67,5 +99,99 @@ class WriteCodeByGenerate(BaseWriteAnalysisCode):
class WriteCodeWithTools(BaseWriteAnalysisCode):
"""Write code with help of local available tools. Choose tools first, then generate code to use the tools"""
async def run(self, context: List[Message], plan: Plan = None, task_guide: str = "") -> str:
return "print('abc')"
@staticmethod
def _parse_recommend_tools(module: str, recommend_tools: list) -> str:
"""
Converts recommended tools to a JSON string and checks tool availability in the registry.
Args:
module (str): The module name for querying tools in the registry.
recommend_tools (list): A list of lists of recommended tools for each step.
Returns:
str: A JSON string with available tools and their schemas for each step.
"""
valid_tools = {}
available_tools = registry.get_all_by_module(module).keys()
for index, tools in enumerate(recommend_tools):
key = f"Step {index + 1}"
tools = [tool for tool in tools if tool in available_tools]
valid_tools[key] = registry.get_schemas(module, tools)
return json.dumps(valid_tools)
async def _tool_recommendation(
self, task: str, data_desc: str, code_steps: str, available_tools: list
) -> list:
"""
Recommend tools for each step of the specified task
Args:
task (str): the task description
data_desc (str): the description of the dataset for the task
code_steps (str): the code steps to generate the full code for the task
available_tools (list): the available tools for the task
Returns:
list: recommended tools for each step of the specified task
"""
prompt = TOOL_RECOMMENDATION_PROMPT.format(
task=task,
data_desc=data_desc,
code_steps=code_steps,
available_tools=available_tools,
)
tool_config = create_func_config(SELECT_FUNCTION_TOOLS)
rsp = await self.llm.aask_code(prompt, **tool_config)
recommend_tools = rsp["recommend_tools"]
return recommend_tools
async def run(
self,
context: List[Message],
plan: Plan = None,
task_guide: str = "",
data_desc: str = "",
) -> str:
task_type = plan.current_task.task_type
task = plan.current_task.instruction
available_tools = registry.get_all_schema_by_module(task_type)
available_tools = [
{k: tool[k] for k in ["name", "description"] if k in tool}
for tool in available_tools
]
task_guide = "\n".join(
[f"Step {step.strip()}" for step in task_guide.split("\n")]
)
recommend_tools = await self._tool_recommendation(
task, task_guide, available_tools
)
recommend_tools = self._parse_recommend_tools(task_type, recommend_tools)
special_prompt = ML_SPECIFIC_PROMPT.get(task_type, "")
module_name = ML_MODULE_MAP[task_type]
output_desc = TOOL_OUTPUT_DESC.get(task_type, "")
all_tasks = ""
completed_code = ""
for i, task in enumerate(plan.tasks):
stats = "DONE" if task.is_finished else "TODO"
all_tasks += f"Subtask {task.task_id}: {task.instruction}({stats})\n"
for task in plan.tasks:
if task.code:
completed_code += task.code + "\n"
prompt = TOO_ORGANIZATION_PROMPT.format(
all_tasks=all_tasks,
completed_code=completed_code,
data_desc=data_desc,
special_prompt=special_prompt,
code_steps=task_guide,
module_name=module_name,
output_desc=output_desc,
available_tools=recommend_tools,
)
tool_config = create_func_config(CODE_GENERATOR_WITH_TOOLS)
rsp = await self.llm.aask_code(prompt, **tool_config)
return rsp["code"]

170
metagpt/prompts/ml_engineer.py Normal file
View file

@ -0,0 +1,170 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2023/11/24 15:43
# @Author : lidanyang
# @File : ml_engineer
# @Desc :
TOOL_RECOMMENDATION_PROMPT = """
## Comprehensive Task Description:
{task}
## Dataset Description:
Details about the dataset for the project:
{data_desc}
This task is divided into several steps, and you need to select the most suitable tools for each step. A tool means a function that can be used to help you solve the task.
## Detailed Code Steps for the Task:
{code_steps}
## List of Available Tools:
{available_tools}
## Tool Selection and Instructions:
- For each code step listed above, choose up to five tools that are most likely to be useful in solving the task.
- If you believe that no tools are suitable for a step, indicate with an empty list.
- Only list the names of the tools, not the full schema of each tool.
- The result should only contain tool names that are in the list of available tools.
- The result list should be in the same order as the code steps.
"""
SELECT_FUNCTION_TOOLS = {
"name": "select_function_tools",
"description": "Given code steps to generate full code for a task, select suitable tools for each step by order.",
"parameters": {
"type": "object",
"properties": {
"recommend_tools": {
"type": "array",
"description": "List of tool names for each code step. Empty list if no tool is suitable.",
"items": {
"type": "array",
"items": {
"type": "string",
},
},
},
},
"required": ["recommend_tools"],
},
}
CODE_GENERATOR_WITH_TOOLS = {
"name": "add_subtask_code",
"description": "Add new code of current subtask to the end of an active Jupyter notebook.",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "The code to be added.",
},
},
"required": ["code"],
},
}
TOO_ORGANIZATION_PROMPT = """
As a senior data scientist, your role involves developing code for a specific sub-task within a larger project. This project is divided into several sub-tasks, which may either be new challenges or extensions of previous work.
## Sub-tasks Overview
Here's a list of all the sub-tasks, indicating their current status (DONE or TODO). Your responsibility is the first TODO task on this list.
{all_tasks}
## Historical Code (Previously Done Sub-tasks):
This code, already executed in the Jupyter notebook, is critical for understanding the background and foundation for your current task.
```python
{completed_code}
```
## Dataset Description:
Details about the dataset for the project:
{data_desc}
## Current Task Notion:
{special_prompt}
## Code Steps for Your Sub-task:
Follow these steps to complete your current TODO task. You may use external Python functions or write custom code as needed. Ensure your code is self-contained.
{code_steps}
When you call a function, you should import the function from `{module_name}` first, e.g.:
```python
from metagpt.tools.functions.libs.feature_engineering import fill_missing_value
```
## Available Functions for Each Step:
Each function is described in JSON format, including the function name and parameters. {output_desc}
{available_tools}
## Your Output Format:
Generate the complete code for every step, listing any used function tools at the beginning of the step:
```python
# Step 1
# Tools used: [function names or 'none']
<your code for this step, without any comments>
# Step 2
# Tools used: [function names or 'none']
<your code for this step, without any comments>
# Continue with additional steps, following the same format...
```end
*** Important Rules ***
- Use only the tools designated for each code step.
- Your output should only include code for the current sub-task. Don't repeat historical code.
- Only mention functions in comments if used in the code.
- Ensure the output new code is executable in the current Jupyter notebook environment, with all historical code executed.
"""
DATA_PREPROCESS_PROMPT = """
In data preprocessing, closely monitor each column's data type. Apply suitable methods for various types (numerical, categorical, datetime, textual, etc.) to ensure the pandas.DataFrame is correctly formatted.
Additionally, ensure that the columns being processed must be the ones that actually exist in the dataset.
"""
FEATURE_ENGINEERING_PROMPT = """
When performing feature engineering, please adhere to the following principles:
- For specific user requests (such as removing a feature, creating a new feature based on existing data), directly generate the corresponding code.
- In cases of unclear user requirements, write feature engineering code that you believe will most improve model performance. This may include feature transformation, combination, aggregation, etc., with a limit of five features at a time.
- Ensure that the feature you're working with is indeed present in the dataset and consider the data type (numerical, categorical, etc.) and application scenario (classification, regression tasks, etc.).
- Importantly, provide detailed comments explaining the purpose of each feature and how it might enhance model performance, especially when the features are generated based on semantic understanding without clear user directives.
"""
CLASSIFICATION_MODEL_PROMPT = """
"""
REGRESSION_MODEL_PROMPT = """
"""
DATA_PREPROCESS_OUTPUT_DESC = "Please note that all functions uniformly output a processed pandas.DataFrame, facilitating seamless integration into the broader workflow."
FEATURE_ENGINEERING_OUTPUT_DESC = "Please note that all functions uniformly output updated pandas.DataFrame with feature engineering applied."
CLASSIFICATION_MODEL_OUTPUT_DESC = ""
REGRESSION_MODEL_OUTPUT_DESC = ""
ML_SPECIFIC_PROMPT = {
"data_preprocess": DATA_PREPROCESS_PROMPT,
"feature_engineering": FEATURE_ENGINEERING_PROMPT,
"classification_model": CLASSIFICATION_MODEL_PROMPT,
"regression_model": REGRESSION_MODEL_PROMPT,
}
TOOL_OUTPUT_DESC = {
"data_preprocess": DATA_PREPROCESS_OUTPUT_DESC,
"feature_engineering": FEATURE_ENGINEERING_OUTPUT_DESC,
"classification_model": CLASSIFICATION_MODEL_OUTPUT_DESC,
"regression_model": REGRESSION_MODEL_OUTPUT_DESC,
}
ML_MODULE_MAP = {
"data_preprocess": "metagpt.tools.functions.libs.machine_learning.data_preprocess",
"feature_engineering": "metagpt.tools.functions.libs.machine_learning.feature_engineering",
"classification_model": "metagpt.tools.functions.libs.machine_learning.ml_model",
"regression_model": "metagpt.tools.functions.libs.machine_learning.ml_model",
}

74
metagpt/roles/ml_engineer.py
View file

@ -21,10 +21,11 @@ STRUCTURAL_CONTEXT = """
{current_task}
"""
def truncate(result: str, keep_len: int = 1000) -> str:
desc = """I truncated the result to only keep the last 1000 characters\n"""
if result.startswith(desc):
result = result[-len(desc):]
result = result[-len(desc) :]
if len(result) > keep_len:
result = result[-keep_len:]
@ -35,10 +36,16 @@ def truncate(result: str, keep_len: int = 1000) -> str:
class AskReview(Action):
async def run(self, context: List[Message], plan: Plan = None):
logger.info("Current overall plan:")
logger.info("\n".join([f"{task.task_id}: {task.instruction}, is_finished: {task.is_finished}" for task in plan.tasks]))
logger.info(
"\n".join(
[
f"{task.task_id}: {task.instruction}, is_finished: {task.is_finished}"
for task in plan.tasks
]
)
)
logger.info("most recent context:")
# prompt = "\n".join(
@ -46,21 +53,26 @@ class AskReview(Action):
# )
prompt = ""
latest_action = context[-1].cause_by.__name__ if context[-1].cause_by else ""
prompt += f"\nPlease review output from {latest_action}:\n" \
"If you want to change a task in the plan, say 'change task task_id, ... (things to change)'\n" \
prompt += (
f"\nPlease review output from {latest_action}:\n"
"If you want to change a task in the plan, say 'change task task_id, ... (things to change)'\n"
"If you confirm the output and wish to continue with the current process, type CONFIRM:\n"
)
rsp = input(prompt)
confirmed = "confirm" in rsp.lower()
return rsp, confirmed
class WriteTaskGuide(Action):
class WriteTaskGuide(Action):
async def run(self, task_instruction: str, data_desc: str = "") -> str:
return ""
class MLEngineer(Role):
def __init__(self, name="ABC", profile="MLEngineer", goal="", auto_run: bool = False):
def __init__(
self, name="ABC", profile="MLEngineer", goal="", auto_run: bool = False
):
super().__init__(name=name, profile=profile, goal=goal)
self._set_react_mode(react_mode="plan_and_act")
self.plan = Plan(goal=goal)
@ -70,7 +82,6 @@ class MLEngineer(Role):
self.auto_run = auto_run
async def _plan_and_act(self):
# create initial plan and update until confirmation
await self._update_plan()
@ -96,8 +107,11 @@ class MLEngineer(Role):
await self._update_plan()
async def _write_and_exec_code(self, max_retry: int = 3):
task_guide = await WriteTaskGuide().run(self.plan.current_task.instruction) if self.use_task_guide else ""
task_guide = (
await WriteTaskGuide().run(self.plan.current_task.instruction)
if self.use_task_guide
else ""
)
counter = 0
success = False
@ -109,22 +123,29 @@ class MLEngineer(Role):
# print("*" * 10)
# breakpoint()
if not self.use_tools:
if not self.use_tools or self.plan.current_task.task_type == "":
# code = "print('abc')"
code = await WriteCodeByGenerate().run(context=context, plan=self.plan, task_guide=task_guide)
code = await WriteCodeByGenerate().run(
context=context, plan=self.plan, task_guide=task_guide
)
cause_by = WriteCodeByGenerate
else:
code = await WriteCodeWithTools().run(context=context, plan=self.plan, task_guide=task_guide)
code = await WriteCodeWithTools().run(
context=context, plan=self.plan, task_guide=task_guide, data_desc=""
)
cause_by = WriteCodeWithTools
self.working_memory.add(Message(content=code, role="assistant", cause_by=cause_by))
self.working_memory.add(
Message(content=code, role="assistant", cause_by=cause_by)
)
result, success = await self.execute_code.run(code)
# truncated the result
print(truncate(result))
# print(result)
self.working_memory.add(Message(content=result, role="user", cause_by=ExecutePyCode))
self.working_memory.add(
Message(content=result, role="user", cause_by=ExecutePyCode)
)
if code.startswith("!pip"):
success = False
@ -138,9 +159,13 @@ class MLEngineer(Role):
async def _ask_review(self):
if not self.auto_run:
context = self.get_useful_memories()
review, confirmed = await AskReview().run(context=context[-5:], plan=self.plan)
review, confirmed = await AskReview().run(
context=context[-5:], plan=self.plan
)
if review.lower() not in ("confirm", "y", "yes"):
self._rc.memory.add(Message(content=review, role="user", cause_by=AskReview))
self._rc.memory.add(
Message(content=review, role="user", cause_by=AskReview)
)
return confirmed
return True
@ -149,7 +174,9 @@ class MLEngineer(Role):
while not plan_confirmed:
context = self.get_useful_memories()
rsp = await WritePlan().run(context, max_tasks=max_tasks)
self.working_memory.add(Message(content=rsp, role="assistant", cause_by=WritePlan))
self.working_memory.add(
Message(content=rsp, role="assistant", cause_by=WritePlan)
)
plan_confirmed = await self._ask_review()
tasks = WritePlan.rsp_to_tasks(rsp)
@ -160,9 +187,13 @@ class MLEngineer(Role):
"""find useful memories only to reduce context length and improve performance"""
user_requirement = self.plan.goal
tasks = json.dumps([task.dict() for task in self.plan.tasks], indent=4, ensure_ascii=False)
tasks = json.dumps(
[task.dict() for task in self.plan.tasks], indent=4, ensure_ascii=False
)
current_task = self.plan.current_task.json() if self.plan.current_task else {}
context = STRUCTURAL_CONTEXT.format(user_requirement=user_requirement, tasks=tasks, current_task=current_task)
context = STRUCTURAL_CONTEXT.format(
user_requirement=user_requirement, tasks=tasks, current_task=current_task
)
context_msg = [Message(content=context, role="user")]
return context_msg + self.working_memory.get()
@ -171,6 +202,7 @@ class MLEngineer(Role):
def working_memory(self):
return self._rc.memory
if __name__ == "__main__":
# requirement = "create a normal distribution and visualize it"
requirement = "run some analysis on iris dataset"

1
metagpt/tools/functions/__init__.py
View file

@ -6,4 +6,3 @@
# @Desc :
from metagpt.tools.functions.register.register import registry
import metagpt.tools.functions.libs.feature_engineering
print(registry.functions)

10
metagpt/utils/common.py
View file

@ -305,3 +305,13 @@ def parse_recipient(text):
pattern = r"## Send To:\s*([A-Za-z]+)\s*?" # hard code for now
recipient = re.search(pattern, text)
return recipient.group(1) if recipient else ""
def create_func_config(func_schema: dict) -> dict:
"""Create new function call config"""
tools = [{"type": "function", "function": func_schema}]
tool_choice = {"type": "function", "function": {"name": func_schema["name"]}}
return {
"tools": tools,
"tool_choice": tool_choice,
}