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, structured turn streaming, and a local flow.chat() REPL.

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.

handle_turn(), stream_turn(), and chat() raise ValueError unless conversational mode is enabled. Applying @ConversationConfig(...) enables it automatically; otherwise set conversational = True.

Quick start

from uuid import uuid4

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


@ConversationConfig(defer_trace_finalization=True)
class SupportFlow(Flow[ConversationState]):
    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

Streaming a turn

Use stream_turn() when a UI or runtime needs structured events for one chat turn. It returns a stream session with ordered frames for Flow routing, LLM chunks, tool activity, and conversation messages.

stream = flow.stream_turn("Where is my order?", session_id=session_id)

with stream:
    for frame in stream.events:
        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
            print(frame.content, end="", flush=True)

result = stream.result

For the full frame contract and channel list, see Streaming Runtime Contract.

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 — user-defined @start methods (if any) → route_conversation (the built-in start/router) → the selected @listen handler. route_conversation also calls the overridable conversation_start() helper.
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) when the visible reply is not the return value, or when you trim history. A public string return is also recorded as assistant and included in the @persist snapshot, so a fresh Flow instance restores it. The user line is already stored by handle_turn — do not append it again in handlers.

Configuration overview

Decorating a Flow subclass with ConversationConfig both attaches the chat defaults and enables conversational mode. See the full field reference below. Override pre-classification per turn with handle_turn(..., intents=..., intent_llm=...).

Lower-level ChatState helpers

ChatState, the legacy ConversationalConfig, and crewai.flow.conversation helpers are still importable for advanced orchestration, tests, or custom wrappers. They are separate from the ConversationState / ConversationConfig API and 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.

ConversationState stores messages as ConversationMessage objects and additionally provides current_user_message, ended, events, and agent_threads. Use conversation_messages when passing its canonical history to an LLM.

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 from crewai.flow.conversation for tests or custom orchestration. These helpers use the legacy ConversationalConfig shape; prepare_conversational_turn() also clears last_intent, unlike handle_turn(), which preserves it as router context.

Intent routing patterns

A. Pre-classify via ConversationConfig (simplest)

Set default_intents and intent_llm. Each handle_turn() pre-classifies the current message. A non-empty result returned by a custom route_turn() takes precedence; otherwise route_conversation uses the current turn's classified intent.

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

Each handle_turn() completes one graph run, and the conversation continues with another handle_turn() using the same session_id. With the default deferred trace lifecycle, that run emits conversation_turn_completed, while FlowFinished is emitted once when finalize_session_traces() closes the session. @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

Opt into the conversational chat graph by setting conversational = True on a Flow subclass or applying @ConversationConfig(...). The base Flow then supplies route_conversation as the built-in start/router plus the converse_turn and end_conversation listeners. The deprecated answer_from_history_turn listener remains available for compatibility. The framework 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.flow import (
    ConversationConfig,
    ConversationState,
)


@ConversationConfig(defer_trace_finalization=True)
class SupportFlow(Flow[ConversationState]):
    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.

With no custom routes, turns fall through to converse. With custom routes and a conversation/router LLM, the framework synthesizes a default RouterConfig; provide one explicitly only to customize its prompt, route list, descriptions, or fallback behavior. Setting default_intents uses the legacy pre-classification path instead.

If no conversation LLM is configured, the built-in converse_turn returns a configuration placeholder rather than generating an answer.

RouterConfig and the auto-built route catalog

from typing import Literal

from pydantic import BaseModel

from crewai import LLM
from crewai.flow 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, and the deprecated answer_from_history compatibility route (phrased for the router LLM).
3. The method's declared description (used by declarative flows and DSL projections).
4. First non-empty line of the @listen(label) handler's docstring.
5. 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."""
    ...

Naming handlers

The string in @listen("…") is a router route label (an event name), not the Python method name. Route labels and method completion events share one trigger namespace, so naming a handler the same as its route causes the handler to re-trigger itself in a loop.

Use a different method name — the docs examples use a handle_* prefix:

@listen("create_video")
def handle_create_video(self) -> str:
    """User wants a new video."""
    ...

Do not mirror the route label on the method:

@listen("create_video")
def create_video(self) -> str:  # rejected at flow instantiation
    ...

…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 user-defined @start methods (if any), then route_conversation as the built-in start/router, then the chosen @listen handler. route_conversation invokes the overridable conversation_start() helper.
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 and persists the updated state.messages so @persist restore includes the assistant turn.

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.

chat(defer_trace_finalization=True) temporarily enables the instance deferral flag for the REPL and restores its prior value on exit.

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.flow 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 non-empty string from route_turn. A falsy return does not invoke _route_with_config() from your override; routing falls through to this turn's pre-classified intent, then the deprecated answer_from_history compatibility path when configured, and finally converse. A previous turn's last_intent is available in router context but is never replayed as a fallback.

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).

Declaring a conversational flow in JSON/YAML

A declarative Flow can be conversational too. Add a top-level conversational block and declare your own routes as methods that listen to a route label:

schema: crewai.flow/v1
name: SupportFlow

conversational:
  system_prompt: You are a terse support assistant.
  llm: gpt-4o-mini
  router:
    llm: gpt-4o-mini

methods:
  handle_order:
    description: Order status, shipping and delivery questions.
    listen: order
    do:
      call: agent
      with:
        role: Support specialist
        goal: Answer order questions accurately
        backstory: Knows the fulfilment pipeline.
        input: "${state.current_user_message}"

Declaring the block is the opt-in — enabled defaults to true. Set enabled: false to keep the configuration while turning chat off. This also disables built-in method synthesis, so the declaration must provide a normal non-conversational graph.

Three things are supplied for you:

Declarative llm, router.llm, and intent_llm fields accept either a model id or a configuration mapping such as {model: openai/gpt-4o-mini, max_tokens: 512}. The conversational block also supports default_intents, visible_agent_outputs, defer_trace_finalization, and the RouterConfig fields shown above. Deprecated answer_from_history_prompt / answer_from_history_llm declarations remain accepted for compatibility.

Run it from Python with the same turn APIs as a class-based conversational Flow:

from crewai.flow import Flow

flow = Flow.from_declaration(path="flow.yaml")

try:
    flow.handle_turn("Where is my order?", session_id="session-1")
finally:
    flow.finalize_session_traces()

Naming routes

Route labels and method names share one trigger namespace, so a handler must not be named after the route it listens to — create_video listening to create_video is rejected when the flow is built. Use a handle_* prefix.

What a declaration cannot express

crewai run opens the chat TUI for a declarative conversational flow — the same one a Python conversational Flow gets. A chat loop needs a terminal, so a headless run exits non-zero with guidance instead of running a single turn; drive it from Python there with handle_turn() or stream_turn(). A declarative method with a human_feedback: block (Python: @human_feedback) runs on a terminal REPL, because the runtime collects feedback with a blocking prompt the TUI cannot service. --inputs is not accepted for a conversational flow — each turn's input is the message you type — and resuming a session by id is not wired into the CLI yet; use flow.handle_turn(message, session_id=...) from Python for that.

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 hides Rich console panels and suppresses method execution events. Flow start/finish events still emit, so the outer Flow lifecycle remains traceable, but individual method spans are omitted.

Conversational Flow trace lifecycle

The conversational Flow uses the same tracing lifecycle: defer_trace_finalization defaults to True, so each handle_turn() keeps the session trace open. Deferred turns also suppress per-turn flow_failed; on a turn error or session abort, finalize the session explicitly. This closes the batch with the session-level FlowFinished event rather than a per-turn FlowFailed event. Always 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

For conversational UIs, use stream_turn() and iterate its ordered StreamFrame objects:

stream = flow.stream_turn("Where is my order?", session_id=session_id)

with stream:
    for frame in stream.events:
        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
            print(frame.content, end="", flush=True)

reply = stream.result

For a non-conversational Flow, setting stream = True makes kickoff() return a StreamSession. Do not set flow.stream = True when using handle_turn(); stream_turn() owns the conversational streaming lifecycle.

Imports

from crewai.flow import (
    ChatState,
    ConversationalConfig,
    ConversationalInputs,
    Flow,
    listen,
    persist,
    router,
    start,
)
from crewai.flow.conversation import prepare_conversational_turn
from crewai.flow import (
    ConversationConfig,
    ConversationState,
    RouterConfig,
)

See also

• Mastering Flow State Management — persistence, Pydantic state, @persist
• Build Your First Flow — flow basics