stroblmeandClaude Fable 5
774b03953a
The first cut had node code call fluksio.log_metric, which was a second, undeclared way for data to leave a node: invisible to validation, absent from the canvas, and stored where the graph could not see it. That is precisely the MLflow discrepancy this framework exists to avoid, so it is gone. A node that produces values over time is a generator. Every yield is a dict keyed by output port, published the instant it happens — same port, same type check, same place on the canvas as any other value — and what it returns is its result. A port doing this declares stream: true, and a run keeps every number one takes, so experiment tracking is a consequence of the graph rather than an API beside it: a chart binds to a training curve the way it binds to a temperature. fluksio.emit writes the same ports imperatively, for where a yield cannot reach — inside a training framework's callback. In a live flow an emission also wakes what is downstream, as a subscriber publishing does; in a run it does not, because a run's graph is scheduled once and mid-node cascades would leave 'finished' with nothing to mean. The enqueued item carries no payload: the value is already in state, and one carrying it would re-apply an old emission after the node returned. Verified on the stack: 30 loss values arrived live on the flow socket during a run, attributed to the node that produced them, and the same node run on the remote worker streamed its curve back across the socket. Also caches remote compile results per worker, so attaching a GPU box does not put a network round trip in every rebuild. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AD8SfVhzXBG2nAfFcVh3iD
416 lines
16 KiB
Python
416 lines
16 KiB
Python
"""The node base class: identity, ports, lifecycle and execution.
|
|
|
|
Every node type in this package derives from :class:`Node`; the type-specific
|
|
modules beside this one add what talking to a particular outside world means.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
import logging
|
|
from collections.abc import Callable, Coroutine, Iterable, Iterator
|
|
from typing import TYPE_CHECKING, Any, TypeAlias
|
|
|
|
from app.flow import logs
|
|
from app.flow.messages import MessageSpec, qualify
|
|
from app.flow.supervision import Supervisor
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
from app.flow.pipeline import Pipeline
|
|
from app.flow.state import StateBackend
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# What a node hands back: the pipeline's state once it is bound, since a
|
|
# trigger runs the graph, and its own outputs when it is not.
|
|
NodeResult: TypeAlias = "StateBackend | dict[str, Any] | None"
|
|
|
|
|
|
class NodeOutputError(TypeError):
|
|
"""A node function returned something that cannot be mapped onto ports."""
|
|
|
|
|
|
class Node:
|
|
"""
|
|
A pipeline node that wraps a function with typed inputs/outputs.
|
|
|
|
Nodes are the fundamental building blocks of a pipeline. Each node encapsulates
|
|
a function that processes data, with explicit message-based inputs (requires)
|
|
and outputs (provides). Nodes can be connected into a directed acyclic graph (DAG)
|
|
where data flows from upstream to downstream nodes.
|
|
|
|
:param f: The function to execute when the node runs.
|
|
:type f: Callable
|
|
:param requires: Input messages this node consumes. Can be a single Message
|
|
or list of Messages. Empty for source nodes.
|
|
:type requires: MessageSpec | list[MessageSpec]
|
|
:param provides: Output messages this node produces. Can be a single Message
|
|
or list of Messages.
|
|
:type provides: MessageSpec | list[MessageSpec]
|
|
:param params: Additional parameters passed to the function during execution.
|
|
:type params: dict
|
|
:param name: Optional name for the node. Defaults to function name.
|
|
:type name: str | None
|
|
|
|
:ivar synchronous: If True, node only executes when all required inputs have
|
|
new versions since last execution. Useful for synchronizing multiple streams.
|
|
:vartype synchronous: bool
|
|
|
|
:example:
|
|
>>> def process_temp(temperature, params):
|
|
... return {"celsius": temperature * 0.5 + 32}
|
|
>>>
|
|
>>> temp_node = Node(
|
|
... f=process_temp,
|
|
... requires=MessageSpec(name="temperature", dtype=DType.FLOAT),
|
|
... provides=MessageSpec(name="celsius", dtype=DType.FLOAT),
|
|
... params={},
|
|
... )
|
|
"""
|
|
|
|
__slots__ = (
|
|
"f",
|
|
"id",
|
|
"flow",
|
|
"name",
|
|
"input_ports",
|
|
"output_ports",
|
|
"requires",
|
|
"provides",
|
|
"params",
|
|
"_pipeline",
|
|
"synchronous",
|
|
"_on_health",
|
|
"supervisor",
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
f: Callable[..., Any],
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
self.f = f
|
|
self._pipeline: Pipeline | None = None
|
|
self._on_health: Callable[[Node, str, str | None], None] | None = None
|
|
# Set by the controller before start(); a node built on its own (tests,
|
|
# previews) runs its loops unsupervised.
|
|
self.supervisor: Supervisor | None = None
|
|
self.params = dict(params) if params else {}
|
|
self.synchronous = bool(self.params.get("synchronous", False))
|
|
|
|
self.input_ports = self._normalize_ports(requires)
|
|
self.output_ports = self._normalize_ports(provides)
|
|
|
|
self.flow = ""
|
|
self.name: str = name or str(getattr(f, "__name__", "node"))
|
|
self.id: str = self.name
|
|
self._index_ports()
|
|
|
|
@staticmethod
|
|
def _normalize_ports(
|
|
msgs: MessageSpec | Iterable[MessageSpec],
|
|
) -> list[MessageSpec]:
|
|
"""Accept a single spec or any iterable of them."""
|
|
if isinstance(msgs, MessageSpec):
|
|
return [msgs]
|
|
return list(msgs or ())
|
|
|
|
def _index_ports(self) -> None:
|
|
"""Index bound ports by message name; unbound ports have no wiring."""
|
|
self.requires: dict[str, MessageSpec] = {
|
|
s.name: s for s in self.input_ports if s.name
|
|
}
|
|
self.provides: dict[str, MessageSpec] = {
|
|
s.name: s for s in self.output_ports if s.name
|
|
}
|
|
|
|
def assign_flow(self, flow: str, node_id: str) -> None:
|
|
"""Place this node in a flow, qualifying its identity and messages.
|
|
|
|
Called once by the loader, before the pipeline is built.
|
|
"""
|
|
self.flow = flow
|
|
self.name = node_id
|
|
self.id = f"{flow}.{node_id}"
|
|
self.input_ports = [
|
|
s.model_copy(update={"name": qualify(flow, s.name)})
|
|
for s in self.input_ports
|
|
]
|
|
self.output_ports = [
|
|
s.model_copy(update={"name": qualify(flow, s.name)})
|
|
for s in self.output_ports
|
|
]
|
|
self._index_ports()
|
|
|
|
@property
|
|
def local_id(self) -> str:
|
|
"""The node's id within its flow."""
|
|
return self.name
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Lifecycle
|
|
#
|
|
# A plain logic node has nothing to start or stop. The ones that talk to the
|
|
# outside — subscriptions, schedules, webhooks — override these, which is
|
|
# how the controller can bring a flow up or down without knowing what any
|
|
# particular node type is.
|
|
# -------------------------------------------------------------------------
|
|
|
|
# Whether running this node twice is the same as running it once. False
|
|
# for anything that reaches outside — a second publish is a second command
|
|
# to the device, which at-least-once delivery must not cause.
|
|
idempotent: bool = True
|
|
|
|
@classmethod
|
|
def instance_key(cls, params: dict[str, Any]) -> str | None:
|
|
"""Which outside thing these parameters point at, if any.
|
|
|
|
Two nodes with the same key talk to the same broker topic, URL or
|
|
bucket, however many flows they sit in — which is what the brain graph
|
|
draws as one neuron. ``None``, the default, means a node of this type
|
|
is only ever itself.
|
|
|
|
Called with the *stored* parameters, so a credential is still an
|
|
unresolved ``{"$secret": ...}`` reference. No key may be built from one:
|
|
an id is not a place for a password.
|
|
"""
|
|
return None
|
|
|
|
async def start(self, app: FastAPI | None = None) -> None:
|
|
"""Begin whatever this node listens to. Called when its flow starts."""
|
|
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
"""Undo :meth:`start`. Called before a rebuild, and must be idempotent."""
|
|
|
|
def _run_supervised(
|
|
self, name: str, factory: Callable[[], Coroutine[Any, Any, None]]
|
|
) -> asyncio.Task[None] | None:
|
|
"""Run a background loop under supervision where there is one.
|
|
|
|
Returns the task only when unsupervised, since that is the one case
|
|
where the caller has to cancel it itself.
|
|
"""
|
|
if self.supervisor is not None:
|
|
self.supervisor.spawn(f"{self.id}:{name}", self.flow, factory)
|
|
return None
|
|
return asyncio.create_task(factory())
|
|
|
|
def report_health(self, status: str, detail: str | None = None) -> None:
|
|
"""Say how this node's connection is doing: ok, degraded or down."""
|
|
if self._on_health is not None:
|
|
self._on_health(self, status, detail)
|
|
|
|
def bind(self, pipeline: Pipeline) -> None:
|
|
"""
|
|
Bind this node to a pipeline for external triggering.
|
|
|
|
Once bound, the node can trigger downstream execution when called.
|
|
This is typically done automatically during pipeline construction.
|
|
|
|
:param pipeline: The pipeline to bind this node to.
|
|
:type pipeline: Pipeline
|
|
"""
|
|
self._pipeline = pipeline
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Node-private state
|
|
#
|
|
# Code you write in a `python` node has no state handle: it is a pure
|
|
# function of its messages, and remembers things by reading a message it
|
|
# also writes. Node *types* the engine ships are a different matter — a
|
|
# rate limiter or a filter-on-change is about what happened last time, and
|
|
# that belongs to the engine rather than to any flow. It lives under a
|
|
# reserved key prefix, beside the timestamps and versions the pipeline
|
|
# already keeps there, so it survives a rebuild and never shows up as a
|
|
# message.
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _state_key(self, key: str) -> str:
|
|
return f"__node__:{self.id}:{key}"
|
|
|
|
def remember(self, key: str, value: Any) -> None:
|
|
"""Keep a value for the next run of this node."""
|
|
if self._pipeline is not None:
|
|
self._pipeline.state[self._state_key(key)] = value
|
|
|
|
def recall(self, key: str, default: Any = None) -> Any:
|
|
"""What this node last remembered under ``key``."""
|
|
if self._pipeline is None:
|
|
return default
|
|
return self._pipeline.state.get(self._state_key(key), default)
|
|
|
|
def forget(self, key: str) -> None:
|
|
if self._pipeline is not None:
|
|
self._pipeline.state.delete(self._state_key(key))
|
|
|
|
def __repr__(self) -> str:
|
|
return self.id
|
|
|
|
def __hash__(self) -> int:
|
|
return hash(self.id)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Port translation
|
|
#
|
|
# The graph speaks qualified message names; node functions speak ports.
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _to_kwargs(self, inputs: dict[str, Any]) -> dict[str, Any]:
|
|
kwargs = {}
|
|
for msg_name, value in inputs.items():
|
|
spec = self.requires.get(msg_name)
|
|
if spec is None:
|
|
continue
|
|
spec.check(value)
|
|
kwargs[spec.port] = value
|
|
return kwargs
|
|
|
|
def _to_messages(self, retval: Any) -> dict[str, Any] | None:
|
|
"""Map a function's port-keyed return value onto message names.
|
|
|
|
Only ``None`` means "nothing to publish". A falsy value of the wrong
|
|
shape — ``0``, ``""``, an empty list — is a mistake worth naming rather
|
|
than silence someone has to debug from an empty canvas.
|
|
"""
|
|
if retval is None:
|
|
return None
|
|
if not isinstance(retval, dict):
|
|
raise NodeOutputError(
|
|
f"'{self.local_id}' returned {type(retval).__name__}. Outputs are "
|
|
"keyed by port, so return a dict like {'out': value}, or None to "
|
|
"publish nothing."
|
|
)
|
|
by_port = {s.port: s for s in self.output_ports if s.name}
|
|
outputs = {}
|
|
for key, value in retval.items():
|
|
spec = by_port.get(key) or self.provides.get(key)
|
|
if spec is None:
|
|
continue
|
|
spec.check(value)
|
|
outputs[spec.name] = value
|
|
return outputs or None
|
|
|
|
def execute(self, inputs: dict[str, Any] | None = None) -> dict[str, Any] | None:
|
|
"""Run the node function and return its outputs by message name.
|
|
|
|
Downstream nodes are not triggered — the pipeline schedules those.
|
|
"""
|
|
kwargs = self._to_kwargs(inputs or {})
|
|
result = self.f(**kwargs, params=self.params)
|
|
if inspect.isgenerator(result):
|
|
# Only when the node runs in this process; out of process the
|
|
# worker has already drained it and sent each yield on ahead.
|
|
result = self._drain(result)
|
|
return self._to_messages(result)
|
|
|
|
def _drain(self, generator: Iterator[Any]) -> Any:
|
|
"""Publish each yield as it happens; the end of it is the result."""
|
|
pending: Any = None
|
|
have_pending = False
|
|
try:
|
|
while True:
|
|
value = next(generator)
|
|
if have_pending:
|
|
self.emit(pending)
|
|
pending, have_pending = value, True
|
|
except StopIteration as stop:
|
|
if stop.value is not None:
|
|
if have_pending:
|
|
self.emit(pending)
|
|
return stop.value
|
|
return pending if have_pending else None
|
|
|
|
def emit(self, values: dict[str, Any] | None) -> None:
|
|
"""Publish on this node's output ports mid-execution.
|
|
|
|
A value produced while a node is still working is a value like any
|
|
other: same ports, same type checking, same place on the canvas. What
|
|
it is *not* is a log — nothing leaves a node except through a port it
|
|
declared, so a training curve is an output of the graph rather than a
|
|
side effect beside it.
|
|
"""
|
|
outputs = self._to_messages(values)
|
|
if outputs and self._pipeline is not None:
|
|
self._pipeline.publish_emission(self, outputs)
|
|
|
|
def trigger(
|
|
self, inputs: dict[str, Any] | None = None, durable: bool | None = None
|
|
) -> NodeResult:
|
|
"""
|
|
Trigger this node externally, executing downstream nodes if dependencies are met.
|
|
|
|
This method is for nodes that receive data from upstream dependencies.
|
|
For trigger/subscriber nodes that inject data into the pipeline, use :meth:`inject`.
|
|
|
|
:param inputs: Input values matching this node's ``requires``.
|
|
:type inputs: dict | None
|
|
:returns: Result of the node execution and downstream propagation.
|
|
:rtype: dict | None
|
|
:raises RuntimeError: If node is not bound to a pipeline.
|
|
"""
|
|
if self._pipeline is None:
|
|
raise RuntimeError("Node must be bound to a pipeline to trigger")
|
|
return self(inputs, durable=durable)
|
|
|
|
def inject(
|
|
self, outputs: dict[str, Any] | None = None, durable: bool | None = None
|
|
) -> NodeResult:
|
|
"""
|
|
Inject data into the pipeline as if this node produced it.
|
|
|
|
This method is for trigger/subscriber nodes that receive external data
|
|
(e.g., HTTP requests, MQTT messages) and need to inject it into the pipeline.
|
|
The data is validated against this node's ``provides`` specification.
|
|
|
|
For source nodes (nodes with no ``requires``), if no outputs are provided,
|
|
the node's function will be executed to generate outputs.
|
|
|
|
:param outputs: Output values matching this node's ``provides``.
|
|
:type outputs: dict | None
|
|
:returns: Result of downstream propagation.
|
|
:rtype: dict | None
|
|
:raises RuntimeError: If node is not bound to a pipeline.
|
|
:raises TypeError: If output values don't match ``provides`` types.
|
|
:raises KeyError: If required output keys are missing.
|
|
"""
|
|
if self._pipeline is None:
|
|
raise RuntimeError("Node must be bound to a pipeline to inject")
|
|
|
|
outputs = outputs or {}
|
|
|
|
# A source node asked to inject nothing produces its own data.
|
|
if not outputs and not self.requires:
|
|
collected = logs.Collector()
|
|
with logs.capture(collected):
|
|
outputs = self.f(params=self.params) or {}
|
|
self._pipeline.publish_log(self, collected, "")
|
|
|
|
return self._pipeline.trigger(self, self._to_messages(outputs), durable=durable)
|
|
|
|
def __call__(
|
|
self, inputs: dict[str, Any] | None = None, durable: bool | None = None
|
|
) -> NodeResult:
|
|
"""
|
|
Execute the node and trigger downstream nodes if bound to a pipeline.
|
|
|
|
Validates inputs against the node's ``requires`` specification, executes
|
|
the wrapped function, validates outputs, and triggers downstream execution
|
|
if the node is bound to a pipeline.
|
|
|
|
:param inputs: Input values keyed by message name. Must match the node's
|
|
``requires`` specification.
|
|
:type inputs: dict | None
|
|
:returns: Node outputs if successful, or pipeline execution results if bound.
|
|
:rtype: dict | None
|
|
"""
|
|
outputs = self.execute(inputs)
|
|
if self._pipeline is None:
|
|
return outputs
|
|
return self._pipeline.trigger(self, outputs, durable=durable)
|