Merge pull request #1 from Stitch-z/feature-tutorial-assistant

Feature/tutorial assistant

examples/write_tutorial.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+# _*_ coding: utf-8 _*_
+"""
+@Time    : 2023/9/4 21:40:57
+@Author  : Stitch-z
+@File    : tutorial_assistant.py
+"""
+import asyncio
+
+from metagpt.roles.tutorial_assistant import TutorialAssistant
+
+
+async def main():
+    topic = "Write a tutorial about MySQL"
+    role = TutorialAssistant(language="Chinese")
+    await role.run(topic)
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

metagpt/actions/write_tutorial.py

@@ -0,0 +1,102 @@
+#!/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 datetime import datetime
+from typing import Dict
+
+import aiofiles
+
+from metagpt.actions import Action
+from metagpt.const import TUTORIAL_PATH
+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
+
+    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, such as {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}
+        """
+        prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
+        directory = await self._aask(prompt=prompt)
+        return json.loads(directory)
+
+
+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)
+
+
+class SaveDocx(Action):
+    """Action class for saving tutorial docx.
+
+    Args:
+        name: The name of the action.
+    """
+
+    def __init__(self, name: str = "", *args, **kwargs):
+        super().__init__(name, *args, **kwargs)
+
+    async def run(self, title: str, content: str, *args, **kwargs) -> str:
+        """Execute the action to save the generated tutorial document to a Markdown file.
+
+        Args:
+            title: The title of tutorial.
+            content: The total content of tutorial.
+
+        Returns:
+            The full filename of tutorial content.
+
+        """
+        current_time = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+        pathname = TUTORIAL_PATH / current_time
+        pathname.mkdir(parents=True, exist_ok=True)
+        filename = f"{pathname}/{title}.md"
+        async with aiofiles.open(filename, mode="w", encoding="utf-8") as writer:
+            await writer.write(content)
+            logger.info(f"Successfully write docx: {filename}")
+            return filename

metagpt/const.py

@@ -33,5 +33,6 @@ API_QUESTIONS_PATH = UT_PATH / "files/question/"
 YAPI_URL = "http://yapi.deepwisdomai.com/"
 TMP = PROJECT_ROOT / 'tmp'
 RESEARCH_PATH = DATA_PATH / "research"
+TUTORIAL_PATH = DATA_PATH / "tutorial_docx"
 
 MEM_TTL = 24 * 30 * 3600

metagpt/prompts/tutorial_assistant.py

@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+# _*_ coding: utf-8 _*_
+"""
+@Time    : 2023/9/4 15:40:40
+@Author  : Stitch-z
+@File    : tutorial_assistant.py
+@Describe : Tutorial Assistant's prompt templates.
+"""
+
+
+DIRECTORY_PROMPT = """
+You are now a seasoned technical professional in the field of the internet. 
+We need you to write a technical tutorial with the topic "{topic}". 
+Please provide the specific table of contents for this tutorial, strictly following the following requirements:
+1. The output must be strictly in the specified language, {language}.
+2. Answer in the dictionary format like {{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}]}}.
+3. The directory should be as specific and sufficient as possible, with a primary and secondary directory.The secondary directory is in the array.
+4. Do not have extra spaces or line breaks.
+5. Each directory title has practical significance.
+"""
+
+CONTENT_PROMPT = """
+You are now a seasoned technical professional in the field of the internet. 
+We need you to write a technical tutorial with the topic "{topic}". 
+Now I will give you the module directory titles for the topic. 
+Please output the detailed principle content of this title in detail. 
+If there are code examples, please provide them according to standard code specifications. 
+Without a code example, it is not necessary.
+
+The module directory titles for the topic is as follows:
+{directory}
+
+Strictly limit output according to the following requirements:
+1. Follow the Markdown syntax format for layout.
+2. If there are code examples, they must follow standard syntax specifications, have document annotations, and be displayed in code blocks.
+3. The output must be strictly in the specified language, {language}.
+4. Do not have redundant output, including concluding remarks.
+5. Don't return the topic "{topic}".
+"""

metagpt/roles/tutorial_assistant.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+# _*_ coding: utf-8 _*_
+"""
+@Time    : 2023/9/4 15:40:40
+@Author  : Stitch-z
+@File    : tutorial_assistant.py
+"""
+
+from typing import Dict
+
+from metagpt.actions.write_tutorial import WriteDirectory, WriteContent, SaveDocx
+from metagpt.logs import logger
+from metagpt.roles import Role
+from metagpt.schema import Message
+
+
+class TutorialAssistant(Role):
+    """Tutorial assistant, input one sentence to generate a tutorial document in markup format.
+
+    Args:
+        name: The name of the role.
+        profile: The role profile description.
+        goal: The goal of the role.
+        constraints: Constraints or requirements for the role.
+        language: The language in which the tutorial documents will be generated.
+    """
+
+    def __init__(
+        self,
+        name: str = "Stitch",
+        profile: str = "Tutorial Assistant",
+        goal: str = "Generate tutorial documents",
+        constraints: str = "Strictly follow Markdown's syntax, with neat and standardized layout",
+        language: str = "Chinese",
+    ):
+        super().__init__(name, profile, goal, constraints)
+        self._init_actions([WriteDirectory(language=language)])
+        self.topic = ""
+        self.main_title = ""
+        self.total_content = ""
+        self.language = language
+
+    async def _think(self) -> None:
+        """Determine the next action to be taken by the role."""
+        if self._rc.todo is None:
+            self._set_state(0)
+            return
+
+        if self._rc.state + 1 < len(self._states):
+            self._set_state(self._rc.state + 1)
+        else:
+            self._rc.todo = None
+
+    async def _handle_directory(self, titles: Dict) -> Message:
+        """Handle the directories for the tutorial document.
+
+        Args:
+            titles: A dictionary containing the titles and directory structure,
+                    such as {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}
+
+        Returns:
+            A message containing information about the directory.
+        """
+        self.main_title = titles.get("title")
+        directory = f"{self.main_title}\n"
+        self.total_content += f"# {self.main_title}"
+        actions = list()
+        for first_dir in titles.get("directory"):
+            actions.append(WriteContent(language=self.language, directory=first_dir))
+            key = list(first_dir.keys())[0]
+            directory += f"- {key}\n"
+            for second_dir in first_dir[key]:
+                directory += f"  - {second_dir}\n"
+        actions.append(SaveDocx())
+        self._init_actions(actions)
+        self._rc.todo = None
+        return Message(content=directory)
+
+    async def _act(self) -> Message:
+        """Perform an action as determined by the role.
+
+        Returns:
+            A message containing the result of the action.
+        """
+        todo = self._rc.todo
+        if type(todo) is WriteDirectory:
+            msg = self._rc.memory.get(k=1)[0]
+            self.topic = msg.content
+            resp = await todo.run(topic=self.topic)
+            logger.info(resp)
+            return await self._handle_directory(resp)
+        elif type(todo) is SaveDocx:
+            filename = await todo.run(title=self.main_title, content=self.total_content)
+            return Message(content=filename, role=self.profile)
+        resp = await todo.run(topic=self.topic)
+        logger.info(resp)
+        if self.total_content != "":
+            self.total_content += "\n\n\n"
+        self.total_content += resp
+        return Message(content=resp, role=self.profile)
+
+    async def _react(self) -> Message:
+        """Execute the assistant's think and actions.
+
+        Returns:
+            A message containing the final result of the assistant's actions.
+        """
+        while True:
+            await self._think()
+            if self._rc.todo is None:
+                break
+            msg = await self._act()
+        return msg