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

rm repetitive tool config for writing code; rm WriteCodeWithoutTools

Browse source
This commit is contained in:
yzlin 2024-02-26 21:10:16 +08:00
parent 0eda2e6581
commit 219d361ca6
11 changed files with 77 additions and 196 deletions
Download patch file Download diff file Expand all files Collapse all files

3
metagpt/actions/__init__.py
View file

@ -23,7 +23,7 @@ from metagpt.actions.write_prd import WritePRD
from metagpt.actions.write_prd_review import WritePRDReview
from metagpt.actions.write_test import WriteTest
from metagpt.actions.mi.execute_nb_code import ExecuteNbCode
from metagpt.actions.mi.write_analysis_code import WriteCodeWithoutTools, WriteCodeWithTools
from metagpt.actions.mi.write_analysis_code import WriteCodeWithTools
from metagpt.actions.mi.write_plan import WritePlan
@ -46,7 +46,6 @@ class ActionType(Enum):
WEB_BROWSE_AND_SUMMARIZE = WebBrowseAndSummarize
CONDUCT_RESEARCH = ConductResearch
EXECUTE_NB_CODE = ExecuteNbCode
WRITE_CODE_WITHOUT_TOOLS = WriteCodeWithoutTools
WRITE_CODE_WITH_TOOLS = WriteCodeWithTools
WRITE_PLAN = WritePlan

4
metagpt/actions/mi/debug_code.py
View file

@ -1,6 +1,6 @@
from __future__ import annotations
from metagpt.actions.mi.write_analysis_code import BaseWriteAnalysisCode
from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.schema import Message
from metagpt.utils.common import create_func_call_config
@ -72,7 +72,7 @@ CODE_REFLECTION = {
}
class DebugCode(BaseWriteAnalysisCode):
class DebugCode(Action):
async def run(
self,
context: list[Message] = None,

41
metagpt/actions/mi/ml_action.py
View file

@ -5,14 +5,13 @@ from typing import Tuple
from metagpt.actions import Action
from metagpt.actions.mi.write_analysis_code import WriteCodeWithTools
from metagpt.prompts.mi.ml_action import (
ML_GENERATE_CODE_PROMPT,
ML_TOOL_USAGE_PROMPT,
PRINT_DATA_COLUMNS,
ML_PROMPT,
UPDATE_DATA_COLUMNS,
USE_NO_TOOLS_EXAMPLE,
USE_TOOLS_EXAMPLE,
)
from metagpt.prompts.mi.write_analysis_code import CODE_GENERATOR_WITH_TOOLS
from metagpt.schema import Message, Plan
from metagpt.utils.common import create_func_call_config, remove_comments
from metagpt.utils.common import remove_comments
class WriteCodeWithToolsML(WriteCodeWithTools):
@ -32,26 +31,17 @@ class WriteCodeWithToolsML(WriteCodeWithTools):
code_context = "\n\n".join(code_context)
# prepare prompt depending on tool availability & LLM call
if tool_schemas:
prompt = ML_TOOL_USAGE_PROMPT.format(
user_requirement=plan.goal,
history_code=code_context,
current_task=plan.current_task.instruction,
column_info=column_info,
tool_type_usage_prompt=tool_type_usage_prompt,
tool_schemas=tool_schemas,
)
prompt = ML_PROMPT.format(
user_requirement=plan.goal,
history_code=code_context,
current_task=plan.current_task.instruction,
column_info=column_info,
tool_type_usage_prompt=tool_type_usage_prompt,
tool_schemas=tool_schemas,
examples=USE_TOOLS_EXAMPLE if tool_schemas else USE_NO_TOOLS_EXAMPLE,
)
else:
prompt = ML_GENERATE_CODE_PROMPT.format(
user_requirement=plan.goal,
history_code=code_context,
current_task=plan.current_task.instruction,
column_info=column_info,
tool_type_usage_prompt=tool_type_usage_prompt,
)
tool_config = create_func_call_config(CODE_GENERATOR_WITH_TOOLS)
rsp = await self.llm.aask_code(prompt, **tool_config)
rsp = await self.llm.aask_code(prompt, language="python")
# Extra output to be used for potential debugging
context = [Message(content=prompt, role="user")]
@ -65,6 +55,5 @@ class UpdateDataColumns(Action):
code_context = [remove_comments(task.code) for task in finished_tasks]
code_context = "\n\n".join(code_context)
prompt = UPDATE_DATA_COLUMNS.format(history_code=code_context)
tool_config = create_func_call_config(PRINT_DATA_COLUMNS)
rsp = await self.llm.aask_code(prompt, **tool_config)
rsp = await self.llm.aask_code(prompt, language="python")
return rsp

62
metagpt/actions/mi/write_analysis_code.py
View file

@ -11,7 +11,6 @@ from typing import Tuple
from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.prompts.mi.write_analysis_code import (
CODE_GENERATOR_WITH_TOOLS,
SELECT_FUNCTION_TOOLS,
TOOL_RECOMMENDATION_PROMPT,
TOOL_USAGE_PROMPT,
@ -22,43 +21,19 @@ from metagpt.tools.tool_registry import validate_tool_names
from metagpt.utils.common import create_func_call_config
class BaseWriteAnalysisCode(Action):
DEFAULT_SYSTEM_MSG: str = """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. Must reuse variables in the lastest other code directly, dont creat it again, it is very import for you. Use !pip install in a standalone block to install missing packages.Usually the libraries you need are already installed.Dont check if packages already imported.**""" # prompt reference: https://github.com/KillianLucas/open-interpreter/blob/v0.1.4/interpreter/system_message.txt
# REUSE_CODE_INSTRUCTION = """ATTENTION: DONT include codes from previous tasks in your current code block, include new codes only, DONT repeat codes!"""
class WriteCodeWithTools(Action):
"""Write code with help of local available tools. Choose tools first, then generate code to use the tools"""
def insert_system_message(self, context: list[Message], system_msg: str = None):
use_tools: bool = True
# selected tools to choose from, listed by their names. An empty list means selection from all tools.
selected_tools: list[str] = []
DEFAULT_SYSTEM_MSG: str = """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. Must reuse variables in the lastest other code directly, dont creat it again, it is very import for you. Use !pip install in a standalone block to install missing packages.Usually the libraries you need are already installed.Dont check if packages already imported.**""" # prompt reference: https://github.com/KillianLucas/open-interpreter/blob/v0.1.4/interpreter/system_message.txt
def _insert_system_message(self, context: list[Message], system_msg: str = None):
system_msg = system_msg or self.DEFAULT_SYSTEM_MSG
context.insert(0, SystemMessage(content=system_msg)) if context[0].role != "system" else None
return context
async def run(self, context: list[Message], plan: Plan = None) -> dict:
"""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.
Returns:
dict: code result in the format of {"code": "print('hello world')", "language": "python"}
"""
raise NotImplementedError
class WriteCodeWithoutTools(BaseWriteAnalysisCode):
"""Ask LLM to generate codes purely by itself without local user-defined tools"""
async def run(self, context: list[Message], plan: Plan = None, system_msg: str = None, **kwargs) -> dict:
messages = self.insert_system_message(context, system_msg)
rsp = await self.llm.aask_code(messages, **kwargs)
return rsp
class WriteCodeWithTools(BaseWriteAnalysisCode):
"""Write code with help of local available tools. Choose tools first, then generate code to use the tools"""
# selected tools to choose from, listed by their names. An empty list means selection from all tools.
selected_tools: list[str] = []
def _get_tools_by_type(self, tool_type: str) -> dict:
"""
Retreive tools by tool type from registry, but filtered by pre-selected tool list
@ -138,18 +113,19 @@ class WriteCodeWithTools(BaseWriteAnalysisCode):
plan: Plan,
**kwargs,
) -> str:
# prepare tool schemas and tool-type-specific instruction
tool_schemas, tool_type_usage_prompt = await self._prepare_tools(plan=plan)
if self.use_tools:
# prepare tool schemas and tool-type-specific instruction
tool_schemas, tool_type_usage_prompt = await self._prepare_tools(plan=plan)
# form a complete tool usage instruction and include it as a message in context
tools_instruction = TOOL_USAGE_PROMPT.format(
tool_schemas=tool_schemas, tool_type_usage_prompt=tool_type_usage_prompt
)
context.append(Message(content=tools_instruction, role="user"))
# form a complete tool usage instruction and include it as a message in context
tools_instruction = TOOL_USAGE_PROMPT.format(
tool_schemas=tool_schemas, tool_type_usage_prompt=tool_type_usage_prompt
)
context.append(Message(content=tools_instruction, role="user"))
# prepare prompt & LLM call
prompt = self.insert_system_message(context)
tool_config = create_func_call_config(CODE_GENERATOR_WITH_TOOLS)
rsp = await self.llm.aask_code(prompt, **tool_config)
prompt = self._insert_system_message(context)
rsp = await self.llm.aask_code(prompt, language="python")
return rsp

36
metagpt/actions/mi/write_plan.py
View file

@ -12,19 +12,17 @@ from typing import Tuple
from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.prompts.mi.write_analysis_code import (
ASSIGN_TASK_TYPE_CONFIG,
ASSIGN_TASK_TYPE_PROMPT,
)
from metagpt.schema import Message, Plan, Task
from metagpt.tools import TOOL_REGISTRY
from metagpt.utils.common import CodeParser, create_func_call_config
from metagpt.utils.common import CodeParser
class WritePlan(Action):
PROMPT_TEMPLATE: str = """
# Context:
__context__
# Available Task Types:
__task_type_desc__
# Task:
Based on the context, write a plan or modify an existing plan of what you should do to achieve the goal. A plan consists of one to __max_tasks__ tasks.
If you are modifying an existing plan, carefully follow the instruction, don't make unnecessary changes. Give the whole plan unless instructed to modify only one task of the plan.
@ -36,46 +34,24 @@ class WritePlan(Action):
"task_id": str = "unique identifier for a task in plan, can be an ordinal",
"dependent_task_ids": list[str] = "ids of tasks prerequisite to this task",
"instruction": "what you should do in this task, one short phrase or sentence",
"task_type": "type of this task, should be one of Available Task Types",
},
...
]
```
"""
async def assign_task_type(self, tasks: list[dict]) -> str:
"""Assign task type to each task in tasks
Args:
tasks (list[dict]): tasks to be assigned task type
Returns:
str: tasks with task type assigned in a json string
"""
task_info = "\n".join([f"Task {task['task_id']}: {task['instruction']}" for task in tasks])
async def run(self, context: list[Message], max_tasks: int = 5, use_tools: bool = False) -> str:
task_type_desc = "\n".join(
[f"- **{tool_type.name}**: {tool_type.desc}" for tool_type in TOOL_REGISTRY.get_tool_types().values()]
) # task type are binded with tool type now, should be improved in the future
prompt = ASSIGN_TASK_TYPE_PROMPT.format(
task_info=task_info, task_type_desc=task_type_desc
) # task types are set to be the same as tool types, for now
tool_config = create_func_call_config(ASSIGN_TASK_TYPE_CONFIG)
rsp = await self.llm.aask_code(prompt, **tool_config)
task_type_list = rsp["task_type"]
logger.info(f"assigned task types: {task_type_list}")
for task, task_type in zip(tasks, task_type_list):
task["task_type"] = task_type
return json.dumps(tasks)
async def run(self, context: list[Message], max_tasks: int = 5, use_tools: bool = False) -> str:
prompt = (
self.PROMPT_TEMPLATE.replace("__context__", "\n".join([str(ct) for ct in context]))
# .replace("__current_plan__", current_plan)
.replace("__max_tasks__", str(max_tasks))
.replace("__max_tasks__", str(max_tasks)).replace("__task_type_desc__", task_type_desc)
)
rsp = await self._aask(prompt)
rsp = CodeParser.parse_code(block=None, text=rsp)
if use_tools:
rsp = await self.assign_task_type(json.loads(rsp))
return rsp