数据截至 2026-01-21(上游 commit 11cdf466d042)

Action 与 ActionNode:工作单元 + 结构化输出引擎

30 秒导读: 上一章讲了 Role 的"观察-思考-行动"循环。这一章讲**"行动"到底是什么**——Action 是角色能做的一件事的封装(写 PRD、写代码、跑测试……),而 ActionNode 是藏在 Action 里的一台结构化输出引擎:它负责把"请给我一段代码 / 一个文件列表 / 一份需求"这种模糊要求,变成一个有类型、能被 Python 直接读取的对象。核心难题只有一个:LLM 只会吐自由文本,而下游程序需要字段。

---

1. 这是什么(零基础也能懂)

一句话定义: Action 是"角色做的一件事"的最小工作单元;ActionNode 是把 LLM 的自由文本逼成结构化数据的引擎。

它们解决什么问题。 想象你让 LLM"写一份产品需求文档"。它会回你一大段 Markdown。可你的程序下一步要拿到其中的"项目名""需求列表""竞品分析"分别去用——你不能对着一坨文本做 output["需求列表"]。于是需要有人负责:

1. 把"要产出哪些字段、每个字段什么类型、什么要求"讲清楚给 LLM;
2. 收到回复后把它解析成字段;
3. 校验字段齐不齐、类型对不对;
4. 中间出错还要自动重试。

这四件事,就是 ActionNode 的全部工作。Action 则是它外面那层壳:给角色一个统一的 run() 入口,让"做一件事"这个动作可以被 Role 的循环调度。

一个直觉类比。 把 Action 想成一台咖啡机上的一个按钮("美式""拿铁"各是一个 Action),按下去就出一杯东西;ActionNode 则是按钮背后那套**"按配方出料 + 质检 + 不合格重做"的管线。你按按钮(run),它保证出来的不是一摊乱七八糟的液体,而是符合规格、可以直接端走**的成品。

用起来什么样。 最简单的 Action 甚至不用写子类,给一句 instruction 就能跑:

# 示意,非源码。演示「一句话指令 → 一个可 run 的 Action」
action = Action(instruction="把这段英文翻译成中文")
result = await action.run(messages)   # 内部自动走 ActionNode 那套引擎
# result 是结构化输出,而不是一坨裸文本

这一步的关键在于:你只提供了"做什么"(instruction),"怎么把文本变成结构化结果"由 Action/ActionNode 自动补齐。下面就来拆这台引擎。

本章只讲抽象基类与结构化引擎;具体业务 Action(WritePRD、WriteCode 等)属于第 4 章:经典 SOP 流水线。

---

2. 顶层全景(它大概怎么转)

先看两者的关系。 Action 是外壳,ActionNode 是内核;一个 Action 内部持有一个(可选的)ActionNode 作为它的"输出引擎"。

        Role 的循环调用
              │  run(messages)
              ▼
   ┌───────────────────────────┐
   │   Action  (工作单元的壳)   │   action.py
   │   · name / i_context       │
   │   · node: ActionNode ──────┼──┐   持有一个节点作输出引擎
   │   · run → _run_action_node │  │
   └───────────────────────────┘  │
                                   ▼
   ┌────────────────────────────────────────────┐
   │        ActionNode  (结构化输出引擎)          │  action_node.py
   │                                              │
   │   compile ──► 拼 prompt(context+示例+指令)  │
   │      │                                       │
   │      ▼                                       │
   │   fill ────► 调 LLM ─► 解析 ─► pydantic 校验 │
   │      │                         ▲             │
   │      │        失败重试(tenacity)            │
   │      ▼                                       │
   │   instruct_content(带类型的结构化对象)      │
   │                                              │
   │   review / revise ──► 自我审校 + 修订闭环    │
   └────────────────────────────────────────────┘

怎么读这张图: 从上到下是一次调用的流向。Role 按下 run;Action 把历史消息整理成 context,委托给内部的 ActionNode;ActionNode 先 compile 出 prompt,再 fill(生成→解析→校验,失败就重试),最后交出一个带类型的结构化对象 instruct_content。

部件一句话职责:

部件	干什么	在哪
Action	角色"做一件事"的壳,提供统一 run 入口	actions/action.py:29
ActionNode	结构化输出引擎:建模型、拼 prompt、填充、校验	actions/action_node.py:135
create_model_class	用 pydantic create_model 动态生成校验用的模型类	action_node.py:246
compile	把 instruction+example 拼进 SIMPLE_TEMPLATE 成 prompt	action_node.py:382
fill	生成→解析→校验的主循环,含四种 FillMode	action_node.py:596
_aask_v1	真正调 LLM 的那步,挂着 tenacity 重试装饰器	action_node.py:428
review/revise	对已产出的结构化结果做审校与自动修订	action_node.py:729 / 816
ActionOutput	早期的输出包装(content + instruct_content)	actions/action_output.py:12
register_action_outcls	让"同名同字段"的动态类复用同一个对象	action_outcls_registry.py:11

主线走一遍(高层): 输入是一串历史消息 → Action.run 把它们拼成 ## History Messages 的 context → 交给 node.fill → ActionNode 用 instruction/example 编译出 prompt → 调 LLM 拿到裸文本 → 按 schema 解析 → 用动态 pydantic 模型校验 → 输出结构化的 instruct_content。

---

3. 核心原理(逐个机制,由浅入深)

3.1 Action:一件事的封装

它要解决的小问题: 让"角色做的每一件事"长得一样——都能被 run、都能拿到上下文、都能挂一个输出引擎。

思路。 Action 继承 SerializationMixin / ContextMixin / pydantic BaseModel,自身几乎不含业务逻辑,只提供几个通用零件:

字段/方法	作用
name	动作名;为空时自动取类名(set_name_if_empty,action.py:69)
i_context	这次动作的输入上下文,类型很宽(dict / 各种 Context / str / None)
prefix	会被塞进 LLM 的 system message(set_prefix)
node	一个 ActionNode,即这个 Action 的输出引擎(默认 None)
run	统一入口:有 node 就走引擎,否则要求子类自己实现

一句指令就能建一个 Action。 最妙的一处在 _init_with_instruction:pydantic 的 "before" 校验器,发现构造参数里有 instruction,就当场替你造一个 ActionNode 挂到 node 上。

# action.py:76-83  符号:_init_with_instruction
@model_validator(mode="before")
@classmethod
def _init_with_instruction(cls, values):
    if "instruction" in values:
        name = values["name"]
        i = values.pop("instruction")
        values["node"] = ActionNode(key=name, expected_type=str, instruction=i, example="", schema="raw")
    return values

这段的意思是:"给指令 = 给引擎"。所以 §1 里那个 Action(instruction=...) 才能直接跑——它其实被塞了一个 schema="raw"(自由文本)的单节点。

run 的分叉。 run 本身不做事,只做路由:

# action.py:110-114  符号:run
async def run(self, *args, **kwargs):
    if self.node:
        return await self._run_action_node(*args, **kwargs)
    raise NotImplementedError("The run method should be implemented in a subclass.")

有 node 就委托给 _run_action_node;没有就抛 NotImplementedError——这正是业务子类(如 WriteCode)覆写 run 的地方(第 4 章)。

_run_action_node 做的事很轻: 把传入的消息列表倒序编号,拼成一段 ## History Messages,当作 req 喂给节点的 fill。

# action.py:103-108  符号:_run_action_node
msgs = args[0]
context = "## History Messages\n"
context += "\n".join([f"{idx}: {i}" for idx, i in enumerate(reversed(msgs))])
return await self.node.fill(req=context, llm=self.llm)

set_prefix 的副作用链。 设 prefix 时会一路把它同步给 LLM 的 system_prompt,并把同一个 LLM 实例交给 node,保证壳与引擎用的是同一个模型配置(action.py:85-91)。

到这里,Action 的故事讲完了——它薄得几乎透明。真正的工程含量,全在它委托出去的那个 node.fill 里。

3.2 ActionNode:把"要求"变成一个 pydantic 模型

它要解决的小问题: 我口头说"要一个 str 类型的字段叫 code,要求是实现某函数"——怎么让程序据此自动生成一个能校验的类型?

思路:节点即字段。 一个 ActionNode 携带四要素:key(字段名)、expected_type(期望类型)、instruction(对 LLM 的要求)、example(示例)。多个节点通过 children 组成一棵树——一棵树 = 一个多字段的结构化 schema。

             root: "WritePRD_AN"
             ├── Project Name   : str
             ├── Requirements   : List[str]
             ├── Competitive    : List[List[str]]
             └── ...            (每个子节点 = 一个输出字段)

核心魔法:用 pydantic create_model 动态建类。 关键不是手写一个 class,而是在运行时根据节点树现造一个 pydantic 模型出来:

# action_node.py:246-282  符号:create_model_class(节选)
@classmethod
@register_action_outcls
def create_model_class(cls, class_name: str, mapping: Dict[str, Tuple[Type, Any]]):
    """基于pydantic v2的模型动态生成,用来检验结果类型正确性"""
    ...
    new_class = create_model(class_name, __validators__=validators, **new_fields)
    return new_class

这里的 mapping 是 {字段名: (类型, Field(...))} 的字典,由 get_mapping(action_node.py:240)从节点树递归收集而来。拿到这个动态类之后,output_class(**parsed_data) 一步就同时完成了结构化封装 + 类型校验——pydantic 会替你检查类型、跑校验器。

还内建了一个"缺字段"校验器。 动态类里注入了一个 mode="before" 的 check_fields(action_node.py:251-267):必填字段(非 Optional)缺了就抛 Missing fields,多出来的字段只 warning。这样 LLM 少答一个字段能被立刻抓住。

反向也能走:从 pydantic 模型建节点树。 from_pydantic(action_node.py:840-867)递归地把一个 pydantic 模型拆成 ActionNode 树——嵌套模型变成嵌套子节点。这让你可以"先定义好数据结构,再自动得到对应的输出引擎"。

3.3 compile:把节点树编译成 prompt

它要解决的小问题: 光有 schema 不够,还得用一段 prompt 把要求讲给 LLM——包括上下文、格式示例、逐字段指令、约束。

思路:一个模板,填四个槽。 所有非 raw 的输出都套用 SIMPLE_TEMPLATE(action_node.py:56-73):

## context
{context}
-----
## format example
{example}
## nodes: "<node>: <type>  # <instruction>"
{instruction}
## constraint
{constraint}
## action
Follow instructions of nodes, generate output and make sure it follows the format example.

compile(action_node.py:382)负责把三块内容各自编译好再填进模板:

槽	由谁生成	内容
instruction	compile_instruction(:369)	每个节点渲染成 type # instruction
example	compile_example(:374)	每个节点的 example,并用 [CONTENT]...[/CONTENT] 标签包裹
constraint	compile 内	语言约束 + "输出必须裹在 [CONTENT][/CONTENT] 里"的格式约束(:412)

一个刻意的细节:example 不走 f-string。 compile_example 的注释解释了原因(action_node.py:377-378):如果把 list 类型的示例经 f-string 转成 str 再 json.dumps,会额外套一层引号,让 ['main.py'](列表)变成 "['main.py']"(字符串),LLM 就会照抄成错误类型。这是踩过坑后留下的注释,很有教学价值。

schema 决定 example 的形态。 schema="json" 时示例是 JSON,schema="markdown" 时是 Markdown;而 schema="raw"(§3.1 那种单指令)则完全不编译模板,只把 context + 语言约束 + instruction 拼一起(action_node.py:393-394)。

3.4 fill:生成→解析→校验的主循环

这是整台引擎的心脏。它把上面所有零件串成一次真正的调用,并处理四种不同的输出风格。

先看 fill 的骨架: 设好 llm 与 context 后,按 mode 分流到不同的填充方式(action_node.py:596-663)。

# action_node.py:630-652  符号:fill(节选,省略部分分支)
self.set_llm(llm)
self.set_context(req)
if self.schema:
    schema = self.schema

if mode == FillMode.CODE_FILL.value:
    result = await self.code_fill(context, function_name, timeout)
    ...
elif mode == FillMode.XML_FILL.value:
    context = self.xml_compile(context=self.context)
    result = await self.xml_fill(context, images=images)
    ...
elif mode == FillMode.SINGLE_FILL.value:
    result = await self.single_fill(context, images=images)
    ...
if strgy == "simple":
    return await self.simple_fill(schema=schema, mode=mode, images=images, timeout=timeout, exclude=exclude)

四种 FillMode(action_node.py:46-49)对应四种"怎么从文本里抠出字段":

FillMode	抠取方式	典型用途
默认(simple_fill)	按 schema 用 JSON/Markdown parser 整体解析	多字段结构化输出(PRD、设计)
CODE_FILL	用 sanitize 从 ``` 代码块里提取代码	生成代码(code_fill,:533)
XML_FILL	正则 <field>...</field> 抠标签,并按字段类型转型	需要类型转换的字段(:553)
SINGLE_FILL	整段回复原样作为单一字段	只要一个自由文本字段(:546)

XML_FILL 尤其能体现"逼成结构化"的手法:它对每个字段跑正则,再按 field_type 把字符串转成 int / bool / list / dict,转不动就落一个安全默认值(如 int 失败给 0,list 失败给 [],见 action_node.py:570-592)。

默认路径 simple_fill → _aask_v1 才是主干: 先 compile 出 prompt,再调 _aask_v1 拿到 (裸文本, 结构化对象),分别存进 self.content 与 self.instruct_content(action_node.py:468-484)。

_aask_v1:调用 + 解析 + 校验,外加重试。 这一步把"文本→对象"真正落地,并且整个函数挂着 tenacity 重试:

# action_node.py:423-452  符号:_aask_v1
@retry(
    wait=wait_random_exponential(min=1, max=20),
    stop=stop_after_attempt(6),
    after=general_after_log(logger),
)
async def _aask_v1(self, prompt, output_class_name, output_data_mapping, ...):
    content = await self.llm.aask(prompt, ...)                 # ① 生成:调 LLM
    output_class = self.create_model_class(output_class_name, output_data_mapping)  # ② 动态建模型
    if schema == "json":
        parsed_data = llm_output_postprocess(output=content, schema=..., req_key=f"[/{TAG}]")  # ③ JSON 解析
    else:
        parsed_data = OutputParser.parse_data_with_mapping(content, output_data_mapping)        # ③ Markdown 解析
    instruct_content = output_class(**parsed_data)             # ④ 校验:构造即校验
    return content, instruct_content

这就是"逼成结构化"的完整闭环: 生成(①)→ 动态建模(②)→ 解析(③)→ 用 pydantic 构造来校验(④)。任何一步抛异常(解析失败、字段缺失、类型不符),@retry 就会指数退避、最多重试 6 次——LLM 偶尔跑偏时,靠"再试一次"把成功率抬上去。这是"让自由文本可被程序消费"的最后一道保险。

一个可留意的坑(inferred): fill 在 CODE_FILL 分支里传的是裸变量 context(action_node.py:636),而该函数并没有名为 context 的局部参数/变量(它设置的是 self.context)。XML_FILL/SINGLE_FILL 分支则先给 context 局部赋了值再用。因此 CODE_FILL 分支看起来会引用到未定义的 context;实际调用方通常通过其他入口触发代码生成,这条直连分支未必被走到。以源码为准。

3.5 review / revise:产出后的自我审校

它要解决的小问题: 结构化结果拿到了,但内容质量未必达标。能不能让引擎自己"挑错 + 改错"?

两步闭环。

• review(action_node.py:729):对每个字段,拿"输出值 vs 该字段的 instruction"做比对,套用 REVIEW_TEMPLATE(:75)让 LLM 只对不达标的字段给出修改意见。支持 AUTO(LLM 自审)或 HUMAN(人审)。
• revise(action_node.py:816):根据 review 的意见,套用 REVISE_TEMPLATE(:105)让 LLM 只改有意见的那些字段,再用 update_instruct_content(:332)把修订合并回结构化对象。

   fill ──► instruct_content
              │
              ▼
   review ──► {字段: 修改意见}     (只列不达标的键)
              │
              ▼
   revise ──► 按意见改这些字段 ──► 回填 instruct_content

一个细节:review 前必须 fill 过。 review/revise 开头都断言 self.llm 已存在、instruct_content 非空(action_node.py:736-739),否则抛 use review after fill——因为审校依赖已经产出的结构化结果。schema="raw" 的自由文本节点因为没有 instruct_content,不支持审校。

---

4. 巧妙之处(可借鉴的技术)

① 动态类去重:同名同字段复用同一个类。 pydantic 的 create_model 每次调用都返回不同的类对象,即使名字和字段完全一样——这会让"两次生成的结果类型不相等"。register_action_outcls(action_outcls_registry.py:11-42)用装饰器把 (类名, 字段mapping) 拼成一个 outcls_id 做缓存键,命中就复用。它还做了两处归一化:对字段 dict 排序、把 typing.List/typing.Dict 文本替换成 list/dict(:31-33),让"等价定义"能命中同一份缓存。

② 用"构造即校验"代替手写校验。 全流程没有一处手写的 if-else 检查字段;把类型/必填/校验器都塞进动态 pydantic 模型,output_class(**parsed_data) 一行同时完成封装与校验。少写校验代码,还天然拿到 pydantic 的错误信息去驱动重试。

③ 重试放在最靠近 LLM 的那层。 @retry 挂在 _aask_v1(而不是更外层的 fill),意味着每次重试都重新走一遍"生成→解析→校验",但不会重复外层的编译等准备工作;指数退避(wait_random_exponential)也避免了对 LLM 的瞬时打爆。

④ 用注释固化踩过的坑。 compile_example 里"别用 f-string,否则 list 会被套引号"的注释(action_node.py:377-378),以及 xml_fill 里逐类型的容错默认值,都是把"格式约束"经验直接写进代码的例子。

---

5. 边界与局限(诚实)

• XML_FILL 用 eval 解析 list/dict。 xml_fill 对 list/dict 字段直接 eval(raw_value)(action_node.py:581、:588),依赖 LLM 输出的是合法 Python 字面量;这既有安全隐含风险,也把"字段值"限定在能被 eval 的形态里。
• compile 只实现了 mode="children"。 docstring 明说 mode="all" 与 mode="root" 是 NotImplemented(action_node.py:388-390),多层节点树主要按"编译所有子节点"这一条路走。
• JSON 指令被刻意弃用。 compile 里有一大段被注释掉的"直接用 pydantic 生成 JSON instruction"的代码,并标注 FIXME: json instruction会带来格式问题(action_node.py:396-407)——说明"让 LLM 严格产 JSON schema"在实践中不稳,才退回 Markdown 指令 + 标签包裹的方案。
• ActionOutput 基本是历史遗留。 actions/action_output.py:12 只是 content + instruct_content 的朴素包装;当前主路径的结构化结果直接挂在 ActionNode.instruct_content 上。

---

6. 代码地图(导航索引)

主题	文件路径	符号名
Action 基类 / 统一 run 入口	metagpt/actions/action.py	Action、run、_run_action_node
一句指令自动建节点	metagpt/actions/action.py	_init_with_instruction
同步 prefix 到 LLM/node	metagpt/actions/action.py	set_prefix
ActionNode 主体(节点树)	metagpt/actions/action_node.py	ActionNode、__init__、add_child
动态生成 pydantic 校验类	metagpt/actions/action_node.py	create_model_class、get_mapping
从 pydantic 反向建节点树	metagpt/actions/action_node.py	from_pydantic
编译 prompt(模板+示例+指令)	metagpt/actions/action_node.py	compile、compile_instruction、compile_example、SIMPLE_TEMPLATE
主循环:生成→解析→校验	metagpt/actions/action_node.py	fill、simple_fill、_aask_v1
四种填充模式	metagpt/actions/action_node.py	FillMode、code_fill、xml_fill、single_fill
自我审校 / 修订闭环	metagpt/actions/action_node.py	review、auto_review、revise、auto_revise、REVIEW_TEMPLATE、REVISE_TEMPLATE
动态类去重缓存	metagpt/actions/action_outcls_registry.py	register_action_outcls
早期输出包装	metagpt/actions/action_output.py	ActionOutput

---

接着往哪读:

• 上游是谁调用 Action:01-role-loop.md(Role 的观察-思考-行动循环)。
• 具体业务 Action(WritePRD/WriteCode…)怎么用这台引擎:04-classic-sop-pipeline.md。
• Action 的产出如何在角色间流转:03-environment-message-bus.md。