MetaGPT-打卡-day2，MetaGPT框架组件学习

文章目录

- Agent组件
- 实现一个单动作的Agent
- 实现一个多动作的Agent
- 技术文档生成助手
- - 其他尝试

今天是第二天的打卡~昨天是关于一些概念的大杂烩，今天的话，就来到了Hello World环节。从单个Agnet到多个Agent，再到组合更复杂的工作流来解决问题。

Agent组件

虽然看过了一些资料，但是还是有些不知道该怎么去理解Agent的概念。从单词本身的翻译来看，很多都是经纪人、代理人的意思，关于称之为“数字人”或者“智能体”的方式，目前还是没有很好的理解到，希望在后面的学习中，能解答自己的困惑。
【By INSCODE AI创作助手】
在大模型中，Agent是指可以独立行动、具有决策能力和交互能力的实体。这些实体可以是人、机器人、虚拟角色等。Agent通常通过感知环境、收集信息、分析数据、制定策略和执行动作来实现自己的目标。

在大模型中，Agent的概念主要用于描述智能系统中的个体或实体，这些个体可以与其他Agent或环境进行交互，通过学习和适应来改进自己的行为和决策能力。

Agent通常具有以下特征：

自主性：Agent可以自主地根据自身的目标和环境条件制定行动计划，并进行决策。
学习能力：Agent可以通过学习和适应来改进自己的行为和决策能力。
感知能力：Agent可以感知环境中的信息，并根据这些信息做出相应的决策。
交互能力：Agent可以与其他Agent或环境进行交互，进行信息的传递和共享。
目标导向性：Agent具有特定的目标或任务，其行动和决策都是为了实现这些目标或完成这些任务。

在大模型中，Agent的概念常常用于建模和仿真复杂的智能系统，如人工智能系统、自动驾驶系统、多智能体系统等。Agent的设计和实现可以基于传统的规则和逻辑，也可以基于机器学习和深度学习等方法。 Agent的研究和应用可以帮助我们更好地理解和处理复杂的现实世界问题。

在MetaGPT中，Agent有如下的定义。

Agent=大语言模型LLM+观察+思考+行动+记忆。

在MetaGPT中定义的一个agent运行流程如下：

在这里插入图片描述

当启动定义好的Agent后，可以通过获取接收到的信息，进行分析，然后决定下一步的操作，然后紧接着执行相应的操作，得到这个环节的结果。

不过这里面感觉需要去解决的问题还很多（不过可以先不考虑这些内容，先让代码跑起来看看效果）：

观察、思考、提出方案等完全依赖大模型的效果
大模型有关于token的限制，这就需要考虑任务的信息量或者解决大模型的限制
新生信息的处理和分析
……

在metaggpt中，Role类是智能体的逻辑抽象，一个角色能够执行特定的动作，拥有记忆、思考、执行行动的能力。

实现一个单动作的Agent

import os

os.environ["ZHIPUAI_API_KEY"] = "秘钥"

import re
import asyncio
from bson import Code
from metagpt.actions import Action
from metagpt.roles import Role
from metagpt.logs import logger
from metagpt.schema import Message

class Coding(Action):
    PROMPT_TEMPLATE="""
    Write a python function that can {instruction} and provide test cases.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """
    
    def __init__(self,name="Coding",context=None,llm=None):
        super().__init__(name,context,llm)
    
    async def run(self,instruction:str,**kwargs):
        prompt=self.PROMPT_TEMPLATE.format(instruction=instruction)
        rsp=await self._aask(prompt)
        code_text=Coding.parse_code(rsp)
        return code_text

    @staticmethod
    def parse_code(text:str):
        pattern =r'```python\n(.*)```'
        match=re.search(pattern,text,re.DOTALL)
        code_text=match.group(1) if match else text
        return code_text
    
class AloneCoder(Role):
    def __init__(self, name="ZhangSan", profile="AloneCoder",**kwargs):
        super().__init__(name, profile, **kwargs)
        self._init_actions(actions=[Coding])
    async def _act(self) -> Message:
        logger.info(f"{self.name} is coding...")
        logger.info(f"{self._setting}:ready to {self._rc.todo}")
        todo=self._rc.todo

        msg=self.get_memories(k=1)[0]
        code_text=await todo.run(msg.content)
        msg=Message(content=code_text,role=self.profile,cause_by=type(todo))
        return msg
    
async def main():
    msg="编写一个函数，计算输入日期范围内的所有星期五"
    role=AloneCoder()
    logger.info(msg)
    result=await role.run(with_message=msg)
    logger.info(result)

asyncio.run(main())

输出：

2024-01-17 14:31:18.604 | INFO     | metagpt.const:get_metagpt_package_root:32 - Package root set to C:\Users\youxi\uenv\GPT\code
2024-01-17 14:31:19.452 | INFO     | __main__:main:53 - 编写一个函数，计算输入日期范围内的所有星期五
2024-01-17 14:31:19.454 | INFO     | __main__:_act:41 - ZhangSan is coding...
2024-01-17 14:31:19.455 | INFO     | __main__:_act:42 - ZhangSan(AloneCoder):ready to Coding
 ```python
from datetime import datetime, timedelta

def find_fridays(start_date, end_date):
    fridays = []
    start_date = datetime.strptime(start_date, '%Y-%m-%d')
end_date = datetime.strptime(end_date, '%Y-%m-%d')

    while start_date <= end_date:
        if start_date.weekday() == 4:  # 判断星期五
            fridays.append(start_date)
        start_date += timedelta(days=1)

    return fridays

# 测试案例
test_case_1 = ('2021-09-01', '2021-09-30')
test_case_2 = ('2021-01-01', '2021-12-31')

print(find_fridays(test_case_1[0], test_case_1[1]))
print(find_fridays(test_case_2[0], test_case_2[1]))
```

运行结果：

```
['2021-09-03', '2021-09-10', '2021-09-17', '2021-09-24']
['2021-01-08', '2021-01-15', '2021-01-22', '2021-01-29', '2021-02-05', '2021-02-12', '2021-02-19', '2021-02-26', '2021-03-05', '2021-03-12', '2021-03-19', '2021-03-26', '2021-04-02', '2021-04-09', '2021-04-16', '2021-04-23', '2021-04-30', '2021-05-07', '2021-05-14', '2021-05-21', '2021-05-28', '2021-06-04', '2021-06-11', '2021-06-18', '2021-06-25', '2021-07-02', '2021-07-09', '2021-07-16', '2021-07-23', '2021-07-30', '2021-08-06', '2021-08-13', '2021-08-20', '2021-08-27', '2021-09-03', '2021-09-10', '2021-09-17', '2021-09-24', '2021-09-30']
``
2024-01-17 14:31:57.378 | INFO     | metagpt.provider.openai_api:update_cost:91 - Total running cost: $0.001 | Max budget: $10.000 | Current cost: $0.001, prompt_tokens: 53, completion_tokens: 793
2024-01-17 14:31:57.380 | INFO     | __main__:main:55 - AloneCoder: 
from datetime import datetime, timedelta

def find_fridays(start_date, end_date):
    fridays = []
    start_date = datetime.strptime(start_date, '%Y-%m-%d')
    end_date = datetime.strptime(end_date, '%Y-%m-%d')

    while start_date <= end_date:
        if start_date.weekday() == 4:  # 判断星期五
            fridays.append(start_date)
        start_date += timedelta(days=1)

    return fridays

# 测试案例
test_case_1 = ('2021-09-01', '2021-09-30')
test_case_2 = ('2021-01-01', '2021-12-31')

print(find_fridays(test_case_1[0], test_case_1[1]))
print(find_fridays(test_case_2[0], test_case_2[1]))
```
运行结果：
```
['2021-09-03', '2021-09-10', '2021-09-17', '2021-09-24']
['2021-01-08', '2021-01-15', '2021-01-22', '2021-01-29', '2021-02-05', '2021-02-12', '2021-02-19', '2021-02-26', '2021-03-05', '2021-03-12', '2021-03-19', '2021-03-26', '2021-04-02', '2021-04-09', '2021-04-16', '2021-04-23', '2021-04-30', '2021-05-07', '2021-05-14', '2021-05-21', '2021-05-28', '2021-06-04', '2021-06-11', '2021-06-18', '2021-06-25', '2021-07-02', '2021-07-09', '2021-07-16', '2021-07-23', '2021-07-30', '2021-08-06', '2021-08-13', '2021-08-20', '2021-08-27', '2021-09-03', '2021-09-10', '2021-09-17', '2021-09-24', '2021-09-30']

`

实现一个多动作的Agent

对于上面仅执行一个动作的话，其实没必要这么折腾。所以Role的抽象，主要是用于动作的组合，通过连接动作，构建工作流来解决任务。

这里结合示例代码来实现一个根据自然语言描述，生成代码并允许的例子。

import os

os.environ["ZHIPUAI_API_KEY"] = "xxxxxxxxxxx"

import re
import asyncio
from metagpt.actions import Action
from metagpt.roles import Role
from metagpt.logs import logger
from metagpt.schema import Message

class Coding(Action):
    PROMPT_TEMPLATE="""
    Write a python function that can {instruction} and provide two runnnable test cases.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """
    
    def __init__(self,name="Coding",context=None,llm=None):
        super().__init__(name,context,llm)
    
    async def run(self,instruction:str,**kwargs):
        prompt=self.PROMPT_TEMPLATE.format(instruction=instruction)
        rsp=await self._aask(prompt)
        code_text=Coding.parse_code(rsp)
        return code_text

    @staticmethod
    def parse_code(text:str):
        pattern =r'```python\n(.*)```'
        match=re.search(pattern,text,re.DOTALL)
        code_text=match.group(1) if match else text
        return code_text
    
class AloneCoder(Role):
    def __init__(self, name="ZhangSan", profile="AloneCoder",**kwargs):
        super().__init__(name, profile, **kwargs)
        self._init_actions(actions=[Coding])
    async def _act(self) -> Message:
        logger.info(f"{self.name} is coding...")
        logger.info(f"{self._setting}:ready to {self._rc.todo}")
        todo=self._rc.todo

        msg=self.get_memories(k=1)[0]
        code_text=await todo.run(msg.content)
        msg=Message(content=code_text,role=self.profile,cause_by=type(todo))
        return msg
    
async def main():
    msg="编写一个函数，计算输入日期范围内的所有星期五"
    role=AloneCoder()
    logger.info(msg)
    result=await role.run(with_message=msg)
    logger.info(result)

asyncio.run(main())

输出：

2024-01-17 15:20:43.206 | INFO     | metagpt.const:get_metagpt_package_root:32 - Package root set to C:\Users\youxi\uenv\GPT\code
2024-01-17 15:20:44.092 | INFO     | __main__:main:74 - 编写一个函数，计算列表中数字的众数
2024-01-17 15:20:44.094 | INFO     | __main__:_act:63 - Yang(RunnableCoder):准备WriteCode
 ```python
def find_mode(data):
    if not data:
        return None
    frequency = {}
    for num in data:
        if num in frequency:
            frequency[num] += 1
        else:
            frequency[num] = 1

    max_count = 0
    mode = None
    for num, count in frequency.items():
        if count > max_count:
            max_count = count
            mode = num

    return mode

# 测试用例
data = [1, 2, 3, 3, 4, 4, 4, 5, 5, 5, 5, 6, 6, 6, 6, 7, 7, 7, 8, 8, 8, 9, 9, 9, 9, 10, 10, 10, 10, 11, 11, 11, 11, 12, 12, 12, 12, 13, 13, 13, 13, 14, 14, 14, 14, 15, 15, 15, 15, 16, 16, 16, 16, 17, 17, 17, 17, 18, 18, 18, 18, 19, 19, 19, 19, 20, 20, 20, 20]  
print(find_mode(data))  # 输出：5
``2024-01-17 15:21:03.128 | INFO     | metagpt.provider.openai_api:update_cost:91 - Total running cost: $0.000 | Max budget: $10.000 | Current cost: $0.000, prompt_tokens: 57, completion_tokens: 391
2024-01-17 15:21:03.130 | INFO     | __main__:_act:63 - Yang(RunnableCoder):准备RunCode
2024-01-17 15:21:03.215 | INFO     | __main__:run:46 - code_result='5\n'
2024-01-17 15:21:03.217 | INFO     | __main__:main:76 - RunnableCoder: 5

相比较单一的Agent，基于多个Agent就可以执行更多有趣的事情了。面对复杂的问题，我们可以把问题拆分开来，一点点的让大模型折腾，不过这样还是有些浪费人力，所以还是交给框架去解决。

技术文档生成助手

思路：由于token限制，不可能一下完成很复杂的内容处理，所以就可以这么样处理：先生成大纲，然后根据大纲生成每个标题的内容，最后再合并起来内容。这里的话，就可以创建两个Action：写大纲，写内容，然后再定义一个角色，输入需求，生成，然后合并结果。

import os
os.environ["ZHIPUAI_API_KEY"] = "xxxxxxxx"
from datetime import datetime
from typing import Dict
import asyncio
from metagpt.actions.write_tutorial import WriteDirectory, WriteContent
from metagpt.const import TUTORIAL_PATH
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.utils.file import File

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):
    """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, including {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}.
        """
        COMMON_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}".
        """

        DIRECTORY_PROMPT = COMMON_PROMPT + """
        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 strictly in the dictionary format like {{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}, {{"dir 2": ["sub dir 3", "sub dir 4"]}}]}}.
        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.
        """
        prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
        resp = await self._aask(prompt=prompt)
        return OutputParser.extract_struct(resp, dict)

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.
        """
        COMMON_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}".
        """
        CONTENT_PROMPT = COMMON_PROMPT + """
        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. Strict requirement not to output the topic "{topic}".
        """
        prompt = CONTENT_PROMPT.format(
            topic=topic, language=self.language, directory=self.directory)
        return await self._aask(prompt=prompt)

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."""
        logger.info(self._rc.state)
        logger.info(self,)
        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"
        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)
        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()
        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

async def main():
    msg = "Git 教程"
    role = TutorialAssistant()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)

asyncio.run(main())