Building agentic workflows

An agentic workflow is a collection of nodes and edges that define the logic of your workflow. To build an agentic workflow, follow these steps:

Import your agents and tools

Ensure that the agents and tools you want to connect in an agentic workflow are already imported.

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.

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

Example 1
Example 2
Example 3
Example 4

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",
    input_schema = Pet,
    output_schema = PetFacts
)
def build_get_pet_facts_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
    check_pet_type_branch: Branch = aflow.branch(evaluator="flow.input.kind.strip().lower() == 'dog'")
    aflow.edge(START, check_pet_type_branch)

    # edges are automatically added when cases are added
    check_pet_type_branch.case(True, dog_fact_node).case(False, cat_fact_node)

    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.

Import the agentic workflow

Import the agentic workflow as a tool using the orchestrate tools import command in the CLI.

BASH

orchestrate tools import -k flow -f <file-path>

Show command flags

Flag	Type	Required	Description
`--kind` / `-k`	string	Yes	The kind of tool to import, this is always `flow` for agentic workflow based tools.
`--file` / `-f`	string	Yes	The path to the file of the agentic workflow you want to import (or a URL containing the file).

Test the agentic workflow locally in the ADK

Test the agentic workflow you created using a Python script before adding it to an agent. For more information, see Testing flows.

Add the agentic workflow to one agent

After successfully testing the agentic workflow, add it to an agent’s specification and update the agent.

Release Notes

Get Started

Build

Deploy

Analyze

watsonx Orchestrate Developer Edition

watsonx Orchestrate ADK MCP Server

Reference

Legal notices