Kenhuang Academy
  • English

      • Flows

      A Flow is the production backbone: event-driven, stateful, branchable. Official core decorators are @start(), @listen(...), and @router(). Combine listeners with or_ / and_. Persist with @persist (crewai.flow.persistence). Human gates use @human_feedback (≥ 1.8.0, crewai.flow.human_feedback). Do not invent decorator names that are absent from the docs.

      Concept-page import:

      from crewai.flow.flow import Flow, listen, start, router, or_, and_

      The Quickstart also uses from crewai.flow import Flow, listen, start. Either import path is valid.

      State and entry points

      Every Flow instance gets a unique state id. Use a dict for unstructured state, or Flow[YourModel] with Pydantic.

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

class ExampleState(BaseModel):
    counter: int = 0
    message: str = ""

class StateExampleFlow(Flow[ExampleState]):
    @start()
    def first_method(self):
        self.state.message = "Hello from first_method"
        self.state.counter += 1

    @listen(first_method)
    def second_method(self):
        self.state.message += " - updated by second_method"
        self.state.counter += 1
        return self.state.message

flow = StateExampleFlow()
print(flow.kickoff())
print(flow.state)
      • @start() — entry point; multiple starts may run when their conditions match (often in parallel).
      • @listen(method) or @listen("method_name") — runs when that method finishes; may take its return value as an argument.
      • kickoff() returns the last completed method’s value. plot() / plot("name") writes an HTML diagram.

      Run a Crew inside a step

      This is the official composition: the Flow owns topic and artifacts; the Crew does autonomous research.

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

class ResearchFlow(Flow):
    @start()
    def prepare(self):
        self.state["topic"] = "AI Agents"

    @listen(prepare)
    def run_crew(self):
        agent = Agent(
            role="Researcher",
            goal="Brief the topic",
            backstory="Concise, sourced writing.",
        )
        task = Task(
            description="Write a briefing on {topic}.",
            expected_output="Markdown briefing.",
            agent=agent,
            output_file="output/report.md",
        )
        crew = Crew(agents=[agent], tasks=[task], process=Process.sequential)
        result = crew.kickoff(inputs={"topic": self.state["topic"]})
        self.state["report"] = result.raw
        return result.raw

      In CLI projects, load_crew(Path("crew.jsonc")) replaces a hand-built Crew(...). After a run, flow.usage_metrics aggregates every LLM call in that kickoff (Crews plus bare LLM.call in Flow methods).

      @router, or_, and_

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

class FlagState(BaseModel):
    success_flag: bool = False

class RouterFlow(Flow[FlagState]):
    @start()
    def start_method(self):
        self.state.success_flag = True

    @router(start_method)
    def second_method(self):
        return "success" if self.state.success_flag else "failed"

    @listen("success")
    def third_method(self):
        print("ok path")

    @listen("failed")
    def fourth_method(self):
        print("fail path")

      @listen(or_(a, b)) fires when either completes. @listen(and_(a, b)) waits for both.

      @persist

      from crewai.flow.flow import Flow, start
from crewai.flow.persistence import persist
from pydantic import BaseModel

class CounterState(BaseModel):
    id: str = ""
    counter: int = 0

@persist  # SQLiteFlowPersistence by default
class CounterFlow(Flow[CounterState]):
    @start()
    def step(self):
        self.state.counter += 1

      kickoff(inputs={"id": ...}) resumes the same UUID. kickoff(restore_from_state_id=...) forks a new state.id. Do not combine that with from_checkpoint. You can also put @persist on a single method.

      vs LangGraph

      This site’s LangGraph chapter uses an explicit StateGraph, checkpoints, and interrupts. CrewAI Flows are decorator-based event graphs that embed Crew role teams natively. Existing LangGraph apps can follow official Moving from LangGraph to CrewAI. New work whose core is role collaboration should start with Flow + Crew.

      Next steps

      评论