博客

  • MetaGPT 教学助手:一键生成技术教程

    在学习新技术时,一份好的教程可以事半功倍。MetaGPT 的教学助手角色,可以帮助你根据简单的描述,自动生成技术教程文档。

    教学助手的功能

    教学助手可以根据用户提供的单句描述,生成一份技术教程文档,并支持自定义语言。

    设计理念

    教学助手的设计思路是:

    1. 使用大型语言模型 (LLM) 生成教程的大纲。
    2. 根据二级标题将大纲分解成多个部分。
    3. 针对每个部分,根据标题生成详细内容。
    4. 最后将标题和内容拼接起来。

    使用分段的方式,可以解决 LLM 模型处理长文本的限制。

    代码实现

    角色定义

    定义教学助手角色类,继承自 Role 基类,并重写 __init__ 初始化方法。__init__ 方法必须包含 nameprofilegoalconstraints 参数。第一行代码使用 super().__init__(name, profile, goal, constraints) 调用父类的构造函数,初始化 Role。使用 self.set_actions([WriteDirectory(language=language)]) 添加初始动作和状态。这里,最初添加了写目录动作。此外,还可以定义自定义参数;这里添加了 language 参数,用于支持自定义语言。使用 self._set_react_mode(react_mode="by_order")set_actions 中的动作执行顺序设置为顺序执行。

    class TutorialAssistant(Role):
        """教学助手,输入一句话生成一份标记格式的教程文档。
    
        Args:
            name: 角色名称。
            profile: 角色简介。
            goal: 角色目标。
            constraints: 角色的约束或要求。
            language: 生成教程文档的语言。
        """
    
        def __init__(
            self,
            name: str = "Stitch",
            profile: str = "教学助手",
            goal: str = "生成教程文档",
            constraints: str = "严格遵循 Markdown 语法,布局整洁规范",
            language: str = "Chinese",
        ):
            super().__init__(name, profile, goal, constraints)
            self.set_actions([WriteDirectory(language=language)])
            self.topic = ""
            self.main_title = ""
            self.total_content = ""
            self.language = language
            self._set_react_mode(react_mode="by_order")

    重写 react 方法

    重写 react 方法。使用 await super().react() 调用 Role 基类的 react 方法。根据 __init__ 方法中设置的 react_mode="by_order",按顺序执行 states 中的每个动作。这里重写的目的是在完成所有动作后执行最终操作,即:将拼接后的教程内容写入 markdown 文件。

    async def react(self) -> Message:
        msg = await super().react()
        root_path = TUTORIAL_PATH / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        await File.write(root_path, f"{self.main_title}.md", self.total_content.encode('utf-8'))
        return msg

    重写 _act 方法

    _act 方法负责执行动作。使用 todo = self.rc.todo 从上下文中获取下一个要执行的动作,然后执行动作的 run 方法。这里,它首先通过 WriteDirectory 获取教程目录结构,然后对目录进行分块,为每个块生成一个 WriteContent 动作,并初始化新添加的动作。这里再次调用 await super().react() 是为了从头开始执行所有新添加的 WriteContent 动作。每个动作的结果用于生成一个 Message(content=resp, role=self.profile) 消息,该消息可以放置在上下文记忆 self.rc.memory 中。该角色不需要存储。

    async def _act(self) -> Message:
        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)
            await self._handle_directory(resp)
            return await super().react()
        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 _handle_directory(self, titles: Dict) -> Message:
        """处理教程文档的目录。
    
        Args:
            titles: 包含标题和目录结构的字典,
                    例如 {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}
    
        Returns:
            包含目录信息的 Message。
        """
        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"
        self.set_actions(actions)

    动作定义

    定义动作,每个动作对应一个类对象。继承自 Action 基类,并重写 __init__ 初始化方法。__init__ 方法包含 name 参数。第一行代码使用 super().__init__(name, *args, **kwargs) 调用父类的构造函数,初始化动作。这里,使用 argskwargs 将其他参数传递给父类构造函数,例如 contextllm

    #!/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.
    """
    
    from typing import Dict
    
    from metagpt.actions import Action
    from metagpt.prompts.tutorial_assistant import DIRECTORY_PROMPT, CONTENT_PROMPT
    from metagpt.utils.common import OutputParser
    
    
    class WriteDirectory(Action):
        """写教程目录的动作类。
    
        Args:
            name: 动作名称。
            language: 输出语言,默认值为 "Chinese"。
        """
    
        def __init__(self, name: str = "", language: str = "Chinese", *args, **kwargs):
            super().__init__(name, *args, **kwargs)
            self.language = language

    重写 run 方法

    run 方法是动作执行的主要函数,使用 self._aask(prompt=prompt) 方法查询 LLM 模型。

    async def run(self, topic: str, *args, **kwargs) -> Dict:
        """执行动作,根据主题生成教程目录。
    
        Args:
            topic: 教程主题。
    
        Returns:
            教程目录信息,包括 {"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 OutputParser.extract_struct(resp, dict)

    其他动作的编写类似:

    class WriteContent(Action):
        """写教程内容的动作类。
    
        Args:
            name: 动作名称。
            directory: 要写入的内容。
            language: 输出语言,默认值为 "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:
            """执行动作,根据目录和主题写入文档内容。
    
            Args:
                topic: 教程主题。
    
            Returns:
                写入的教程内容。
            """
            prompt = CONTENT_PROMPT.format(topic=topic, language=self.language, directory=self.directory)
            return await self._aask(prompt=prompt)

    角色执行结果

    输入示例

    • MySQL 教程
    • Redis 教程
    • Hive 教程

    执行命令示例

    提供相应的执行命令示例。

    执行结果

    生成的教程文档位于项目的 /data/tutorial_docx 目录中。

    注意:

    该角色目前不支持互联网搜索功能。内容生成依赖于 LLM 大模型训练的数据。

    总结

    MetaGPT 的教学助手角色可以帮助你快速生成技术教程文档,并支持自定义语言。它可以节省你大量时间和精力,让你专注于更重要的工作。

    更多学习资源

  • MetaGPT 研究员:让 AI 帮你搜集信息、撰写报告

    在信息爆炸的时代,获取准确可靠的信息并将其整理成简洁易懂的报告,是一项非常重要的技能。MetaGPT 的研究员角色,可以帮助你完成这项任务。

    研究员的角色

    MetaGPT 的研究员角色可以模拟人类在互联网上进行研究的过程。它能够:

    • 分析研究问题,将其分解成多个适合搜索引擎查询的子问题。
    • 使用搜索引擎搜索子问题,并根据标题、原始 URL、摘要等信息评估每个结果的相关性和可靠性。
    • 决定是否需要进一步浏览网页。
    • 点击需要进一步探索的网页,评估内容是否对研究问题有帮助,提取相关信息并记录。
    • 将所有记录的相关信息汇总,并撰写一份解决研究问题的报告。

    研究员的架构

    MetaGPT 的研究员角色包含三个主要动作:

    • CollectLinks: 从搜索引擎收集链接。
    • WebBrowseAndSummarize: 浏览网页并提供摘要。
    • ConductResearch: 进行研究并生成研究报告。

    动作定义

    CollectLinks

    CollectLinks 动作用于搜索互联网以获取相关问题并检索 URL 地址列表。由于用户输入的问题可能不直接适合搜索引擎查询,CollectLinks 动作首先将用户的问题分解成多个适合搜索的子问题。然后,它使用搜索引擎进行搜索。

    实现使用了 tools 模块中的 SearchEngine,支持通过 serpapi/google/serper/ddg 进行搜索。实现细节可以在 metagpt/actions/research.py 文件中找到,以下是对 CollectLinks.run 方法的基本解释:

    class CollectLinks(Action):
    
        async def run(
            self,
            topic: str,
            decomposition_nums: int = 4,
            url_per_query: int = 4,
            system_text: str | None = None,
        ) -> dict[str, list[str]]:
            """运行动作以收集链接。
    
            Args:
                topic: 研究主题。
                decomposition_nums: 要生成的搜索问题的数量。
                url_per_query: 每个搜索问题要收集的 URL 数量。
                system_text: 系统文本。
    
            Returns:
                一个字典,包含搜索问题作为键,收集的 URL 作为值。
            """
            system_text = system_text if system_text else RESEARCH_TOPIC_SYSTEM.format(topic=topic)
            # 将研究问题分解成多个子问题
            keywords = await self._aask(SEARCH_TOPIC_PROMPT, [system_text])
            try:
                keywords = OutputParser.extract_struct(keywords, list)
                keywords = parse_obj_as(list[str], keywords)
            except Exception as e:
                logger.exception(f"fail to get keywords related to the research topic \"{topic}\" for {e}")
                keywords = [topic]
    
            # 使用搜索引擎搜索子问题
            results = await asyncio.gather(*(self.search_engine.run(i, as_string=False) for i in keywords))
    
            # 浏览搜索引擎结果,并过滤掉与研究问题无关的结果
            def gen_msg():
                while True:
                    search_results = "\n".join(f"#### Keyword: {i}\n Search Result: {j}\n" for (i, j) in zip(keywords, results))
                    prompt = SUMMARIZE_SEARCH_PROMPT.format(decomposition_nums=decomposition_nums, search_results=search_results)
                    yield prompt
                    remove = max(results, key=len)
                    remove.pop()
                    if len(remove) == 0:
                        break
            prompt = reduce_message_length(gen_msg(), self.llm.model, system_text, CONFIG.max_tokens_rsp)
            logger.debug(prompt)
            queries = await self._aask(prompt, [system_text])
            try:
                queries = OutputParser.extract_struct(queries, list)
                queries = parse_obj_as(list[str], queries)
            except Exception as e:
                logger.exception(f"fail to break down the research question due to {e}")
                queries = keywords
            ret = {}
    
            # 对搜索结果进行排序,并获取 TopK URL
            for query in queries:
                ret[query] = await self._search_and_rank_urls(topic, query, url_per_query)
            return ret

    WebBrowseAndSummarize

    WebBrowseAndSummarize 动作负责浏览网页并对其内容进行摘要。MetaGPT 在 tools 模块中提供了 WebBrowserEngine,它支持通过 playwright/selenium 进行网页浏览。WebBrowseAndSummarize 动作使用 WebBrowserEngine 进行网页浏览。

    实现细节可以在 metagpt/actions/research.py 文件中找到,以下是对 WebBrowseAndSummarize.run 方法的基本解释:

    class WebBrowseAndSummarize(Action):
        async def run(
            self,
            url: str,
            *urls: str,
            query: str,
            system_text: str = RESEARCH_BASE_SYSTEM,
        ) -> dict[str, str]:
            """运行动作以浏览网页并提供摘要。
    
            Args:
                url: 要浏览的主要 URL。
                urls: 要浏览的其他 URL。
                query: 研究问题。
                system_text: 系统文本。
    
            Returns:
                一个字典,包含 URL 作为键,其摘要作为值。
            """
            # 网页浏览和内容提取
            contents = await self.web_browser_engine.run(url, *urls)
            if not urls:
                contents = [contents]
    
            # 网页内容摘要
            summaries = {}
            prompt_template = WEB_BROWSE_AND_SUMMARIZE_PROMPT.format(query=query, content="{}")
            for u, content in zip([url, *urls], contents):
                content = content.inner_text
                chunk_summaries = []
                for prompt in generate_prompt_chunk(content, prompt_template, self.llm.model, system_text, CONFIG.max_tokens_rsp):
                    logger.debug(prompt)
                    summary = await self._aask(prompt, [system_text])
                    if summary == "Not relevant.":
                        continue
                    chunk_summaries.append(summary)
    
                if not chunk_summaries:
                    summaries[u] = None
                    continue
    
                if len(chunk_summaries) == 1:
                    summaries[u] = chunk_summaries[0]
                    continue
    
                content = "\n".join(chunk_summaries)
                prompt = WEB_BROWSE_AND_SUMMARIZE_PROMPT.format(query=query, content=content)
                summary = await self._aask(prompt, [system_text])
                summaries[u] = summary
            return summaries

    ConductResearch

    ConductResearch 动作负责撰写研究报告。它使用 WebBrowseAndSummarize 动作的摘要数据作为上下文,然后生成研究报告。

    实现细节可以在 metagpt/actions/research.py 文件中找到,以下是对 ConductResearch.run 方法的基本解释:

    class ConductResearch(Action):
        async def run(
            self,
            topic: str,
            content: str,
            system_text: str = RESEARCH_BASE_SYSTEM,
        ) -> str:
            """运行动作以进行研究并生成研究报告。
    
            Args:
                topic: 研究主题。
                content: 研究内容。
                system_text: 系统文本。
    
            Returns:
                生成的 research report。
            """
            prompt = CONDUCT_RESEARCH_PROMPT.format(topic=topic, content=content)
            logger.debug(prompt)
            self.llm.auto_max_tokens = True
            return await self._aask(prompt, [system_text])

    研究员角色

    研究员角色将 CollectLinksWebBrowseAndSummarizeConductResearch 动作结合在一起,实现了搜索互联网并汇总报告的功能。因此,在初始化时需要使用 set_actions 方法将这三个动作添加到角色中。由于这些动作按照 CollectLinks -> WebBrowseAndSummarize -> ConductResearch 的顺序执行,因此需要在 react/_act 方法中定义这些动作的执行逻辑。

    实现细节可以在 metagpt/roles/researcher.py 文件中找到,以下是对 Researcher 类的基本解释:

    class Researcher(Role):
        def __init__(
            self,
            name: str = "David",
            profile: str = "Researcher",
            goal: str = "Gather information and conduct research",
            constraints: str = "Ensure accuracy and relevance of information",
            language: str = "en-us",
            **kwargs,
        ):
            super().__init__(name, profile, goal, constraints, **kwargs)
    
            # 添加 `CollectLinks`、`WebBrowseAndSummarize` 和 `ConductResearch` 动作
            self.set_actions([CollectLinks(name), WebBrowseAndSummarize(name), ConductResearch(name)])
    
            # 设置按顺序执行
            self._set_react_mode(react_mode="by_order")
            self.language = language
            if language not in ("en-us", "zh-cn"):
                logger.warning(f"The language `{language}` has not been tested, it may not work.")
    
        async def _act(self) -> Message:
            logger.info(f"{self._setting}: ready to {self.rc.todo}")
            todo = self.rc.todo
            msg = self.rc.memory.get(k=1)[0]
            if isinstance(msg.instruct_content, Report):
                instruct_content = msg.instruct_content
                topic = instruct_content.topic
            else:
                topic = msg.content
    
            research_system_text = get_research_system_text(topic, self.language)
            # 搜索互联网并检索 URL 信息
            if isinstance(todo, CollectLinks):
                links = await todo.run(topic, 4, 4)
                ret = Message(content="", instruct_content=Report(topic=topic, links=links), role=self.profile, cause_by=todo)
            # 浏览网页并对其内容进行摘要
            elif isinstance(todo, WebBrowseAndSummarize):
                links = instruct_content.links
                todos = (todo.run(*url, query=query, system_text=research_system_text) for (query, url) in links.items())
                summaries = await asyncio.gather(*todos)
                summaries = list((url, summary) for i in summaries for (url, summary) in i.items() if summary)
                ret = Message(content="", instruct_content=Report(topic=topic, summaries=summaries), role=self.profile, cause_by=todo)
            # 生成 research report
            else:
                summaries = instruct_content.summaries
                summary_text = "\n---\n".join(f"url: {url}\nsummary: {summary}" for (url, summary) in summaries)
                content = await self.rc.todo.run(topic, summary_text, system_text=research_system_text)
                ret = Message(content="", instruct_content=Report(topic=topic, content=content), role=self.profile, cause_by=type(self.rc.todo))
            self.rc.memory.add(ret)
            return ret
    
        async def react(self) -> Message:
            msg = await super().react()
            report = msg.instruct_content
            # 输出 report
            self.write_report(report.topic, report.content)
            return msg

    使用说明

    依赖和配置

    研究员角色依赖于 SearchEngineWebBrowserEngine。以下是安装和配置这些组件的简要说明。

    SearchEngine

    支持 serpapi/google/serper/ddg 搜索引擎。它们的区别如下:

    名称默认引擎附加依赖包安装
    serpapipip install metagpt[search-google]
    google×google-api-python-clientpip install metagpt[search-google]
    serper×
    ddg×duckduckgo-searchpip install metagpt[search-ddg]

    配置:

    WebBrowserEngine

    支持 playwright/selenium 引擎。要使用它们,必须安装额外的依赖项。它们的区别如下:

    名称默认引擎附加依赖包安装异步支持的平台
    playwrightplaywright, beautifulsoup4pip install metagpt[playwright]Native部分支持的平台
    selenium×selenium, webdriver_manager, beautifulsoup4pip install metagpt[selenium]线程池几乎所有平台

    配置:

    • playwright:
      • browser.engine: 设置为 playwright
      • browser.browser_type: 支持 chromium/firefox/webkit,默认值为 chromium。更多信息: Playwright BrowserTypes
    • selenium:
      • browser.engine: 设置为 selenium
      • browser.browser_type: 支持 chrome/firefox/edge/ie,默认值为 chrome。更多信息: Selenium BrowserTypes

    运行示例和结果

    metagpt.roles.researcher 模块提供了一个命令行界面,用于执行研究员的功能。以下是一个示例:

    python3 -m metagpt.roles.researcher "tensorflow vs. pytorch"

    日志输出:log.txt
    报告输出:dataiku vs. datarobot.md

    总结

    MetaGPT 的研究员角色可以帮助你快速高效地从互联网上搜集信息并撰写研究报告。它可以节省你大量时间和精力,让你专注于更重要的工作。

    更多学习资源

人生梦想 - 关注前沿的计算机技术 acejoy.com 🐾 步子哥の博客 🐾 背多分论坛 🐾 借一步网
Page Stats: PV: 1 | UV: 1
Last updated: 2025-07-11 09:59:31
沪ICP备2024052574号-1