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

研究员角色

研究员角色将 CollectLinks、WebBrowseAndSummarize 和 ConductResearch 动作结合在一起，实现了搜索互联网并汇总报告的功能。因此，在初始化时需要使用 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

使用说明

依赖和配置

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

SearchEngine

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

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

配置：

WebBrowserEngine

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

名称默认引擎附加依赖包安装异步支持的平台
playwright√playwright, 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 的研究员角色可以帮助你快速高效地从互联网上搜集信息并撰写研究报告。它可以节省你大量时间和精力，让你专注于更重要的工作。

更多学习资源

发表评论