fix conflict

metagpt/actions/design_api.py

@@ -92,30 +92,30 @@ class WriteDesign(Action):
             pass  # Folder does not exist, but we don't care
         workspace.mkdir(parents=True, exist_ok=True)
 
-    def _save_prd(self, docs_path, resources_path, context):
+    async def _save_prd(self, docs_path, resources_path, context):
         prd_file = docs_path / "prd.md"
         if context[-1].instruct_content and context[-1].instruct_content.dict()["Competitive Quadrant Chart"]:
             quadrant_chart = context[-1].instruct_content.dict()["Competitive Quadrant Chart"]
-            mermaid_to_file(quadrant_chart, resources_path / "competitive_analysis")
+            await mermaid_to_file(quadrant_chart, resources_path / "competitive_analysis")
 
         if context[-1].instruct_content:
             logger.info(f"Saving PRD to {prd_file}")
             prd_file.write_text(json_to_markdown(context[-1].instruct_content.dict()))
 
-    def _save_system_design(self, docs_path, resources_path, system_design):
+    async def _save_system_design(self, docs_path, resources_path, system_design):
         data_api_design = system_design.instruct_content.dict()[
             "Data structures and interface definitions"
         ]  # CodeParser.parse_code(block="Data structures and interface definitions", text=content)
         seq_flow = system_design.instruct_content.dict()[
             "Program call flow"
         ]  # CodeParser.parse_code(block="Program call flow", text=content)
-        mermaid_to_file(data_api_design, resources_path / "data_api_design")
-        mermaid_to_file(seq_flow, resources_path / "seq_flow")
+        await mermaid_to_file(data_api_design, resources_path / "data_api_design")
+        await mermaid_to_file(seq_flow, resources_path / "seq_flow")
         system_design_file = docs_path / "system_design.md"
         logger.info(f"Saving System Designs to {system_design_file}")
         system_design_file.write_text((json_to_markdown(system_design.instruct_content.dict())))
 
-    def _save(self, context, system_design):
+    async def _save(self, context, system_design):
         if isinstance(system_design, ActionOutput):
             ws_name = system_design.instruct_content.dict()["Python package name"]
         else:
@@ -126,12 +126,12 @@ class WriteDesign(Action):
         resources_path = workspace / "resources"
         docs_path.mkdir(parents=True, exist_ok=True)
         resources_path.mkdir(parents=True, exist_ok=True)
-        self._save_prd(docs_path, resources_path, context)
-        self._save_system_design(docs_path, resources_path, system_design)
+        await self._save_prd(docs_path, resources_path, context)
+        await self._save_system_design(docs_path, resources_path, system_design)
 
     async def run(self, context):
         prompt = PROMPT_TEMPLATE.format(context=context, format_example=FORMAT_EXAMPLE)
         # system_design = await self._aask(prompt)
         system_design = await self._aask_json_v1(prompt, "system_design", OUTPUT_MAPPING)
-        self._save(context, system_design)
+        await self._save(context, system_design)
         return system_design

metagpt/actions/write_tutorial.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+# _*_ coding: utf-8 _*_
+"""
+@Time    : 2023/9/4 15:40:40
+@Author  : Stitch-z
+@File    : tutorial_assistant.py
+@Describe : Actions of the tutorial assistant, including writing directories and document content.
+"""
+import json
+from typing import Dict
+
+from metagpt.actions import Action
+from metagpt.logs import logger
+from metagpt.prompts.tutorial_assistant import DIRECTORY_PROMPT, CONTENT_PROMPT
+
+
+class WriteDirectory(Action):
+    """Action class for writing tutorial directories.
+
+    Args:
+        name: The name of the action.
+        language: The language to output, default is "Chinese".
+    """
+
+    def __init__(self, name: str = "", language: str = "Chinese", *args, **kwargs):
+        super().__init__(name, *args, **kwargs)
+        self.language = language
+
+    @staticmethod
+    async def _handle_resp(resp: str) -> Dict:
+        """Process string results and convert them to JSON format.
+
+        Args:
+            resp: The directory results returned by gpt.
+
+        Returns:
+            The parsed dictionary, such as {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}.
+
+        Raises:
+            Exception: If no matching dictionary section is found.
+            json.JSONDecodeError: If the dictionary part cannot be parsed as JSON.
+        """
+        start = resp.find('{')
+        end = resp.rfind('}')
+        if start != -1 and end != -1 and end > start:
+            directory_str = resp[start:end + 1]
+            logger.info(f"Successfully parsed json: {str(directory_str)}")
+            try:
+                return json.loads(directory_str)
+            except json.JSONDecodeError as e:
+                logger.error(f"Json parsing error: {e}")
+                raise e
+        else:
+            raise Exception("No matching dictionary section found.")
+
+    async def run(self, topic: str, *args, **kwargs) -> Dict:
+        """Execute the action to generate a tutorial directory according to the topic.
+
+        Args:
+            topic: The tutorial topic.
+
+        Returns:
+            the tutorial directory information, including {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}.
+        """
+        prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
+        resp = await self._aask(prompt=prompt)
+        return await self._handle_resp(resp)
+
+
+class WriteContent(Action):
+    """Action class for writing tutorial content.
+
+    Args:
+        name: The name of the action.
+        directory: The content to write.
+        language: The language to output, default is "Chinese".
+    """
+
+    def __init__(self, name: str = "", directory: str = "", language: str = "Chinese", *args, **kwargs):
+        super().__init__(name, *args, **kwargs)
+        self.language = language
+        self.directory = directory
+
+    async def run(self, topic: str, *args, **kwargs) -> str:
+        """Execute the action to write document content according to the directory and topic.
+
+        Args:
+            topic: The tutorial topic.
+
+        Returns:
+            The written tutorial content.
+        """
+        prompt = CONTENT_PROMPT.format(topic=topic, language=self.language, directory=self.directory)
+        return await self._aask(prompt=prompt)
+