Flows

Visão Geral

O CrewAI Flows é um recurso poderoso projetado para simplificar a criação e o gerenciamento de fluxos de trabalho de IA. Os flows permitem que desenvolvedores combinem e coordenem tarefas de codificação e crews de forma eficiente, proporcionando uma estrutura robusta para a construção de automações de IA sofisticadas.

Os flows permitem que você crie fluxos de trabalho estruturados e orientados por eventos. Eles oferecem uma forma integrada de conectar múltiplas tarefas, gerenciar estado e controlar o fluxo de execução nas suas aplicações de IA. Com flows, você pode facilmente projetar e implementar processos de múltiplas etapas que exploram todo o potencial das capacidades do CrewAI.

Criação Simplificada de Fluxos de Trabalho: Conecte facilmente múltiplas crews e tarefas para criar workflows de IA complexos.
Gerenciamento de Estado: Flows facilitam muito o gerenciamento e o compartilhamento de estados entre diferentes tarefas do seu fluxo de trabalho.
Arquitetura Orientada a Eventos: Construído sobre um modelo orientado a eventos, permitindo fluxos dinâmicos e responsivos.
Controle de Fluxo Flexível: Implemente lógica condicional, loops e ramificações dentro dos seus fluxos.

Primeiros Passos

Vamos criar um Flow simples no qual você usará a OpenAI para gerar uma cidade aleatória em uma tarefa e, em seguida, usará essa cidade para gerar uma curiosidade em outra tarefa.

from crewai.flow.flow import Flow, listen, start
from dotenv import load_dotenv
from litellm import completion

load_dotenv()

class ExampleFlow(Flow):
    model = "gpt-4o-mini"

    @start()
    def generate_city(self):
        print("Starting flow")
        # Cada estado do flow recebe automaticamente um ID único
        print(f"Flow State ID: {self.state['id']}")

        response = completion(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": "Return the name of a random city in the world.",
                },
            ],
        )

        random_city = response["choices"][0]["message"]["content"]
        # Armazena a cidade no nosso estado
        self.state["city"] = random_city
        print(f"Random City: {random_city}")

        return random_city

    @listen(generate_city)
    def generate_fun_fact(self, random_city):
        response = completion(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": f"Tell me a fun fact about {random_city}",
                },
            ],
        )

        fun_fact = response["choices"][0]["message"]["content"]
        # Armazena a curiosidade no nosso estado
        self.state["fun_fact"] = fun_fact
        return fun_fact



flow = ExampleFlow()
flow.plot()
result = flow.kickoff()

print(f"Generated fun fact: {result}")

Na ilustração acima, criamos um Flow simples que gera uma cidade aleatória usando a OpenAI e depois cria uma curiosidade sobre essa cidade. O Flow consiste em duas tarefas: generate_city e generate_fun_fact. A tarefa generate_city é o ponto de início do Flow, enquanto a tarefa generate_fun_fact fica escutando o resultado da tarefa generate_city.

Cada instância de Flow recebe automaticamente um identificador único (UUID) em seu estado, que auxilia no rastreamento e gerenciamento das execuções. O estado também pode armazenar dados adicionais (como a cidade gerada e a curiosidade) que permanecem durante toda a execução do flow.

Ao executar o Flow, ele irá:

Gerar um ID único para o estado do flow
Gerar uma cidade aleatória e armazená-la no estado
Gerar uma curiosidade sobre essa cidade e armazená-la no estado
Imprimir os resultados no console

O ID único do estado e os dados armazenados podem ser úteis para rastrear execuções do flow e manter contexto entre as tarefas.

Nota: Certifique-se de configurar seu arquivo .env para armazenar sua OPENAI_API_KEY. Essa chave é necessária para autenticar as requisições à API da OpenAI.

@start()

O decorador @start() é utilizado para marcar um método como ponto inicial de um Flow. Quando um Flow é iniciado, todos os métodos decorados com @start() são executados em paralelo. É possível ter múltiplos métodos start em um Flow, e todos eles serão executados quando o Flow iniciar.

@listen()

O decorador @listen() é utilizado para marcar um método como ouvinte da saída de outra tarefa do Flow. O método decorado com @listen() será executado quando a tarefa especificada emitir uma saída. O método pode acessar a saída da tarefa à qual está escutando como argumento.

Utilização

O decorador @listen() pode ser usado de várias formas:

Escutando um Método pelo Nome: Você pode passar o nome do método ao qual deseja escutar como string. Quando esse método concluir, o método ouvinte será chamado.
```
@listen("generate_city")
def generate_fun_fact(self, random_city):
    # Implementação
```
Escutando um Método Diretamente: Você pode passar o próprio método. Quando esse método concluir, o método ouvinte será chamado.
```
@listen(generate_city)
def generate_fun_fact(self, random_city):
    # Implementação
```

Saída de um Flow

Acessar e manipular a saída de um Flow é essencial para integrar seus workflows de IA a aplicações ou sistemas maiores. O CrewAI Flows fornece mecanismos fáceis para recuperar a saída final, acessar resultados intermediários e gerenciar o estado geral do seu Flow.

Recuperando a Saída Final

Ao executar um Flow, a saída final é determinada pelo último método concluído. O método kickoff() retorna a saída desse método final.

Veja como acessar a saída final:

Code

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

class OutputExampleFlow(Flow):
    @start()
    def first_method(self):
        return "Output from first_method"

    @listen(first_method)
    def second_method(self, first_output):
        return f"Second method received: {first_output}"


flow = OutputExampleFlow()
flow.plot("my_flow_plot")
final_output = flow.kickoff()

print("---- Final Output ----")
print(final_output)

---- Final Output ----
Second method received: Output from first_method

Flow Visual image

Neste exemplo, o second_method é o último método a ser concluído, logo sua saída será a saída final do Flow. O método kickoff() retorna essa saída, que é impressa no console. O método plot() irá gerar o arquivo HTML para visualizar o fluxo.

Acessando e Atualizando o Estado

Além de recuperar a saída final, você pode acessar e atualizar o estado dentro do seu Flow. O estado pode ser usado para armazenar e compartilhar dados entre diferentes métodos do Flow. Após a execução do Flow, você pode acessar o estado para recuperar informações adicionadas ou alteradas durante o processo.

Veja um exemplo de como atualizar e acessar o estado:

Code

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

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()
flow.plot("my_flow_plot")
final_output = flow.kickoff()
print(f"Final Output: {final_output}")
print("Final State:")
print(flow.state)

Final Output: Hello from first_method - updated by second_method
Final State:
counter=2 message='Hello from first_method - updated by second_method'

Flow Visual image

Neste exemplo, o estado é atualizado tanto por first_method quanto por second_method. Após o término da execução, é possível acessar o estado final e observar as atualizações realizadas por esses métodos.

Ao garantir que a saída do método final seja retornada e oferecer acesso ao estado, o CrewAI Flows facilita a integração dos resultados dos seus workflows de IA em aplicações maiores, além de permitir o gerenciamento e o acesso ao estado durante toda a execução do Flow.

Métricas de Uso do Flow

Após a execução de um Flow, você pode acessar a propriedade usage_metrics para visualizar o consumo agregado de tokens em todas as chamadas de LLM realizadas durante a execução — incluindo chamadas das Crews orquestradas pelo Flow, chamadas dentro de tools de Agents, e invocações diretas de LLM.call(...) feitas a partir de métodos do Flow. Esse é o equivalente, do lado do SDK, ao total exibido na interface do CrewAI Enterprise.

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

class UsageMetricsFlow(Flow):
    @start()
    def run_first_crew(self):
        self.state.first_result = FirstCrew().crew().kickoff()

    @listen(run_first_crew)
    def call_llm_directly(self):
        # Chamada direta de LLM — também contabilizada por flow.usage_metrics
        llm = LLM(model="openai/gpt-4o-mini")
        self.state.summary = llm.call("Resuma os principais pontos.")

    @listen(call_llm_directly)
    def run_second_crew(self):
        self.state.second_result = SecondCrew().crew().kickoff()

flow = UsageMetricsFlow()
flow.kickoff()

print(flow.usage_metrics)
# UsageMetrics(total_tokens=8579, prompt_tokens=6210, completion_tokens=2369,
#              cached_prompt_tokens=0, reasoning_tokens=0,
#              cache_creation_tokens=0, successful_requests=5)

Cada campo do UsageMetrics retornado representa a soma de todas as chamadas de LLM feitas em uma única invocação de flow.kickoff(). Os contadores são resetados a cada novo kickoff() (e em cada iteração de kickoff_for_each), de modo que execuções sucessivas não duplicam o total. A propriedade é segura para ser lida em qualquer momento após o kickoff(); lê-la durante a execução retorna o total parcial acumulado até aquele instante.

Gerenciamento de Estado em Flows

Gerenciar o estado de forma eficaz é fundamental para construir fluxos de trabalho de IA confiáveis e de fácil manutenção. O CrewAI Flows oferece mecanismos robustos para o gerenciamento de estado tanto não estruturado quanto estruturado, permitindo que o desenvolvedor escolha a abordagem que melhor se adapta à sua aplicação.

Gerenciamento de Estado Não Estruturado

No gerenciamento de estado não estruturado, todo o estado é armazenado no atributo state da classe Flow. Essa abordagem oferece flexibilidade, permitindo que o desenvolvedor adicione ou modifique atributos do estado conforme necessário sem precisar definir um esquema rígido. Mesmo com estados não estruturados, os flows do CrewAI geram e mantêm automaticamente um identificador único (UUID) para cada instância de estado.

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

class UnstructuredExampleFlow(Flow):

    @start()
    def first_method(self):
        # O estado inclui automaticamente um campo 'id'
        print(f"State ID: {self.state['id']}")
        self.state['counter'] = 0
        self.state['message'] = "Hello from structured flow"

    @listen(first_method)
    def second_method(self):
        self.state['counter'] += 1
        self.state['message'] += " - updated"

    @listen(second_method)
    def third_method(self):
        self.state['counter'] += 1
        self.state['message'] += " - updated again"

        print(f"State after third_method: {self.state}")


flow = UnstructuredExampleFlow()
flow.plot("my_flow_plot")
flow.kickoff()

Flow Visual image

Nota: O campo id é gerado e preservado automaticamente durante toda a execução do flow. Não é necessário gerenciá-lo ou defini-lo manualmente, e ele permanecerá mesmo ao atualizar o estado com novos dados.

Pontos-Chave:

Flexibilidade: É possível adicionar atributos dinamicamente ao self.state sem restrições pré-definidas.
Simplicidade: Ideal para fluxos de trabalho diretos em que a estrutura do estado é mínima ou varia bastante.

Gerenciamento de Estado Estruturado

No gerenciamento de estado estruturado, utilizam-se esquemas pré-definidos para garantir consistência e segurança de tipos em todo o workflow. Ao usar modelos como o BaseModel da Pydantic, os desenvolvedores podem definir a forma exata do estado, melhorando a validação e fornecendo auto-complete nos ambientes de desenvolvimento.

Cada estado nos flows do CrewAI recebe automaticamente um identificador único (UUID) para ajudar no rastreamento e gerenciamento. Esse ID é gerado e mantido automaticamente pelo sistema de flows.

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


class ExampleState(BaseModel):
    # Nota: o campo 'id' é adicionado automaticamente a todos os estados
    counter: int = 0
    message: str = ""


class StructuredExampleFlow(Flow[ExampleState]):

    @start()
    def first_method(self):
        # Acesse o ID gerado automaticamente, se necessário
        print(f"State ID: {self.state.id}")
        self.state.message = "Hello from structured flow"

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

    @listen(second_method)
    def third_method(self):
        self.state.counter += 1
        self.state.message += " - updated again"

        print(f"State after third_method: {self.state}")


flow = StructuredExampleFlow()
flow.kickoff()

Flow Visual image

Pontos-Chave:

Esquema Definido: ExampleState deixa claro a estrutura do estado, aumentando a legibilidade e a manutenção do código.
Segurança de Tipos: O uso da Pydantic garante que os atributos do estado tenham os tipos certos, reduzindo os erros em tempo de execução.
Auto-Completar: IDEs conseguem oferecer auto-completar e checagem de erros, graças ao modelo definido do estado.

Escolhendo entre Estado Não Estruturado e Estruturado

Use Estado Não Estruturado quando:
- O estado do fluxo é simples ou altamente dinâmico.
- Flexibilidade é mais importante do que uma definição rígida do estado.
- Prototipagem rápida é necessária sem a sobrecarga de definição de esquemas.
Use Estado Estruturado quando:
- O flow exige uma estrutura de estado bem definida e consistente.
- Segurança de tipos e validação são importantes para a confiabilidade da aplicação.
- É desejado usar recursos da IDE como auto-completar e checagem de tipos para uma melhor experiência de desenvolvimento.

Ao oferecer as duas opções de gerenciamento de estado, o CrewAI Flows permite que desenvolvedores criem fluxos de IA que sejam ao mesmo tempo flexíveis e robustos, atendendo a uma ampla variedade de requisitos de aplicação.

Persistência de Flow

O decorador @persist permite a persistência automática do estado nos flows do CrewAI, garantindo que você mantenha o estado do flow entre reinicializações ou execuções diferentes do workflow. Esse decorador pode ser aplicado tanto ao nível de classe, quanto ao nível de método, oferecendo flexibilidade sobre como gerenciar a persistência do estado.

Persistência no Nível de Classe

Quando aplicado no nível da classe, o decorador @persist garante a persistência automática de todos os estados dos métodos do flow:

@persist  # Usa SQLiteFlowPersistence por padrão
class MyFlow(Flow[MyState]):
    @start()
    def initialize_flow(self):
        # Este método terá seu estado persistido automaticamente
        self.state.counter = 1
        print("Initialized flow. State ID:", self.state.id)

    @listen(initialize_flow)
    def next_step(self):
        # O estado (incluindo self.state.id) é recarregado automaticamente
        self.state.counter += 1
        print("Flow state is persisted. Counter:", self.state.counter)

Persistência no Nível de Método

Para um controle mais granular, você pode aplicar @persist em métodos específicos:

class AnotherFlow(Flow[dict]):
    @persist  # Persiste apenas o estado deste método
    @start()
    def begin(self):
        if "runs" not in self.state:
            self.state["runs"] = 0
        self.state["runs"] += 1
        print("Method-level persisted runs:", self.state["runs"])

Forking de Estado Persistido

@persist suporta dois modos distintos de hidratação em kickoff / kickoff_async:

kickoff(inputs={"id": <uuid>}) — resume: carrega o snapshot mais recente do UUID informado e continua escrevendo sob o mesmo flow_uuid. O histórico se estende.
kickoff(restore_from_state_id=<uuid>) — fork: carrega o snapshot mais recente do UUID informado, hidrata o estado da nova execução a partir dele, e atribui um novo state.id (auto-gerado, ou inputs["id"] se fixado). As escritas do @persist da nova execução vão para o novo state.id; o histórico do flow de origem é preservado.

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
class CounterFlow(Flow[CounterState]):
    @start()
    def step(self):
        self.state.counter += 1
        print(f"[id={self.state.id}] counter={self.state.counter}")

# Execução 1: estado novo, counter 0 -> 1, persistido sob flow_1.state.id
flow_1 = CounterFlow()
flow_1.kickoff()

# Fork: hidrata do snapshot mais recente de flow_1, mas usa um state.id NOVO
flow_2 = CounterFlow()
flow_2.kickoff(restore_from_state_id=flow_1.state.id)
# flow_2.state.counter começa em 1 (hidratado), e step() incrementa para 2.
# flow_2.state.id != flow_1.state.id; o histórico de flow_1 não é alterado.

Se o restore_from_state_id informado não corresponder a nenhum estado persistido, o kickoff retorna silenciosamente ao comportamento padrão — o mesmo comportamento do inputs["id"] quando não encontrado. Combinar restore_from_state_id com from_checkpoint lança um ValueError; escolha uma única fonte de hidratação. Fixar inputs["id"] durante o fork compartilha uma chave de persistência com outro flow — geralmente você quer apenas restore_from_state_id.

Como Funciona

Identificação Única do Estado
- Cada estado do flow recebe automaticamente um UUID único
- O ID é preservado entre atualizações do estado e chamadas de métodos
- Suporta tanto estados estruturados (Pydantic BaseModel) quanto não estruturados (dicionário)
Backend SQLite Padrão
- O SQLiteFlowPersistence é o backend de armazenamento padrão
- Os estados são salvos automaticamente em um banco de dados SQLite local
- O tratamento de erros é robusto, oferecendo mensagens claras caso ocorram falhas nas operações de banco de dados
Tratamento de Erros
- Mensagens de erro abrangentes para operações de banco de dados
- Validação automática do estado ao salvar e carregar
- Feedback claro quando houver problemas de persistência

Considerações Importantes

Tipos de Estado: São suportados tanto estados estruturados (Pydantic BaseModel) quanto não estruturados (dicionário)
ID Automático: O campo id é adicionado automaticamente se não estiver presente
Recuperação de Estado: Flows que falharem ou forem reiniciados podem recarregar automaticamente seu estado anterior
Implementação Personalizada: Você pode fornecer sua própria implementação de FlowPersistence para necessidades de armazenamento especializadas

Vantagens Técnicas

Controle Preciso Através de Acesso de Baixo Nível
- Acesso direto às operações de persistência para casos avançados
- Controle detalhado via decoradores de persistência no nível do método
- Inspeção de estado e recursos de depuração embutidos
- Visibilidade total das mudanças e operações de persistência do estado
Maior Confiabilidade
- Recuperação automática do estado após falhas no sistema ou reinicializações
- Atualizações de estado baseadas em transações para garantir integridade dos dados
- Mensagens de erro abrangentes e claras
- Validação robusta durante operações de salvar e carregar estado
Arquitetura Extensível
- Backend de persistência personalizável através da interface FlowPersistence
- Suporte para soluções de armazenamento especializadas além do SQLite
- Compatibilidade tanto com estados estruturados (Pydantic) quanto não estruturados (dict)
- Integração perfeita com os padrões de flow existentes no CrewAI

A arquitetura de persistência enfatiza precisão técnica e opções de personalização, permitindo que desenvolvedores mantenham controle total sobre o gerenciamento de estado enquanto se beneficiam dos recursos de confiabilidade integrados.

Controle de Flow

Lógica Condicional: `or`

A função or_ nos flows permite escutar múltiplos métodos e acionar o método ouvinte quando qualquer um dos métodos especificados gerar uma saída.

Code

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

class OrExampleFlow(Flow):

    @start()
    def start_method(self):
        return "Hello from the start method"

    @listen(start_method)
    def second_method(self):
        return "Hello from the second method"

    @listen(or_(start_method, second_method))
    def logger(self, result):
        print(f"Logger: {result}")



flow = OrExampleFlow()
flow.plot("my_flow_plot")
flow.kickoff()

Logger: Hello from the start method
Logger: Hello from the second method

Flow Visual image

Ao executar esse Flow, o método logger será acionado pela saída tanto do start_method quanto do second_method. A função or_ serve para escutar vários métodos e disparar o método ouvinte quando qualquer um emitir um resultado.

Lógica Condicional: `and`

A função and_ nos flows permite escutar múltiplos métodos e acionar o método ouvinte apenas quando todos os métodos especificados emitirem uma saída.

Code

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

class AndExampleFlow(Flow):

    @start()
    def start_method(self):
        self.state["greeting"] = "Hello from the start method"

    @listen(start_method)
    def second_method(self):
        self.state["joke"] = "What do computers eat? Microchips."

    @listen(and_(start_method, second_method))
    def logger(self):
        print("---- Logger ----")
        print(self.state)

flow = AndExampleFlow()
flow.plot()
flow.kickoff()

---- Logger ----
{'greeting': 'Hello from the start method', 'joke': 'What do computers eat? Microchips.'}

Flow Visual image

Ao executar esse Flow, o método logger só será disparado quando ambos start_method e second_method emitirem uma saída. A função and_ é usada para escutar vários métodos e acionar o método ouvinte apenas quando todas as condições forem atendidas.

Router

O decorador @router() nos flows permite definir lógica de roteamento condicional baseada na saída de um método. Você pode especificar diferentes rotas conforme a saída do método, permitindo controlar o fluxo de execução de forma dinâmica.

Code

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

class ExampleState(BaseModel):
    success_flag: bool = False

class RouterFlow(Flow[ExampleState]):

    @start()
    def start_method(self):
        print("Starting the structured flow")
        random_boolean = random.choice([True, False])
        self.state.success_flag = random_boolean

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

    @listen("success")
    def third_method(self):
        print("Third method running")

    @listen("failed")
    def fourth_method(self):
        print("Fourth method running")


flow = RouterFlow()
flow.plot("my_flow_plot")
flow.kickoff()

Starting the structured flow
Third method running
Fourth method running

Flow Visual image

No exemplo, o start_method gera um valor booleano aleatório e armazena no estado. O second_method usa o decorador @router() para decidir o roteamento conforme o valor booleano. Se o valor for True, retorna "success", senão retorna "failed". Os métodos third_method e fourth_method escutam a saída do second_method e executam com base no valor retornado.

Ao executar esse Flow, a saída será diferente dependendo do valor booleano aleatório gerado pelo start_method.

Human in the Loop (feedback humano)

O decorador @human_feedback permite fluxos de trabalho human-in-the-loop, pausando a execução do flow para coletar feedback de um humano. Isso é útil para portões de aprovação, revisão de qualidade e pontos de decisão que requerem julgamento humano.

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

class ReviewFlow(Flow):
    @start()
    @human_feedback(
        message="Você aprova este conteúdo?",
        emit=["approved", "rejected", "needs_revision"],
        llm="gpt-4o-mini",
        default_outcome="needs_revision",
    )
    def generate_content(self):
        return "Conteúdo para revisão..."

    @listen("approved")
    def on_approval(self, result: HumanFeedbackResult):
        print(f"Aprovado! Feedback: {result.feedback}")

    @listen("rejected")
    def on_rejection(self, result: HumanFeedbackResult):
        print(f"Rejeitado. Motivo: {result.feedback}")

Quando emit é especificado, o feedback livre do humano é interpretado por um LLM e mapeado para um dos outcomes especificados, que então dispara o decorador @listen correspondente.

Você também pode usar @human_feedback sem roteamento para simplesmente coletar feedback:

@start()
@human_feedback(message="Algum comentário sobre esta saída?")
def my_method(self):
    return "Saída para revisão"

@listen(my_method)
def next_step(self, result: HumanFeedbackResult):
    # Acesse o feedback via result.feedback
    # Acesse a saída original via result.output
    pass

Acesse todo o feedback coletado durante um flow via self.last_human_feedback (mais recente) ou self.human_feedback_history (todo o feedback em uma lista).

Para um guia completo sobre feedback humano em flows, incluindo feedback assíncrono/não-bloqueante com providers customizados (Slack, webhooks, etc.), veja Feedback Humano em Flows.

Adicionando Agentes aos Flows

Os agentes podem ser integrados facilmente aos seus flows, oferecendo uma alternativa leve às crews completas quando você precisar executar tarefas simples e focadas. Veja um exemplo de como utilizar um agente em um flow para realizar uma pesquisa de mercado:

import asyncio
from typing import Any, Dict, List

from crewai_tools import SerperDevTool
from pydantic import BaseModel, Field

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


# Define um formato de saída estruturado
class MarketAnalysis(BaseModel):
    key_trends: List[str] = Field(description="List of identified market trends")
    market_size: str = Field(description="Estimated market size")
    competitors: List[str] = Field(description="Major competitors in the space")


# Define o estado do flow
class MarketResearchState(BaseModel):
    product: str = ""
    analysis: MarketAnalysis | None = None


# Cria uma classe de flow
class MarketResearchFlow(Flow[MarketResearchState]):
    @start()
    def initialize_research(self) -> Dict[str, Any]:
        print(f"Starting market research for {self.state.product}")
        return {"product": self.state.product}

    @listen(initialize_research)
    async def analyze_market(self) -> Dict[str, Any]:
        # Cria um agente para pesquisa de mercado
        analyst = Agent(
            role="Market Research Analyst",
            goal=f"Analyze the market for {self.state.product}",
            backstory="You are an experienced market analyst with expertise in "
            "identifying market trends and opportunities.",
            tools=[SerperDevTool()],
            verbose=True,
        )

        # Define a consulta de pesquisa
        query = f"""
        Research the market for {self.state.product}. Include:
        1. Key market trends
        2. Market size
        3. Major competitors

        Format your response according to the specified structure.
        """

        # Executa a análise com formato de saída estruturado
        result = await analyst.kickoff_async(query, response_format=MarketAnalysis)
        if result.pydantic:
            print("result", result.pydantic)
        else:
            print("result", result)

        # Retorna a análise para atualizar o estado
        return {"analysis": result.pydantic}

    @listen(analyze_market)
    def present_results(self, analysis) -> None:
        print("\nMarket Analysis Results")
        print("=====================")

        if isinstance(analysis, dict):
            # Se recebemos um dict com a chave 'analysis', extrai o objeto de análise real
            market_analysis = analysis.get("analysis")
        else:
            market_analysis = analysis

        if market_analysis and isinstance(market_analysis, MarketAnalysis):
            print("\nKey Market Trends:")
            for trend in market_analysis.key_trends:
                print(f"- {trend}")

            print(f"\nMarket Size: {market_analysis.market_size}")

            print("\nMajor Competitors:")
            for competitor in market_analysis.competitors:
                print(f"- {competitor}")
        else:
            print("No structured analysis data available.")
            print("Raw analysis:", analysis)


# Exemplo de uso
async def run_flow():
    flow = MarketResearchFlow()
    flow.plot("MarketResearchFlowPlot")
    result = await flow.kickoff_async(inputs={"product": "AI-powered chatbots"})
    return result


# Executa o flow
if __name__ == "__main__":
    asyncio.run(run_flow())

Flow Visual image

Esse exemplo demonstra diversos recursos fundamentais do uso de agentes em flows:

Saída Estruturada: O uso de modelos Pydantic para definir o formato esperado da saída (MarketAnalysis) garante segurança de tipos e dados estruturados em todo o flow.
Gerenciamento de Estado: O estado do flow (MarketResearchState) mantém o contexto entre as etapas e armazena entradas e saídas.
Integração de Ferramentas: Os agentes podem usar ferramentas (como WebsiteSearchTool) para potencializar suas habilidades.

Adicionando Crews aos Flows

Criar um flow com múltiplas crews no CrewAI é simples.

Você pode gerar um novo projeto CrewAI que já inclui toda a estrutura para criar um flow com várias crews executando o seguinte comando:

crewai create flow name_of_flow

Esse comando irá gerar um novo projeto CrewAI com a estrutura de pastas necessária. O projeto gerado inclui uma crew pré-criada chamada poem_crew, já funcional. A crew embutida inicial usa a estrutura clássica Python/YAML; novas crews independentes criadas com crewai create crew usam a estrutura JSON-first.

Estrutura de Pastas

Após rodar o comando crewai create flow name_of_flow, você verá uma estrutura parecida com:

Diretório/Arquivo	Descrição
`name_of_flow/`	Diretório raiz do flow.
├── `crews/`	Contém diretórios para crews específicas.
│ └── `poem_crew/`	Diretório da “poem_crew” com configurações e scripts.
│ ├── `config/`	Arquivos de configuração da “poem_crew”.
│ │ ├── `agents.yaml`	YAML que define os agentes da “poem_crew”.
│ │ └── `tasks.yaml`	YAML que define as tarefas da “poem_crew”.
│ ├── `poem_crew.py`	Script da funcionalidade da “poem_crew”.
├── `tools/`	Ferramentas adicionais usadas no flow.
│ └── `custom_tool.py`	Implementação de ferramenta customizada.
├── `main.py`	Script principal do flow.
├── `README.md`	Descrição do projeto e instruções.
├── `pyproject.toml`	Arquivo de configurações e dependências do projeto.
└── `.gitignore`	Arquivos e pastas a serem ignorados no controle de versão.

Construindo suas Crews

Na pasta crews, você pode definir múltiplas crews. Cada crew tem sua própria pasta, com arquivos de configuração e o arquivo de definição da crew. Por exemplo, a pasta poem_crew contém:

config/agents.yaml: Define os agentes da crew.
config/tasks.yaml: Define as tarefas da crew.
poem_crew.py: Contém a definição da crew, incluindo agentes, tarefas, etc.

Você pode copiar, colar e editar a poem_crew para criar outras crews clássicas embutidas.

Para crews embutidas JSON-first, use uma pasta com crew.jsonc e agents/*.jsonc:

crews/
└── research_crew/
    ├── agents/
    │   └── researcher.jsonc
    └── crew.jsonc

Depois carregue a crew em uma etapa do Flow:

from pathlib import Path
from crewai.project import load_crew

crew, default_inputs = load_crew(
    Path(__file__).parent / "crews" / "research_crew" / "crew.jsonc"
)
result = crew.kickoff(inputs={**default_inputs, "topic": "AI Agents"})

Conectando Crews no `main.py`

No arquivo main.py, você cria seu flow e conecta as crews. É possível definir o fluxo usando a classe Flow e os decoradores @start e @listen para definir a ordem de execução.

Veja um exemplo de como conectar a poem_crew no arquivo main.py:

#!/usr/bin/env python
from random import randint

from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, start
from .crews.poem_crew.poem_crew import PoemCrew

class PoemState(BaseModel):
    sentence_count: int = 1
    poem: str = ""

class PoemFlow(Flow[PoemState]):

    @start()
    def generate_sentence_count(self):
        print("Generating sentence count")
        self.state.sentence_count = randint(1, 5)

    @listen(generate_sentence_count)
    def generate_poem(self):
        print("Generating poem")
        result = PoemCrew().crew().kickoff(inputs={"sentence_count": self.state.sentence_count})

        print("Poem generated", result.raw)
        self.state.poem = result.raw

    @listen(generate_poem)
    def save_poem(self):
        print("Saving poem")
        with open("poem.txt", "w") as f:
            f.write(self.state.poem)

def kickoff():
    poem_flow = PoemFlow()
    poem_flow.kickoff()


def plot():
    poem_flow = PoemFlow()
    poem_flow.plot("PoemFlowPlot")

if __name__ == "__main__":
    kickoff()
    plot()

Neste exemplo, a classe PoemFlow define um fluxo que gera a quantidade de frases, usa a PoemCrew para gerar um poema e, depois, salva o poema em um arquivo. O flow inicia com o método kickoff(), e o gráfico é gerado pelo método plot().

Flow Visual image

Executando o Flow

(Opcional) Antes de rodar o flow, instale as dependências executando:

crewai install

Após instalar as dependências, ative o ambiente virtual com:

source .venv/bin/activate

Com o ambiente ativado, execute o flow usando um dos comandos:

crewai run

uv run kickoff

O flow será executado, e você verá a saída no console.

Plotando Flows

Visualizar seus fluxos de trabalho de IA proporciona insights valiosos sobre a estrutura e os caminhos de execução dos flows. O CrewAI oferece uma ferramenta de visualização poderosa que permite gerar plots interativos dos flows, facilitando o entendimento e a otimização dos workflows de IA.

O que são Plots?

No CrewAI, plots são representações gráficas dos fluxos de trabalho de IA. Eles mostram as tarefas, suas conexões e o fluxo de dados entre elas. Essa visualização ajuda a compreender a sequência de operações, identificar gargalos e garantir que a lógica do workflow está alinhada com o esperado.

Como Gerar um Plot

O CrewAI oferece duas formas práticas de gerar plots dos seus flows:

Opção 1: Usando o método `plot()`

Se estiver trabalhando diretamente com uma instância do flow, basta chamar o método plot() do objeto. Isso criará um arquivo HTML com o plot interativo do seu flow.

# Considerando que você já tem uma instância do flow
flow.plot("my_flow_plot")

Esse comando gera um arquivo chamado my_flow_plot.html no diretório atual. Abra esse arquivo em um navegador para visualizar o plot interativo.

Opção 2: Usando a Linha de Comando

Em projetos CrewAI estruturados, é possível gerar um plot pela linha de comando. Isso é útil para projetos maiores, onde você deseja visualizar toda a configuração do flow.

crewai flow plot

O comando gera um arquivo HTML com o plot do flow, semelhante ao método plot(). Basta abrir o arquivo no navegador para explorar o workflow.

Entendendo o Plot

O plot gerado mostra nós representando as tarefas do seu flow, com setas indicando o fluxo de execução. A visualização é interativa, permitindo zoom, navegação e detalhes ao passar o mouse nos nós.

Ao visualizar seus flows, você tem clareza do formato do workflow, facilitando debug, otimização e comunicação dos seus processos de IA para outras pessoas.

Conclusão

A plotagem dos flows é um recurso poderoso do CrewAI para aprimorar o design e o gerenciamento de fluxos de IA complexos. Usando o método plot() ou a linha de comando, você obtém uma visão visual dos workflows, benefício tanto para desenvolvimento quanto para apresentação.

Próximos Passos

Se você deseja explorar exemplos adicionais de flows, acompanhe alguns exemplos em nosso repositório de exemplos. Aqui estão quatro sugestões específicas de flows, cada uma demonstrando casos de uso distintos para você escolher conforme seu problema:

Email Auto Responder Flow: Este exemplo demonstra um loop infinito, onde um job de background roda continuamente automatizando respostas de email. É ideal para tarefas rotineiras sem intervenção manual. Ver Exemplo
Lead Score Flow: Destaca como adicionar feedback humano e manipular diferentes ramos condicionais usando router. Um ótimo aprendizado para workflows com decisão dinâmica e supervisão humana. Ver Exemplo
Write a Book Flow: Exemplo ideal para encadear múltiplas crews, onde a saída de uma é usada por outra. Uma crew faz um sumário do livro inteiro, outra gera capítulos… Tudo conectado para entregar um livro completo. Perfeito para processos longos e coordenados. Ver Exemplo
Meeting Assistant Flow: Demonstra como transmitir um evento para desencadear múltiplas ações posteriores. Exemplo: ao finalizar uma reunião, atualizar um Trello, enviar mensagem no Slack e salvar resultados ao mesmo tempo. Indicado para gerenciamento completo de tarefas e notificações. Ver Exemplo

Explore esses exemplos para descobrir como aproveitar CrewAI Flows em diferentes contextos – desde automação de tarefas repetitivas até o gerenciamento de processos dinâmicos com decisões e feedback humano.

Além disso, confira nosso vídeo no YouTube sobre como utilizar flows no CrewAI abaixo!

Executando Flows

Existem duas formas de executar um flow:

Usando a API do Flow

Você pode executar um flow programaticamente criando uma instância da sua classe de flow e chamando o método kickoff():

# Exemplo de execução de flow em português
flow = ExemploFlow()
resultado = flow.kickoff()

Usando a CLI

A partir da versão 0.103.0, é possível executar flows usando o comando crewai run:

crewai run

O comando detecta automaticamente se seu projeto é um flow (com base na configuração type = "flow" no pyproject.toml) e executa conforme o esperado. Esse é o método recomendado para executar flows pelo terminal.

O comando legado crewai flow kickoff está deprecated. Use crewai run para crews e flows.

Flows

Visão Geral

Primeiros Passos

@start()

@listen()

Utilização

Saída de um Flow

Recuperando a Saída Final

Acessando e Atualizando o Estado

Métricas de Uso do Flow

Gerenciamento de Estado em Flows

Gerenciamento de Estado Não Estruturado

Gerenciamento de Estado Estruturado

Escolhendo entre Estado Não Estruturado e Estruturado

Persistência de Flow

Persistência no Nível de Classe

Persistência no Nível de Método

Forking de Estado Persistido

Como Funciona

Considerações Importantes

Vantagens Técnicas

Controle de Flow

Lógica Condicional: or

Lógica Condicional: and

Router

Human in the Loop (feedback humano)

Adicionando Agentes aos Flows

Adicionando Crews aos Flows

Estrutura de Pastas

Construindo suas Crews

Conectando Crews no main.py

Executando o Flow

Plotando Flows

O que são Plots?

Como Gerar um Plot

Opção 1: Usando o método plot()

Opção 2: Usando a Linha de Comando

Entendendo o Plot

Conclusão

Próximos Passos

Executando Flows

Usando a API do Flow

Usando a CLI

Lógica Condicional: `or`

Lógica Condicional: `and`

Conectando Crews no `main.py`

Opção 1: Usando o método `plot()`