53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
"""Node executor protocol and registry."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Protocol, runtime_checkable
|
|
|
|
from deerflow.workflows.errors import WorkflowError
|
|
from deerflow.workflows.ports import ACTIVE_PORTS
|
|
from deerflow.workflows.runtime.context import RunContext
|
|
from deerflow.workflows.schemas import NodeResult, NodeType, WorkflowNode
|
|
|
|
|
|
@runtime_checkable
|
|
class NodeExecutor(Protocol):
|
|
async def execute(self, node: WorkflowNode, ctx: RunContext) -> NodeResult: ...
|
|
|
|
|
|
class NodeRegistry:
|
|
"""Maps node type → executor. Unknown types fail with a typed error."""
|
|
|
|
def __init__(self) -> None:
|
|
self._executors: dict[str, NodeExecutor] = {}
|
|
|
|
def register(self, node_type: NodeType | str, executor: NodeExecutor) -> None:
|
|
self._executors[str(node_type)] = executor
|
|
|
|
def get(self, node_type: str) -> NodeExecutor:
|
|
executor = self._executors.get(node_type)
|
|
if executor is None:
|
|
raise WorkflowError(
|
|
"WORKFLOW_RESOURCE_MISSING",
|
|
f"节点类型暂不支持执行：{node_type}",
|
|
details={"nodeType": node_type},
|
|
)
|
|
return executor
|
|
|
|
def supported(self) -> list[str]:
|
|
return sorted(self._executors)
|
|
|
|
|
|
def require_dep(value: Any, *, node: WorkflowNode, what: str) -> Any:
|
|
if value is None:
|
|
raise WorkflowError(
|
|
"WORKFLOW_RESOURCE_MISSING",
|
|
f"运行环境缺少{what}能力，无法执行该节点",
|
|
node_id=node.id,
|
|
details={"capability": what},
|
|
)
|
|
return value
|
|
|
|
|
|
__all__ = ["ACTIVE_PORTS", "NodeExecutor", "NodeRegistry", "require_dep"]
|