本节课的任务
了解ActionNode，基于ActionNode（1）重写技术文档助手（2）完成一个短篇小说写作

ActionNode

关于为什么采用ActionNode的数据结构，参考官方文档为什么采用ActionNode的数据结构

这里我说一下在使用上比较重要的点：

• 为了结构化SOP（标准流程）输出而设计

• 为后续支COT、TOT、GOT等技术，解决非线性的复杂的SOP提供抽象数据结构

  
  
  image.png

比如像编写技术教程、编写小说这些场景中，如果只通过一个Prompt是很难实现的，同时也会受限于LLM的TOKEN限制，这个时候就可以考虑使用ActionNode。

在ActionNode基类中，也配置了更多格式检查和格式规范工具，让CoT执行过程中，内容的传递更加结构化。
ActionNode可以被视为一组动作树，而且仍然需要基于Action类进行构建。

ActionNode数据结构如下：

class ActionNode:
    """ActionNode is a tree of nodes."""

    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

ActionNode常用方法：

• add_child 增加子ActionNode
• add_children 批量增加子ActionNode
• from_children直接从一系列的子nodes初始化
• compile 编译ActionNode的Prompt模板
• _aask_v1使用ActionOutput封装调aask访问模型的输出
• fill 对ActionNode进行填槽，即实现执行传入的prompt并获取结果返回，并将结果存储在自身中

定义单个ActionNode

# 命令文本
DIRECTORY_STRUCTION = """
    您现在是互联网领域的经验丰富的技术专业人员。
    我们需要您撰写一个技术教程。
    """

# 实例化一个ActionNode，输入对应的参数
DIRECTORY_WRITE = ActionNode(
    # ActionNode的名称
    key="Directory Write",
    # 期望输出的格式
    expected_type=str,
    # 命令文本
    instruction=DIRECTORY_STRUCTION,
    # 例子输入，在这里我们可以留空
    example="",
 )

基于ActionNode重写技术文档助手

主要是重写WriteDirectory和WriteContent两个Action，分别修改为WriteDirectoryWithActionNode和WriteContentWithActionNode，
首先定义ActionNode，这里每个Action只定义了一个ActionNode，可能不能完全体现ActionNode的强大功能，除了格式化、对模型结果校验之外。

# 命令文本
DIRECTORY_STRUCTION = """
    You are now a seasoned technical professional in the field of the internet. 
    We need you to write a technical tutorial".
    """

DIRECTORY_PROMPT = """
The topic of tutorial is {topic}. 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.
"""
# 实例化一个ActionNode，输入对应的参数
DIRECTORY_WRITE = ActionNode(
    # ActionNode的名称
    key="DirectoryWrite",
    # 期望输出的格式
    expected_type=str,
    # 命令文本
    instruction=DIRECTORY_STRUCTION,
    # 例子输入，在这里我们可以留空
    example="",
)
CONTENT_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}".
"""
CONTENT_WRITE = ActionNode(
    # ActionNode的名称
    key="ContentWrite",
    # 期望输出的格式
    expected_type=str,
    # 命令文本
    instruction=DIRECTORY_STRUCTION,
    # 例子输入，在这里我们可以留空
    example="",
)

两个Action修改：

class WriteDirectoryWithActionNode(Action):
    """Action class for writing tutorial directories.

    Args:
        name: The name of the action.
        language: The language to output, default is "Chinese".
    """
    name: str = "WriteDirectoryWithActionNode"
    language: str = "Chinese"

    def __init__(self, name: str = "", language: str = "Chinese", *args, **kwargs):
        super().__init__()
        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"]}]}.
        """

        # 我们设置好prompt，作为ActionNode的输入
        prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
        # resp = await self._aask(prompt=prompt)
        # return OutputParser.extract_struct(resp, dict)
        # 直接调用ActionNode.fill方法，注意输入llm
        # 该方法会返回self，也就是一个ActionNode对象
        resp_node = await DIRECTORY_WRITE.fill(context=prompt, llm=self.llm, schema="raw")
        # 选取ActionNode.content，获得我们期望的返回信息
        resp = resp_node.content
        return OutputParser.extract_struct(resp, dict)


class WriteContentWithActionNode(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".
    """

    name: str = "WriteContentWithActionNode"
    directory: dict = dict()
    language: str = "Chinese"

    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.
        """
        prompt = CONTENT_PROMPT.format(topic=topic, language=self.language, directory=self.directory)
        # return await self._aask(prompt=prompt)
        resp_node = await CONTENT_WRITE.fill(context=prompt, llm=self.llm, schema="raw")
        resp = resp_node.content
        return resp

其实，主要修改的是run的调用部分，以WriteDirectoryWithActionNode为例：
原来是:

        prompt = CONTENT_PROMPT.format(topic=topic, language=self.language, directory=self.directory)
        return await self._aask(prompt=prompt)

修改后：

        prompt = CONTENT_PROMPT.format(topic=topic, language=self.language, directory=self.directory)
        resp_node = await CONTENT_WRITE.fill(context=prompt, llm=self.llm, schema="raw")
        resp = resp_node.content
        return resp

好像代码还多了， 主要是调用了fill方法，当ActionNode比较多时，就能体现这个结构的好处了，可简单理解为可以把一个大的Action拆分为小的Action， ActionNode是SOP调度的最小单位。

注意事项

调用ActionNode的fill函数时，当前需要schema="raw"，否则代码中对大模型返回的结果校验会出现错误：

2024-01-23 01:25:39.504 | ERROR    | metagpt.utils.common:log_it:438 - Finished call to 'metagpt.actions.action_node.ActionNode._aask_v1' after 16.313(s), this was the 1st time calling it. exp: 1 validation error for DirectoryWrite_AN
  Value error, Missing fields: {'DirectoryWrite'} [type=value_error, input_value={'Erlang编程教程': {'...'9.4 部署与维护']}}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.5/v/value_error

这个校验有点 奇怪， 不知道是否和模型有关系，当前用的是glm-4，后面使用gpt4测试一下。

生成《Erlang编程教程》

async def main():
    topic = "Write a tutorial about Erlang Programming"
    role = TutorialAssistantWithActionNode(language="Chinese")
    await role.run(topic)
if __name__ == "__main__":
    asyncio.run(main())

运行后生成的技术文档见《Erlang编程教程》- 由MetaGPT自动生成
。

《Erlang编程教程》- 由MetaGPT自动生成

完整代码

#!/usr/bin/env python3
# _*_ coding: utf-8 _*_

import asyncio
from typing import Dict
from datetime import datetime

from metagpt.actions import Action
from metagpt.actions.action_node import ActionNode
from metagpt.utils.common import OutputParser

from metagpt.const import TUTORIAL_PATH
from metagpt.logs import logger
from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message
from metagpt.utils.file import File

# 命令文本
DIRECTORY_STRUCTION = """
    You are now a seasoned technical professional in the field of the internet. 
    We need you to write a technical tutorial".
    """

DIRECTORY_PROMPT = """
The topic of tutorial is {topic}. 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.
"""
# 实例化一个ActionNode，输入对应的参数
DIRECTORY_WRITE = ActionNode(
    # ActionNode的名称
    key="DirectoryWrite",
    # 期望输出的格式
    expected_type=str,
    # 命令文本
    instruction=DIRECTORY_STRUCTION,
    # 例子输入，在这里我们可以留空
    example="",
)


class WriteDirectoryWithActionNode(Action):
    """Action class for writing tutorial directories.

    Args:
        name: The name of the action.
        language: The language to output, default is "Chinese".
    """
    name: str = "WriteDirectoryWithActionNode"
    language: str = "Chinese"

    def __init__(self, name: str = "", language: str = "Chinese", *args, **kwargs):
        super().__init__()
        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"]}]}.
        """

        # 我们设置好prompt，作为ActionNode的输入
        prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
        # resp = await self._aask(prompt=prompt)
        # return OutputParser.extract_struct(resp, dict)
        # 直接调用ActionNode.fill方法，注意输入llm
        # 该方法会返回self，也就是一个ActionNode对象
        resp_node = await DIRECTORY_WRITE.fill(context=prompt, llm=self.llm, schema="raw")
        # 选取ActionNode.content，获得我们期望的返回信息
        resp = resp_node.content
        return OutputParser.extract_struct(resp, dict)


CONTENT_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}".
"""
CONTENT_WRITE = ActionNode(
    # ActionNode的名称
    key="ContentWrite",
    # 期望输出的格式
    expected_type=str,
    # 命令文本
    instruction=DIRECTORY_STRUCTION,
    # 例子输入，在这里我们可以留空
    example="",
)


class WriteContentWithActionNode(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".
    """

    name: str = "WriteContentWithActionNode"
    directory: dict = dict()
    language: str = "Chinese"

    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.
        """
        prompt = CONTENT_PROMPT.format(topic=topic, language=self.language, directory=self.directory)
        # return await self._aask(prompt=prompt)
        resp_node = await CONTENT_WRITE.fill(context=prompt, llm=self.llm, schema="raw")
        resp = resp_node.content
        return resp


class TutorialAssistantWithActionNode(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.
    """

    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"

    topic: str = ""
    main_title: str = ""
    total_content: str = ""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._init_actions([WriteDirectoryWithActionNode(language=self.language)])
        self._set_react_mode(react_mode=RoleReactMode.BY_ORDER.value)

    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(WriteContentWithActionNode(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)
        return Message()

    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 WriteDirectoryWithActionNode:
            msg = self.rc.memory.get(k=1)[0]
            self.topic = msg.content
            resp = await todo.run(topic=self.topic)
            logger.info(resp)
            await self._handle_directory(resp)
            return await super().react()
        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:
        msg = await super().react()
        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"))
        msg.content = str(root_path / f"{self.main_title}.md")
        return msg


async def main():
    topic = "Write a tutorial about Erlang Programming"
    role = TutorialAssistantWithActionNode(language="Chinese")
    await role.run(topic)


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

生成小说

命令文本

DIRECTORY_STRUCTION = """
You are now a seasoned writer specializing in anti-corruption and crime themes. 
We need you to write a novel with an anti-corruption theme. 
The storyline should be full of suspense, with numerous twists and turns, 
multiple interwoven plotlines, genuine emotions, and it should be eye-opening. 
You can take inspiration from "狂飙" (Furious Storm).
"""

生成的小说《反腐风暴》

《反腐风暴》

第一章：暗流涌动

第一节：谜影重重

在城市的繁华背后，一股暗流正在涌动。政府官员、商界大亨、黑帮分子，各色人物粉墨登场，他们之间错综复杂的关系网，如同重重迷雾，让人看不清真相。一起看似普通的交通事故，却牵扯出一连串的贪腐案件。在这迷影重重的局面下，一群勇敢的记者和检察官，决心揭开这片黑幕。

第二节：初露端倪

随着调查的深入，一些不起眼的线索逐渐浮出水面。一笔巨额资金的流动，一份被篡改的合同，一串神秘的电话号码，每一个线索都像一把钥匙，打开一扇通往真相的门。在这个过程中，有人选择沉默，有人选择背叛，也有人选择坚持。初露端倪的真相，让所有人都感到震惊，这个城市的黑暗面，远比想象中的要庞大。然而，勇敢的斗士们并未退缩，他们知道，只有揭开真相，才能让阳光重新照耀这座城市。

第二章：权力的游戏

第三节：幕后黑手

权力之下，必有操纵之手。在这座繁华都市的背面，一群神秘人物正在暗中布局。他们利用金钱、美色、权力，编织起一张张错综复杂的关系网，将这座城市变成他们的棋盘。这些幕后黑手，或是权势熏心的政客，或是富可敌国的商人，他们为了利益，不择手段。

在这场权力的游戏中，他们通过各种手段，控制着城市的资源和项目，将巨额利益输送至自己的腰包。而那些被卷入这场游戏的人，要么成为他们的棋子，要么被无情地淘汰。正义与邪恶，在这里模糊了界限，黑白不再分明。

第四节：利益输送

利益输送是这场权力游戏的最终目的。通过巧立名目，幕后黑手们将国家资产转化为私人财富。他们利用政策漏洞，进行权钱交易，将工程项目、土地开发等变成自己的提款机。而那些被蒙在鼓里的百姓，却为这些人的奢华生活埋单。

在这条黑暗的利益链中，每个人都可能是受害者，也可能是帮凶。为了揭露这背后的罪行，一群勇敢的记者、检察官和警察，开始了一场与黑暗势力斗智斗勇的较量。他们冒着生命危险，挖掘证据，将那些幕后黑手绳之以法。然而，在这场斗争中，他们不仅要面对来自黑暗势力的压力，还要承受来自亲朋好友的误解和背叛。

在这场没有硝烟的战争中，正义与邪恶的较量愈发激烈。而最终，光明能否战胜黑暗，还这座城市一个清白，仍是一个未知数。

第三章：正义的曙光

第五节：线索追踪

在这个章节中，主人公林浩，一位刚正不阿的检察官，开始追踪一条看似不起眼的线索。这条线索牵扯出一连串的腐败案件，背后隐藏着一个庞大的犯罪网络。林浩深知，每一条线索都是揭开真相的关键，他不敢有丝毫懈怠。通过对账本、通讯记录的仔细分析，他渐渐发现了一些规律，线索开始串联起来，如同拼图一般，逐渐勾勒出一个惊人的真相。

林浩在追踪线索的过程中，遭遇了来自各方的阻力和威胁，但他并未退缩。他坚信，真相是维护正义最有力的武器。在这场与腐败势力斗智斗勇的较量中，林浩展现出了非凡的勇气和智慧。

第六节：反腐斗士

随着调查的深入，林浩逐渐成为这场反腐斗争的先锋。他联合了一群志同道合的战友，共同揭露腐败分子的罪行。在这一节中，林浩和他的团队面对着越来越大的压力，但他们始终坚守正义，誓要将腐败分子绳之以法。

在一场激烈的较量中，林浩成功锁定了一名关键证人。这名证人曾深受腐败分子迫害，对腐败网络了如指掌。在林浩的鼓励和感召下，证人鼓起勇气，揭露了腐败势力的罪行。最终，在大量证据面前，腐败分子无处遁形，正义的曙光终于驱散了黑暗。

这一章节展示了主人公林浩坚定的信念和勇敢的行动，他成为了一名真正的反腐斗士，为维护社会公平正义付出了艰辛的努力。

第四章：生死较量

第七节：危机四伏

在这个章节中，主人公林浩面临着前所未有的危机。一方面，他深入调查的贪腐案件即将浮出水面，另一方面，隐藏在暗处的敌人也开始蠢蠢欲动。每一步都如同踩在刀尖上，稍有不慎便是粉身碎骨。

林浩在调查过程中发现了一份关键的证据，足以证明副市长陈明涉嫌严重贪腐。但他不知道，这份证据早已被对手察觉，一场精心策划的陷阱正等待着他。与此同时，他的家人也受到了威胁，这让林浩陷入了两难境地。

在紧张的氛围中，林浩与他的团队展开了一场与时间的赛跑。他们必须利用有限的信息，揭开敌人的真面目，同时确保自己的安全。危机四伏，每一个决策都可能决定生死。

第八节：生死时速

随着调查的深入，林浩逐渐摸清了敌人的底细，一场决定胜负的较量即将展开。他联手警方，制定了一个大胆的计划，旨在将贪腐分子一网打尽。

在这个章节中，林浩与时间赛跑，每一秒都可能改变局势。他必须在对手察觉之前，将关键证据公之于众，让罪犯无处可逃。在这场生死时速的较量中，林浩展现出了过人的智慧和勇气。

最终，在一场惊心动魄的行动中，林浩和他的团队成功地将陈明等贪腐分子绳之以法。这场胜利来之不易，但它证明了正义终将战胜邪恶。这场生死较量，让人们对反腐败斗争充满信心。

第五章：真相大白

第九节：真相揭露

随着调查的深入，掩藏在繁华都市背后的一张张丑陋面孔逐渐浮出水面。原来，这起看似平常的经济案件背后，隐藏着一个庞大的腐败网络。权力与金钱的交易，法律与道德的沉沦，让这座城市蒙上了一层阴影。

主人公林涛，在经历了无数次的挫折和威胁后，终于找到了关键证据。那是一份记录了所有非法交易的名单，上面涉及到了许多政府要员和商界巨头。真相揭露的那一刻，林涛感受到了从未有过的压力和责任。他知道，这份证据足以引发一场社会地震，但同时也可能让他付出生命的代价。

第十节：正义必胜

在经过一番激烈的思想斗争后，林涛决定将证据公之于众。一场声势浩大的反腐风暴随之展开，那些曾经不可一世的人物纷纷落马。尽管林涛在揭露真相的过程中遭遇了重重阻力和生命危险，但他始终坚持正义，坚信邪恶无法战胜正义。

最终，在广大人民群众的支持和关注下，腐败分子被绳之以法，正义得到了伸张。林涛也因为勇敢揭露真相而成为人们心中的英雄。这场反腐斗争虽然取得了阶段性胜利，但林涛明白，这只是开始，反腐败永远在路上。而他，也将继续在这条道路上，为正义而战。

第六章：反腐之路

第十一节：制度建设

反腐制度建设是遏制腐败现象蔓延的重要措施。本节重点探讨如何构建科学、严密、有效的反腐制度体系。

1. 完善法律法规：加强反腐败立法，形成一套完整的法律体系，为反腐败工作提供法律依据。
2. 强化监督机制：设立独立的反腐监督机构，加强对公职人员的监督，确保权力运行在阳光下。
3. 廉政教育：从源头上预防腐败，加强公职人员的廉政教育，提高他们的廉洁自律意识。
4. 选拔任用制度改革：建立公正、公平、透明的选拔任用制度，防止“带病提拔”现象发生。
5. 激励与问责：建立激励机制，鼓励公职人员廉洁奉公；同时，对腐败行为实行严格问责，形成高压态势。

第十二节：任重道远

反腐斗争是一场持久战，需要全社会共同努力，任重而道远。

1. 加强国际合作：与其他国家分享反腐经验，共同打击跨国腐败犯罪。
2. 公民参与：鼓励公民积极参与反腐斗争，发挥舆论监督作用，共同揭露腐败现象。
3. 深化体制改革：继续深化政治体制改革，消除腐败滋生的土壤。
4. 科技创新：利用大数据、人工智能等科技手段，提高反腐工作的针对性和有效性。
5. 坚定信心：面对反腐斗争的严峻形势，我们要坚定信心，持之以恒，推动反腐工作不断取得新成效。

在这场反腐之路的征程中，制度建设是基石，任重道远是我们的信念。只有不断深化反腐斗争，才能为国家和民族的繁荣昌盛创造一个风清气正的环境。