Skip to content

feat(dynaflow): add per-node retry policy, error fallback edges & state resumability #338

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Kaap10 merged 1 commit into from
Oct 13, 2026
Merged

feat(dynaflow): add per-node retry policy, error fallback edges & state resumability #338

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions src/dynavec/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,14 +81,17 @@
ApprovalNode,
CodeNode,
Edge,
GraphRunState,
LLMNode,
LoopNode,
Node,
NodePort,
NodeRegistry,
NodeSchema,
NodeStatus,
PortType,
RetrieverNode,
RetryPolicy,
RouterNode,
SimpleGraph,
SubGraphNode,
Expand Down Expand Up @@ -203,10 +206,13 @@
"Node",
"NodePort",
"NodeSchema",
"NodeStatus",
"RetryPolicy",
"PortType",
"NodeRegistry",
"node_registry",
"Edge",
"GraphRunState",
"SimpleGraph",
"RetrieverNode",
"LLMNode",
Expand Down
7 changes: 5 additions & 2 deletions src/dynavec/nodes/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

from __future__ import annotations

from .base import Node, NodePort, NodeSchema, PortType
from .base import Node, NodePort, NodeSchema, NodeStatus, PortType, RetryPolicy
from .builtin import (
ApprovalNode,
CodeNode,
Expand All @@ -14,14 +14,16 @@
ToolNode,
WebhookNode,
)
from .graph import Edge, SimpleGraph
from .graph import Edge, GraphRunState, SimpleGraph
from .registry import NodeRegistry, node_registry

__all__ = [
"Node",
"NodePort",
"NodeSchema",
"NodeStatus",
"PortType",
"RetryPolicy",
"NodeRegistry",
"node_registry",
"RetrieverNode",
Expand All @@ -34,5 +36,6 @@
"WebhookNode",
"SubGraphNode",
"Edge",
"GraphRunState",
"SimpleGraph",
]
62 changes: 57 additions & 5 deletions src/dynavec/nodes/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,52 @@

import uuid
from abc import ABC, abstractmethod
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any, Literal

PortType = Literal["string", "number", "boolean", "array", "object", "any"]
NodeStatus = Literal["pending", "running", "retrying", "succeeded", "failed", "skipped"]


@dataclass
class RetryPolicy:
"""Configurable retry policy for node execution with exponential backoff and jitter."""

max_attempts: int = 3
base_delay: float = 0.05
max_delay: float = 2.0
jitter: bool = True
retry_on: tuple[type[Exception], ...] | Callable[[Exception], bool] | None = None

def should_retry(self, exc: Exception) -> bool:
"""Evaluate if an exception is eligible for retry under this policy."""
if self.retry_on is None:
return True
if isinstance(self.retry_on, tuple):
return isinstance(exc, self.retry_on)
if callable(self.retry_on):
return bool(self.retry_on(exc))
return True

def to_dict(self) -> dict[str, Any]:
"""Serialize retry policy to dictionary."""
return {
"max_attempts": self.max_attempts,
"base_delay": self.base_delay,
"max_delay": self.max_delay,
"jitter": self.jitter,
}

@classmethod
def from_dict(cls, data: dict[str, Any]) -> RetryPolicy:
"""Deserialize retry policy from dictionary."""
return cls(
max_attempts=data.get("max_attempts", 3),
base_delay=data.get("base_delay", 0.05),
max_delay=data.get("max_delay", 2.0),
jitter=data.get("jitter", True),
)


@dataclass
Expand Down Expand Up @@ -78,23 +120,33 @@ def __init__(
self,
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
self.id = node_id or f"{self.node_type}_{uuid.uuid4().hex[:8]}"
self.config = dict(config or {})
self.retry_policy = retry_policy
self.status: NodeStatus = "pending"
self.attempts: int = 0
self.last_error: str | None = None

@abstractmethod
def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
"""Execute the node logic given an inputs dictionary, returning outputs."""

def to_dict(self) -> dict[str, Any]:
"""Serialize node to a ReactFlow-compatible node object."""
data: dict[str, Any] = {
"label": self.schema.label,
"config": self.config,
"category": self.schema.category,
"status": self.status,
}
if self.retry_policy is not None:
data["retry_policy"] = self.retry_policy.to_dict()

return {
"id": self.id,
"type": self.node_type,
"data": {
"label": self.schema.label,
"config": self.config,
"category": self.schema.category,
},
"data": data,
"position": {"x": 0, "y": 0},
}
39 changes: 31 additions & 8 deletions src/dynavec/nodes/builtin.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
from typing import Any

from ..chat.base import ChatModel, Message
from .base import Node, NodePort, NodeSchema
from .base import Node, NodePort, NodeSchema, RetryPolicy
from .registry import node_registry

# --- 1. RetrieverNode ---
Expand Down Expand Up @@ -68,8 +68,9 @@ def __init__(
retriever_fn: Callable[..., Any] | None = None,
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(node_id=node_id, config=config)
super().__init__(node_id=node_id, config=config, retry_policy=retry_policy)
self.retriever_fn = retriever_fn

def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
Expand Down Expand Up @@ -163,8 +164,9 @@ def __init__(
model: ChatModel | Callable[[list[Message]], Any],
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(node_id=node_id, config=config)
super().__init__(node_id=node_id, config=config, retry_policy=retry_policy)
self.model = model

def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
Expand Down Expand Up @@ -249,8 +251,9 @@ def __init__(
tool: Callable[..., Any],
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(node_id=node_id, config=config)
super().__init__(node_id=node_id, config=config, retry_policy=retry_policy)
self.tool = tool

def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
Expand Down Expand Up @@ -335,8 +338,9 @@ def __init__(
route_fn: Callable[[Any], str] | None = None,
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(node_id=node_id, config=config)
super().__init__(node_id=node_id, config=config, retry_policy=retry_policy)
self.route_fn = route_fn

def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
Expand Down Expand Up @@ -391,8 +395,9 @@ def __init__(
code_fn: Callable[[dict[str, Any]], dict[str, Any]],
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(node_id=node_id, config=config)
super().__init__(node_id=node_id, config=config, retry_policy=retry_policy)
self.code_fn = code_fn

def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
Expand Down Expand Up @@ -460,6 +465,14 @@ class ApprovalNode(Node):
},
)

def __init__(
self,
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(node_id=node_id, config=config, retry_policy=retry_policy)

def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
return {
"approved": bool(inputs.get("approved", True)),
Expand Down Expand Up @@ -511,8 +524,9 @@ def __init__(
map_fn: Callable[[Any], Any] | None = None,
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(node_id=node_id, config=config)
super().__init__(node_id=node_id, config=config, retry_policy=retry_policy)
self.map_fn = map_fn

def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
Expand Down Expand Up @@ -571,6 +585,14 @@ class WebhookNode(Node):
},
)

def __init__(
self,
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(node_id=node_id, config=config, retry_policy=retry_policy)

def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
return {"payload": inputs.get("payload", {})}

Expand Down Expand Up @@ -613,8 +635,9 @@ def __init__(
subgraph: Any = None,
node_id: str | None = None,
config: dict[str, Any] | None = None,
retry_policy: RetryPolicy | None = None,
) -> None:
super().__init__(node_id=node_id, config=config)
super().__init__(node_id=node_id, config=config, retry_policy=retry_policy)
self.subgraph = subgraph

def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
Expand Down
Loading
Loading