diff --git a/metagpt/actions/intent_detect.py b/metagpt/actions/intent_detect.py deleted file mode 100644 index aa430a0bb..000000000 --- a/metagpt/actions/intent_detect.py +++ /dev/null @@ -1,436 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -""" -This script is designed to classify intentions from complete conversation content. - -Usage: - This script can be used to classify intentions from a conversation. It utilizes models for detecting intentions - from the text provided and categorizes them accordingly. If the intention of certain words or phrases is unclear, - it prompts the user for clarification. - -Dependencies: - This script depends on the metagpt library, pydantic, and other utilities for message parsing and interaction. - -""" -import json -import re -from typing import List - -from pydantic import BaseModel, Field - -from metagpt.actions import Action -from metagpt.logs import logger -from metagpt.schema import Message -from metagpt.utils.common import parse_json_code_block - - -class SOPItem(BaseModel): - """ - Represents an item in a Standard Operating Procedure (SOP). - - Attributes: - description (str): The description or title of the SOP. - sop (List[str]): The steps or details of the SOP. - """ - - description: str - sop: List[str] - - -SOP_CONFIG = [ - SOPItem( - description="Intentions related to or including software development, such as developing or building software, games, programming, app, websites, etc. Excluding bug fixes, report any issues.", - sop=[ - "Writes a PRD based on software requirements.", - "Writes a design to the project repository, based on the PRD of the project.", - "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.", - # "Run QA test on the project repository.", - "Stage and commit changes for the project repository using Git.", - ], - ), - SOPItem( - description="Error message, issues, fix bug, exception description", - sop=[ - "Fix bugs in the project repository.", - "Stage and commit changes for the project repository using Git.", - ], - ), - SOPItem( - 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.", - ], - ), -] - - -class IntentDetectClarification(BaseModel): - """ - Represents clarifications for unclear intentions. - - Attributes: - ref (str): The reference to the original words. - clarification (str): A question for the user to clarify the intention of the unclear words. - """ - - ref: str - clarification: str - - -class IntentDetectIntentionRef(BaseModel): - """ - Represents intentions along with their references. - - Attributes: - intent (str): The intention from the "Intentions" section. - refs (List[str]): List of original text references from the "Dialog" section that match the intention. - """ - - intent: str - refs: List[str] - - -class IntentDetectIntentionSOP(BaseModel): - """ - Represents an intention mapped to a Standard Operating Procedure (SOP). - - Attributes: - intention (IntentDetectIntentionRef): Reference to the intention. - sop (SOPItem, optional): Standard Operating Procedure (SOP) item related to the intention. - """ - - intention: IntentDetectIntentionRef - sop: SOPItem = None - - -class IntentDetectResult(BaseModel): - """ - Represents the result of intention detection. - - Attributes: - clarifications (List[IntentDetectClarification]): List of clarifications for unclear intentions. - intentions (List[IntentDetectIntentionSOP]): List of intentions mapped to Standard Operating Procedures (SOPs). - """ - - clarifications: List[IntentDetectClarification] = Field(default_factory=list) - intentions: List[IntentDetectIntentionSOP] = Field(default_factory=list) - - -class IntentDetect(Action): - """ - Action class for intention detection. - - Attributes: - _dialog_intentions (IntentDetectDialogIntentions): Instance of IntentDetectDialogIntentions. - Dialog intentions for matching user intentions. - _references (IntentDetectReferences): Instance of IntentDetectReferences. - References to intentions and unreferenced content. - _intent_to_sops (List[IntentSOP]): List of IntentSOP objects. - Mapping of intentions to Standard Operating Procedures (SOPs). - result (IntentDetectResult): Instance of IntentDetectResult. - Result object containing the outcome of intention detection. - """ - - class IntentDetectDialogIntentions(BaseModel): - class IntentDetectIntention(BaseModel): - ref: str - intent: str - - intentions: List[IntentDetectIntention] - clarifications: List[IntentDetectClarification] - - class IntentDetectReferences(BaseModel): - class IntentDetectUnrefs(BaseModel): - ref: str - reason: str - - intentions: List[IntentDetectIntentionRef] - unrefs: List[IntentDetectUnrefs] - - class IntentSOP(BaseModel): - intent: str - sop: str - sop_index: int - reason: str - - _dialog_intentions: IntentDetectDialogIntentions = None - _references: IntentDetectReferences = None - _intent_to_sops: List[IntentSOP] = None - result: IntentDetectResult = None - - async def run(self, with_messages: List[Message] = None, **kwargs) -> Message: - """ - Runs the intention detection action. - - Args: - with_messages (List[Message]): List of messages representing the conversation content. - **kwargs: Additional keyword arguments. - """ - msg_markdown = self._message_to_markdown(with_messages) - intentions = await self._get_intentions(msg_markdown) - await self._get_references(msg_markdown, intentions) - await self._get_sops() - await self._merge() - - return Message( - content=self.result.model_dump_json(), role="assistant", cause_by=self, instruct_content=self.result - ) - - async def _get_intentions(self, msg_markdown: str) -> List[str]: - rsp = await self.llm.aask( - msg_markdown, - system_msgs=[ - "You are a tool that can classify user intentions.", - "Detect and classify the intention of each word spoken by the user in the conversation.", - "If the user's intention is not clear, create a request for the user to clarify the intention of" - " the unclear words.", - "Return a markdown object with:\n" - '- an "intentions" key containing a list of JSON objects, where each object contains:\n' - ' - a "ref" key containing the original words reference;\n' - ' - an "intent" key explaining the intention of the referenced word;\n' - '- a "clarifications" key containing a list of JSON objects, where each object contains:\n' - ' - a "ref" key containing the original words reference;\n' - ' - a "clarification" key containing a question, in the tone of an assistant, prompts the user to provide more details about the intention regarding the unclear word(s) referenced in the user\'s description.', - ], - stream=False, - ) - logger.debug(rsp) - json_blocks = parse_json_code_block(rsp) - if not json_blocks: - return [] - self._dialog_intentions = self.IntentDetectDialogIntentions.model_validate_json(json_blocks[0]) - return [i.intent for i in self._dialog_intentions.intentions] - - async def _get_references(self, msg_markdown: str, intentions: List[str]): - intention_list = "\n".join([f"- {i}" for i in intentions]) - prompt = f"## Dialog\n{msg_markdown}\n---\n## Intentions\n{intention_list}\n" - rsp = await self.llm.aask( - prompt, - system_msgs=[ - "You are a tool that categorizes text content by intent.", - "Place the original text from the `Dialog` section under the matching intent of `Intentions` section.", - "Allow different intents to reference the same original text.", - "Return a markdown JSON object with:\n" - '- an "intentions" key containing a list of JSON objects, where each object contains:\n' - ' - a "intent" key containing the intention from "Intentions" section;\n' - ' - a "refs" key containing a list of strings of original text from the "Dialog" section that match' - " the intention.\n" - '- a "unrefs" key containing a list of JSON objects, where each object contains:\n' - ' - a "ref" key containing the unreferenced original text.\n' - ' - a "reason" key explaining why it is unreferenced.\n', - ], - stream=False, - ) - logger.debug(rsp) - json_blocks = parse_json_code_block(rsp) - if not json_blocks: - return [] - self._references = self.IntentDetectReferences.model_validate_json(json_blocks[0]) - - async def _get_sops(self): - intention_list = "" - for i, v in enumerate(self._references.intentions): - intention_list += f"{i + 1}. intent: {v.intent}\n" - for j in v.refs: - intention_list += f" - ref: {j}\n" - sop_list = "" - for i, v in enumerate(SOP_CONFIG): - sop_list += f"{i + 1}. {v.description}\n" - prompt = f"## Intentions\n{intention_list}\n---\n## SOPs\n{sop_list}\n" - rsp = await self.llm.aask( - prompt, - system_msgs=[ - "You are a tool that matches user intentions with Standard Operating Procedures (SOPs).", - 'You search for matching SOPs under "SOPs" based on user intentions in "Intentions" and their related original descriptions.', - 'Inspect each intention in "Intentions".', - "Return a markdown JSON list of objects, where each object contains:\n" - '- an "intent" key containing the intention from the "Intentions" section;\n' - '- a "sop" key containing the SOP description from the "SOPs" section; filled with an empty string if no match.\n' - '- a "sop_index" key containing the int type index of SOP description from the "SOPs" section; filled with 0 if no match.\n' - '- a "reason" key explaining why it is matching/mismatching.\n', - ], - stream=False, - ) - logger.debug(rsp) - json_blocks = parse_json_code_block(rsp) - vv = json.loads(json_blocks[0]) - self._intent_to_sops = [self.IntentSOP.model_validate(i) for i in vv] - - async def _merge(self): - self.result = IntentDetectResult(clarifications=self._dialog_intentions.clarifications) - distinct = {} - # Consolidate intentions under the same SOP. - for i in self._intent_to_sops: - if i.sop_index == 0: # 1-based index - refs = self._get_intent_ref(i.intent) - item = IntentDetectIntentionSOP(intention=IntentDetectIntentionRef(intent=i.intent, refs=refs)) - self.result.intentions.append(item) - continue - distinct[i.sop_index] = [i.intent] + distinct.get(i.sop_index, []) - - merge_intents = {} - intent_to_sops = {i.intent: i.sop_index for i in self._intent_to_sops if i.sop_index != 0} - for sop_index, intents in distinct.items(): - if len(intents) > 1: - merge_intents[sop_index] = intents - continue - # Merge single intention - refs = self._get_intent_ref(intents[0]) - item = IntentDetectIntentionSOP(intention=IntentDetectIntentionRef(intent=intents[0], refs=refs)) - sop_index = intent_to_sops.get(intents[0]) - item.sop = SOP_CONFIG[sop_index - 1] # 1-based index - self.result.intentions.append(item) - - # Merge repetitive intentions into one - for sop_index, intents in merge_intents.items(): - intent_ref = IntentDetectIntentionRef(intent="\n".join(intents), refs=[]) - for i in intents: - refs = self._get_intent_ref(i) - intent_ref.refs.extend(refs) - intent_ref.refs = list(set(intent_ref.refs)) - item = IntentDetectIntentionSOP(intention=intent_ref) - item.sop = SOP_CONFIG[sop_index - 1] # 1-based index - self.result.intentions.append(item) - - def _get_intent_ref(self, intent: str) -> List[str]: - refs = [] - for i in self._references.intentions: - if i.intent == intent: - refs.extend(i.refs) - return refs - - @staticmethod - def _message_to_markdown(messages) -> str: - markdown = "" - for i in messages: - content = i.content.replace("\n", " ") - markdown += f"> {i.role}: {content}\n>\n" - return markdown - - -class LightIntentDetect(IntentDetect): - async def run(self, with_messages: List[Message] = None, **kwargs) -> Message: - """ - Runs the intention detection action. - - Args: - with_messages (List[Message]): List of messages representing the conversation content. - **kwargs: Additional keyword arguments. - """ - msg_markdown = self._message_to_markdown(with_messages) - await self._get_intentions(msg_markdown) - await self._get_sops() - await self._merge() - - return Message(content="", role="assistant", cause_by=self) - - async def _get_sops(self): - intention_list = "" - for i, v in enumerate(self._dialog_intentions.intentions): - intention_list += f"{i + 1}. intent: {v.intent}\n - ref: {v.ref}\n" - sop_list = "" - for i, v in enumerate(SOP_CONFIG): - sop_list += f"{i + 1}. {v.description}\n" - prompt = f"## Intentions\n{intention_list}\n---\n## SOPs\n{sop_list}\n" - rsp = await self.llm.aask( - prompt, - system_msgs=[ - "You are a tool that matches user intentions with Standard Operating Procedures (SOPs).", - 'You search for matching SOPs under "SOPs" based on user intentions in "Intentions" and their related original descriptions.', - 'Inspect each intention in "Intentions".', - "Return a markdown JSON list of objects, where each object contains:\n" - '- an "intent" key containing the intention from the "Intentions" section;\n' - '- a "sop" key containing the SOP description from the "SOPs" section; filled with an empty string if no match.\n' - '- a "sop_index" key containing the int type index of SOP description from the "SOPs" section; filled with 0 if no match.\n' - '- a "reason" key explaining why it is matching/mismatching.\n', - ], - stream=False, - ) - logger.debug(rsp) - json_blocks = parse_json_code_block(rsp) - vv = json.loads(json_blocks[0]) - self._intent_to_sops = [self.IntentSOP.model_validate(i) for i in vv] - - async def _merge(self): - self.result = IntentDetectResult(clarifications=[]) - distinct = {} - # Consolidate intentions under the same SOP. - for i in self._intent_to_sops: - if i.sop_index == 0: # 1-based index - ref = self._get_intent_ref(i.intent) - item = IntentDetectIntentionSOP(intention=IntentDetectIntentionRef(intent=i.intent, refs=[ref])) - self.result.intentions.append(item) - continue - distinct[i.sop_index] = [i.intent] + distinct.get(i.sop_index, []) - - merge_intents = {} - intent_to_sops = {i.intent: i.sop_index for i in self._intent_to_sops if i.sop_index != 0} - for sop_index, intents in distinct.items(): - if len(intents) > 1: - merge_intents[sop_index] = intents - continue - # Merge single intention - ref = self._get_intent_ref(intents[0]) - item = IntentDetectIntentionSOP(intention=IntentDetectIntentionRef(intent=intents[0], refs=[ref])) - sop_index = intent_to_sops.get(intents[0]) # 1-based - if sop_index: - item.sop = SOP_CONFIG[sop_index - 1] # 1-based index - self.result.intentions.append(item) - - # Merge repetitive intentions into one - for sop_index, intents in merge_intents.items(): - intent_ref = IntentDetectIntentionRef(intent="\n".join(intents), refs=[]) - for i in intents: - ref = self._get_intent_ref(i) - intent_ref.refs.append(ref) - intent_ref.refs = list(set(intent_ref.refs)) - item = IntentDetectIntentionSOP(intention=intent_ref) - item.sop = SOP_CONFIG[sop_index - 1] # 1-based index - self.result.intentions.append(item) - - def _get_intent_ref(self, intent: str) -> str: - refs = [] - for i in self._dialog_intentions.intentions: - if i.intent == intent: - refs.append(i.ref) - return "\n".join(refs) - - -class SentenceIntentDetect(IntentDetect): - sop: List[str] = None - - async def run(self, with_messages: List[Message] = None, **kwargs) -> Message: - """ - Runs the intention detection action. - - Args: - with_messages (List[Message]): List of messages representing the conversation content. - **kwargs: Additional keyword arguments. - """ - msg_markdown = self._message_to_markdown(with_messages) - self.sop = await self._get_intentions(msg_markdown) - return Message(content="", role="assistant", cause_by=self) - - async def _get_intentions(self, msg_markdown: str) -> List[str]: - prompt = f"## Dialog\n{msg_markdown}\n" - prompt += "## Intentions\n" - for i, v in enumerate(SOP_CONFIG): - prompt += f"{i + 1}. {v.description}\n" - prompt += f"{len(SOP_CONFIG) + 1}. Others" - rsp = await self.llm.aask( - prompt, - system_msgs=[ - 'You are a tool for selecting a suitable intention from the "Intentions" section.', - 'Select the intention that matches the conversation in the "Dialog" section from the "Intentions" section.', - 'If no matching intention is found, choose "Others".', - "Return the integer index of your choice.", - ], - stream=False, - ) - logger.debug(rsp) - idx = int(re.findall(r"\b\d+\b", rsp)[0]) - 1 - if idx < len(SOP_CONFIG): - return SOP_CONFIG[idx].sop - return [] diff --git a/metagpt/roles/di/mgx.py b/metagpt/roles/di/mgx.py index 0bfef5820..b2caa930b 100644 --- a/metagpt/roles/di/mgx.py +++ b/metagpt/roles/di/mgx.py @@ -4,7 +4,7 @@ import asyncio from typing import Dict, List -from metagpt.actions.intent_detect import SentenceIntentDetect +from metagpt.actions.di.detect_intent import DetectIntent from metagpt.logs import logger from metagpt.roles.di.data_interpreter import DataInterpreter from metagpt.schema import Message @@ -14,25 +14,11 @@ class MGX(DataInterpreter): use_intent: bool = True intents: Dict = {} - async def _intent_detect(self, user_msgs: List[Message] = None, **kwargs): - todo = SentenceIntentDetect(context=self.context) - await todo.run(user_msgs) - logger.info(f"intent_desp is {todo.sop}") - - # Extract intent and sop prompt - intention_ref = "\n".join([i.content for i in user_msgs]) - if todo.sop: - self.intents[intention_ref] = todo.sop - logger.debug(f"refs: {intention_ref}, sop: {todo.sop}") - sop_str = "\n".join([f"- {i}" for i in todo.sop]) - markdown = ( - f"### User Requirement Detail\n```text\n{intention_ref}\n````\n" - f"### Knowledge\nTo meet user requirements, the following standard operating procedure(SOP) must be" - f" used. SOP descriptions cannot be modified; user requirements can only be appended to the end of corresponding steps.\n" - f"{sop_str}" - ) - return markdown - return intention_ref + async def _detect_intent(self, user_msgs: List[Message] = None, **kwargs): + todo = DetectIntent(context=self.context) + request_with_sop, sop_type = await todo.run(user_msgs) + logger.info(f"{sop_type} {request_with_sop}") + 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.""" @@ -41,7 +27,7 @@ class MGX(DataInterpreter): goal = self.rc.memory.get()[-1].content # retreive latest user requirement if self.use_intent: # add mode user_message = Message(content=goal, role="user") - goal = await self._intent_detect(user_msgs=[user_message]) + goal = await self._detect_intent(user_msgs=[user_message]) logger.info(f"Goal is {goal}") await self.planner.update_plan(goal=goal) diff --git a/tests/metagpt/actions/test_intent_detect.py b/tests/metagpt/actions/test_intent_detect.py index c3cd6eaf8..7b3352f61 100644 --- a/tests/metagpt/actions/test_intent_detect.py +++ b/tests/metagpt/actions/test_intent_detect.py @@ -1,16 +1,5 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- -import json - -import pytest - -from metagpt.actions.intent_detect import ( - IntentDetect, - LightIntentDetect, - SentenceIntentDetect, -) -from metagpt.logs import logger -from metagpt.schema import Message DEMO_CONTENT = [ { @@ -137,54 +126,3 @@ DEMO2_CONTENT = [ DEMO3_CONTENT = [ {"role": "user", "content": "git clone 'https://github.com/spec-first/connexion' and format to MetaGPT project"} ] - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - "content", - [json.dumps(DEMO1_CONTENT), json.dumps(DEMO_CONTENT), json.dumps(DEMO2_CONTENT), json.dumps(DEMO3_CONTENT)], -) -# @pytest.mark.skip -async def test_intent_detect(content: str, context): - action = IntentDetect(context=context) - messages = [Message.model_validate(i) for i in json.loads(content)] - rsp = await action.run(messages) - assert isinstance(rsp, Message) - assert action._dialog_intentions - assert action._references - assert action._intent_to_sops - assert action.result - logger.info(action.result.model_dump_json()) - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - "content", - [json.dumps(DEMO1_CONTENT), json.dumps(DEMO_CONTENT), json.dumps(DEMO2_CONTENT), json.dumps(DEMO3_CONTENT)], -) -# @pytest.mark.skip -async def test_light_intent_detect(content: str, context): - action = LightIntentDetect(context=context) - messages = [Message.model_validate(i) for i in json.loads(content)] - rsp = await action.run(messages) - assert isinstance(rsp, Message) - assert action._dialog_intentions - assert action._intent_to_sops - assert action.result - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - "content", - [json.dumps(DEMO1_CONTENT), json.dumps(DEMO_CONTENT), json.dumps(DEMO2_CONTENT), json.dumps(DEMO3_CONTENT)], -) -# @pytest.mark.skip -async def test_sentence_intent(content: str, context): - action = SentenceIntentDetect(context=context) - messages = [Message.model_validate(i) for i in json.loads(content)] - await action.run(messages) - assert action.sop is not None - - -if __name__ == "__main__": - pytest.main([__file__, "-s"])