Advanced Flows: Branching and Routers

Advanced Flows: Branching and Routers

In Chapter 7, we introduced Flows as a way to chain events linearly. However, real-world problems are rarely straight lines; they are decision trees. You might need to check if a topic is sensitive before writing about it, or you might want to generate text and code simultaneously.

In this chapter, we master the @router decorator for conditional logic and explore how to run tasks in parallel within a Flow.

Routers: Conditional Branching

A Router allows your Flow to decide which path to take dynamically based on the current state. Think of it as a traffic controller standing at an intersection. Instead of @listen, you use @router on a method that returns the name of the next method to execute.

Example: The Content Gatekeeper

Imagine a flow that generates a blog post. Before writing, we want to check if the topic is tech-related. If yes, we write code examples; if no, we write an essay.

from crewai.flow.flow import Flow, start, listen, router
from pydantic import BaseModel

class ContentState(BaseModel):
    topic: str = ""
    is_tech: bool = False
    final_content: str = ""

class ContentFlow(Flow[ContentState]):

    @start()
    def analyze_topic(self):
        # Logic to determine if topic is technical
        # For this example, we simulate a check
        self.state.topic = "Python Decorators"
        self.state.is_tech = True
        print(f"Analyzing topic: {self.state.topic}")

    @router(analyze_topic)
    def route_based_on_type(self):
        if self.state.is_tech:
            return "write_tech_tutorial"
        else:
            return "write_general_essay"

    @listen("write_tech_tutorial")
    def write_tech_tutorial(self):
        self.state.final_content = "Here is a Python code example..."
        print("Writing technical content...")

    @listen("write_general_essay")
    def write_general_essay(self):
        self.state.final_content = "Once upon a time..."
        print("Writing essay...")

In the code above, route_based_on_type does not perform the task itself; it strictly directs the flow. It returns a string that matches the method name of the desired next step.

Parallel Execution and Aggregation

Sometimes efficiency is key. If you need to research three different websites, doing them one by one is slow. Flows allow you to trigger multiple methods from a single starting point.

If multiple methods listen to the same output (or start point), they run in parallel. To bring them back together, you use logic within the @listen decorator.

• @listen(or_=[method_a, method_b]): Runs as soon as either A or B finishes.
• @listen(and_=[method_a, method_b]): Waits until both A and B finish before running.

    @start()
    def kickoff(self):
        print("Starting research...")

    @listen(kickoff)
    def research_competitor_A(self):
        # simulate work
        return "Data A"

    @listen(kickoff)
    def research_competitor_B(self):
        # simulate work
        return "Data B"

    @listen(and_=[research_competitor_A, research_competitor_B])
    def compile_report(self):
        # This runs only after BOTH research tasks are done
        print("Compiling final report from A and B...")

By combining Routers and Parallel execution, you can build non-linear, sophisticated AI pipelines that mimic complex human workflows.