Building agentic workflows

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.

Release Notes

Get Started

Build

Analyze

watsonx Orchestrate Developer Edition

watsonx Orchestrate ADK MCP Server

Reference

Legal notices