feat: +software_development_intent_detect

metagpt/actions/intent_detect.py

@@ -157,15 +157,7 @@ class IntentDetect(Action):
         intentions = await self._get_intentions(msg_markdown)
         await self._get_references(msg_markdown, intentions)
         await self._get_sops()
-        self.result = IntentDetectResult(clarifications=self._dialog_intentions.clarifications)
-        sops = {i.description: i for i in SOP_CONFIG}
-        intent_to_sops = {i.intent: i.sop for i in self._intent_to_sops if i.sop != ""}
-        for i in self._references.intentions:
-            item = IntentDetectIntentionSOP(intention=i)
-            key = intent_to_sops.get(i.intent)
-            if key:
-                item.sop = sops.get(key)
-            self.result.intentions.append(item)
+        await self._merge()
 
         return Message(
             content=self.result.model_dump_json(), role="assistant", cause_by=self, instruct_content=self.result
@@ -251,6 +243,45 @@ class IntentDetect(Action):
         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)
+        sops = {i.description: i for i in SOP_CONFIG}
+        intent_to_sops = {i.intent: i.sop for i in self._intent_to_sops if i.sop != ""}
+
+        distinct = {}
+        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=ref)
+                self.result.intentions.append(item)
+                continue
+            distinct[i.sop_index] = [i.intent] + distinct.get(i.sop_index, [])
+
+        merge_intents = {}
+        for sop_index, intents in distinct.items():
+            if len(intents) > 1:
+                merge_intents[sop_index] = intents
+                continue
+            intent_ref = self._get_intent_ref(intents[0])
+            item = IntentDetectIntentionSOP(intention=intent_ref)
+            key = intent_to_sops.get(intents[0])
+            if key:
+                item.sop = sops.get(key)
+            self.result.intentions.append(item)
+
+        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.extend(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):
+        mappings = {i.intent: i for i in self._references.intentions}
+        return mappings[intent]
+
     @staticmethod
     def _message_to_markdown(messages) -> str:
         markdown = ""
@@ -258,3 +289,50 @@ class IntentDetect(Action):
             content = i.content.replace("\n", " ")
             markdown += f"> {i.role}: {content}\n>\n"
         return markdown
+
+
+class LightIntentDetect(IntentDetect):
+    result: List[SOPItem] = 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)
+        await self._get_intentions(msg_markdown)
+        await self._get_sops()
+
+        distinct = {i.sop_index - 1: SOP_CONFIG[i.sop_index - 1] for i in self._intent_to_sops if i.sop_index > 0}
+        self.result = list(distinct.values())
+        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]

metagpt/tools/libs/dialog.py

@@ -6,7 +6,11 @@ This script defines tools for dialog.
 
 from typing import List
 
-from metagpt.actions.intent_detect import IntentDetect, IntentDetectResult
+from metagpt.actions.intent_detect import (
+    IntentDetect,
+    IntentDetectResult,
+    LightIntentDetect,
+)
 from metagpt.context import Context
 from metagpt.schema import Message
 from metagpt.tools.tool_registry import register_tool
@@ -74,3 +78,39 @@ async def intent_detect(messages: List[Message]) -> IntentDetectResult:
     action = IntentDetect(context=ctx)
     await action.run(messages)
     return action.result
+
+
+@register_tool(tags=["dialog", "software development intent detect"])
+async def software_development_intent_detect(messages: List[Message]) -> List[str]:
+    """Detects software development intent from a list of dialog messages.
+
+    Args:
+        messages (List[Message]): A list of dialog messages.
+
+    Returns:
+        IntentDetectResult: The result of intent detection.
+
+    Example:
+        >>> # Create messages
+        >>> dialog = [
+        >>>     {"role":"user", "content":"user queries ..."},
+        >>>     {"role":"assistant", "content": "assistant answers ..."},
+        >>>     ...
+        >>> ]
+        >>> from metagpt.schema import Message
+        >>> messages = [Message.model_validate(i) for i in dialog]
+        >>> result = await software_development_intent_detect(messages)
+        >>> print(result)
+        [
+            "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 codes to the project repository, based on the project plan of the project.",
+            "Run QA test on the project repository.",
+            "Stage and commit changes for the project repository using Git."
+        ]
+    """
+    ctx = Context()
+    action = LightIntentDetect(context=ctx)
+    await action.run(messages)
+    return action.result[0].sop if action.result else []