【AI的未来 - AI Agent系列】【MetaGPT】4. ActionNode从理论到实战

文章目录

    • 0. ActionNode基础
      • 0.1 官方解释
      • 0.2 我的理解
      • 0.3 ActionNode的数据结构
      • 0.4 如何使用ActionNode
    • 1. ActionNode简单实战
      • 1.1 思考并返回特定格式的数字
        • 1.1.1 定义两个ActionNode
        • 1.1.2 为这两个动作节点设置一个父节点
        • 1.1.3 定义一个Action来承载上面的ActionNode
      • 1.2 逐个打印数字
      • 1.3 实现Role,执行Action
    • 2. 完整代码和运行效果
      • 2.1 完整代码
      • 2.2 运行效果

0. ActionNode基础

0.1 官方解释

在MG框架0.5版本中,新增加了ActionNode类,为Agent的动作执行提供更强的能力
ActionNode可以被视为一组动作树,根据类内定义,一个动作树的父节点可以访问所有的子动作节点;也就是说,定义了一个完整的动作树之后,可以从父节点按树的结构顺序执行每一个子动作节点。因此,动作的执行也可以突破0.4版本框架中,需要在Role的_react内循环执行的限制,达到更好的CoT效果。

0.2 我的理解

如下图,ActionNode需要内置在Action中使用,之前可能是Action就是一个动作,执行完一个就执行完下一个。有了ActionNode后,Action就不只是一个动作,而可能是一系列动作。这一系列动作可以组织成链式结构,或者树状结构或者更复杂的结构。

在这里插入图片描述

0.3 ActionNode的数据结构

schema: str  # raw/json/markdown, default: ""

# Action Context
context: str  # all the context, including all necessary info
llm: BaseLLM  # LLM with aask interface
children: dict[str, "ActionNode"]

# Action Input
key: str  # Product Requirement / File list / Code
expected_type: Type  # such as str / int / float etc.
# context: str  # everything in the history.
instruction: str  # the instructions should be followed.
example: Any  # example for In Context-Learning.

# Action Output
content: str
instruct_content: BaseModel

其中几个重要的成员变量(以我目前浅显使用过的例子来看):

  • instruction:一般是prompt的部分内容
  • key:一般是ActionNode的名字
  • schema:指定该ActionNode的输出格式,指定为json或markdown之后会有严格的校验
  • expected_type:期望返回格式,例如str
  • example:类似prompt中的few-shot,给几个期望输出的例子

0.4 如何使用ActionNode

下面是官方教程给的例子:

# 定义单个子节点ActionNode
UI_DESIGN_DESC = ActionNode(
    key="UI Design Desc",
    expected_type=str,
    instruction="place the design objective here",
    example="Snake games are classic and addictive games with simple yet engaging elements. Here are the main elements"
    " commonly found in snake games",
)

# 定义完所有ActionNode之后,将其存放在一个List里面
NODES = [
    UI_DESIGN_DESC,
    SELECTED_ELEMENTS,
    HTML_LAYOUT,
    CSS_STYLES,
    ANYTHING_UNCLEAR,
]

# 将上述List的所有子节点传入父节点UI_DESIGN_NODE中
UI_DESIGN_NODE = ActionNode.from_children("UI_DESIGN", NODES)

从上面代码看到,使用ActionNode的步骤很简单,如下
(1)定义一系列ActionNode
(2)根据一系列ActionNode构造父ActionNode

之后,在Action.run方法中,调用父节点的执行方法fill,获取所有子节点按顺序执行之后的结果

ui_describe = await UI_DESIGN_NODE.fill(prompt)

上面我重点标出了"在Action.run方法中",说明ActionNode一定是依附在Action中运行的,在Action外不能独立运行?

1. ActionNode简单实战

上面了解了ActionNode的定义方法,以及运行fill函数,下面以一个简单的例子,实战一下。实战内容为 《MetaGPT智能体开发入门》课程中的例子:打印前10个斐波那契数列的数字,实现内容如下:

  • (1)LLM要能以特定的可解析的格式来返回斐波那契数列
  • (2)通过格式解析实现逐个打印数字的效果。

不重要:斐波那契数列是一个数列,每个数都是前两个数的和,通常以0和1开始

1.1 思考并返回特定格式的数字

将第一个实现内容(LLM要能以特定的可解析的格式来返回斐波那契数列)拆解:

  • (1)思考前10个斐波那契数列的数字是什么
  • (2)思考到的数字按特定格式输出

这就可以用两个ActionNode来实现。下面我们具体来实现。

1.1.1 定义两个ActionNode
# 将思考斐波那契数列的10个数字作为prompt输入,在这里我们将“思考需要生成的数字列表”作为命令(instruction)写入
# 将期望返回格式(expected_type)设置为str,无需设置例子(example)
SIMPLE_THINK_NODE = ActionNode(
    key="Simple Think Node",
    expected_type=str,
    instruction="""
            Think about what list of numbers you need to generate
            """,
    example=""
)

# 在这里通过命令(instruction)来规定需要生成的数字列表格式,提供例子(example)来帮助LLM理解
SIMPLE_CHECK_NODE = ActionNode(
    key="Simple CHECK Node",
    expected_type=str,
    instruction="""
            Please provide the number list for me, strictly following the following requirements:
            1. Answer strictly in the list format like [1,2,3,4]
            2. Do not have extra spaces or line breaks.
            Return the list here:
            """,
    example="[1,2,3,4]"
            "[4,5,6]",
 )
1.1.2 为这两个动作节点设置一个父节点
class THINK_NODES(ActionNode):
    def __init__(self, name="Think Nodes", expected_type=str, instruction="", example=""):
        super().__init__(key=name, expected_type=expected_type, instruction=instruction, example=example)
        self.add_children([SIMPLE_THINK_NODE, SIMPLE_CHECK_NODE])    # 初始化过程,将上面实现的两个子节点加入作为THINK_NODES类的子节点

    async def fill(self, context, llm, schema="raw", mode="auto", strgy="complex"):
        self.set_llm(llm)
        self.set_context(context)
        if self.schema:
            schema = self.schema

        if strgy == "simple":
            return await self.simple_fill(schema=schema, mode=mode)
        elif strgy == "complex":
            # 这里隐式假设了拥有children
            child_context = context    # 输入context作为第一个子节点的context
            for _, i in self.children.items():
                i.set_context(child_context)    # 为子节点设置context
                child = await i.simple_fill(schema=schema, mode=mode)
                child_context = child.content    # 将返回内容(child.content)作为下一个子节点的context

            self.content = child_context    # 最后一个子节点返回的内容设置为父节点返回内容(self.content)
            return self

为什么需要设置父节点?

  • ActionNode的 fill 方法,有一个参数叫“strgy”,当我们将这个参数设置为“complex”时,这个方法会按顺序执行每一个子节点,并将上一个子节点返回的内容作为下一个子节点的prompt。为了将两个动作节点串联起来,形成一个简单的CoT效果,我们需要设置一个父节点。
1.1.3 定义一个Action来承载上面的ActionNode

前文已经说了,ActionNode的运行需要依赖Action的动作,所以这里需要定义一个Action:

class ThinkAction(Action):
    def __init__(self, name="ThinkAction", context=None, llm=None):
        super().__init__()
        self.node = THINK_NODES()    # 初始化Action时,初始化一个THINK_NODE实例并赋值给self.node

    async def run(self, instruction) -> list:
        PROMPT = """
            You are now a number list generator, follow the instruction {instruction} and 
            generate a number list to be printed please.
            """

        prompt = PROMPT.format(instruction=instruction)
        rsp_node = await self.node.fill(context=prompt, llm=self.llm, schema="raw",
                                        strgy="complex")  # 1. 运行子节点,获取返回(返回格式为ActionNode)(注意设置 schema="raw") 2. 注意strgy为complex,表示执行所有子节点,如果是"simple", 则只会执行父节点本身
        rsp = rsp_node.content  # 获取返回的文本内容,返回的是ActionNode,通过.content来获取实际内容
        rsp_match = self.find_in_brackets(rsp)  # 按列表格式解析返回的文本内容,定位“[”与“]”之间的内容

        try:
            rsp_list = list(map(int, rsp_match[0].split(',')))  # 按列表格式解析返回的文本内容,按“,”对内容进行分割,并形成一个python语法中的列表
            return rsp_list
        except:
            return []

    @staticmethod
    def find_in_brackets(s):
        pattern = r'\[(.*?)\]'
        match = re.findall(pattern, s)
        return match

这个Action的几个重点关注点:

  • (1)初始化中self.node = THINK_NODES(),将ActionNode依附在Action中。
  • (2)Action的run方法中执行ActionNode的动作:await self.node.fill(context=prompt, llm=self.llm, schema="raw", strgy="complex")
    • 其中注意schema为"raw"
    • strgy为“complex”,表示会执行完 THINK_NODES()中的所有子节点才会返回

1.2 逐个打印数字

上面的程序将前10个数字解析出来并形成了Python中的list数组。下面实现逐个打印的Action

class SimplePrint(Action):
    input_num: int = 0

    def __init__(self, name="SimplePrint", input_num:int=0):
        super().__init__()
        self.input_num = input_num

    async def run(self):
        print(str(self.input_num) + "\n")
        return str(self.input_num)

1.3 实现Role,执行Action

有了Action,需要有个Role执行它。Role的代码和细节注释如下:

class Printer(Role):

    def __init__(self, name="TXXZ", profile="Printer", goal="Print the number", constraints=""):
        super().__init__()
        self._init_actions([ThinkAction]) ## 1. 将Action加入Role的执行列表

    async def _think(self) -> None:
        """Determine the action"""
        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 _prepare_print(self, num_list:list) -> Message:
        """Add actions"""
        actions = list()
        for num in num_list: ## 2. 对于Action返回的数组,逐个添加SimplePrint动作
            actions.append(SimplePrint(input_num=num))

        self._init_actions(actions) ## 4. 这里第一个action变成了SimplePrint动作
        self.rc.todo = None ## 3. 为None时,_think函数会回到第一个action执行
        return Message(content=str(num_list))

    async def _act(self) -> Message:
        """Action"""
        todo = self.rc.todo
        if type(todo) is ThinkAction :
            msg = self.rc.memory.get(k=1)[0]
            self.goal = msg.content
            resp = await todo.run(instruction=self.goal) # 7. 个人感觉这里的goal有和没有都没关系,虽然作为prompt传入ThinkAction,但是这里并不是打印Action,与任务无关
            return await self._prepare_print(resp) ## 5. ActionNode都执行完了,返回的是个数组,逐个去添加打印Action

        resp = await todo.run() ## 6. 执行打印Action
        return Message(content=resp, role=self.profile)

    async def _react(self) -> Message:
        while True:
            await self._think()
            if self.rc.todo is None:
                break
            msg = await self._act()
        return msg

2. 完整代码和运行效果

2.1 完整代码

# 加载 .env 到环境变量
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())

import asyncio
import re

from metagpt.actions.action import Action, ActionNode
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message

# 将思考斐波那契数列的10个数字作为prompt输入,在这里我们将“思考需要生成的数字列表”作为命令(instruction)写入
# 将期望返回格式(expected_type)设置为str,无需设置例子(example)
SIMPLE_THINK_NODE = ActionNode(
    key="Simple Think Node",
    expected_type=str,
    instruction="""
            Think about what list of numbers you need to generate
            """,
    example=""
)

# 在这里通过命令(instruction)来规定需要生成的数字列表格式,提供例子(example)来帮助LLM理解
SIMPLE_CHECK_NODE = ActionNode(
    key="Simple CHECK Node",
    expected_type=str,
    instruction="""
            Please provide the number list for me, strictly following the following requirements:
            1. Answer strictly in the list format like [1,2,3,4]
            2. Do not have extra spaces or line breaks.
            Return the list here:
            """,
    example="[1,2,3,4]"
            "[4,5,6]",
 )


class THINK_NODES(ActionNode):
    def __init__(self, name="Think Nodes", expected_type=str, instruction="", example=""):
        super().__init__(key=name, expected_type=expected_type, instruction=instruction, example=example)
        self.add_children([SIMPLE_THINK_NODE, SIMPLE_CHECK_NODE])    # 初始化过程,将上面实现的两个子节点加入作为THINK_NODES类的子节点

    async def fill(self, context, llm, schema="raw", mode="auto", strgy="complex"):
        self.set_llm(llm)
        self.set_context(context)
        if self.schema:
            schema = self.schema

        if strgy == "simple":
            return await self.simple_fill(schema=schema, mode=mode)
        elif strgy == "complex":
            # 这里隐式假设了拥有children
            child_context = context    # 输入context作为第一个子节点的context
            for _, i in self.children.items():
                i.set_context(child_context)    # 为子节点设置context
                child = await i.simple_fill(schema=schema, mode=mode)
                child_context = child.content    # 将返回内容(child.content)作为下一个子节点的context

            self.content = child_context    # 最后一个子节点返回的内容设置为父节点返回内容(self.content)
            return self


class SimplePrint(Action):
    """
    Action that print the num inputted
    """
    input_num: int = 0

    def __init__(self, name="SimplePrint", input_num:int=0):
        super().__init__()

        self.input_num = input_num

    async def run(self):
        print(str(self.input_num) + "\n")
        return str(self.input_num)

class ThinkAction(Action):
    """
    Action that think
    """

    def __init__(self, name="ThinkAction", context=None, llm=None):
        super().__init__()
        self.node = THINK_NODES()    # 初始化Action时,初始化一个THINK_NODE实例并赋值给self.node

    async def run(self, instruction) -> list:
        PROMPT = """
            You are now a number list generator, follow the instruction {instruction} and 
            generate a number list to be printed please.
            """

        prompt = PROMPT.format(instruction=instruction)
        rsp_node = await self.node.fill(context=prompt, llm=self.llm, schema="raw",
                                        strgy="complex")  # 运行子节点,获取返回(返回格式为ActionNode)(注意设置 schema="raw" )
        rsp = rsp_node.content  # 获取返回的文本内容

        rsp_match = self.find_in_brackets(rsp)  # 按列表格式解析返回的文本内容,定位“[”与“]”之间的内容

        try:
            rsp_list = list(map(int, rsp_match[0].split(',')))  # 按列表格式解析返回的文本内容,按“,”对内容进行分割,并形成一个python语法中的列表

            return rsp_list
        except:
            return []

    @staticmethod
    def find_in_brackets(s):
        pattern = r'\[(.*?)\]'
        match = re.findall(pattern, s)
        return match

class Printer(Role):

    def __init__(self, name="Jerry", profile="Printer", goal="Print the number", constraints=""):
        super().__init__()

        self._init_actions([ThinkAction])
        # self.num_list = list()

    async def _think(self) -> None:
        """Determine the action"""
        # logger.info(self.rc.state)

        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 _prepare_print(self, num_list:list) -> Message:
        """Add actions"""
        actions = list()

        for num in num_list:
            actions.append(SimplePrint(input_num=num))

        self._init_actions(actions)
        self.rc.todo = None
        return Message(content=str(num_list))

    async def _act(self) -> Message:
        """Action"""
        todo = self.rc.todo

        if type(todo) is ThinkAction :
            msg = self.rc.memory.get(k=1)[0]
            self.goal = msg.content
            resp = await todo.run(instruction=self.goal)
            # logger.info(resp)

            return await self._prepare_print(resp)

        resp = await todo.run()
        # logger.info(resp)

        return Message(content=resp, role=self.profile)

    async def _react(self) -> Message:
        """"""
        while True:
            await self._think()

            if self.rc.todo is None:
                break
            msg = await self._act()

        return msg


async def main():
    msg = "Provide the first 10 numbers of the Fibonacci series"
    role = Printer()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)


if __name__ == '__main__':
    asyncio.run(main())

2.2 运行效果

在这里插入图片描述

本文代码是经过修改的,可以直接运行。

  • 前提:使用 MetaGPT 0.6+ 的版本,我这里用的是github最新代码(2024-01-16),源码编译的。

未完待续… 请移步下篇文章 【AI的未来 - AI Agent系列】【MetaGPT】4.1 细说我在ActionNode实战中踩的那些坑
带你绕过很多坑!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/330956.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

Baichuan2百川模型部署的bug汇总

1.4bit的量化版本最好不要在Windows系统中运行&#xff0c;大概原因报错原因是bitsandbytes不支持window&#xff0c;bitsandbytes-windows目前仅支持8bit量化。 2. 报错原因是机器没有足够的内存和显存&#xff0c;offload_folder设置一个文件夹来保存那些离线加载到硬盘的权…

spring springfox-swagger2 2.7.0配置

springboot版本 <parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.0.0.RELEASE</version><relativePath/> <!-- lookup parent from repository -->…

什么是小程序?特点和技术架构详解

小程序是一种新的移动应用程序格式&#xff0c;一种结合了 Web 技术以及客户端技术的混合解决方案。 传统的原生应用运行起来比较流畅&#xff0c;但是也有天然的基因缺陷&#xff1a; 不支持动态化&#xff0c;发布周期长需要开发Android和iOS两套代码&#xff0c;开发成本高…

使用原生input模拟器样式正常,但是真机上 input框溢出

目录 一、问题 二、解决方法 三、总结 tiips:如嫌繁琐&#xff0c;直接移步总结即可&#xff01; 一、问题 1.使用原生input写了一个搜索框&#xff0c;在模拟器和pc上一切正常。但是打包放到手机上&#xff0c;样式就有问题&#xff1a;这个搜索框的布局是正常的&#xf…

pycharm管理仓库(Manager Repository)不见了

经常使用pycharm的大佬们都知道&#xff0c;pycharm中内置了很多库和包&#xff0c;很好用 但是下载来用自带的源很麻烦&#xff0c;于是就用国内的源 可以当我们添加管理仓库的时候&#xff0c;却没有了按钮&#xff0c;如何解决呢&#xff1f; 回到pycharm的主界面&#xf…

【C++】STL 算法 - 累加填充算法 ( 元素累加算法 - accumulate 函数 | 元素填充算法 - fill 函数 )

文章目录 一、元素累加算法 - accumulate 函数1、函数原型分析2、代码示例 二、元素填充算法 - fill 函数1、函数原型分析2、代码示例 一、元素累加算法 - accumulate 函数 1、函数原型分析 在 C 语言 的 标准模板库 ( STL , STL Standard Template Library ) 中 , 提供了 accu…

Unity3D代码混淆方案详解

背景 Unity引擎使用Mono运行时&#xff0c;而C#语言易受反编译影响&#xff0c;存在代码泄露风险。本文通过《QQ乐团》项目实践&#xff0c;提出一种适用于Unity引擎的代码混淆方案&#xff0c;以保护代码逻辑。 引言 在Unity引擎下&#xff0c;为了防止代码被轻易反编译&a…

AVL树 -- C++实现

AVL树 – C实现 1. AVL树的概念 二叉搜索树虽可以缩短查找的效率&#xff0c;但如果数据有序或接近有序二叉搜索树将退化为单支树&#xff0c;查找元素相当于在顺序表中搜索元素&#xff0c;效率低下。因此&#xff0c;两位俄罗斯的数学家G.M.Adelson-Velskii和E.M.Landis在1…

组件v-model(.sync)记录使用(vue3)

示例&#xff08;演示地址&#xff09; 以下是Vue3中使用v-model实现组件的双向数据绑定的示例代码&#xff1a; 首先&#xff0c;让我们来了解一下Vue3中v-model的用法。在Vue3中&#xff0c;v-model 指令可以用于自定义组件上&#xff0c;用于实现组件的双向数据绑定。与Vue2…

为什么选择HelpLook而不是Document360:知识库工具分析

在现今的信息化时代&#xff0c;企业们越来越倾向于使用知识库工具来收集、组织和分享他们的知识资源。HelpLook和Document360是市面上两款不错的知识库管理工具。那如果非要在他们之中选一个&#xff0c;还是建议HelpLook。以下是一些对比分析&#xff0c;希望可以给大家提供参…

Python通过HTTP代理进行数据传输

在Python中&#xff0c;HTTP代理是一种常用的技术&#xff0c;用于控制和修改HTTP请求和响应。通过使用HTTP代理&#xff0c;我们可以更好地控制网络请求的行为&#xff0c;提高安全性、隐私性和效率。下面我们将详细介绍如何在Python中使用HTTP代理进行数据传输。 一、设置HT…

tinyxml2

使用tinyxml2&#xff0c;得知道一些xml基础 xml tutorial--菜鸟 tinyxml2类对象 链接 结构 XMLNode 什么是节点 节点&#xff1a;元素、声明、文本、注释等。 XMLDocument xml文档(文件)对象。 作用&#xff1a; 加载xml文件&#xff0c; tinyxml2作用 先定义两个宏 …

Elasticsearch各种高级文档操作

本文来记录下Elasticsearch各种文档操作 文章目录 初始化文档数据查询所有文档匹配查询文档关键字精确查询文档多关键字精确查询文档字段匹配查询文档指定查询字段查询文档过滤字段查询文档概述指定想要显示的字段示例指定不想要显示的字段示例 组合查询文档范围查询文档概述使…

Redis+lua 解决秒杀超卖问题

商品超卖的原因是在高并发场景下,业务层面扣库存和库存的判断会出现并发情况&#xff0c;但是使用lua 脚本&#xff0c;就会避免超卖的发生。 1.在项目resources文件夹创建一个名为 stock.lua的脚本 if (redis.call(hexists, KEYS[1], KEYS[2]) 1) thenlocal stock tonumber…

HarmonyOS—构建第一个ArkTS应用(Stage模型)

创建ArkTS工程 构建第一个页面 若首次打开DevEco Studio&#xff0c;请点击Create Project创建工程。如果已经打开了一个工程&#xff0c;请在菜单栏选择File > New > Create Project来创建一个新工程。选择Application应用开发&#xff0c;选择模板“Empty Ability”&am…

Intel Quartus II IP之DP1.4 工程的创建与使用

前述&#xff1a; Win10电脑安装了Quartus 21.4&#xff0c;这可以满足绝大多数工程&#xff0c;特别是对于简单调用fifo/ram等的工程&#xff0c;但是想要学习Quartus的HDMI/DP等高速接口类IP&#xff0c;首先需要创建HDMI/DP IP的设计demo工程&#xff0c;此时还需要安装Ecl…

android studio从空白开始

对我来说&#xff0c;真正的第一步是清理电脑C盘。从剩余8G清理到25G&#xff0c;把原来看不顺眼又不敢删的文件夹和软件全删了&#xff0c;删爽了的后果就是&#xff0c;用两天的时间在把一些环境配置慢慢装回来&#xff0c;node.js&#xff0c;jdk&#xff0c;npm。努力把它们…

SG3225HBN(晶体振荡器)

SG3225HBN晶体振荡器是一款低抖动、HCSL输出晶振&#xff0c;频率范围100MHz ~ 325MHz&#xff0c;3.2 2.5 1.05mm封装常规有源晶振&#xff0c;具有小尺寸&#xff0c;高稳定性&#xff0c;可以在G&#xff1a;-40℃至 85℃&#xff0c;H&#xff1a;-40℃至 105℃ 的温度内…

网页版短信平台主要功能有哪些|短信系统开发搭建源码

网页版短信平台主要功能有哪些|短信系统开发搭建源码 网页版短信平台是一种方便用户进行在线短信管理和发送的工具。它具有多项实用的功能&#xff0c;使用户能够更高效地进行短信通讯。以下是网页版短信平台的主要功能&#xff1a; 1. **短信发送和接收&#xff1a;** 用户可…

linux第一个小程序 --- 进度条【简洁】

行缓冲区的概念 结果&#xff1a;先输入hello world然后休眠三秒后结束 当去掉’\n“ 后&#xff0c;结果就变成了先休眠三秒&#xff0c;然后打印hello world后结束。 该现象就证明了缓冲区的存在。 当缓冲区中遇到’‘\n’或者缓冲区被写满后才会被打印出来&#xff0c;在第…