The ConditionalNode class branches workflow execution based on a single condition, routing to one set of nodes when the condition is true and another when it is false.
Constructor
from fibonacci import ConditionalNode
node = ConditionalNode(
id="check_sentiment",
name="Check Sentiment",
left_value="{{analyze_sentiment}}",
operator="contains",
right_value="positive",
true_branch=["celebrate"],
false_branch=["investigate"],
dependencies=["analyze_sentiment"]
)
Parameters
| Parameter | Type | Default | Description |
|---|
id | str | Required | Unique node identifier (lowercase letters, numbers, underscores, hyphens) |
name | str | Required | Human-readable node name |
left_value | str | Required | Left side of the condition — supports template variables like {{node_id}} or {{input.field}} |
operator | str | Required | Comparison operator (see table below) |
right_value | str | "" | Right side of the condition — leave empty for is_empty / is_not_empty operators |
true_branch | list[str] | [] | Node IDs to execute when the condition is true |
false_branch | list[str] | [] | Node IDs to execute when the condition is false |
dependencies | list[str] | [] | Node IDs this node waits for before running |
enable_retry | bool | False | Retry on failure |
max_retries | int | 3 | Maximum retry attempts (used when enable_retry=True) |
retry_delay | float | 1.0 | Initial delay in seconds between retries |
Supported Operators
The operator field accepts exactly these 10 values:
| Operator | Description | Example |
|---|
equals | Exact string or numeric match | "completed" equals "completed" |
not_equals | Not equal | "pending" not_equals "completed" |
contains | Left value contains right value as a substring | "error on line 5" contains "error" |
not_contains | Left value does not contain right value | "success" not_contains "error" |
greater_than | Numeric greater-than | 100 greater_than 50 |
less_than | Numeric less-than | 25 less_than 50 |
starts_with | Left value starts with right value | "error: msg" starts_with "error" |
ends_with | Left value ends with right value | "report.pdf" ends_with ".pdf" |
is_empty | Left value is empty or null (right_value is ignored) | "" is_empty |
is_not_empty | Left value has a non-empty value (right_value is ignored) | "some text" is_not_empty |
Only these 10 operators are supported. Using any other string will raise a ValidationError at workflow build time.
Basic Examples
Simple True/False Branch
from fibonacci import Workflow, LLMNode, ToolNode, ConditionalNode
wf = Workflow(name="feedback-router")
# Step 1: Classify the feedback
classify = LLMNode(
id="classify",
name="Classify Feedback",
instruction="Classify this feedback as positive or negative. Reply with one word only.\n\n{{input.feedback}}"
)
# Step 2: Route based on result
router = ConditionalNode(
id="router",
name="Route Feedback",
left_value="{{classify}}",
operator="contains",
right_value="negative",
true_branch=["escalate"],
false_branch=["log_ok"],
dependencies=["classify"]
)
# True branch
escalate = ToolNode(
id="escalate",
name="Escalate to Support",
tool="slack_send_message",
params={
"channel": "#support",
"message": "Negative feedback: {{input.feedback}}"
},
dependencies=["router"]
)
# False branch
log_ok = ToolNode(
id="log_ok",
name="Log Positive Feedback",
tool="google_sheets_append",
params={
"spreadsheet_id": "{{input.sheet_id}}",
"range": "Log!A:B",
"values": [["{{input.feedback}}", "positive"]]
},
dependencies=["router"]
)
wf.add_nodes([classify, router, escalate, log_ok])
Numeric Threshold
check_score = ConditionalNode(
id="check_score",
name="Score Threshold",
left_value="{{evaluate}}",
operator="greater_than",
right_value="7",
true_branch=["publish"],
false_branch=["revise"],
dependencies=["evaluate"]
)
String Prefix/Suffix Check
check_type = ConditionalNode(
id="check_type",
name="Check File Type",
left_value="{{input.filename}}",
operator="ends_with",
right_value=".pdf",
true_branch=["process_pdf"],
false_branch=["unsupported_format"]
)
Empty Value Guard
check_data = ConditionalNode(
id="check_data",
name="Data Present?",
left_value="{{fetch_data}}",
operator="is_not_empty",
right_value="", # ignored for is_empty / is_not_empty
true_branch=["process_data"],
false_branch=["send_no_data_alert"]
)
Chaining Conditions (AND / OR Logic)
The ConditionalNode evaluates a single condition. Chain multiple nodes together to build AND / OR logic.
AND Logic — chain nodes
# First check: sentiment negative?
check_sentiment = ConditionalNode(
id="check_sentiment",
name="Negative Sentiment?",
left_value="{{analyze}}",
operator="contains",
right_value="negative",
true_branch=["check_urgency"], # only continue AND chain when true
false_branch=["standard_response"],
dependencies=["analyze"]
)
# Second check: urgency high? (only reached when sentiment IS negative)
check_urgency = ConditionalNode(
id="check_urgency",
name="High Urgency?",
left_value="{{analyze_urgency}}",
operator="equals",
right_value="high",
true_branch=["escalate"], # negative AND high urgency
false_branch=["soft_response"],
dependencies=["check_sentiment"]
)
OR Logic — multiple conditions pointing to the same branch
# Escalate if sentiment is negative
check_neg = ConditionalNode(
id="check_neg",
name="Negative?",
left_value="{{analyze.sentiment}}",
operator="equals",
right_value="negative",
true_branch=["escalate"],
false_branch=["check_critical"],
dependencies=["analyze"]
)
# OR escalate if category is complaint
check_critical = ConditionalNode(
id="check_critical",
name="Complaint?",
left_value="{{analyze.category}}",
operator="equals",
right_value="complaint",
true_branch=["escalate"], # same escalate node
false_branch=["normal_flow"],
dependencies=["check_neg"]
)
Per-Node Condition (Conditional Execution)
Every node type supports .with_condition() to skip execution unless a condition is met:
# Only send email if the user opted in
send_email = ToolNode(
id="send_email",
name="Send Email",
tool="gmail_send",
params={...}
).with_condition(
left_value="{{input.email_opt_in}}",
operator="equals",
right_value="true"
)
# Only run expensive analysis on large payloads
deep_analysis = LLMNode(
id="deep_analysis",
name="Deep Analysis",
instruction="..."
).with_condition(
left_value="{{input.data_size}}",
operator="greater_than",
right_value="1000"
)
Retry Configuration
Use .with_retry() to enable retries on the conditional node itself:
router = ConditionalNode(
id="router",
name="Route Request",
left_value="{{classify}}",
operator="equals",
right_value="urgent",
true_branch=["urgent_handler"],
false_branch=["standard_handler"],
dependencies=["classify"]
).with_retry(max_retries=3, delay=1.0)
YAML Configuration
nodes:
- id: router
name: Route by Sentiment
type: condition
left_value: "{{analyze}}"
operator: contains
right_value: "negative"
true_branch:
- escalate
false_branch:
- log_ok
dependencies:
- analyze
Complete Example
from fibonacci import Workflow, LLMNode, ToolNode, ConditionalNode
wf = Workflow(name="support-ticket-router")
# Classify ticket
classify = LLMNode(
id="classify",
name="Classify Ticket",
instruction="""Classify this support ticket.
Respond with exactly one word: urgent, billing, technical, or general.
Ticket: {{input.message}}"""
)
# Route urgent tickets immediately
check_urgent = ConditionalNode(
id="check_urgent",
name="Urgent?",
left_value="{{classify}}",
operator="equals",
right_value="urgent",
true_branch=["page_on_call"],
false_branch=["check_billing"],
dependencies=["classify"]
)
# Route billing tickets
check_billing = ConditionalNode(
id="check_billing",
name="Billing Issue?",
left_value="{{classify}}",
operator="equals",
right_value="billing",
true_branch=["billing_team"],
false_branch=["tech_team"],
dependencies=["check_urgent"]
)
# Urgent: page on-call engineer
page_on_call = ToolNode(
id="page_on_call",
name="Page On-Call",
tool="slack_send_message",
params={
"channel": "#on-call",
"message": "🚨 Urgent ticket: {{input.message}}"
},
dependencies=["check_urgent"]
)
# Billing team
billing_team = ToolNode(
id="billing_team",
name="Notify Billing",
tool="slack_send_message",
params={
"channel": "#billing",
"message": "Billing ticket: {{input.message}}"
},
dependencies=["check_billing"]
)
# Tech support (default)
tech_team = LLMNode(
id="tech_team",
name="Tech Response",
instruction="Provide a helpful technical response to: {{input.message}}",
dependencies=["check_billing"]
)
wf.add_nodes([classify, check_urgent, check_billing, page_on_call, billing_team, tech_team])
result = wf.run(input_data={"message": "My payment failed!"})
print(result.output_data)
