Logic Nodes

Logic Nodes enable dynamic branching in aiXplain pipelines. They let you handle inputs or results differently depending on conditions.

Router Nodes: Check data type or simple conditions (e.g., TEXT vs. AUDIO) and route data to different branches.
Decision Nodes: Evaluate a numeric score (e.g., sentiment) and branch the pipeline based on thresholds (e.g., positive vs. negative).

This example demonstrates how to create a pipeline that routes input data into different processing paths and then makes a decision based on a numeric score. Specifically, it:

Uses a Router Node to branch the flow if the input is TEXT or AUDIO.
Processes text with a sentiment model (directly) and audio with a speech recognition model (to convert audio to text), then sentiment.
Applies a Decision Node to classify the sentiment as positive or negative.
Ends with output nodes for each classification outcome.

Create the Pipeline

This example pipeline consists of:

1 input node (accepting text or audio).
2 processing nodes (speech recognition, sentiment).
1 router node (distinguishes text vs. audio).
1 decision node (checks sentiment score).
2 output nodes (for positive or negative results).

from aixplain.factories import PipelineFactory
from aixplain.enums import DataType
from aixplain.modules.pipeline.designer.enums import (Operation, RouteType )
from aixplain.modules.pipeline.designer.nodes import (Router, Route, Decision)

# Create Pipeline & Input Node
pipeline = PipelineFactory.init("Pipeline Example")
input_node = pipeline.input()  # Pipeline’s main entry point

#  Router Node (TEXT vs. AUDIO)

router_node = Router(
    routes=[
        Route(value=DataType.TEXT, path=[], operation=Operation.EQUAL, type=RouteType.CHECK_TYPE),
        Route(value=DataType.AUDIO, path=[], operation=Operation.EQUAL, type=RouteType.CHECK_TYPE)],
    pipeline=pipeline
)

# Link pipeline input to the router’s input
input_node.outputs.input.link(router_node.inputs.input)

# Create output params for the router
router_node.outputs.create_param("text_route", data_type=DataType.TEXT)
router_node.outputs.create_param("audio_route", data_type=DataType.AUDIO)

#  Add Processing Nodes

translation_node = pipeline.translation(asset_id="60ddefbe8d38c51c5885f98a")
speech_node = pipeline.speech_recognition(asset_id="66060729709303af00f9a755")
sentiment_node = pipeline.sentiment_analysis(asset_id="61728750720b09325cbcdc36")

# Link Router to Processing Nodes

router_node.link(translation_node, from_param="text_route", to_param="text")
router_node.link(speech_node, from_param="audio_route", to_param="source_audio")

# For AUDIO, pass recognized text to sentiment
speech_node.outputs.data.link(sentiment_node.inputs.text)

# For TEXT, pass translation output to sentiment
translation_node.outputs.data.link(sentiment_node.inputs.text)

# Decision Node for Sentiment

decision_node = Decision(
    routes=[
        Route(
            value=0.5,
            path=[],
            operation=Operation.GREATER_THAN,
            type=RouteType.CHECK_TYPE,
        ),
    ],
    pipeline=pipeline
)

# Link the sentiment score to the decision node input
sentiment_node.outputs.score.link(decision_node.inputs.comparison)

# Create named outputs for positive/negative
decision_node.outputs.create_param("positive", data_type=DataType.TEXT)
decision_node.outputs.create_param("negative", data_type=DataType.TEXT)

# Add Output Nodes

positive_output = pipeline.output()
negative_output = pipeline.output()

decision_node.link(positive_output, from_param="positive", to_param="output")
decision_node.link(negative_output, from_param="negative", to_param="output")

# Save & Run

pipeline.save(save_as_asset=True)
print("Pipeline saved successfully!")

# Provide text => goes to translation -> sentiment -> decision
result_text = pipeline.run("Hello world", version="3.0")
print("Text Input =>", result_text)

Guide to Building the Example Pipeline

Initialize the Pipeline

from aixplain.factories import PipelineFactory
from aixplain.enums import DataType
from aixplain.modules.pipeline.designer.enums import Operation, RouteType
from aixplain.modules.pipeline.designer.nodes import Router, Route, Decision

pipeline = PipelineFactory.init("Pipeline Example")

Add Input and Output Nodes

We add one input node for any data (text or audio) and two output nodes to capture final results:

Input Nodes

input_node = pipeline.input()  # Single input node (accepts text or audio)

Output Nodes

positive_output = pipeline.output()
negative_output = pipeline.output()

Link Nodes

Router Node: Distinguishes TEXT vs. AUDIO

router_node = Router(
    routes=[
        Route(value=DataType.TEXT,  path=[], operation=Operation.EQUAL, type=RouteType.CHECK_TYPE),
        Route(value=DataType.AUDIO, path=[], operation=Operation.EQUAL, type=RouteType.CHECK_TYPE),
    ],
    pipeline=pipeline
)


# Connect pipeline input -> router
input_node.outputs.input.link(router_node.inputs.input)

# Manually create named outputs for the router
router_node.outputs.create_param("text_route", data_type=DataType.TEXT)
router_node.outputs.create_param("audio_route", data_type=DataType.AUDIO)

Text Path:

# Router -> sentiment (if TEXT)
router_node.link(sentiment_node, from_param="text_route", to_param="text")

Audio Path:

# Router -> speech (if AUDIO)
router_node.link(speech_node, from_param="audio_route", to_param="source_audio")

# Then speech node’s “data” (recognized text) -> sentiment
speech_node.outputs.data.link(sentiment_node.inputs.text)

Decision Node:

# sentiment_node outputs a 'score' param
sentiment_node.outputs.score.link(decision_node.inputs.comparison)

# Create named outputs on the decision
decision_node.outputs.create_param("positive", data_type=DataType.TEXT)
decision_node.outputs.create_param("negative", data_type=DataType.TEXT)

# Link decision to final outputs
decision_node.link(positive_output, from_param="positive", to_param="output")
decision_node.link(negative_output, from_param="negative", to_param="output")

View Node Parameters

You can inspect the available inputs/outputs for a node by printing:

print("Speech Node Inputs:", speech_node.inputs.__dict__)
print("Sentiment Node Outputs:", sentiment_node.outputs.__dict__)

Use this to confirm param names (e.g., source_audio, text, score).

Validate Pipeline

Confirm that:

The pipeline starts with input_node.
Ends with two output nodes (positive_output, negative_output).
Every node in between is linked in (receives data) and linked out (passes data forward).

If everything is correct, pipeline.validate() or a call to pipeline.save() will succeed.

Save and Run

pipeline.save(save_as_asset=True)  # Raises an error if the pipeline is invalid

# Provide text or audio input
result_text = pipeline.run("Hello world", version="3.0")
print("Text Input =>", result_text)

Using Router and Decision nodes together allows you to dynamically branch the pipeline flow based on data types (TEXT vs. AUDIO) or numeric conditions (score thresholds), creating richer AI workflows.

Create the Pipeline​

Guide to Building the Example Pipeline​

Initialize the Pipeline​

Add Input and Output Nodes​

Input Nodes​

Output Nodes​

Link Nodes​

View Node Parameters​

Validate Pipeline​

Save and Run​