Building agentic workflows - IBM watsonx Orchestrate ADK

Define a Python function with the @flow decorator

Use the @flow decorator to define your agentic workflow. In the decorator, specify the agentic workflow’s name, display_name, description, input_schema, output_schema, initiators, schedulable, llm_model, and agent_conversation_memory_turns_limit.

The function must take a single parameter of type Flow and return a Flow object.
Construct the agentic workflow using a combination of tool(), agent(), and edge-building functions like sequence() or edge().
For Branch and Loop nodes, use a Python expression in the evaluator to define the branching or looping condition. For more information, see Agentic workflow expressions.
To support multiple languages, use aflow.target_locales(['fr', 'es', 'de']) to specify target languages for user activities. After you import the agentic workflow, export translations to CSV, translate them, and import them back. For details, see Multi-language support.
When you use forms in user nodes, connect the form buttons to the next node in the flow using the edge() function with the button_label parameter. This connection enables users to trigger the transition to the next node when they click a button on the form.

The following tabs show some code snippet examples of agentic workflow:

Example 1
Example 2
Example 3
Example 4
Example 5

Example 1: Sequencing two tools

Python

@flow(
    name = "hello_message_flow",
    input_schema=Name,
    output_schema=str
)
def build_hello_message_flow(aflow: Flow = None) -> Flow:
    """ Based on the first and last name of a person, combine into a single name and create a simple hello world message. """

    combine_names_node = aflow.tool(combine_names)
    get_hello_message_node = aflow.tool(get_hello_message)

    aflow.edge(START, combine_names_node).edge(combine_names_node, get_hello_message_node).edge(get_hello_message_node, END)

    return aflow

For the full example and complete code, see hello_message_flow.

Example 2: Sequencing two agents

Python

@flow(
    name="ibm_knowledge_to_emails",
    description="This flow will send a random fact about IBM to a group of people",
    input_schema=FlowInput,
    output_schema=FlowOutput
)
def build_ibm_knowledge_to_emails(aflow: Flow) -> Flow:
    """ Retrieve a random fact about IBM and send it out to an email list. """

    ask_agent_for_ibm_knowledge = aflow.agent(
        name="ask_agent_for_ibm_knowledge",
        agent="ibm_agent",
        display_name="ask_agent_for_ibm_knowledge",
        message="Please retrieve a random fact about IBM on a topic.",
        description="Get a random fact about IBM.",
        input_schema=IBMAgentInput,
        output_schema=IBMAgentOutput,
    )

    ask_agent_to_send_email_node = aflow.agent(
        name="ask_agent_to_send_email",
        agent="email_agent",
        display_name="ask_agent_to_send_email",
        message="Please send email based on the following email addresses and based on a fact about IBM.'",
        description="This agent will send email content to a list of email addresses",
        input_schema=EmailAgentInput,
        output_schema=EmailAgentOutput,
    )
    aflow.sequence(START, ask_agent_for_ibm_knowledge, ask_agent_to_send_email_node, END)

    return aflow

For the full example and complete code, see ibm_knowledge_to_emails.

Example 3: Branching

Python

@flow(
        name = "get_pet_facts_if_else",
        input_schema = Pet,
        output_schema = PetFacts
)
def build_get_pet_facts_if_else_flow(aflow: Flow) -> Flow:
    """ Based on the request, we will return the list of facts about the pet. A pet can be either a cat or a dog. """
    
    dog_fact_node = aflow.tool("getDogFact")
    cat_fact_node = aflow.tool("getCatFact")

    # create a branch with conditions
    check_pet_kind_conditions_branch: Branch = aflow.conditions()
    check_pet_kind_conditions_branch.condition(
        expression="flow.input.kind.strip().lower() == 'dog'", to_node=dog_fact_node
        ).condition(expression="flow.input.kind.strip().lower() == 'cat'", to_node=cat_fact_node
        ).condition(to_node=dog_fact_node, default=True)
    aflow.edge(START, check_pet_kind_conditions_branch)


    aflow.edge(dog_fact_node, END)
    aflow.edge(cat_fact_node, END)

    return aflow

For the full example and complete code, see get_pet_facts.

Example 4: Foreach

Python

@flow(
    name="send_invitation_to_customer",
    input_schema=CustomerName,
    output_schema=None
)
def build_send_invitation_to_customer_flow(aflow: Flow) -> Flow:
    """ Given a list of customers, we will iterate through the list and send email to each """

    get_customer_list_node = aflow.tool(get_emails_from_customer)

    # calling add_foreach will create a subflow, and we can add more node to the subflow
    foreach_flow: Flow = aflow.foreach(item_schema = CustomerRecord)
    node2 = foreach_flow.tool(send_invitation_email)
    foreach_flow.sequence(START, node2, END)

    # add the foreach flow to the main flow
    aflow.edge(START, get_customer_list_node)
    aflow.edge(get_customer_list_node, foreach_flow)
    aflow.edge(foreach_flow, END)

    return aflow

For the full example and complete code, see foreach_email.

Example 5: Connecting form buttons to the next node

Python

@flow(
    name="feedback_form_flow",
    input_schema=None,
)
def build_feedback_form_flow(aflow: Flow) -> Flow:
    """ Create a flow with a feedback form that connects to a confirmation message. """

    # Create a user flow
    user_flow = aflow.userflow()
    user_flow.spec.display_name = "Feedback Form"

    # Create the main form
    feedback_form = user_flow.form(name="feedback_form", display_name="Customer Feedback")

    # Add form fields
    feedback_form.text_input_field(
        name="customer_name",
        label="Name",
        required=True,
        placeholder_text="Enter your name"
    )
    feedback_form.text_input_field(
        name="feedback_text",
        label="Feedback",
        required=True,
        placeholder_text="Share your feedback"
    )

    # Create a confirmation message node
    confirmation_message = user_flow.field(
        direction="output",
        name="confirmation",
        display_name="Confirmation",
        kind=UserFieldKind.Text,
        text="Thank you for your feedback!"
    )

    # Connect the form button to the confirmation message
    user_flow.edge(START, feedback_form)
    user_flow.edge(feedback_form, confirmation_message, button_label="Submit")
    user_flow.edge(confirmation_message, END)

    # Add the user flow to the main flow
    aflow.sequence(START, user_flow, END)

    return aflow

When you use forms in user nodes, the button_label parameter in the edge() function connects a specific button on the form to the next node. This connection enables the flow to transition to the appropriate next step when users click the button.