Flows Conversacionais

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, pontes para UI e um REPL local flow.chat() para flows conversacionais.

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

APIs de turno

Use flow.handle_turn(message, session_id=...) para cada mensagem de usuário em REST, WebSocket, testes e UIs customizadas. Use flow.chat() quando quiser um loop de chat local no terminal para um Flow conversacional.

Flow.kickoff() não aceita os argumentos nomeados user_message= ou session_id=. Para flows conversacionais, handle_turn() guarda a mensagem pendente e chama kickoff(inputs={"id": session_id}) internamente.

API	Uso
handle_turn(message, session_id=...)	Wrapper ergonômico de um turno para Flow conversacional
chat()	REPL local no terminal para Flow conversacional
kickoff(inputs={...})	Execução avançada do flow sem tratamento de turno conversacional
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 handle_turn (SSE / WebSocket)

Início rápido

from uuid import uuid4

from crewai import Flow
from crewai.flow import listen
from crewai.experimental.conversational import (
    ConversationConfig,
    ConversationState,
)


@ConversationConfig(defer_trace_finalization=True)
class SupportFlow(Flow[ConversationState]):
    conversational = True

    def route_turn(self, context):
        message = (self.state.current_user_message or "").lower()
        if "pedido" in message or "order" in message:
            return "order"
        if "tchau" in message or "goodbye" in message:
            return "goodbye"
        return "help"

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

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

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


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

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

Ciclo de vida do turno

Cada handle_turn 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_assistant_message(reply) para que o próximo turno inclua a resposta do assistente. A linha do usuário já é salva por handle_turn — 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.

Flow conversacional (experimental)

Warning

Funcionalidade experimental. A superfície do Flow conversacional (conversational = True, handle_turn, ConversationConfig, RouterConfig, ConversationState, o grafo embutido + helpers) vive em crewai.experimental e pode mudar de formato antes de graduar. Fixe a versão do CrewAI se depende de comportamento específico e acompanhe o changelog para mudanças quebradoras. Feedback / issues bem-vindos.

Habilite o grafo conversacional definindo conversational = True em uma subclasse de Flow. O Flow base passa a expor 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 isto quando quiser um chat multi-turno com router LLM e handlers por rota sem cablar o ciclo de vida na mão. Use Flow[ChatState] (o padrão de mais baixo nível acima) quando precisar de controle total.

Exemplo rápido

from crewai import LLM, Flow
from crewai.flow import listen
from crewai.experimental.conversational import (
    ConversationConfig,
    ConversationState,
    RouterConfig,
)


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(Flow[ConversationState]):
    conversational = True

    @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()

Para um chat local no terminal, use chat():

def kickoff() -> None:
    SupportFlow().chat()

chat() envolve handle_turn() em um REPL, sai com exit / quit, ignora linhas em branco por padrão e chama finalize_session_traces() quando a sessão termina.

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. Flow.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ê.

Chame handle_turn() para mensagens de chat. Chamar kickoff(inputs={"id": ...}) diretamente executa o grafo sem aplicar o wrapper de turno conversacional.

chat() para REPLs locais

flow.chat() é o wrapper de terminal pronto para uso em cima de handle_turn():

flow = SupportFlow()
flow.chat()

Ele cobre o loop local comum:

1. Solicita uma mensagem do usuário.
2. Para com exit / quit, EOFError ou KeyboardInterrupt.
3. Chama handle_turn(message, session_id=...).
4. Imprime o resultado do assistente.
5. Finaliza traces de sessão adiados em um bloco finally.

Customize o comportamento do terminal com I/O injetável:

flow.chat(
    session_id="demo-session",
    prompt="You: ",
    assistant_prefix="Assistant: ",
    exit_commands=("exit", "quit", "bye"),
)

Para apps web, workers em background, testes e transportes customizados, continue usando handle_turn() diretamente.

Comportamento customizado do router

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

class SupportFlow(Flow[ConversationState]):
    conversational = True

    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.

flow.chat(session_id=session_id)

flow.chat() chama finalize_session_traces() para você. Quando você controla o loop com handle_turn() ou kickoff(...), chame finalize_session_traces() quando a sessão terminar.

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 Flow conversacional

O Flow conversacional 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