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"})

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

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

Environments

Agents

Tools and Connections

Large Language Models (LLMs)

Knowledge Bases

Webchats

Tutorials

API's reference

Legal notices