Conversational Flows

Overview

Conversational apps treat each user line as a new flow run with the same session id. CrewAI adds helpers for message history, optional intent routing, deferred tracing, UI bridges, and a local flow.chat() REPL for conversational flows.

Turn APIs

Use flow.handle_turn(message, session_id=...) for every user message from REST, WebSocket, tests, and custom UIs. Use flow.chat() when you want a local terminal chat loop for a conversational Flow.

Flow.kickoff() does not accept user_message= or session_id= keyword arguments. For conversational flows, handle_turn() stores the pending message and calls kickoff(inputs={"id": session_id}) internally after resetting per-turn execution state.

Quick start

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 "order" in message:
            return "order"
        if "bye" in message or "goodbye" in message:
            return "goodbye"
        return "help"

    @listen("order")
    def handle_order(self):
        reply = "Your order is on the way."
        self.append_assistant_message(reply)
        return reply

    @listen("help")
    def handle_help(self):
        reply = "How can I help?"
        self.append_assistant_message(reply)
        return reply

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


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

try:
    flow.handle_turn("Where is my order?", session_id=session_id)
    flow.handle_turn("What about returns?", session_id=session_id)
finally:
    flow.finalize_session_traces()  # one trace link for the whole chat

Turn lifecycle

Each handle_turn runs this pipeline:

1. Turn setup — stores the pending user message, resolves the session id, resets per-turn execution tracking, and calls kickoff(inputs={"id": session_id}).
2. State restore — if inputs["id"] exists and @persist is configured, loads the latest snapshot.
3. FlowStarted — emitted on the first deferred session turn only.
4. Pending turn hydration — appends the user message to state.messages, sets current_user_message / last_user_message, and optionally classifies when intents / default_intents + intent_llm are set.
5. Graph execution — conversation_start → route_conversation → the selected @listen handler.
6. End of run — per-turn flow_finished and trace finalization are skipped when deferral is enabled; nested Agent.kickoff() / crews do not close the parent batch either.

Handlers should call append_assistant_message(reply) so the next turn’s conversation_messages includes assistant text. The user line is already stored by handle_turn — do not append it again in handlers.

ConversationConfig (class-level defaults)

Decorate your conversational Flow subclass with ConversationConfig.

Override pre-classification per turn with handle_turn(..., intents=..., intent_llm=...).

Lower-level ChatState helpers

ChatState, ConversationalConfig, and crewai.flow.conversation helpers are still importable for advanced orchestration, tests, or custom wrappers. They do not add user_message= or session_id= keyword arguments to Flow.kickoff().

from crewai.flow import ChatState


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

ConversationalInputs is a TypedDict for conventional kickoff(inputs={...}) keys: id, user_message, last_intent.

Flow conversational API

handle_turn parameters

kickoff parameters

Flow.kickoff() accepts inputs, input_files, from_checkpoint, and restore_from_state_id. Pass inputs={"id": session_id} when you need raw flow execution, but use handle_turn() when the call represents a chat message.

Instance attributes

Methods and properties

Module helpers (crewai.flow.conversation)

Importable for tests or custom orchestration:

Intent routing patterns

A. Pre-classify via ConversationConfig (simplest)

Set default_intents and intent_llm. Each handle_turn() runs classification before routing; read self.state.last_intent in route_turn().

B. Classify inside route_turn (richer prompts)

Set default_intents=None so handle_turn() only appends the user message. In route_turn(), call classify_intent with a custom prompt or descriptions:

def route_turn(self, context):
    intent = self.classify_intent(
        self._routing_prompt(self.state.current_user_message),
        ("GREETING", "ORDER", "RESEARCH", "GOODBYE"),
        llm="gpt-4o-mini",
    )
    self.state.last_intent = intent
    return intent

Use @listen("RESEARCH") (or similar) for steps that run Agent.kickoff() with tools — not bare LLM.call() — when you need web research or multi-step tool use.

When the flow finishes but the user keeps chatting

FlowFinished means this graph run completed. The conversation continues with another handle_turn() and the same session_id. @persist restores messages, flags, and context.

Persist pattern: prefer @persist on a single terminal step (for example finalize) rather than on the whole Flow class. Class-level persist saves after every method; load_state uses the latest row, which may be a mid-run snapshot (for example right after bootstrap) and miss handler updates from the same turn.

Do not use @human_feedback for follow-up chat lines unless a human must approve a specific step output before it is shown.

Conversational Flow (experimental)

Opt into the conversational chat graph by setting conversational = True on a Flow subclass. The base Flow then ships a built-in @start / @router / converse_turn / end_conversation graph, manages state.messages, can drive a router LLM, and keeps the trace batch open across turns. You write the custom routes; the framework owns the rest.

Use this when you want a multi-turn chat with a router and per-route handlers without wiring the lifecycle yourself. Use Flow[ChatState] (the lower-level pattern above) when you need full control.

Quick example

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: dict) -> str | None:
        message = (self.state.current_user_message or "").lower()
        if "search" in message or "news" in message:
            return "INTERNET_SEARCH"
        if "docs" in message or "crewai" in message:
            return "CREWAI_DOCS"
        return "converse"

    @listen("INTERNET_SEARCH")
    def handle_internet_search(self) -> str:
        """Fresh web research, current news, real-time lookups."""
        reply = "I would run the web research route here."
        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."""
        reply = "I would look up the CrewAI docs here."
        self.append_assistant_message(reply)
        return reply


flow = SupportFlow()
try:
    flow.handle_turn("What can you do?")              # routes to converse
    flow.handle_turn("Search the web for AI news.")   # routes to INTERNET_SEARCH
    flow.handle_turn("Check the CrewAI docs.")         # routes to CREWAI_DOCS
finally:
    flow.finalize_session_traces()

For a local terminal chat, use chat():

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

chat() wraps handle_turn() in a REPL, exits on exit / quit, skips blank lines by default, and calls finalize_session_traces() when the session ends.

ConversationConfig

Class decorator that attaches per-class chat defaults.

RouterConfig and the auto-built route catalog

from typing import Literal

from pydantic import BaseModel

from crewai import LLM
from crewai.experimental.conversational import RouterConfig


class MyRoute(BaseModel):
    intent: Literal["INTERNET_SEARCH", "CREWAI_DOCS", "converse"]


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

router_config = RouterConfig(
    prompt="Optional domain framing (policy, voice, persona).",
    response_format=MyRoute,        # optional; auto-generated otherwise
    llm=ROUTER_LLM,                  # falls back to ConversationConfig.llm
    routes=["INTERNET_SEARCH", "CREWAI_DOCS"],   # optional; inferred from listeners
    route_descriptions={
        "INTERNET_SEARCH": "Override the docstring for this one route.",
    },
    default_intent="converse",       # used when LLM call fails or no LLM available
    fallback_intent="converse",      # used when LLM returns an invalid route
    intent_field="intent",
)

The router prompt that gets sent to the LLM is built automatically. For each route the framework picks a description with this precedence:

1. RouterConfig.route_descriptions[label] — explicit override.
2. Flow.builtin_route_descriptions[label] — framework-canned text for converse, end, answer_from_history (phrased for the router LLM).
3. First non-empty line of the @listen(label) handler's docstring.
4. Empty (the route is listed without a description).

So in practice, adding a new route is @listen("X") + a one-line docstring:

from crewai.flow import listen


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

…and the router LLM sees:

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 is for domain framing (assistant persona, business rules, voice). The route catalog is auto-built — don't list routes in prompt; they'll drift the moment you add a handler.

Built-in routes

You can override any of these by defining a same-named handler in your subclass.

handle_turn() semantics

flow.handle_turn(message) runs one turn:

1. Resets per-execution tracking (_completed_methods, _method_outputs) so the graph re-runs — without this, repeated kickoff calls on the same flow instance would short-circuit on turn 2+ because Flow.kickoff_async treats inputs={"id": ...} as a checkpoint restore.
2. Appends the user message to state.messages, sets current_user_message / last_user_message. last_intent is preserved from the prior turn so the router LLM can use it as a signal.
3. Runs conversation_start → route_conversation → the chosen @listen handler.
4. The router stores its decision in state.last_intent (visible to the next turn's router context).
5. If your handler returned a string and didn't already call append_assistant_message, handle_turn appends it for you.

Call handle_turn() for chat messages. Calling kickoff(inputs={"id": ...}) directly runs the flow graph without applying the conversational turn wrapper.

chat() for local REPLs

flow.chat() is the batteries-included terminal wrapper around handle_turn():

flow = SupportFlow()
flow.chat()

It handles the common local loop:

1. Prompts for a user message.
2. Stops on exit / quit, EOFError, or KeyboardInterrupt.
3. Calls handle_turn(message, session_id=...).
4. Prints the assistant result.
5. Finalizes deferred session traces in a finally block.

Customize the terminal behavior with injectable I/O:

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

For web apps, background workers, tests, and custom transports, keep using handle_turn() directly.

Custom router behavior

To run side effects (event bus setup, telemetry) on every routing decision, override route_turn:

from typing import Any

from crewai import Flow
from crewai.experimental.conversational import ConversationState


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)

To bypass the LLM router entirely and pick a route programmatically, return a string from route_turn; returning None falls back to _route_with_config(...).

append_assistant_message and append_agent_result

Inside a @listen(label) handler, choose:

• self.append_assistant_message(text) — adds a user-visible assistant turn to state.messages. The next turn's converse_turn sees it.
• self.append_agent_result(agent_name, result, visibility="private") — records a structured event in state.events and a thread in state.agent_threads[agent_name]. Public visibility also calls append_assistant_message for you. Use private results for scratch work that shouldn't pollute the canonical history.

ConversationConfig.visible_agent_outputs can promote specific agents' private results to public globally ("all", or a list of agent names).

Tracing across turns

With defer_trace_finalization=True (default in ConversationConfig):

• One trace batch for the whole chat session.
• flow_started on the first turn only; flow_finished once in finalize_session_traces().
• Per-turn kickoff does not print “Trace batch finalized”.
• Nested work (Agent.kickoff(), crews, Exa tools) appends to the parent batch; inner AgentExecutor flows do not close the session batch early.

flow.chat(session_id=session_id)

flow.chat() calls finalize_session_traces() for you. When you own the loop with handle_turn(), call finalize_session_traces() when the session ends.

suppress_flow_events=True only hides Rich console panels; trace and method events still emit for observability.

Conversational Flow trace lifecycle

The experimental conversational Flow uses the same tracing lifecycle: defer_trace_finalization defaults to True, so each handle_turn() keeps the session trace open. Always finalize at the end of the session — wrap your REPL/loop in try/finally and call flow.finalize_session_traces() on exit. Without it, the trace batch stays open and the final conversation may never export.

Streaming

Set stream = True on the Flow class. kickoff(...) will then emit assistant_delta (and related) events through the standard event bus.

Imports

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

See also

• Mastering Flow State Management — persistence, Pydantic state, @persist
• Build Your First Flow — flow basics
• Demo: lib/crewai/runner_conversational_flow_simple.py — minimal REPL with RESEARCH + Exa agent