Skip to main content
A workflow ties multiple nodes together. To create a workflow, decorate a function with @workflow(). This function defines the data flow between your nodes.

​
Example

from typing import Annotated
from pipeline_ui import node, workflow, TextParameter, NodePosition, ImageOutput, ImageInput
from pipeline_ui.modules.node._response.response_media_schema import ImageResponse

@node()
def get_prompt(
    prompt: str = TextParameter(default="A beautiful image", description="Image prompt")
) -> Annotated[str, NodeOutput(name="prompt", description="Encoded prompt")]:
    return prompt

@node()
def generate_image(
    prompt: str = NodeInput(python_type=str, description="The encoded prompt")
) -> Annotated["Image", ImageOutput(name="image", description="Generated image")]:
    # Imagine your image generation code here
    from PIL import Image
    return Image.new("RGB", (256, 256), (255, 255, 255))

@workflow(positions=[
    NodePosition(x=-100, y=-200, node="get_prompt"),
    NodePosition(x=150, y=-200, node="generate_image")
])
def simple_workflow(
    prompt: str = TextParameter(default="A beautiful image", description="Image prompt")
) -> Annotated["Image", ImageResponse(name="image", description="Final output", render=True)]:
    encoded = get_prompt(prompt)
    image = generate_image(encoded)
    return image

​
Tips

  • Use the positions parameter in the workflow decorator to specify node locations on the canvas.
  • Ensure that every node used in your workflow is connected (all input and output pins must be linked).

​
Finding more examples

You can find more examples in the examples folder. Each file is a self-contained workflow that you can run.