What are Agentic Flows in CrewAI? - Analytics Vidhya

Streamline your AI workflows with CrewAI Flows! This powerful framework provides structured patterns for orchestrating interactions between AI agents, enabling developers to seamlessly integrate coding tasks and Crews for robust AI automation. CrewAI's Agentic Flows offer event-driven workflows, simplifying task coordination, state management, and execution control within your AI applications.

Table of Contents

• What are Crews?
• Understanding Flows
  • Workflow Control and Sequencing
  • Efficient State Management
  • Flexible Input Handling
  • Event-Driven Architecture and Dynamic Adjustment
  • Task Routing and Conditional Execution
• Flows in Practice: A Movie Recommendation Example
  • Setup and Installation
  • Handling Warnings
  • Loading Environment Variables
  • Importing Necessary Modules
  • Defining the Agent
  • Defining Tasks
  • Creating Crews for Each Genre
  • Defining Genres and GenreState
  • Building the MovieRecommendationFlow
  • Visualizing the Flow
  • Initiating the Flow
• Conclusion
• Frequently Asked Questions

What are Crews?

CrewAI's Crews facilitate the orchestration of AI agents for automated task completion. They enable smooth collaboration between agents to solve complex problems. But why "Flows"? Because CrewAI Flows provide structured patterns for managing these agent interactions, defining how agents communicate and work together to achieve specific goals. Flows are essentially sequences of tasks, where the output of one task can trigger the next. The system offers flexible mechanisms for managing state and conditional execution.

Understanding Flows

Flows operate on an event-driven model, reacting to specific triggers and conditions. This allows for dynamic workflow adjustments based on task execution results, streamlining complex AI processes.

Workflow Control and Sequencing

CrewAI Flows allow developers to structure task sequences and control information flow between tasks. Tasks can be chained together, creating a logical order of operations. Conditional execution of tasks based on prior task outputs is also supported.

Efficient State Management

Structured state management, often using Pydantic's BaseModel, ensures data consistency and structure between tasks. This provides type safety, validation, and easier management of complex data states.

Flexible Input Handling

Flows accept inputs to initialize or update their state at any point during execution. Inputs can be provided at the start, during, or after execution, depending on workflow needs.

Event-Driven Architecture and Dynamic Adjustment

CrewAI Flows dynamically adjust based on task results. Tasks can listen for outputs from preceding steps, creating a reactive system where new tasks are triggered based on previous outputs. The @listen() and @router() decorators provide this flexibility, enabling conditional and dynamic task linking. The @start() decorator marks the flow's starting point.

{{TABLE_PLACEHOLDER21}}`or<td>Triggers a listener when any specified method emits an output.</td>

Decorators and Conditional Logic	Description
@listen()	Creates listener methods triggered by specific events or task outputs.
@router()	Enables conditional routing, allowing different execution paths based on prior step outputs. Useful for managing success/failure outcomes.
and_`	Triggers a listener only when all specified methods emit outputs.

Task Routing and Conditional Execution

Flows utilize routing to control execution based on conditions. The @router() decorator allows methods to select execution paths based on prior task results. For example, a method might check a previous task's output and choose a path based on whether specific conditions are met.

Flows in Practice: A Movie Recommendation Example

Let's create an agentic system using CrewAI Flows to recommend movies based on genre.

Setup and Installation

<code>!pip install crewai -U
!pip install crewai-tools</code>

Handling Warnings

<code>import warnings
warnings.filterwarnings('ignore')</code>

Loading Environment Variables (Replace placeholders with your actual keys)

<code>import os
os.environ["OPENAI_API_KEY"] = 'YOUR_OPENAI_API_KEY'
os.environ['OPENAI_MODEL_NAME'] = 'gpt-4o-mini-2024-07-18'
os.environ["SERPER_API_KEY"]='YOUR_SERPER_API_KEY'</code>

Importing Necessary Modules

<code>from crewai import Agent, Task, Crew
from crewai.flow.flow import listen, start, and_, or_, router
from crewai_tools import SerperDevTool
from crewai import Flow
from pydantic import BaseModel</code>

Defining the Agent

A single agent will be used for all tasks. This agent uses a Google search tool.

<code>movie_agent = Agent(
role="Recommend popular movie specific to the genre",
goal="Provide a list of movies based on user preferences",
backstory="You are a cinephile, "
"you recommend good movies to your friends, "
"the movies should be of the same genre",
tools=[SerperDevTool()],
verbose=True
)</code>

Defining Tasks

<code>action_task = Task(name="ActionTask", description="Recommends a popular action movie", expected_output="A list of 10 popular movies", agent=movie_agent)
comedy_task = Task(name="ComedyTask", description="Recommends a popular comedy movie", expected_output="A list of 10 popular movies", agent=movie_agent)
drama_task = Task(name="DramaTask", description="Recommends a popular drama movie", expected_output="A list of 10 popular movies", agent=movie_agent)
sci_fi_task = Task(name="SciFiTask", description="Recommends a sci-fi movie", expected_output="A list of 10 popular movies", agent=movie_agent)</code>

Creating Crews for Each Genre

<code>action_crew = Crew(agents=[movie_agent], tasks=[action_task], verbose=True)
comedy_crew = Crew(agents=[movie_agent], tasks=[comedy_task], verbose=True)
drama_crew = Crew(agents=[movie_agent], tasks=[drama_task], verbose=True)
sci_fi_crew = Crew(agents=[movie_agent], tasks=[sci_fi_task], verbose=True)</code>

Defining Genres and GenreState

<code>GENRES = ["action", "comedy", "drama", "sci-fi"]
class GenreState(BaseModel):
    genre: str = ""</code>

Building the MovieRecommendationFlow

This class inherits from the Flow class and uses state functionality.

<code>class MovieRecommendationFlow(Flow[GenreState]):
    @start()
    def input_genre(self):
        genre = input("Enter a genre: ")
        print(f"Genre input received: {genre}")
        self.state.genre = genre
        return genre

    @router(input_genre)
    def route_to_crew(self):
        genre = self.state.genre
        if genre not in GENRES:
            raise ValueError(f"Invalid genre: {genre}")
        if genre == "action":
            return "action"
        elif genre == "comedy":
            return "comedy"
        elif genre == "drama":
            return "drama"
        elif genre == "sci-fi":
            return "sci-fi"

    @listen("action")
    def action_movies(self, genre):
        recommendations = action_crew.kickoff()
        return recommendations

    @listen("comedy")
    def comedy_movies(self, genre):
        recommendations = comedy_crew.kickoff()
        return recommendations

    @listen("drama")
    def drama_movies(self, genre):
        recommendations = drama_crew.kickoff()
        return recommendations

    @listen("sci-fi")
    def sci_fi_movies(self, genre):
        recommendations = sci_fi_crew.kickoff()
        return recommendations

    @listen(or_("action_movies", "comedy_movies", "drama_movies", "sci_fi_movies"))
    def finalize_recommendation(self, recommendations):
        print("Final movie recommendations:")
        return recommendations</code>

The @listen, @router, or_, and @start decorators manage the flow's execution.

Visualizing the Flow

<code>flow = MovieRecommendationFlow()
flow.plot() #This will generate a file, you'll need to display it separately (e.g., using an image display function in your environment)</code>

Initiating the Flow

<code>recommendations = await flow.kickoff_async()</code>

Conclusion

CrewAI's event-driven workflows simplify AI task orchestration. The flexible and adaptive nature of CrewAI Flows, combined with features like @listen(), @router(), and state management, makes them powerful tools for building efficient and dynamic AI applications.

Frequently Asked Questions

Q1. How do I pass inputs to a Flow? Use flow.kickoff(inputs={"counter": 10}).

Q2. What's the difference between @start() and @listen()? @start() marks flow starting points; @listen() marks methods triggered by task completion.

Q3. How do I visualize my flow? Use flow.plot().

Q4. Can I incorporate human feedback? Yes, CrewAI Flows support human-in-the-loop feedback.

The above is the detailed content of What are Agentic Flows in CrewAI? - Analytics Vidhya. For more information, please follow other related articles on the PHP Chinese website!