Structured Intelligence: Introduction to Flows

While Crews are powerful for orchestrating tasks, real-world applications often require more than just a chain of AI responses. You might need to run code, wait for events, perform conditional logic, or manage state between different AI operations. This is where Flows come in.

Flows represent the modern, event-driven way to build AI applications in CrewAI. They allow you to combine structured programming logic with the flexibility of AI agents.

State Management

In a Flow, State is the memory of your application. It allows different steps of your workflow to share data. Unlike a simple function chain, a Flow maintains a state object that is accessible to all methods within that flow. This is crucial for passing the output of one agent as the input to a piece of Python logic, or vice versa.

Defining a Flow

To create a Flow, you inherit from the Flow class and define your logic using decorators. The two most critical decorators are:

@start(): Marks the method that initiates the flow. When you run .kickoff(), this method executes first.
@listen(method_name): Marks a method that waits for another method to finish. It receives the output of the method it is listening to.

Building Your First Flow

Let’s look at a simple example where we generate a city name and then generate a fun fact about it. Notice how we don’t just use Agents; we could mix in standard Python code if we wanted to.

from crewai.flow.flow import Flow, start, listen
from crewai import Agent, Task, Crew

class CityFunFactFlow(Flow):

    @start()
    def generate_city(self):
        # Logic to pick a city (could be an Agent or simple code)
        print("Step 1: Picking a city...")
        return "Tokyo"

    @listen(generate_city)
    def generate_fact(self, city_name):
        # This method waits for generate_city to finish
        print(f"Step 2: Generating fact for {city_name}...")

        # We can use a Crew inside a Flow step!
        agent = Agent(role='Traveler', goal='Share facts', backstory='...')
        task = Task(
            description=f'Tell me a fun fact about {city_name}',
            expected_output='A one sentence fact.',
            agent=agent
        )
        crew = Crew(agents=[agent], tasks=[task])
        result = crew.kickoff()
        return result.raw

# Running the flow
flow = CityFunFactFlow()
final_output = flow.kickoff()
print(f"Final Result: {final_output}")

This architecture allows you to build complex, non-linear workflows (DAGs) where multiple methods can listen to the same start event, running in parallel, and re-converging later. It bridges the gap between traditional software engineering patterns and AI orchestration.