feat: merge gitlab:mgx_ops

metagpt/actions/intent_detect.py

@@ -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 []

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)

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"])