Chapter 15 · Flow入门: Event-Driven Workflows

Chapter goals: Understand CrewAI Flow's event-driven model, master the Flow class and @start / @listen decorators, pass data between methods via return values and state, embed Crews inside Flows, and make the right Flow-vs-Crew选型 based on the scenario.

15.1 Crew's Ceiling and Flow's Positioning

Crews excel at "a group of roles collaborating to complete one goal," but they have two天然短板:

Weak control flow: sequential/hierarchical two流程难以 express "if A fails, go to B" or "run X and Y in parallel then merge";
Scattered state management: cross-Crew shared data can only pass through task outputs层层递传.

Flow is CrewAI's answer — structured, event-driven workflows: you define steps with plain Python methods, declare "who triggers whom" with decorators, and flow data through state and return values. The official three key values: simplify multi-Crew/task orchestration,内置 state management, flexible control flow (conditions, branches).

15.2 First Flow: @start and @listen

python

# first_flow.py —— minimal Flow: generate a city name → generate a fun fact
import os
from crewai import LLM
from crewai.flow.flow import Flow, listen, start

llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)

class CityFlow(Flow):

    @start()                      # mark the流程 entry
    def generate_city(self):
        print("Flow started")
        # every Flow's state auto-gains a unique ID
        print(f"State ID: {self.state['id']}")
        result = llm.call("Return any city name in the world.")
        self.state["city"] = result          # write to shared state
        return result                        # return value is also passed to listeners

    @listen(generate_city)        # listen to the previous step: executes after it completes
    def generate_fun_fact(self, random_city):
        fun_fact = llm.call(f"Tell one interesting fact about {random_city}.")
        self.state["fun_fact"] = fun_fact
        return fun_fact


flow = CityFlow()
flow.plot("city_flow_plot")       # generate visualization HTML (optional)
result = flow.kickoff()           # execute; returns the last-completed method's output
print(f"Generated fun fact: {result}")

Key points unpacked:

@start() marks the entry method; on Flow启动, all qualifying @start methods execute (often in parallel when there are multiple);
@listen(xxx) declares "execute me when xxx completes"; the return value of the listened method is passed as an argument;
kickoff() returns the last-completed method's output — above, that's generate_fun_fact's return value;
self.state is the共享 state container shared by all methods; when unstructured, it's a dict with an auto-added id key (UUID).

@listen also supports the string form: @listen("generate_city") is equivalent to passing the method object — useful for forward-reference scenarios within a class.

15.3 Structured State: TypedDict / Pydantic

Dict-style state is flexible but error-prone on typos. Add a generic参数 to Flow to upgrade to structured state — Pydantic BaseModel is recommended:

python

# state_flow.py —— Pydantic structured state
from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, start

class PoemState(BaseModel):
    sentence_count: int = 1     # note: the id field is auto-added and maintained
    poem: str = ""

class PoemFlow(Flow[PoemState]):

    @start()
    def prepare(self):
        self.state.sentence_count = 4      # attribute access + IDE autocomplete
        self.state.poem = ""

    @listen(prepare)
    def write_poem(self):
        text = llm.call(f"Write a short poem with {self.state.sentence_count} sentences.")
        self.state.poem = text
        return text

flow = PoemFlow()
final = flow.kickoff()
print(flow.state)      # view final state: sentence_count=4 poem='...'

Selection heuristic (official stance): state is simple and fluid,追求 rapid prototyping → unstructured dict; need type safety, validation, and IDE hints → Pydantic structured state. Production projects should almost always use the latter.

15.4 Calling LLM and Embedding Crews Inside Flow Methods

Flow methods freely mix three AI invocation granularities:

python

# mixed_flow.py —— all three invocation styles coexist
import os
from pydantic import BaseModel
from crewai import Agent, Task, Crew, Process, LLM
from crewai.flow.flow import Flow, listen, start

llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)

class ReportState(BaseModel):
    topic: str = ""
    outline: str = ""
    report: str = ""

class ReportFlow(Flow[ReportState]):

    @start()
    def set_topic(self):
        self.state.topic = "AI Agents in Customer Service Systems"

    @listen(set_topic)
    def make_outline(self):
        # Method 1: direct LLM call — lightweight single-step tasks
        self.state.outline = llm.call(
            f"Draft a three-section outline for '{self.state.topic}', output only the outline."
        )
        return self.state.outline

    @listen(make_outline)
    def write_report(self):
        # Method 2: single Agent kickoff directly —更省事 when no team collaboration needed
        writer = Agent(role="Writer", goal="Write per outline",
                       backstory="Technical writing expert", llm=llm)
        t = Task(description=f"Write the report per outline:\n{self.state.outline}",
                 expected_output="An ~800-word Markdown report.",
                 agent=writer)
        single = Crew(agents=[writer], tasks=[t],
                      process=Process.sequential)
        out = single.kickoff()
        self.state.report = out.raw
        return out.raw


if __name__ == "__main__":
    result = ReportFlow().kickoff()
    print(result)

The third method wraps a full multi-role Crew as a function called by Flow steps (expanded in the next chapter with a scaffold project形态). Selection criterion: single-step → LLM.call or single Agent; needs multi-role collaboration and tool division → Crew.

15.5 Async Methods and kickoff_async

Flow methods can be plain functions or async def; correspondingly use await flow.kickoff_async() to启动:

python

import asyncio
from crewai.flow.flow import Flow, listen, start

class AsyncFlow(Flow):

    @start()
    async def fetch_data(self):           # async method: suitable for IO-bound work
        await asyncio.sleep(0.5)          # simulate an async request
        self.state["data"] = "raw data"
        return "ok"

    @listen(fetch_data)
    def process(self, status):            # sync methods can also listen to async methods
        return f"{status}: {self.state['data']}"

asyncio.run(AsyncFlow().kickoff_async())

When integrating with web services (FastAPI endpoints触发 workflows), kickoff_async avoids blocking the event loop — the标准 posture for production deployment.

15.6 Flow vs Crew再辨析

Dimension	Crew	Flow
Abstraction level	Role-collaboration team	Workflow scaffolding
Control flow	sequential / hierarchical	Arbitrary Python logic + conditional routing
State	Task output chain	Explicit state (persistable)
Suitable for	"One thing handed to one team"	"One business pipeline containing multiple teams/steps"

Heuristic: start with the simplest方案 — single Q&A needs no框架 concept; multi-role one-shot collaboration → Crew; when you hit any need for conditional branching, parallel merge, breakpoint recovery, or跨-phase shared state, wrap the Crew inside a Flow.

Chapter Summary

Flow uses @start() to declare entry and @listen() to declare dependencies, forming an event-driven DAG; kickoff() returns the last-completed method's output;
Two data paths: method return value → listener argument; self.state global共享 (dict or Flow[PydanticModel] generic);
Flow methods can混合 LLM.call, single Agent, and full Crew at three granularities;
Async methods paired with kickoff_async() — required for服务化 deployment;
Selection:协作 one-shot → Crew; branching/parallel/state/recovery needs → Flow wrapping Crew.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What does flow.kickoff() return?

2. How does a @listen(generate_city) method get the previous step's data?

3. To give state field validation and IDE autocomplete, what should you do?

4. Which scenario is best served by Crew alone without introducing Flow?

🛠️ Hands-On Exercises

Convert the two-role writing crew from Chapter 5 into a Flow: @start receives the topic → embed a Crew to produce the article → @listen counts words and writes to article.md.
Rewrite the chapter's CityFlow with Pydantic structured state, explicitly declaring city and fun_fact fields in state, and print the完整 state at the end.
Write a Flow with two @start methods (e.g., fetch weather and news simultaneously), observe their execution order, store both results in state, then merge with a single @listen(and_(...)) method (previewed in the next chapter's merge技巧).

Return values and state alone aren't enough — next learn conditional routing and persistence: Chapter 16 · Flow Advanced: State Management and Persistence.

Chapter 15 · Flow入门: Event-Driven Workflows ​

15.1 Crew's Ceiling and Flow's Positioning ​

15.2 First Flow: @start and @listen ​

15.3 Structured State: TypedDict / Pydantic ​

15.4 Calling LLM and Embedding Crews Inside Flow Methods ​

15.5 Async Methods and kickoff_async ​

15.6 Flow vs Crew再辨析 ​

Chapter Summary ​

🧪 随堂测验

🛠️ Hands-On Exercises ​

Chapter 15 · Flow入门: Event-Driven Workflows

15.1 Crew's Ceiling and Flow's Positioning

15.2 First Flow: @start and @listen

15.3 Structured State: TypedDict / Pydantic

15.4 Calling LLM and Embedding Crews Inside Flow Methods

15.5 Async Methods and kickoff_async

15.6 Flow vs Crew再辨析

Chapter Summary

🛠️ Hands-On Exercises