What are Agentic Flows in CrewAI?

Wish to simplify the creation and administration of AI workflows? CrewAI flows provide structured patterns for orchestrating AI agent interactions. They permit builders to successfully mix coding duties and Crews, providing a strong framework for growing AI automation. With Agentic Flows in CrewAI, you’ll be able to design structured, event-driven workflows that streamline activity coordination, handle state, and management execution inside your AI functions.

What are Crews?

Crews from crewAI allow the orchestration of AI brokers for activity automation. It facilitates seamless collaboration between brokers to resolve 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 realize duties. These Flows include a sequence of duties the place the output of 1 activity 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 may design workflows that dynamically modify based mostly on the execution outcomes from completely different duties, enabling seamless execution of complicated AI processes. 

Workflow Management

In CrewAI Flows, builders can construction the movement of duties and management how info passes between them. Duties may be chained collectively, making a logical movement 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 are able to use Structured State Administration, which makes use of a predefined schema, usually by means of Pydantic’s BaseModel, to make sure that the info handed between duties follows a selected construction. It gives the advantages of kind security, validation, and simpler administration of complicated information 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 may be supplied firstly, throughout, or after execution. 

Occasion-Pushed Structure

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

Activity Routing

Flows additionally allow using 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 technique would possibly examine the output of a earlier activity and determine 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 films based mostly on style. This might be a easy agentic system that may 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 crucial modules

from crewai import Agent, Activity, Crew
from crewai.movement.movement 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 common film particular to the style",
    objective="Present a listing of films based mostly on consumer preferences",
    backstory="You're a cinephile, "
              "you advocate good films to your folks, "
              "the flicks must be of the identical style",
    instruments=[SerperDevTool()],
    verbose=True
)

Outline duties

action_task = Activity(
    title="ActionTask",
    description="Recommends a well-liked motion film",
    expected_output="A listing of 10 common films",
    agent=movie_agent
)
comedy_task = Activity(
    title="ComedyTask",
    description="Recommends a well-liked comedy film",
    expected_output="A listing of 10 common films",
    agent=movie_agent
)
drama_task = Activity(
    title="DramaTask",
    description="Recommends a well-liked drama film",
    expected_output="A listing of 10 common films",
    agent=movie_agent
)
sci_fi_task = Activity(
    title="SciFiTask",
    description="Recommends a sci-fi film",
    expected_output="A listing of 10 common films",
    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 kind 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 acquired: {style}")
        self.state.style = style
        return style
    @router(input_genre)
    def route_to_crew(self):
        style = self.state.style
        if style not in GENRES:
            elevate 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("Remaining film suggestions:")
        return suggestions

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

The router is getting used right here to determine which methodology 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_( ) methodology returns an output.

(Observe: use and_( ) if you need the strategy to attend till all outputs are acquired).

Plot the movement

movement = MovieRecommendationFlow()
movement.plot()

Show the movement 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. Primarily based on the “enter style,” solely one of many 4 duties is executed, and when considered one of them returns an output, the “Finalize Advice” will get triggered. 

Kickoff the movement

suggestions = await movement.kickoff_async()

I entered sci-fi as enter when requested.

After all the 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 activity orchestration. By leveraging structured flows, builders can design environment friendly, dynamic AI programs 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 may be successfully used to tailor AI functions, with duties reacting to consumer inputs and dynamically adjusting to situations.

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

Steadily Requested Query

I am a tech fanatic, graduated from Vellore Institute of Expertise. I am working as a Information Science Trainee proper now. I’m very a lot taken with Deep Studying and Generative AI.