文章目录
- 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实战中踩的那些坑
带你绕过很多坑!