A flow is a collection of nodes and edges that define the logic of your workflow. To build a flow, follow these steps:
1
Import your agents and tools
Ensure that the agents and tools you want to connect in a flow are already imported.
2
Define a Python function with the @flow decorator
Use the @flow decorator to define your flow. In the decorator, specify the flow’s name, description, input_schema, and output_schema.
The function must take a single parameter of type Flow and return a Flow object.
Construct the flow 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 Flow expressions.
The following tabs show some code snippet examples of flows:
Example 1: Sequencing two tools
[Python]
Copy
Ask AI
@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
@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
@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
@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]
Copy
Ask AI
@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.
3
Import the flow
Import the flow as a tool using the orchestrate tools import command in the CLI.
