自定义指令集

AmritaSense 的内置指令集已经覆盖了条件分支、循环、异常处理等核心控制流。但当这些基本指令的组合反复出现、形成固定模式时，就可以通过 SelfCompileInstruction 将其封装为新的指令。这种扩展不侵入解释器，只在编译期展开为标准节点组合，运行时与内置指令完全等效。

自编译指令接口：SelfCompileInstruction

SelfCompileInstruction 是一个抽象基类，定义了所有自编译指令的统一入口：

from abc import ABC, abstractmethod
from amrita_sense.node.abc_base import AbstractComposeOriginal


class SelfCompileInstruction(ABC):
    @abstractmethod
    def extract(self) -> AbstractComposeOriginal:
        """将自定义指令展开为底层节点组合"""
        pass

契约只要求“返回某种源组合”（AbstractComposeOriginal）。实践中 extract() 返回默认实现 NodeCompose 即可——框架会像处理手写编排一样，自动递归渲染你返回的组合。

核心概念

• 编译期展开：extract() 在 render() 阶段被调用，不是在运行时。展开后的 NodeCompose 被递归渲染，最终生成与手写编排完全相同的 NodeComposeRendered
• 透明性：对工作流的其他部分而言，自定义指令和内置指令没有区别。它们共享同一套别名系统、同一套寻址机制
• 组合性：自定义指令内部可以包含其他指令（包括其他自定义指令），嵌套深度没有限制

实现要求

1. 必须实现 extract() 方法
2. extract() 只能依赖编译期已知的信息（构造参数）
3. 如果展开后的结构包含跳转，地址计算必须在 extract() 内完成
4. 返回的 NodeCompose 会被自动递归渲染，无需手动调用 render()

实现模式：extract() 与地址计算

实现自定义指令的核心工作，是将一个“意图”映射为一个“节点数组”。这个映射包括三个步骤：

步骤一：确定节点列表

将指令的语义拆解为具体的节点序列。例如，一个“重试”指令可以拆解为：执行节点 -> 检查结果 -> 若失败且未达上限则跳回 -> 若成功或达上限则继续。

步骤二：计算跳转地址

如果展开后的结构包含跳转（JMP、ConditionJumpNode 等），需要在 extract() 内根据节点列表的长度计算偏移量。所有地址必须是静态确定的整数。

步骤三：返回 NodeCompose

将节点列表包装为 NodeCompose(*nodes) 返回。框架会自动处理后续的递归渲染。

简单示例：带日志的节点包装器

class LoggedNode(SelfCompileInstruction):
    def __init__(self, node: BaseNode, name: str):
        self._node = node
        self._name = name

    def extract(self) -> NodeCompose:
        @Node()
        def log_start():
            print(f"[{self._name}] 开始执行")

        @Node()
        def log_end():
            print(f"[{self._name}] 执行完毕")

        return NodeCompose(log_start, self._node, log_end)

使用：

workflow = start >> LoggedNode(process_data, "数据处理") >> end

等价于手写：

workflow = start >> log_start >> process_data >> log_end >> end

案例一：重试器

将一个可能失败的节点包装为支持重试的逻辑，是自编译指令的典型应用场景。

需求

• 执行目标节点
• 若节点抛出异常，自动重试
• 超过最大重试次数后，执行降级节点

实现

from amrita_sense.instructions import IF, NATIVE_WHILE, Try


class RetryClause(SelfCompileInstruction):
    def __init__(
        self, node: BaseNode, max_retries: int = 3, fallback: BaseNode | None = None
    ):
        self._node = node
        self._max = max_retries
        self._fallback = fallback

    def extract(self) -> NodeCompose:
        retries = 0
        ok = False

        @Node()
        def under_max() -> bool:
            return not ok and retries < self._max

        @Node()
        def on_error() -> None:
            nonlocal retries
            retries += 1

        @Node()
        def on_success() -> None:
            nonlocal ok
            ok = True

        @Node()
        def failed() -> bool:
            return not ok

        # 循环体是组合，因此必须用 NATIVE_WHILE（WHILE 只接受单节点）。
        body = Try(self._node).CATCH(Exception, on_error).THEN(on_success)
        loop = NATIVE_WHILE(under_max).ACTION(body)

        if self._fallback is not None:
            return loop >> IF(failed, self._fallback)
        return loop.extract()

使用：

RetryClause(call_api, max_retries=3, fallback=use_cache)

关键点

• 循环体是一个组合（Try(...).CATCH(...).THEN(...)），因此循环必须用 NATIVE_WHILE。基于 call_sub 的 WHILE 只接受单节点，参见单节点循环体。
• 原生循环不在循环体外包 try/except，所以循环通过条件退出而非 BreakLoop：on_success 把 ok 置真、或 on_error 用尽重试次数后，under_max 即返回假。
• IF(failed, ...) 只在所有尝试都失败时执行降级节点。
• 跳转地址由内置指令自动计算，RetryClause 无需手动管理偏移量
• 用户看到的只是 RetryClause(...)，底层展开细节完全透明

案例二：条件执行包装器

将“条件满足时执行某节点，否则跳过”这个常见模式封装为单一指令。

具体实现

class ExecuteWhen(SelfCompileInstruction):
    def __init__(self, condition: Node[bool], action: BaseNode):
        self._cond = condition
        self._action = action

    def extract(self) -> NodeCompose:
        # IF 不带 ELSE 完全合法——条件为假时分支自然跳过，无需 ELSE(NOP)
        return IF(self._cond, self._action).extract()

使用：

ExecuteWhen(has_data, process_data)

等价于 IF(has_data, process_data)——条件为假时工作流直接跳过该动作（无需 ELSE(NOP)）。

扩展：带否则分支的版本

class ExecuteWhenElse(SelfCompileInstruction):
    def __init__(self, condition: Node[bool], action: BaseNode, otherwise: BaseNode):
        self._cond = condition
        self._action = action
        self._other = otherwise

    def extract(self) -> NodeCompose:
        return NodeCompose(IF(self._cond, self._action).ELSE(self._other))

反汇编标注：__sdb_dis__ / __sdb_cmt__

自定义指令会展开成用户从未写过的节点，因此调试器清单默认显示的是框架内部节点。节点可以用两个软约束魔术属性描述自己所在的那一行，由 REPL 调试器的反汇编器读取：

属性	作用
__sdb_dis__	指令列显示的助记符。
__sdb_cmt__	非 None 时覆盖 ; 后的注释内容（默认为节点的 tag）。

两者都通过普通 getattr 在反汇编时读取——输入是已编译的产物，因此像跳转目标这样的操作数此时早已解析完毕。值依赖编译结果时优先用 @property：它每次列清单都重新读取实例状态，重编译后无需任何重新赋值步骤。

class RetryJump(BaseNode):
    """计数器未达上限时跳回循环体。"""

    __sdb_cmt__ = "retry back-edge"

    def __init__(self, limit: int) -> None:
        self._limit = limit
        self._body: list[int] = []
        self._init(self.__call__, tag=None, wrap_to_async=False, address_able=True)

    @property
    def __sdb_dis__(self) -> str:
        #  每次列清单重新读取，因此重编译不需要额外的赋值步骤
        return f"RETRY {self._limit} -> {self._body or '?'}"

    def __call__(self, pc: WorkflowInterpreter) -> None: ...

    def _post_compile(self, compose: NodeComposeRendered) -> None:
        #  操作数在此处解析——property 会自动取到新值
        self._body = compose.calc.resolve_alias("body")

四个值得记住的细节：

• @property 是数据描述符，因此该节点会拒绝 self.__sdb_dis__ = ...——这正是让取值保持单一来源的机制。固定助记符请用类属性；操作数只存在于闭包中时（PUSH_RET、INT 等）才用实例属性。
• 两个名字都有双尾下划线，不会触发名字改写，因此在类体内写 self.__sdb_dis__ = ... 是安全的。
• 每次 DLL apply() 变基后 _post_compile 都会重新执行，而 property 会自动取到新操作数——因此助记符里嵌入的地址在槽位重链接后依然正确。
• 节点永远不知道自己的地址，因此相对于所在段的操作数只能写成 #N（near_to 槽位）或 +N（offset 偏移），无法写成完整地址。

清单格式与内置指令集声明的助记符见 REPL 调试。

自定义指令的设计原则

1. 封装模式，而非封装逻辑：自定义指令应封装反复出现的编排模式（如重试、条件执行、超时保护），而非具体的业务逻辑。业务逻辑应留在节点内部

2. 利用现有指令：优先通过组合 IF、WHILE、Try 等内置指令来构建自定义指令，而非直接管理跳转偏移。只有在内置指令无法表达所需控制流时，才需要手动计算地址

3. 保持透明：自定义指令展开后的结构应与手写编排一致，不影响调试、挂起、中断等机制的正常工作

4. 命名语义化：指令名称应直接传达其控制流意图（如 Retry、Timeout、Parallel），让编排链读起来像自然语言

5. 标注助记符：给节点一个 __sdb_dis__，让调试器清单显示 RETRY 3 -> [1, 0] 而不是匿名内部节点——自定义指令恰恰是最需要它的那一类