Workflow Edges
Edges define the connections between workflow nodes — what happens next after each node completes. Edges can be unconditional (always go to the same next node) or conditional (choose the next node based on state).
Edge Types
Unconditional edge: always go to the same next node.
lookup_order → check_eligibility
Conditional edge: choose the next node based on state or the node's return value.
check_eligibility → issue_refund (if amount < 100)
check_eligibility → human_approval (if amount ≥ 100)
check_eligibility → reject_refund (if ineligible)
Terminal edge: the workflow ends.
send_confirmation → END
Loop edge: go back to a previous node.
wait_for_approval → check_eligibility (after approval received)
Fallback edge: triggered on failure.
lookup_order → [failure] → escalate_to_human
Edge Map Implementation
# Static edge map — defines all allowed transitions
edge_map = {
"START": {"*": "lookup_order"},
"lookup_order": {"found": "check_eligibility",
"not_found": "notify_not_found"},
"check_eligibility": {"auto_refund": "issue_refund",
"needs_approval": "request_human_approval",
"rejected": "reject_refund"},
"request_human_approval":{"approved": "issue_refund",
"rejected": "reject_refund"},
"issue_refund": {"success": "send_confirmation",
"error": "escalate_to_human"},
"reject_refund": {"*": "send_rejection_email"},
"notify_not_found": {"*": "END"},
"send_confirmation": {"*": "END"},
"send_rejection_email": {"*": "END"},
"escalate_to_human": {"*": "END"},
}
def get_next_node(edge_map: dict, current_node: str, condition: str = "*") -> str:
"""Resolve the next node from the edge map."""
transitions = edge_map.get(current_node, {})
# Try specific condition first, then wildcard
return transitions.get(condition, transitions.get("*", "END"))
# Test edge resolution
print(get_next_node(edge_map, "lookup_order", "found")) # check_eligibility
print(get_next_node(edge_map, "check_eligibility", "auto_refund")) # issue_refund
print(get_next_node(edge_map, "send_confirmation")) # END