Node system cn

本文档系统阐述 Jianmu 节点体系的三大支柱——类继承层次、端口绑定机制与依赖注入架构——为后续深入各内置节点与预设模式建立统一的认知基座。阅读完本文后，你将能够理解自定义节点的编写方式、节点间数据如何通过状态共享流动，以及运行时服务如何透明地注入到行为树的每个角落。

节点体系在架构中的位置

在 架构分层总览 中已经阐明：Jianmu 是一个三层架构——运行时引擎层驱动执行、节点组合层编排逻辑、能力模块层提供模型/工具/记忆等可复用能力。节点体系恰好处于三层交汇的枢纽位置：它从能力模块层消费模型客户端、工具、上下文构建器，通过运行时引擎层注入的 StateManager 与 RunContext 读写状态，最终在 ReactiveRunner 的事件驱动调度下完成每一个 tick。

Jianmu 的节点模型建立在 py_trees 行为树框架之上，并通过自定义 mixin 与异步基类对其进行了深度扩展。整个节点体系围绕四个核心文件展开：

文件	职责
jianmu/engine/behaviour.py	JianmuNodeMixin、同步 Behaviour、异步 AsyncBehaviour、Decorator
jianmu/engine/deps.py	RunContext、InjectPayload、inject_runtime_deps 注入分发
jianmu/node/base.py	语义别名导出（Node = Behaviour, AsyncNode = AsyncBehaviour）
jianmu/node/composites.py	Jianmu 特化组合节点 LoopUntilSuccess

---

类继承层次

Jianmu 的节点类体系基于双继承模式：每个节点同时继承 py_trees 的行为基类（获取树执行语义）和 JianmuNodeMixin（获取端口绑定与依赖注入能力）。以下是完整的继承图谱：

classDiagram
    direction TB

    class JianmuNodeMixin {
        +namespace: str
        +ctx: RunContext
        +state_manager: StateManager
        -_input_bindings: dict
        -_output_bindings: dict
        +inject(payload)
        +bind(inputs, outputs)
        +read_port(name, default)
        +write_port(name, value)
        +write_ports(values)
        +append_port_messages(name, messages)
        +model_client ModelClient
        +interaction InteractionController
        +recovery CheckpointRecoveryFacade
    }

    class Behaviour {
        +__init__(name, namespace)
    }

    class AsyncBehaviour {
        +async_task: Task
        +initialise()
        +tick()
        +update() Status
        +update_async()* Status
        +terminate(new_status)
    }

    class Decorator {
        +__init__(name, child, namespace)
    }

    class LoopUntilSuccess {
        +max_iterations: int
        +abort_condition: Callable
        +iteration_count: int
    }

    class Timeout {
        +duration: float
        +expiry_time: float
    }

    class FlattenedAgentNode {
        +_is_compiled: bool
        +_ns: str
        +_compile_subtree()* bool
    }

    class py_trees_Behaviour {
        <<py_trees>>
        +name: str
        +status: Status
        +parent: Composite
        +children: list
        +setup()
        +tick()
        +stop()
    }

    class py_trees_Decorator {
        <<py_trees>>
        +decorated: Behaviour
        +child: Behaviour
    }

    py_trees_Behaviour <|-- JianmuNodeMixin : mixin
    py_trees_Behaviour <|-- Behaviour
    JianmuNodeMixin <|-- Behaviour
    Behaviour <|-- AsyncBehaviour

    py_trees_Decorator <|-- Decorator
    JianmuNodeMixin <|-- Decorator
    Decorator <|-- LoopUntilSuccess
    Decorator <|-- Timeout
    Decorator <|-- FlattenedAgentNode

    note for AsyncBehaviour "所有内置 LLM/Tool/Swarm 节点\n的最终基类"
    note for FlattenedAgentNode "SkillNode, SwarmNode\n共用的延迟编译基类"

在这个层次中，用户面向的两个核心入口是：

• Node (Behaviour)：同步节点基类，实现 update() -> Status 即可。适用场景为纯条件判断（StateCondition）、日志输出（Log）等不需要异步 IO 的轻量节点。
• AsyncNode (AsyncBehaviour)：异步节点基类，实现 update_async() -> Status 即可。绝大多数内置节点（LLM 调用、工具执行、Swarm 通信）都继承自此，因为需要异步等待网络 IO 或子进程。

Decorator 同时继承 py_trees.decorators.Decorator 与 JianmuNodeMixin，使其既能包装子节点（控制流语义）又能访问端口绑定和运行时上下文。LoopUntilSuccess、Timeout 以及技能系统中的 FlattenedAgentNode 都从此派生。

---

JianmuNodeMixin：端口绑定与运行时操作

JianmuNodeMixin 是注入到每个 Jianmu 节点中的共享能力层。它不是独立实例化的类，而是通过 Python 多重继承混入 Behaviour / Decorator，为节点提供五个维度的运行时能力：依赖注入接收、端口绑定解析、状态读写、模型/交互属性解析与薄的 checkpoint 恢复 facade。

字段初始化

每个 Jianmu 节点在构造时通过 _init_jianmu_fields() 初始化一套私有字段：

self.namespace: str | None       # 状态命名空间，用于子树隔离
self.ctx: RunContext | None       # 运行时依赖容器（注入后填充）
self.state_manager: StateManager | None  # 状态管理器（注入后填充）
self._wake_up: Callable | None    # 唤醒回调（注入后填充）
self._input_bindings: dict[str, str]   # 端口名 → 状态字段映射
self._output_bindings: dict[str, str]  # 端口名 → 状态字段映射

这些字段在构造时为空，直到 ReactiveRunner 调用 inject() 方法后才会被填充为运行时实际对象。

bind()：端口绑定链式 API

bind() 是节点间数据流的声明式配置入口。它接收 inputs 和 outputs 两个字典，将逻辑端口名映射到状态字段键名，并返回 self 以支持方法链式调用：

NormalizeTextNode(name="NormalizeText").bind(
    inputs={"source_text": "text"},           # 逻辑端口 source_text → 状态字段 text
    outputs={"normalized_text": "normalized_text"},  # 输出端口 normalized_text → 状态字段 normalized_text
)

内部实现中，bind() 会调用 _normalize_binding_key() 对键值进行标准化——自动去除 state. 前缀并处理空值回退。绑定映射被存储在 _input_bindings 和 _output_bindings 两个字典中。

端口读写 API

节点通过 逻辑端口名 读写数据，底层由绑定映射自动路由到正确的状态字段：

方法	方向	说明
read_port(port_name, default, state_key)	输入	按绑定解析端口 → 从 StateManager 读取
write_port(port_name, value, signal, state_key)	输出	按绑定解析端口 → 写入单个状态字段
write_ports(values, signal, state_keys)	输出	批量写入多个端口（一次状态更新）
append_port_messages(port_name, messages, state_key)	输出	追加消息到消息列表端口（通过 StateMessageStore）

解析链路为：port_name → resolve_input_port() / resolve_output_port() → _binding_map() → _normalize_binding_key() → 最终状态字段名 → read_state() / write_state()。

read_state() 和 write_state() 会自动在节点的 namespace 下进行命名空间限定，实现子树之间的状态隔离。

模型客户端、交互控制器与恢复 facade

三个惰性属性让节点无需关心依赖来源：

• model_client：优先返回显式设置 _explicit_model_client，否则回退到注入的 ctx.model_client。这意味着子类可以在构造函数接收 model_client 参数并赋值给 _explicit_model_client 来覆盖运行时注入的默认客户端。
• interaction：返回一个绑定到当前 StateManager 的 InteractionController 实例，用于挂起/恢复流程（审批、用户输入）。首次访问时创建并缓存。
• recovery：返回绑定到当前节点的 checkpoint-owned CheckpointRecoveryFacade。普通自定义节点应把它视为唯一恢复入口：业务进度写入 Jianmu state，只有直接执行需要 replay 或 reconcile 的外部副作用时才使用 recovery task records。

旧的 dump_resume_state() / restore_resume_state() hook 是少数内置节点的内部兼容路径，不是应用层自定义节点的扩展模型。

面向应用层的推荐规则更窄：普通自定义节点应把真正的业务进度落到普通 Jianmu state 中；只有当节点自己拥有非幂等的外部副作用、并且需要 replay 或 reconcile 语义时，才通过 recovery facade 记录 checkpoint-owned task records。

---

AsyncNode：异步生命周期

AsyncBehaviour（即 AsyncNode）是 Jianmu 中最关键的基类。它通过 asyncio.Task 机制将 py_trees 的同步 tick 模型桥接到异步世界。

生命周期状态机

stateDiagram-v2
    [*] --> INVALID: 树启动
    INVALID --> initialise: py_trees 进入节点
    initialise --> RUNNING: 创建 asyncio.Task(update_async)
    RUNNING --> RUNNING: tick() 检测 task 完成 → 重新 initialise
    RUNNING --> SUCCESS: update_async 返回 SUCCESS
    RUNNING --> FAILURE: update_async 返回 FAILURE / 抛异常
    RUNNING --> INVALID: task 被取消 (CancelledError)
    SUCCESS --> [*]
    FAILURE --> [*]
    INVALID --> initialise: 树重新 tick

四个关键方法

initialise() —— 当 py_trees 首次进入节点或节点被重置后调用。核心逻辑：

1. 取消尚未完成的旧 async_task
2. 获取当前事件循环 asyncio.get_running_loop()
3. 创建新任务 loop.create_task(self.update_async())
4. 绑定完成回调：async_task.add_done_callback(lambda _: wake_up()) —— 这是事件驱动架构的核心：异步任务完成时自动唤醒 ReactiveRunner 的事件循环，无需轮询

tick() —— py_trees 每帧调用的入口。重写了父类方法以处理异步任务完成后的重新启动：如果当前状态为 RUNNING 且 async_task 已完成，则检查其结果——若结果也是 RUNNING（表示节点需要继续执行），则调用 initialise() 重新创建任务。这使异步节点可以跨多个 tick 持续执行。

update() —— 将异步任务状态映射回 py_trees Status： - async_task 为 None → FAILURE - async_task 未完成 → RUNNING - async_task 完成 → 返回任务的 Status 结果；若类型不匹配则 FAILURE - CancelledError → INVALID

terminate() —— 节点被中断时取消飞行中的异步任务，清理 async_task 引用。

编写自定义异步节点

用户只需继承 AsyncNode 并实现 update_async() 方法。节点内部通过 self.read_port() 和 self.write_port() 与状态交互，完全不用关心依赖注入的细节：

from jianmu.node.base import AsyncNode
from py_trees.common import Status

class NormalizeTextNode(AsyncNode):
    async def update_async(self) -> Status:
        raw_text = str(self.read_port("source_text", default=""))
        normalized = " ".join(raw_text.strip().lower().split())
        self.write_port("normalized_text", normalized)
        return Status.SUCCESS

---

依赖注入体系

依赖注入是连接运行时引擎与节点层的隐形管线。它由三个核心数据结构和一个分发函数构成，在 ReactiveRunner 的构造和每次 run() 调用时触发。

核心数据结构

flowchart LR
    subgraph "构造阶段"
        RunContext[RunContext<br/>model_client<br/>approval_manager<br/>sandbox<br/>constraints<br/>runtime_event_bus<br/>prompt_runtime]
        StateManager[StateManager<br/>状态存储 + 变更通知]
        WakeUp[_on_wake_signal<br/>唤醒回调]
    end

    RunContext --> InjectPayload
    StateManager --> InjectPayload
    WakeUp --> InjectPayload

    InjectPayload[InjectPayload<br/>frozen dataclass]

    subgraph "注入目标"
        Node1[AgentLLMNode]
        Node2[ToolExecutor]
        Node3[StateCondition]
    end

    InjectPayload -->|inject_runtime_deps| Node1
    InjectPayload -->|inject_runtime_deps| Node2
    InjectPayload -->|inject_runtime_deps| Node3

RunContext 是一个 @dataclass，承载不应序列化到 Agent 状态中的运行时服务对象：

字段	类型	消费方
model_client	ModelClient	LLM 节点、EvaluationNode
approval_manager	ApprovalManager	Guard 系统、ToolExecutor
sandbox	沙箱句柄	ToolNode、ToolExecutor
constraints	Constraints	Guard 策略检查
runtime	不透明扩展	宿主应用自定义数据
prompt_runtime	PromptRuntimeContext	上下文构建器（技能目录、引导文件）
runtime_event_bus	RuntimeEventBus	所有节点的遥测事件发布

InjectPayload 是一个 @dataclass(frozen=True) 的不可变载体，聚合了 RunContext、StateManager 和 wake_up 回调。其不可变性确保注入过程中不会被意外修改。

self.recovery 由注入后的 state manager、run context 以及节点的 restore-stable locator 派生。恢复 schema 和 replay 判断位于 jianmu.memory.checkpoint，不放在节点基类或 runtime loop 里。

注入时机与分发

ReactiveRunner 在两个关键时机执行注入：

1. 构造时（__init__）：首次遍历整棵行为树，为所有 Jianmu 节点注入依赖。使用 root.iterate() 遍历每个节点，调用 inject_runtime_deps()。
2. 每次 run() 调用时：重新注入以刷新 wake_up 回调引用（避免跨运行周期的过期引用），同时注入可能被之前 _event_loop 清理清除的回调。

注入分发函数 inject_runtime_deps() 使用 isinstance(node, JianmuNodeMixin) 进行类型检查：Jianmu 原生节点通过 inject() 方法接收依赖；非 Jianmu 节点（纯 py_trees 节点）被安全忽略。

_event_loop 退出时（finally 块），runner 会注入一个 wake_up=None 的清理载荷，防止节点在运行结束后继续持有活跃的回调引用。

延迟编译节点的注入传播

对于 FlattenedAgentNode（SkillNode、SwarmNode 的基类），子树在首次 initialise() 时才被编译。因此需要在编译完成后手动传播依赖注入：

def _inject_dependencies_to_pipeline(self, pipeline):
    payload = InjectPayload(
        context=self.ctx,
        state_manager=self.state_manager,
        wake_up=self._wake_up,
    )
    for n in pipeline.iterate():
        inject_runtime_deps(n, payload)

这确保了延迟创建的子节点也能获得完整的运行时依赖。

---

端口绑定实战

端口绑定机制的设计哲学是：节点声明逻辑端口 → bind() 映射到状态字段 → port API 自动路由。这种间接层带来了三个关键收益：

1. 节点可复用：同一个 ToolNode 可以通过不同的绑定配置读写不同的状态字段
2. 状态结构解耦：节点不需要知道状态 Schema 的字段名，状态 Schema 的变更不影响节点实现
3. 可视化友好：端口名提供了清晰的节点间数据流语义

绑定解析全链路

端口绑定的核心解析路径如下：

flowchart TD
    A[read_port 'source_text'] --> B{_input_bindings<br/>有映射?}
    B -->|有| C[resolve_input_port<br/>'_input_bindings'映射]
    B -->|无| D[fallback: state_key 参数<br/>或 port_name 自身]
    C --> E[_normalize_binding_key<br/>去 state. 前缀]
    D --> E
    E --> F[read_state resolved_key<br/>在 namespace 下]
    F --> G[StateManager.get<br/>key, namespace=...]

_normalize_binding_key() 是规范化的守门人：自动去除 state. 前缀（兼容旧版写法），处理 None 和空字符串的回退。这意味着 "state.messages" 和 "messages" 在绑定映射中等价。

测试验证的绑定场景覆盖

从 test_port_bindings.py 中可以看到，每个内置节点都经过了完整的端口绑定测试：

节点类型	输入端口绑定	输出端口绑定	验证要点
SimpleLLMNode	messages → custom field	done, final_answer, messages	响应写入自定义状态键
AgentLLMNode	messages, usage	done, final_answer, actions, rounds, usage	多轮对话 + usage 累积
ToolNode	input → custom field	output → custom field	工作流模式工具调用
ToolExecutor	actions, usage	messages, tool_effects	Agent 循环工具执行
StateCondition	score → custom field	—	声明式条件比较
EvaluationNode	messages, answer, best_score	score, reflection, highest_score, best_answer	多字段评分流水线
SpawnAgent	role, task, agent_id, constraints	spawned_agent_id	Swarm agent 孵化
SendMessage	runtime, to_agent_id, message, agent_id, topic	—	Swarm 消息发送

测试覆盖了绑定解析的正确性、缺少绑定时的回退行为、以及 ToolNode(execute=False) 的跳过执行场景。

---

命名空间隔离

每个节点携带一个可选的 namespace 字符串，它在状态读写时作为前缀被 StateManager._normalize_namespace() 处理（转换为 <namespace>. 格式）。命名空间隔离在以下场景中至关重要：

• FlattenedAgentNode 子树：SkillNode 和 SwarmNode 在 initialise() 时自动生成 skill_runtime.<name>_<struct_id> 格式的命名空间——struct_id 从节点在树中的路径位置推导，确保同一类型的多个扁平化代理不会相互干扰。
• 并行分支：Parallel 组合下的多个分支可以通过不同的 namespace 操作各自的状态分片，避免竞态。
• 生命周期清理：FlattenedAgentNode.terminate() 在节点退出时调用 state_manager.clear(self._ns) 彻底清理该命名空间下的所有运行时数据。

命名空间的规范化规则严格：不允许以 . 结尾，不允许首尾空白，空值和 None 均被视为全局命名空间。

---

内置节点类型总览

所有内置节点均继承自 AsyncNode（或对于同步节点继承自 Node），天然享有端口绑定与依赖注入能力。以下是完整的内置节点目录：

分类	节点	基类	核心职责
LLM	SimpleLLMNode	AsyncNode	单次 LLM 调用，构建上下文 → 调用模型 → 持久化响应
LLM	AgentLLMNode	SimpleLLMNode	Agent 模式 LLM 调用，附加 tool_calls 提取、usage 累积、rounds 计数
工具	ToolNode	AsyncNode	工作流中确定性调用单个工具，通过 input_key/output_key 绑定
工具	ToolExecutor	AsyncNode	Agent 循环中执行 LLM 产出的 ToolCall 列表，支持 Guard 拦截
技能	SkillNode	FlattenedAgentNode	延迟编译技能子树为装饰器节点
Swarm	SpawnAgent	AsyncNode	孵化新 Agent，写入 spawned_agent_id；支持 reuse_if_present 幂等模式
Swarm	SendMessage	AsyncNode	向指定 agent/group/topic 发送消息
Swarm	WaitMessage	AsyncNode	等待来自特定发送者的消息
Swarm	WaitForever	AsyncNode	无限期等待（用于守护 Agent）
Swarm	SwarmNode	FlattenedAgentNode	延迟编译 Swarm 协作子树
条件	StateCondition	Node（同步）	谓词或声明式状态条件判断
评估	EvaluationNode	AsyncNode	Actor-Evaluator 模式中的评分节点
工具	Log	Node（同步）	日志输出，可选广播到 Studio
工具	Wait	AsyncNode	异步等待固定时长
工具	Timeout	Decorator	装饰器：对子节点施加超时限制

这些节点的详细设计分别在各自的专题文档中展开：AgentLLMNode 与 SimpleLLMNode 见 LLM 节点，ToolNode 与 ToolExecutor 见 工具与技能节点，Swarm 节点见 Swarm 节点。

---

组合节点与装饰器

Jianmu 从 py_trees 继承了 Sequence、Selector、Parallel 三种标准组合节点，并添加了 Jianmu 特化的 LoopUntilSuccess 装饰器：

组合节点	来源	语义
Sequence	py_trees	顺序执行子节点，任一失败则整体失败（AND 逻辑）
Selector	py_trees	依次尝试子节点，首个成功则整体成功（OR 逻辑 / 回退）
Parallel	py_trees	并行执行子节点，策略可配置（SuccessOnAll、SuccessOnOne 等）
LoopUntilSuccess	Jianmu	装饰器：重试子节点直到成功或达到 max_iterations

这些组合节点本身不是 Jianmu 节点（不继承 JianmuNodeMixin），但它们作为容器承载 Jianmu 子节点时，子节点仍通过 ReactiveRunner 的整树遍历获得依赖注入。在 行为树执行内核 中有更深入的讲解。

LoopUntilSuccess 特别值得关注：它在子节点失败时自动将子节点重置为 INVALID 并触发状态信号唤醒 runner，从而实现重试循环。它还支持 abort_condition 回调用于外部中断（如预算耗尽），并在达到最大迭代次数时写入结构化的 termination 元数据（InteractionKeys.TERMINATION）。

---

@node 装饰器：从函数到节点的捷径

jianmu/node/decorator.py 提供了 @node 装饰器，允许将普通 Python 函数（同步或异步）快速转化为 FunctionNode（AsyncNode 子类）：

@node(name="my_processor")
def process(state):
    return {"result": state["input"] * 2}

装饰器内部自动提取函数名与 docstring 作为节点名和描述，生成的 WrappedNode 是一个类，可以像普通节点一样实例化并嵌入行为树。FunctionNode.update_async() 获取当前状态快照，调用包装函数，并将返回的字典合并回状态。

这是原型阶段最高效的节点创建方式，但生产环境中推荐继承 AsyncNode 以获得更精细的生命周期控制。

---

总结

Jianmu 的节点体系通过三层设计——py_trees 执行语义 + JianmuNodeMixin 运行时能力 + 依赖注入管线——实现了关注点的清晰分离：

• 节点开发者 只需关注 update_async() 中的业务逻辑，通过 read_port / write_port 操作数据，无需关心依赖来源。
• 工作流组装者 通过 bind() 声明端口映射，自由编排节点间的数据流向。
• 运行时引擎 通过 InjectPayload 统一向整棵树注入 RunContext、StateManager 和 wake_up 回调，实现事件驱动的异步执行。

接下来，你可以深入 ReAct 节点工厂 了解预设模式如何将 AgentLLMNode + ToolExecutor + LoopUntilSuccess 组装为完整的 Agent 循环，或阅读 自定义节点 获取编写生产级节点的详细指南。