Flows Conversacionais

Crie apps de chat multi-turno com kickoff por turno, histórico de mensagens, roteamento de intenção, tracing e pontes WebSocket.

​

Visão geral

Apps conversacionais tratam cada linha do usuário como uma nova execução do flow com o mesmo id de sessão. A CrewAI oferece helpers para histórico de mensagens, classificação opcional de intenção, tracing adiado e pontes para UI — sem uma API chat() separada em Flow.

Conceito	Implementação
Id de sessão	kickoff(session_id=...) → inputs["id"] → state.id
Linha do usuário	kickoff(user_message=...) acrescenta em state.messages antes do grafo rodar
Fim do turno	FlowFinished só para esta execução; o chat segue no próximo kickoff
Trace da sessão	ConversationalConfig(defer_trace_finalization=True) + finalize_session_traces()

​

Um ponto de entrada: kickoff

Use flow.kickoff(user_message=..., session_id=...) para cada mensagem (REST, WebSocket, CLI). Não crie um wrapper chat() customizado em Flow.

API	Uso
kickoff(user_message=..., session_id=...)	Cada mensagem do usuário
kickoff_async(...)	Mesmos parâmetros; entrada async nativa
ask()	Prompt bloqueante dentro de um passo (wizard, esclarecimento)
@human_feedback	Aprovar/rejeitar saída de um passo — não a próxima linha do chat
ChatSession.handle_turn(...)	Camada de transporte sobre kickoff (SSE / WebSocket)

​

Início rápido

from uuid import uuid4

from crewai.flow import (
    ChatState,
    ConversationalConfig,
    Flow,
    listen,
    or_,
    persist,
    router,
    start,
)
from crewai.flow.persistence import SQLiteFlowPersistence


class SupportFlow(Flow[ChatState]):
    conversational_config = ConversationalConfig(
        default_intents=["order", "help", "goodbye"],
        intent_llm="gpt-4o-mini",
        defer_trace_finalization=True,
    )

    @start()
    def bootstrap(self):
        if not self.state.session_ready:
            self.state.session_ready = True
        return "ready"

    @router(bootstrap)
    def route(self):
        # last_intent definido em prepare_conversational_turn quando default_intents está setado
        return self.state.last_intent or "help"

    @listen("order")
    def handle_order(self):
        reply = "Seu pedido está a caminho."
        self.append_message("assistant", reply)
        return reply

    @listen("help")
    def handle_help(self):
        reply = "Como posso ajudar?"
        self.append_message("assistant", reply)
        return reply

    @listen("goodbye")
    def handle_goodbye(self):
        reply = "Até logo!"
        self.append_message("assistant", reply)
        return reply

    @persist(SQLiteFlowPersistence("support.db"))
    @listen(or_(handle_order, handle_help, handle_goodbye))
    def finalize(self):
        return self.state.model_dump()


session_id = str(uuid4())
flow = SupportFlow()

flow.kickoff(user_message="Onde está meu pedido?", session_id=session_id)
flow.kickoff(user_message="E as devoluções?", session_id=session_id)
flow.finalize_session_traces()  # um link de trace para o chat inteiro

​

Ciclo de vida do turno

Cada kickoff com user_message executa este pipeline:

1. _configure_conversational_kickoff — mescla session_id / user_message em inputs, aplica ConversationalConfig, habilita tracing adiado quando configurado.
2. Restauração de estado — se inputs["id"] existe e @persist está configurado, carrega o snapshot mais recente.
3. FlowStarted — emitido apenas no primeiro turno da sessão adiada.
4. prepare_conversational_turn — acrescenta a mensagem do usuário em state.messages, define last_user_message, limpa last_intent, classifica opcionalmente quando intents / default_intents + intent_llm estão definidos.
5. Execução do grafo — @start → @router → handlers @listen.
6. Fim da execução — flow_finished por turno e finalização de trace são ignorados com adiamento; Agent.kickoff() / crews aninhados também não fecham o batch pai.

Os handlers devem chamar append_message("assistant", reply) para que o próximo turno inclua a resposta do assistente. A linha do usuário já é salva no kickoff — não acrescente de novo nos handlers.

​

ConversationalConfig (padrões em nível de classe)

Defina na subclasse de Flow como conversational_config: ClassVar[ConversationalConfig | None].

Campo	Padrão	Propósito
default_intents	None	Rótulos de outcome para classificação automática antes do kickoff
intent_llm	None	Modelo para classificação (obrigatório quando há intents)
interactive_prompt	"You: "	Prompt para kickoff(interactive=True)
interactive_timeout	None	Timeout por linha no modo interativo
exit_commands	exit, quit	Palavras que encerram o modo interativo
defer_trace_finalization	True	Manter um batch de trace aberto entre turnos

Sobrescreva por kickoff com intents= e intent_llm=.

​

ChatState (formato persistido recomendado)

from crewai.flow import ChatState


class MyChatState(ChatState):
    # Herdados: id, messages, last_user_message, last_intent, session_ready
    research_turn_count: int = 0
    custom_flag: bool = False

Campo	Função
id	UUID da sessão (igual a session_id / inputs["id"])
messages	list de {role, content} para histórico de LLM
last_user_message	Última linha do usuário neste turno
last_intent	Rótulo de rota após classificação (se usado)
session_ready	Flag de bootstrap único (permissões, caches, etc.)

ConversationalInputs é um TypedDict para kickoff(inputs={...}): id, user_message, last_intent.

​

API conversacional em Flow

​

Parâmetros de kickoff / kickoff_async

Parâmetro	Propósito
user_message	Texto deste turno (ou {"role": "user", "content": "..."})
session_id	UUID da conversa → inputs["id"] / state.id
intents	Rótulos de outcome para classify_intent antes do kickoff
intent_llm	LLM para classificação (obrigatório com intents)
interactive	Loop CLI via ask() (só demos locais)
interactive_prompt	Prompt no modo interativo
interactive_timeout	Timeout de ask() por linha
exit_commands	Palavras que encerram o modo interativo
inputs	Campos extras de estado (mesclados com chaves conversacionais)
restore_from_state_id	Hidratação fork de outro flow persistido

​

Atributos de instância

Atributo	Propósito
conversational_config	Padrões ConversationalConfig em nível de classe
defer_trace_finalization	Flag de instância; definida automaticamente a partir do config no kickoff
suppress_flow_events	Oculta painéis Rich no console; tracing ainda registra eventos
stream	Habilita streaming; use com ChatSession.handle_turn(..., stream=True)

​

Métodos e propriedades

Nome	Descrição
append_message(role, content, **extra)	Acrescenta em state.messages (roles: user, assistant, system, tool)
conversation_messages	Histórico somente leitura para chamadas LLM
classify_intent(text, outcomes, *, llm, context=None)	Mapeia texto a um outcome (mesma lógica de @human_feedback)
receive_user_message(text, *, outcomes=None, llm=None)	Acrescenta mensagem do usuário; opcionalmente define last_intent
finalize_session_traces()	Emite flow_finished adiado e finaliza o batch de trace da sessão
_should_defer_trace_finalization()	Se este flow adia finalização de trace por turno
input_history	Trilha de auditoria de prompts e respostas de ask()

​

Helpers do módulo (crewai.flow.conversation)

Importáveis para testes ou orquestração customizada:

Função	Descrição
normalize_kickoff_inputs(inputs, user_message=..., session_id=...)	Mescla kwargs conversacionais em inputs
get_conversation_messages(flow)	Lê mensagens do estado ou buffer interno
append_message(flow, role, content, **extra)	Igual ao método de instância
prepare_conversational_turn(flow, ...)	Hidratação do turno (geralmente chamado pelo kickoff)
receive_user_message(flow, text, ...)	Igual ao método de instância
set_state_field(flow, name, value)	Define campo em estado dict ou Pydantic
get_conversational_config(flow)	Lê conversational_config da classe
input_history_to_messages(entries)	Converte input_history para formato de mensagens LLM

​

Padrões de roteamento de intenção

​

A. Pré-classificar via ConversationalConfig (mais simples)

Defina default_intents e intent_llm. Cada kickoff classifica antes do @router; leia self.state.last_intent em route().

​

B. Classificar dentro do @router (prompts mais ricos)

Defina default_intents=None para o kickoff só acrescentar a mensagem. Em route(), chame classify_intent com prompt ou descrições customizadas:

@router(bootstrap)
def route(self):
    intent = self.classify_intent(
        self._routing_prompt(self.state.last_user_message),
        ("GREETING", "ORDER", "RESEARCH", "GOODBYE"),
        llm=self.conversational_config.intent_llm or "gpt-4o-mini",
    )
    self.state.last_intent = intent
    return intent

Use @listen("RESEARCH") (ou similar) para passos com Agent.kickoff() e ferramentas — não LLM.call() puro — quando precisar de pesquisa web ou uso multi-etapa de tools.

​

Quando o flow termina mas o usuário continua conversando

FlowFinished significa que esta execução do grafo terminou. A conversa segue com outro kickoff e o mesmo session_id. @persist restaura messages, flags e contexto.

Padrão de persistência: prefira @persist em um único passo terminal (por exemplo finalize) em vez de na classe Flow inteira. Persist em nível de classe salva após cada método; load_state usa a linha mais recente, que pode ser snapshot no meio da execução e perder atualizações dos handlers no mesmo turno.

Não use @human_feedback para linhas de chat de follow-up, a menos que um humano precise aprovar uma saída específica antes de exibi-la.

​

ConversationalFlow de alto nível (experimental)

crewai.experimental.ConversationalFlow é uma subclasse opinativa que cuida do encanamento por turno para você: já vem com um grafo embutido @start / @router / converse_turn / end_conversation, gerencia state.messages, dirige o LLM de roteamento e mantém o batch de trace aberto entre os turnos. Você escreve as rotas customizadas; o framework cuida do resto.

Use-a quando quiser um chat multi-turno com router LLM e handlers por rota sem cablar o ciclo de vida na mão. Desça para Flow[ChatState] (acima) quando precisar de controle total.

​

Exemplo rápido

from crewai import LLM
from crewai.experimental import ConversationConfig, ConversationalFlow, RouterConfig
from crewai.flow import listen


ROUTER_LLM = LLM(model="gpt-4o-mini")


@ConversationConfig(
    system_prompt="A multi-agent assistant for ordinary chat and tool-backed tasks.",
    llm=ROUTER_LLM,
    router=RouterConfig(),  # rotas + descrições auto-descobertas pelos handlers @listen
)
class SupportFlow(ConversationalFlow):
    @listen("INTERNET_SEARCH")
    def handle_internet_search(self) -> str:
        """Fresh web research, current news, real-time lookups."""
        ...
        self.append_assistant_message(reply)
        return reply

    @listen("CREWAI_DOCS")
    def handle_crewai_docs(self) -> str:
        """Look up the CrewAI documentation for framework/API questions."""
        ...
        self.append_assistant_message(reply)
        return reply


flow = SupportFlow()
try:
    flow.handle_turn("O que você pode fazer?")               # roteia para converse (built-in)
    flow.handle_turn("Pesquise na web por notícias de IA.")  # roteia para INTERNET_SEARCH
    flow.handle_turn("Resuma o primeiro resultado.")         # volta para converse
finally:
    flow.finalize_session_traces()

​

ConversationConfig

Decorador de classe que anexa os defaults de chat por classe.

Campo	Padrão	Propósito
system_prompt	slices.conversational_system_prompt (i18n)	System message usado pelo converse_turn embutido. Passe "" para desativar totalmente.
llm	None	LLM de conversa (usado pelo converse_turn e como fallback do router).
router	None	RouterConfig para roteamento por LLM. Sem ele, o flow sempre cai em converse.
answer_from_history_prompt	padrão do framework	System message para a rota opcional answer_from_history.
answer_from_history_llm	None	Habilita o atalho answer_from_history quando definido.
intent_llm	None	LLM para o caminho legado intents=/default_intents.
default_intents	None	Labels de outcome para pré-classificação legada.
visible_agent_outputs	None	"all" ou lista de nomes de agentes cujos append_agent_result() devem virar mensagens públicas.
defer_trace_finalization	True	Mantém um único batch de trace aberto entre chamadas de handle_turn().

​

RouterConfig e o catálogo de rotas auto-gerado

RouterConfig(
    prompt="Enquadramento de domínio opcional (política, voz, persona).",
    response_format=MyRoute,        # opcional; auto-gerado caso contrário
    llm=ROUTER_LLM,                  # usa ConversationConfig.llm como fallback
    routes=["INTERNET_SEARCH", "CREWAI_DOCS"],   # opcional; inferido dos listeners
    route_descriptions={
        "INTERNET_SEARCH": "Sobrescreve a docstring só desta rota.",
    },
    default_intent="converse",       # usado quando a chamada ao LLM falha ou não há LLM
    fallback_intent="converse",      # usado quando o LLM retorna rota inválida
    intent_field="intent",
)

O prompt do router é montado automaticamente. Para cada rota o framework escolhe a descrição nesta precedência:

1. RouterConfig.route_descriptions[label] — override explícito.
2. ConversationalFlow.builtin_route_descriptions[label] — texto canônico do framework para converse, end, answer_from_history (otimizado para o LLM de routing).
3. Primeira linha não vazia da docstring do handler @listen(label).
4. Vazio (a rota aparece no catálogo sem descrição).

Na prática, adicionar uma rota é @listen("X") + uma docstring de uma linha:

@listen("INTERNET_SEARCH")
def handle_internet_search(self) -> str:
    """Fresh web research, current news, real-time lookups."""
    ...

…e o LLM de routing vê:

Routes:
- CREWAI_DOCS: Look up the CrewAI documentation for framework/API questions.
- INTERNET_SEARCH: Fresh web research, current news, real-time lookups.
- converse: Ordinary chat, follow-ups, summaries, clarifications…
- end: User signals the conversation is finished (goodbye, exit, done).

RouterConfig.prompt é para enquadramento de domínio (persona do assistente, regras de negócio, voz). O catálogo de rotas é auto-gerado — não liste rotas em prompt; elas vão sair de sincronia assim que você adicionar um handler.

​

Rotas embutidas

Rota	Handler	Propósito
converse	converse_turn	Handler de chat padrão. Chama ConversationConfig.llm com o system prompt + histórico canônico.
end	end_conversation	Define state.ended = True e emite uma resposta de encerramento.
answer_from_history	answer_from_history_turn	Opcional. Cai aqui quando ConversationConfig.answer_from_history_llm está definido e a mensagem pode ser respondida só pelo histórico.

Você pode sobrescrever qualquer uma definindo um handler com o mesmo nome na subclasse.

​

Semântica de handle_turn()

flow.handle_turn(message) roda um turno:

1. Reseta o tracking por execução (_completed_methods, _method_outputs) para o grafo re-rodar — sem isso, chamadas repetidas de kickoff na mesma instância dariam curto-circuito no turno 2+ porque Flow.kickoff_async trata inputs={"id": ...} como restauração de checkpoint.
2. Anexa a mensagem do usuário em state.messages, define current_user_message / last_user_message. last_intent é preservado do turno anterior para que o LLM de routing possa usá-lo como sinal.
3. Roda conversation_start → route_conversation → o handler @listen escolhido.
4. O router grava sua decisão em state.last_intent (visível para o contexto de routing do próximo turno).
5. Se seu handler retornou uma string e ainda não chamou append_assistant_message, handle_turn anexa para você.

Você também pode chamar flow.kickoff(user_message=..., session_id=...) diretamente — a mesma lógica de reset/run é acionada. handle_turn é o wrapper ergonômico.

​

Comportamento customizado do router

Para rodar efeitos colaterais (setup de event bus, telemetria) em toda decisão de routing, sobrescreva route_turn:

class SupportFlow(ConversationalFlow):
    def route_turn(self, context: dict[str, Any]) -> str | None:
        self.event_bus = MyBus(self)
        return super().route_turn(context)

Para ignorar o router LLM e escolher uma rota programaticamente, retorne uma string de route_turn; retornar None cai no _route_with_config(...).

​

append_assistant_message e append_agent_result

Dentro de um handler @listen(label), escolha:

• self.append_assistant_message(text) — adiciona um turno de assistente visível ao usuário em state.messages. O converse_turn do próximo turno vai vê-lo.
• self.append_agent_result(agent_name, result, visibility="private") — registra um evento estruturado em state.events e uma thread em state.agent_threads[agent_name]. Visibilidade pública também chama append_assistant_message automaticamente. Use resultados privados para trabalho de bastidor que não deve poluir o histórico canônico.

ConversationConfig.visible_agent_outputs pode promover globalmente os resultados privados de agentes específicos para públicos ("all" ou lista de nomes).

​

Tracing entre turnos

Com defer_trace_finalization=True (padrão em ConversationalConfig):

• Um batch de trace para toda a sessão de chat.
• flow_started só no primeiro turno; flow_finished uma vez em finalize_session_traces().
• kickoff por turno não exibe “Trace batch finalized”.
• Trabalho aninhado (Agent.kickoff(), crews, tools Exa) acrescenta ao batch pai; flows internos de AgentExecutor não fecham o batch da sessão cedo.

try:
    while True:
        line = input("You: ").strip()
        if not line:
            break
        flow.kickoff(user_message=line, session_id=session_id)
finally:
    flow.finalize_session_traces()

ChatSession.close() chama finalize_session_traces() quando o adiamento está habilitado.

suppress_flow_events=True só oculta painéis do console; eventos de trace e método ainda são emitidos.

​

Ciclo de vida de trace do ConversationalFlow

A ConversationalFlow experimental usa o mesmo ciclo de vida de tracing: defer_trace_finalization é True por padrão, então cada handle_turn() mantém o trace da sessão aberto. Sempre finalize ao fim da sessão — envolva seu loop em try/finally e chame flow.finalize_session_traces() na saída. Sem isso, o batch fica aberto e a última conversa pode nunca ser exportada.

​

Streaming

Defina stream = True na classe Flow. kickoff(...) então emitirá assistant_delta (e eventos relacionados) pelo event bus padrão.

​

Imports

from crewai.flow import (
    ChatState,
    ConversationalConfig,
    ConversationalInputs,
    Flow,
    listen,
    persist,
    router,
    start,
)

​

Veja também

• Dominando o Gerenciamento de Estado em Flows — persistência, estado Pydantic, @persist
• Construa Seu Primeiro Flow — fundamentos de flow
• Demo: lib/crewai/runner_conversational_flow_simple.py — REPL mínimo com RESEARCH + agente Exa