Esta página se muestra en inglés. Una traducción al español está en curso.
WorkflowsMay 7, 2026·3 min de lectura

CrewAI Flows — Event-Driven Multi-Agent Orchestration

CrewAI Flows is the event-driven orchestration layer on top of Crews. Decorators @start, @listen, @router build a typed state machine for multi-agent.

Listo para agents

Este activo puede ser leído e instalado directamente por agents

TokRepo expone un comando CLI universal, contrato de instalación, metadata JSON, plan según adaptador y contenido raw para que los agents evalúen compatibilidad, riesgo y próximos pasos.

Native · 98/100Política: permitir
Superficie agent
Cualquier agent MCP/CLI
Tipo
Skill
Instalación
Single
Confianza
Confianza: New
Entrada
Asset
Comando CLI universal
npx tokrepo install 19033507-8f3f-4730-a50d-f35c71ef2972
Introducción

CrewAI Flows is the event-driven orchestration layer that complements CrewAI's role-based Crews. Decorators (@start, @listen, @router) build a typed state machine — perfect for production agents where Crews alone become hard to reason about. Best for: multi-step agent pipelines, branching workflows, anything beyond "3 agents talk in a loop". Works with: CrewAI 0.70+, Python 3.10+. Setup time: 2 minutes.


Hello-flow

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

class ResearchFlow(Flow):

    @start()
    def find_topic(self):
        return "open-source vector databases 2026"

    @listen(find_topic)
    def search_web(self, topic):
        # use a Crew of search-specialist agents
        return search_crew.kickoff({"topic": topic})

    @listen(search_web)
    def summarize(self, results):
        return summarizer_crew.kickoff({"results": results})

flow = ResearchFlow()
final = flow.kickoff()

Branching with @router

class BugFlow(Flow):

    @start()
    def classify(self):
        return classifier_crew.kickoff()

    @router(classify)
    def route(self, classification):
        return "fix" if classification.is_bug else "explain"

    @listen("fix")
    def fix_bug(self):
        return fixer_crew.kickoff()

    @listen("explain")
    def explain(self):
        return explainer_crew.kickoff()

State persistence

class StatefulFlow(Flow[ResearchState]):
    initial_state = ResearchState(topic="", findings=[])

    @start()
    def init(self):
        self.state.topic = "agent frameworks"

State survives across method calls and persists to disk if you set persistence=SQLiteFlowPersistence(). Useful for long-running agents that need recovery.


FAQ

Q: Crew vs Flow — when to pick which? A: Crew is best for natural agent-to-agent collaboration (researcher + writer + editor). Flow is best for explicit sequencing, branching, and recovery — when you'd otherwise hand-roll a state machine.

Q: Is CrewAI Flows free? A: Yes — CrewAI is open-source under MIT license. The framework, Crews, and Flows are all free. CrewAI also offers an enterprise hosted platform with monitoring and UI.

Q: Can a Flow contain Crews and vice versa? A: Yes — Flow methods commonly call crew.kickoff() to delegate sub-tasks to a Crew. The reverse (Crew calling a Flow) is less common but possible by exposing the Flow as a tool.


Quick Use

  1. pip install 'crewai[tools]'>=0.70
  2. from crewai.flow.flow import Flow, start, listen
  3. Define a Flow subclass with @start and @listen methods, call flow.kickoff()

Intro

CrewAI Flows is the event-driven orchestration layer that complements CrewAI's role-based Crews. Decorators (@start, @listen, @router) build a typed state machine — perfect for production agents where Crews alone become hard to reason about. Best for: multi-step agent pipelines, branching workflows, anything beyond "3 agents talk in a loop". Works with: CrewAI 0.70+, Python 3.10+. Setup time: 2 minutes.


Hello-flow

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

class ResearchFlow(Flow):

    @start()
    def find_topic(self):
        return "open-source vector databases 2026"

    @listen(find_topic)
    def search_web(self, topic):
        # use a Crew of search-specialist agents
        return search_crew.kickoff({"topic": topic})

    @listen(search_web)
    def summarize(self, results):
        return summarizer_crew.kickoff({"results": results})

flow = ResearchFlow()
final = flow.kickoff()

Branching with @router

class BugFlow(Flow):

    @start()
    def classify(self):
        return classifier_crew.kickoff()

    @router(classify)
    def route(self, classification):
        return "fix" if classification.is_bug else "explain"

    @listen("fix")
    def fix_bug(self):
        return fixer_crew.kickoff()

    @listen("explain")
    def explain(self):
        return explainer_crew.kickoff()

State persistence

class StatefulFlow(Flow[ResearchState]):
    initial_state = ResearchState(topic="", findings=[])

    @start()
    def init(self):
        self.state.topic = "agent frameworks"

State survives across method calls and persists to disk if you set persistence=SQLiteFlowPersistence(). Useful for long-running agents that need recovery.


FAQ

Q: Crew vs Flow — when to pick which? A: Crew is best for natural agent-to-agent collaboration (researcher + writer + editor). Flow is best for explicit sequencing, branching, and recovery — when you'd otherwise hand-roll a state machine.

Q: Is CrewAI Flows free? A: Yes — CrewAI is open-source under MIT license. The framework, Crews, and Flows are all free. CrewAI also offers an enterprise hosted platform with monitoring and UI.

Q: Can a Flow contain Crews and vice versa? A: Yes — Flow methods commonly call crew.kickoff() to delegate sub-tasks to a Crew. The reverse (Crew calling a Flow) is less common but possible by exposing the Flow as a tool.


Source & Thanks

Built by crewAIInc. Licensed under MIT.

crewAIInc/crewAI — ⭐ 30,000+

🙏

Fuente y agradecimientos

Built by crewAIInc. Licensed under MIT.

crewAIInc/crewAI — ⭐ 30,000+

Discusión

Inicia sesión para unirte a la discusión.
Aún no hay comentarios. Sé el primero en compartir tus ideas.

Activos relacionados