What are Agentic Flows in CrewAI?

Need to simplify the creation and administration of AI workflows? CrewAI flows provide structured patterns for orchestrating AI agent interactions. They allow builders to successfully mix coding duties and Crews, providing a robust framework for creating AI automation. With Agentic Flows in CrewAI, you may design structured, event-driven workflows that streamline job coordination, handle state, and management execution inside your AI functions.

What are Crews?

Crews from crewAI allow the orchestration of AI brokers for job automation. It facilitates seamless collaboration between brokers to unravel complicated issues. So why “flows” you ask? CrewAI flows are structured patterns for orchestrating AI agent interactions. They outline how brokers collaborate and talk to attain duties. These Flows encompass a sequence of duties the place the output of 1 job can set off the following, and the system gives versatile mechanisms for state administration and conditional execution management. 

What are Flows?

Flows are event-driven, that means that they react to particular triggers or situations. You possibly can design workflows that dynamically modify based mostly on the execution outcomes from totally different duties, enabling seamless execution of complicated AI processes. 

Workflow Management

In CrewAI Flows, builders can construction the circulate of duties and management how info passes between them. Duties could be chained collectively, making a logical circulate of operations. This construction allows builders to outline a sequence of operations the place sure duties are executed conditionally based mostly on the output of earlier duties.

State Administration

We will use Structured State Administration, which makes use of a predefined schema, sometimes via Pydantic’s BaseModel, to make sure that the information handed between duties follows a particular construction. It gives the advantages of sort security, validation, and simpler administration of complicated knowledge states.

Enter Flexibility

Flows can settle for inputs to initialize or replace their state at any level within the execution. Relying on the workflow’s necessities, these inputs could be supplied at the start, throughout, or after execution. 

Occasion-Pushed Structure

CrewAI Flows enable workflows to regulate based mostly on the outcomes from totally different duties dynamically. Duties can pay attention for outputs from earlier steps, enabling a reactive system the place new duties could be triggered based mostly on the outputs of earlier ones. The decorators @pay attention() and @router() allow this degree of flexibility, permitting builders to hyperlink duties conditionally and dynamically based mostly on their outcomes. Keep in mind that the @begin() decorator is used to mark the start line of a Stream.

Decorators and Conditional Logic	Description
@pay attention()	Creates listener strategies triggered by particular occasions or job outputs.
@router()	Permits conditional routing inside the circulate, permitting totally different execution paths based mostly on the outputs of prior steps. Helpful for managing success or failure outcomes.
or_	Triggers a listener when any of the required strategies emit an output. Perfect for continuing after any job completion.
and_	Ensures a listener is triggered solely when all specified strategies emit outputs. Helpful for ready till a number of situations or duties are accomplished earlier than continuing.

Process Routing

Flows additionally allow the usage of routing to manage how execution proceeds based mostly on particular situations. The @router() decorator facilitates this by permitting strategies to decide on their execution path based mostly on the outcomes of prior duties. For instance, a way would possibly examine the output of a earlier job and resolve whether or not to proceed alongside one route or one other, relying on whether or not sure situations are met.

Additionally learn: Constructing Collaborative AI Brokers With CrewAI

Flows in Motion

Let’s construct an agentic system utilizing flows from CrewAI that recommends motion pictures based mostly on style. This shall be a easy agentic system that can assist us perceive what’s working behind it. 

Installations

!pip set up crewai -U
!pip set up crewai-tools

Warning management

import warnings
warnings.filterwarnings('ignore')

Load surroundings variables

Go to Serper and get your Serper api-key for Google search. We’ll be utilizing the 4o-mini mannequin from OpenAI.

import os
os.environ["OPENAI_API_KEY"] = ‘’
os.environ['OPENAI_MODEL_NAME'] = 'gpt-4o-mini-2024-07-18'
os.environ["SERPER_API_KEY"]=''

Import essential modules

from crewai import Agent, Process, Crew
from crewai.circulate.circulate import pay attention, begin, and_, or_, router
from crewai_tools import SerperDevTool
from crewai import Stream
from pydantic import BaseModel

Outline agent

To maintain issues easy, we’ll outline a single agent that each one the duties will use. This agent can have a google search software that each one the duties can use. 

movie_agent = Agent(
    position="Suggest widespread film particular to the style",
    purpose="Present a listing of films based mostly on person preferences",
    backstory="You're a cinephile, "
              "you suggest good motion pictures to your pals, "
              "the films needs to be of the identical style",
    instruments=[SerperDevTool()],
    verbose=True
)

Outline duties

action_task = Process(
    identify="ActionTask",
    description="Recommends a preferred motion film",
    expected_output="An inventory of 10 widespread motion pictures",
    agent=movie_agent
)
comedy_task = Process(
    identify="ComedyTask",
    description="Recommends a preferred comedy film",
    expected_output="An inventory of 10 widespread motion pictures",
    agent=movie_agent
)
drama_task = Process(
    identify="DramaTask",
    description="Recommends a preferred drama film",
    expected_output="An inventory of 10 widespread motion pictures",
    agent=movie_agent
)
sci_fi_task = Process(
    identify="SciFiTask",
    description="Recommends a sci-fi film",
    expected_output="An inventory of 10 widespread motion pictures",
    agent=movie_agent
)

Outline crews for every style

action_crew = Crew(
    brokers=[movie_agent],
    duties=[action_task],
    verbose=True,
)
comedy_crew = Crew(
    brokers=[movie_agent],
    duties=[comedy_task],
    verbose=True
)
drama_crew = Crew(
    brokers=[movie_agent],
    duties=[drama_task],
    verbose=True
)
sci_fi_crew = Crew(
    brokers=[movie_agent],
    duties=[sci_fi_task],
    verbose=True
)

Outline genres and GenreState

GENRES = ["action", "comedy", "drama", "sci-fi"]
class GenreState(BaseModel):
    style: str = ""

Outline MovieRecommendationFlow

We outline a category inheriting from the Stream class and we will optionally use state performance so as to add or modify state attributes, right here we now have already outlined GenreState with style attribute of sort string. (Discover that our pydantic mannequin “GenreState” is handed in sq. brackets)

class MovieRecommendationFlow(Stream[GenreState]):
    @begin()
    def input_genre(self):
        style = enter("Enter a style: ")
        print(f"Style enter obtained: {style}")
        self.state.style = style
        return style
    @router(input_genre)
    def route_to_crew(self):
        style = self.state.style
        if style not in GENRES:
            increase ValueError(f"Invalid style: {style}")
        if style == "motion":
            return "motion"
        elif style == "comedy":
            return "comedy"
        elif style == "drama":
            return "drama"
        elif style == "sci-fi":
            return "sci-fi"
    @pay attention("motion")
    def action_movies(self, style):
        suggestions = action_crew.kickoff()
        return suggestions
    @pay attention("comedy")
    def comedy_movies(self, style):
        suggestions = comedy_crew.kickoff()
        return suggestions
    @pay attention("drama")
    def drama_movies(self, style):
        suggestions = drama_crew.kickoff()
        return suggestions
    @pay attention("sci-fi")
    def sci_fi_movies(self, style):
        suggestions = sci_fi_crew.kickoff()
        return suggestions
    @pay attention(or_("action_movies", "comedy_movies", "drama_movies", "sci_fi_movies"))
    def finalize_recommendation(self, suggestions):
        print("Closing film suggestions:")
        return suggestions

We should always specify the circulate of execution by @pay attention decorator by passing the earlier technique and the output of the earlier technique could be handed as an argument within the subsequent technique. (Word: The circulate ought to begin with the @begin decorator)

The router is getting used right here to resolve which technique ought to execute based mostly on the output of the strategy contained in the @router decorator. 

The  or_ specifies that the strategy ought to execute as quickly as any of the strategies within the or_( ) technique returns an output.

(Word: use and_( ) in order for you the strategy to attend till all outputs are obtained).

Plot the circulate

circulate = MovieRecommendationFlow()
circulate.plot()

Show the circulate diagram

from IPython.core.show import show, HTML
with open('/content material/crewai_flow.html', 'r') as f:
    html_content = f.learn()
show(HTML(html_content))

This diagram ought to provide you with a greater understanding of what’s occurring. Based mostly on the “enter style,” solely one of many 4 duties is executed, and when one in every of them returns an output, the “Finalize Advice” will get triggered. 

Kickoff the circulate

suggestions = await circulate.kickoff_async()

I entered sci-fi as enter when requested.

After the complete execution of the Stream, that is the output I received for the sci-fi film suggestions I wished. 

Additionally learn: CrewAI Multi-Agent System for Writing Article from YouTube Movies

Conclusion

On this article, we’ve explored how CrewAI’s event-driven workflows simplify the creation and administration of AI job orchestration. By leveraging structured flows, builders can design environment friendly, dynamic AI methods that allow seamless coordination between duties and brokers. With highly effective instruments just like the @pay attention(), @router(), and state administration mechanisms, CrewAI permits for versatile and adaptive workflows. The hands-on instance of a film advice system highlights how Agentic Flows in CrewAI could be successfully used to tailor AI functions, with duties reacting to person inputs and dynamically adjusting to situations.

Additionally, if you’re searching for an AI Agent course on-line then, discover: Agentic AI Pioneer Program.

Incessantly Requested Query