标签： AGI

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-client`	`pip install metagpt[search-google]`
serper	×
ddg	×	`duckduckgo-search`	`pip install metagpt[search-ddg]`

配置：

serpapi:
- search.engine: 设置为 serpapi
- search.api_key: 从 https://serpapi.com/ 获取
- search.params: 要传递给搜索引擎的附加参数，例如：
google:
- search.engine: 设置为 google
- search.api_key: 从 https://console.cloud.google.com/apis/credentials 获取
- search.cse_id: 从 https://programmablesearchengine.google.com/controlpanel/create 获取
serper:
- search.engine: 设置为 serper
- search.api_key: 从 https://serper.dev/ 获取
ddg:
- search.engine: 设置为 ddg

WebBrowserEngine

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

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

更多学习资源

MetaGPT 文档：了解更多关于 MetaGPT 的信息。
MetaGPT Github：查看 MetaGPT 的源代码和示例。
MetaGPT 论文：深入了解 MetaGPT 的技术细节。

2024 年 6 月 19 日

MetaGPT 数据可视化：让数据说话
数据可视化是将数据以图表、图形等视觉形式呈现的过程，它可以帮助我们发现数据中的模式、趋势和相关性，并提供洞察和理解。通过数据可视化，我们可以更好地理解数据的含义，传达和解释结果，并支持数据驱动的决策和沟通。

示例：可视化鸢尾花数据集

任务： 使用 DataInterpreter 对 sklearn 的鸢尾花数据集进行简单的分析和可视化。

代码：
```
python examples/di/data_visualization.py
```
examples/di/data_visualization.py 文件中的代码如下：
```
import asyncio
from metagpt.logs import logger
from metagpt.roles.di.data_interpreter import DataInterpreter
from metagpt.utils.recovery_util import save_history

async def main(requirement: str = ""):

    di = DataInterpreter()
    rsp = await di.run(requirement)
    logger.info(rsp)
    save_history(role=di)


if __name__ == "__main__":

    requirement = "对 sklearn 的鸢尾花数据集进行数据分析，并包含一个图表"
    asyncio.run(main(requirement))
```
执行上述代码后，生成的计划和代码将分别保存在 data/output/current_time/plan.json 和 data/output/current_time/code.ipynb 文件中。

执行结果

DataInterpreter 提出了以下解决方案任务：
```
[
  {
    "task_id": "1",
    "dependent_task_ids": [],
    "instruction": "从 sklearn 加载鸢尾花数据集。"
  },
  {
    "task_id": "2",
    "dependent_task_ids": ["1"],
    "instruction": "对鸢尾花数据集进行探索性数据分析。"
  },
  {
    "task_id": "3",
    "dependent_task_ids": ["2"],
    "instruction": "创建图表来可视化鸢尾花数据集的特征。"
  }
]
```
DataInterpreter 能够将问题分解成逻辑任务，并按照加载数据、分析数据和绘制图表步骤进行执行。

DataInterpreter 生成的代码如下：
```
# ----------------------------------task1------------------------------------
from sklearn.datasets import load_iris
iris_data = load_iris()
iris_data.keys()
!pip install scikit-learn
from sklearn.datasets import load_iris
iris_data = load_iris()
iris_data.keys()
# ----------------------------------task2------------------------------------
import pandas as pd

# 从鸢尾花数据集创建 DataFrame
iris_df = pd.DataFrame(iris_data['data'], columns=iris_data['feature_names'])
iris_df['species'] = pd.Categorical.from_codes(iris_data['target'], iris_data['target_names'])

# 汇总统计
summary_statistics = iris_df.describe()

# 检查缺失值
missing_values = iris_df.isnull().sum()

(summary_statistics, missing_values)
# ----------------------------------task3------------------------------------
import matplotlib.pyplot as plt
import seaborn as sns

# 使用 seaborn 的 pairplot 可视化数据集特征
sns.set(style='whitegrid', context='notebook')
iris_pairplot = sns.pairplot(iris_df, hue='species', height=2.5)
plt.show()
```
在执行任务 1 时，由于环境中缺少 scikit-learn，第一次执行时发生了错误。但是，DataInterpreter 能够分析并解决这个问题，通过安装 scikit-learn 来解决。在任务 3 中，DataInterpreter 使用 seaborn 的 pairplot 函数创建了一个散点图矩阵，它可视化了数据集不同特征之间的关系，并使用颜色区分了不同物种的数据点。最后，使用 plt.show() 显示图表。

以下是 DataInterpreter 运行代码后绘制的图表。很明显，代码成功执行并生成了一个漂亮的可视化表格，它可以帮助我们更有效地分析数据集的特征。

总结

这个例子展示了 DataInterpreter 如何使用数据可视化来帮助我们分析和理解数据。它能够自动生成代码，并根据我们的需求选择合适的可视化方法。

更多学习资源
- MetaGPT 文档：了解更多关于 MetaGPT 的信息。
- MetaGPT Github：查看 MetaGPT 的源代码和示例。
- MetaGPT 论文：深入了解 MetaGPT 的技术细节。
2024 年 6 月 19 日