Testing flows

After importing a flow tool, you can test it using a main Python script before add it to an agent. The following steps guide you through testing a flow, including defining, compiling, and running the flow:

Create a test script

Flows are asynchronous, and so are the functions used to run them.Start by creating a script that defines a function to invoke your flow. In the example below, the script hello_message_flow.py contains a function build_hello_message_flow, decorated with @flow, which returns the Flow object you want to test.

Python

import asyncio

from .hello_message_flow import build_hello_message_flow

async def main():

    my_flow_definition = build_hello_message_flow()

if __name__ == "__main__":
    asyncio.run(main())

Compile the flow

Next, compile the flow. This step generates the flow model and deploys it to the engine. It returns a CompiledFlow instance, which you can use to start a flow run. Use compile_deploy() to compile and deploy the flow.

Python

compiled_flow = await my_flow_definition.compile_deploy()

If you only want to generate the JSON model without deploying it, use compile() instead.

Start a run

Flows are asynchronous, and the engine communicates with your client using events. You can start a run using methods on the CompiledFlow object. Depending on your needs, you can choose from the following approaches:

A simple client

The simplest way to run a flow is to invoke it and wait for the result. Use CompiledFlow.invoke() to start the flow. You can optionally pass input data.

Python

flow_run = await compiled_flow.invoke({"first_name": "John", "last_name": "Doe"})

Show optional parameters to invoke() function

Parameter	Type	Description
input_data	`dict`	Input data to pass to the flow.
on_flow_end_handler	`callable`	Callback executed when the flow completes successfully.
on_flow_error_handler	`callable`	Callback executed if the flow encounters an error.
debug	`bool`	Enables debug mode if set to `True`.

To monitor the flow and retrieve the output:

Python

from ibm_watsonx_orchestrate.experimental.flow_builder.flows.flow import FlowRunStatus

# Wait for the flow to complete
while flow_run.status != FlowRunStatus.COMPLETED and flow_run.status != FlowRunStatus.FAILED:
    await asyncio.sleep(1)  # sleep 1 second between checks
    
print(f"Flow ended with status: {flow_run.status}")
print(f"Flow output: {flow_run.output}" if flow_run.status == FlowRunStatus.COMPLETED else f"Flow error: {flow_run.error}")

Note: Always use asynchronous waits to avoid blocking the event loop.

Using event handlers

You can provide custom event handlers to invoke() for more control over flow completion and error handling.

Python

flow_run = None

def on_flow_end(result):
    """
    Callback function to be called when the flow is completed.
    """
    print(f"Custom Handler: flow `{flow_run.name}` completed with result: {result}")

def on_flow_error(error):
    """
    Callback function to be called when the flow fails.
    """
    print(f"Custom Handler: flow `{flow_run.name}` failed: {error}")

async def main():

    my_flow_definition = build_hello_message_flow()
    compiled_flow = await my_flow_definition.compile_deploy()
    
    global flow_run
    flow_run = await compiled_flow.invoke({"first_name": "John", "last_name": "Doe"}, on_flow_end_handler=on_flow_end, on_flow_error_handler=on_flow_error)

Processing raw events

To handle all emitted events, use invoke_events(), which returns an asynchronous iterator yielding (FlowEvent, FlowRun) tuples.

Python

from ibm_watsonx_orchestrate.experimental.flow_builder.types import FlowEventType

async for event, run in compiled_flow.invoke_events({"first_name": "John", "last_name": "Doe"}):
    if not event:
        continue
    
    if event.kind == FlowEventType.ON_FLOW_END:
        on_flow_end(event.context.data.output)
        print(f"Flow `{run.name}` completed with result: {run.output}")
        break

    elif event.kind == FlowEventType.ON_FLOW_ERROR:
        print(f"Flow `{run.name}` failed with error: {run.error}")
        break

Show optional parameters to invoke() function

Parameter	Type	Description
input_data	`dict`	Input data to pass to the flow.
filters	`Sequence[Union[FlowEventType, TaskEventType]]`	Callback executed when the flow completes successfully.
debug	`bool`	Callback executed if the flow encounters an error.

You can listen for task and flow events, as defined by FlowEventType and TaskEventType in ibm_watsonx_orchestrate.experimental.flow_builder.types.

Release Notes

Get Started

Agents

Connections

Large Language Models (LLMs)

Tools

Orchestrate copilot

Knowledge Bases

Evaluation Framework

Web chats

watsonx Orchestrate Developer Edition

Legal notices