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.

Concept	Implementation
Session id	`handle_turn(..., session_id=...)` → `kickoff(inputs={"id": ...})` → `state.id`
User line	`handle_turn(message)` appends to `state.messages` before the graph runs
Turn complete	`FlowFinished` for this run only; chat continues on the next `handle_turn`
Full-session trace	`ConversationConfig(defer_trace_finalization=True)` + `finalize_session_traces()`

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.

API	Use for
`handle_turn(message, session_id=...)`	Ergonomic one-turn wrapper for conversational `Flow`
`stream_turn(message, session_id=...)`	Stream one conversational turn as ordered runtime frames
`chat()`	Local terminal REPL for conversational `Flow`
`kickoff(inputs={...})`	Advanced flow execution without conversational turn handling
`ask()`	Blocking prompt inside one step (wizard, clarification)
`@human_feedback`	Approve/reject a step output — not the next chat line
`ChatSession.handle_turn(...)`	Transport layer over `handle_turn` (SSE / WebSocket)

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

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.data.get("chunk", ""), end="", flush=True)

result = stream.result

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

Turn lifecycle

Each handle_turn runs this pipeline:

Turn setup — stores the pending user message, resolves the session id, resets per-turn execution tracking, and calls kickoff(inputs={"id": session_id}).
State restore — if inputs["id"] exists and @persist is configured, loads the latest snapshot.
FlowStarted — emitted on the first deferred session turn only.
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.
Graph execution — conversation_start → route_conversation → the selected @listen handler.
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.

Field	Default	Purpose
`system_prompt`	Framework default	System message used by the built-in `converse_turn`.
`llm`	`None`	Conversation LLM used by `converse_turn` and as router fallback.
`router`	`None`	`RouterConfig` for LLM-driven routing.
`intent_llm`	`None`	LLM for `intents=` / `default_intents` pre-classification.
`default_intents`	`None`	Outcome labels for pre-classification.
`defer_trace_finalization`	`True`	Keep one trace batch open across `handle_turn()` calls.

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

Field	Role
`id`	Session UUID (same as `inputs["id"]`)
`messages`	`list` of `{role, content}` for LLM history
`last_user_message`	Latest user line for this turn
`last_intent`	Route label after classification (if used)
`session_ready`	One-time bootstrap flag (permissions, caches, etc.)

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

`Flow` conversational API

`handle_turn` parameters

Parameter	Purpose
`message`	This turn’s text
`session_id`	Conversation UUID → `inputs["id"]` / `state.id`
`intents`	Outcome labels for pre-kickoff `classify_intent`
`intent_llm`	LLM for classification (required with `intents`)
`**kickoff_kwargs`	Forwarded to `kickoff()` for options like `input_files`, `from_checkpoint`, and `restore_from_state_id`

`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

Attribute	Purpose
`conversational`	Set to `True` to enable the conversational graph and `handle_turn()`
`defer_trace_finalization`	Instance flag; set automatically from config on `handle_turn()`
`suppress_flow_events`	Hides console flow panels; tracing still records method/flow events
`stream`	Enable streaming; use with `ChatSession.handle_turn(..., stream=True)`

Methods and properties

Name	Description
`append_assistant_message(content)`	Append a user-visible assistant reply to `state.messages`
`append_message(role, content, **extra)`	Lower-level append to `state.messages`
`conversation_messages`	Read-only history for LLM calls
`classify_intent(text, outcomes, *, llm, context=None)`	Map text to one outcome (same collapse logic as `@human_feedback`)
`receive_user_message(text, *, outcomes=None, llm=None)`	Append user message; optionally set `last_intent`
`finalize_session_traces()`	Emit deferred `flow_finished` and finalize the session trace batch
`_should_defer_trace_finalization()`	Whether this flow defers per-turn trace finalization
`input_history`	Audit trail of `ask()` prompts and responses

Module helpers (`crewai.flow.conversation`)

Importable for tests or custom orchestration:

Function	Description
`normalize_kickoff_inputs(inputs, user_message=..., session_id=...)`	Merge conversational kwargs into `inputs`
`get_conversation_messages(flow)`	Read messages from state or internal buffer
`append_message(flow, role, content, **extra)`	Same as instance method
`prepare_conversational_turn(flow, user_message=..., intents=..., intent_llm=..., config=...)`	Lower-level turn hydration for custom wrappers
`receive_user_message(flow, text, ...)`	Same as instance method
`set_state_field(flow, name, value)`	Set a field on dict or Pydantic state
`get_conversational_config(flow)`	Read class `conversational_config`
`input_history_to_messages(entries)`	Convert `input_history` to LLM message format

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)

This is an experimental feature. The conversational Flow surface (conversational = True, handle_turn, ConversationConfig, RouterConfig, ConversationState, the built-in graph + helpers) lives under crewai.experimental and may change shape before it graduates. Pin your CrewAI version if you depend on specific behavior, and watch the changelog for breaking updates. Open issues / feedback welcome.

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.

Field	Default	Purpose
`system_prompt`	`slices.conversational_system_prompt` from i18n	System message used by the built-in `converse_turn`. Pass `""` to opt out entirely.
`llm`	`None`	Conversation LLM (used by `converse_turn` and as router fallback).
`router`	`None`	`RouterConfig` for LLM-driven routing. Without it, the flow always falls through to `converse`.
`answer_from_history_prompt`	Framework default	System message for the optional `answer_from_history` route.
`answer_from_history_llm`	`None`	Enables the `answer_from_history` short-circuit when set.
`intent_llm`	`None`	LLM for legacy `intents=`/`default_intents` pre-classification.
`default_intents`	`None`	Outcome labels for legacy pre-classification.
`visible_agent_outputs`	`None`	`"all"`, or a list of agent names whose `append_agent_result()` calls should be promoted to public assistant messages.
`defer_trace_finalization`	`True`	Keep one trace batch open across `handle_turn()` calls.

`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:

RouterConfig.route_descriptions[label] — explicit override.
Flow.builtin_route_descriptions[label] — framework-canned text for converse, end, answer_from_history (phrased for the router LLM).
First non-empty line of the @listen(label) handler’s docstring.
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

Route	Handler	Purpose
`converse`	`converse_turn`	Default chat handler. Calls `ConversationConfig.llm` with the system prompt + canonical message history.
`end`	`end_conversation`	Sets `state.ended = True` and emits a terminator reply.
`answer_from_history`	`answer_from_history_turn`	Optional. Routes here when `ConversationConfig.answer_from_history_llm` is set and the message can be answered from existing history.

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:

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.
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.
Runs conversation_start → route_conversation → the chosen @listen handler.
The router stores its decision in state.last_intent (visible to the next turn’s router context).
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:

Prompts for a user message.
Stops on exit / quit, EOFError, or KeyboardInterrupt.
Calls handle_turn(message, session_id=...).
Prints the assistant result.
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,
)

Get Started

Guides

Core Concepts

MCP Integration

Tools

Observability

Learn

Telemetry

Conversational Flows

Overview

Turn APIs

Quick start

Streaming a turn

Turn lifecycle

`ConversationConfig` (class-level defaults)

Lower-level `ChatState` helpers

`Flow` conversational API

`handle_turn` parameters

`kickoff` parameters

Instance attributes

Methods and properties

Module helpers (`crewai.flow.conversation`)

Intent routing patterns

A. Pre-classify via `ConversationConfig` (simplest)

B. Classify inside `route_turn` (richer prompts)

When the flow finishes but the user keeps chatting

Conversational `Flow` (experimental)

Quick example

`ConversationConfig`

`RouterConfig` and the auto-built route catalog

Built-in routes

`handle_turn()` semantics

`chat()` for local REPLs

Custom router behavior

`append_assistant_message` and `append_agent_result`

Tracing across turns

Conversational `Flow` trace lifecycle

Streaming

Imports

See also

​Overview

​Turn APIs

​Quick start

​Streaming a turn

​Turn lifecycle

​ConversationConfig (class-level defaults)

​Lower-level ChatState helpers

​Flow conversational API

​handle_turn parameters

​kickoff parameters

​Instance attributes

​Methods and properties

​Module helpers (crewai.flow.conversation)

​Intent routing patterns

​A. Pre-classify via ConversationConfig (simplest)

​B. Classify inside route_turn (richer prompts)

​When the flow finishes but the user keeps chatting

​Conversational Flow (experimental)

​Quick example

​ConversationConfig

​RouterConfig and the auto-built route catalog

​Built-in routes

​handle_turn() semantics

​chat() for local REPLs

​Custom router behavior

​append_assistant_message and append_agent_result

​Tracing across turns

​Conversational Flow trace lifecycle

​Streaming

​Imports

​See also

Overview

Turn APIs

Quick start

Streaming a turn

Turn lifecycle

`ConversationConfig` (class-level defaults)

Lower-level `ChatState` helpers

`Flow` conversational API

`handle_turn` parameters

`kickoff` parameters

Instance attributes

Methods and properties

Module helpers (`crewai.flow.conversation`)

Intent routing patterns

A. Pre-classify via `ConversationConfig` (simplest)

B. Classify inside `route_turn` (richer prompts)

When the flow finishes but the user keeps chatting

Conversational `Flow` (experimental)

Quick example

`ConversationConfig`

`RouterConfig` and the auto-built route catalog

Built-in routes

`handle_turn()` semantics

`chat()` for local REPLs

Custom router behavior

`append_assistant_message` and `append_agent_result`

Tracing across turns

Conversational `Flow` trace lifecycle

Streaming

Imports

See also