Conversational Flows

Conversational Flows (대화형 플로우)

채팅 앱은 사용자가 메시지를 보낼 때마다 상태가 이어져야 해요. CrewAI의 Conversational Flows는 각 사용자 라인을 같은 세션 id의 새로운 Flow 실행으로 취급해서, 메시지 히스토리와 의도 라우팅, 트레이싱, 구조화 스트리밍을 한 번에 제공합니다. 지원 챗봇이나 대화형 에이전트를 만들 때 handle_turn()을 쓰면 턴 단위로 유연하게 다룰 수 있어요.

출처: 공식문서

본문

핵심 개념

Conventional (일반) 크루와 달리, 대화형 앱의 각 사용자 라인은 고유한 실행이지만 세션 id는 공유합니다.

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	`conversation_turn_completed`; with default trace deferral, `FlowFinished` waits for `finalize_session_traces()`
Full-session trace	`ConversationConfig(defer_trace_finalization=True)` + `finalize_session_traces()`

턴 API

REST, WebSocket, 테스트, 커스텀 UI에서 오는 모든 사용자 메시지에는 **flow.handle_turn(message, session_id=...)**을 쓰고, 로컬 터미널 채팅 루프가 필요하면 **flow.chat()**를 씁니다.

Flow.kickoff()는 user_message=나 session_id= 키워드 인자를 받지 않아요. 대화형 Flow에서 handle_turn()은 대기 중인 메시지를 저장하고, 턴별 실행 상태를 리셋한 뒤 내부적으로 kickoff(inputs={"id": session_id})를 호출합니다.

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

handle_turn(), stream_turn(), chat()는 conversational 모드가 켜져 있지 않으면 ValueError를 발생시킵니다. @ConversationConfig(...)를 적용하면 자동으로 켜지고, 그 외에는 conversational = True로 설정하면 됩니다.

빠른 시작

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):
        # ... return a response string
        return "Your order is on the way!"

    @listen("help")
    def handle_help(self):
        return "How can I help you today?"

    @listen("goodbye")
    def handle_goodbye(self):
        return "Goodbye! Thanks for chatting."


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

이 예시에서 route_turn이 현재 사용자 메시지를 분석해 어떤 라우트로 보낼지 결정하고, 각 @listen("...")이 해당 라우트의 응답을 처리합니다.

턴 수명주기

각 handle_turn()은 Flow 그래프의 한 실행을 돌립니다. 대화 히스토리는 state.messages에 누적되고, state.current_user_message가 현재 턴의 사용자 입력을 담습니다. 기본 트레이스 지연 설정에서는 FlowFinished가 finalize_session_traces()를 기다립니다.

설정 개요

ConversationConfig로 토큰, 지연 트레이스, 의도 라우터 LLM 등을 클래스 단위로 설정합니다.

하위 레벨 `ChatState` 헬퍼

ChatState, ConversationalConfig, crewai.flow.conversation 헬퍼는 고급 오케스트레이션·테스트·커스텀 래퍼에서 여전히 import 가능합니다. ConversationalInputs는 kickoff(inputs={...})의 기존 키(id, user_message, last_intent)를 위한 TypedDict입니다.

의도 라우팅 패턴

ConversationConfig로 사전 분류하거나, route_turn 안에서 더 풍부한 프롬프트로 분류할 수 있습니다.

사용자가 계속 채팅하면

@ConversationConfig(defer_trace_finalization=True)를 쓰지 않으면 턴이 끝날 때마다 트레이스가 종료됩니다. 전체 세션을 하나의 트레이스로 묶으려면 지연 트레이스 + finalize_session_traces() 조합을 사용하세요.

대화형 `Flow`

conversational = True로 옵트인하면, 기본 Flow가 내장 @start/@router/converse_turn/end_conversation 그래프를 제공하고 state.messages를 관리하며, 라우터 LLM을 구동하고 턴 간 트레이스 배치를 열어 둡니다. 커스텀 라우트만 작성하면 프레임워크가 나머지를 관리해요.

RouterConfig와 자동 라우트 카탈로그 — 라우터에 보내는 프롬프트가 자동 생성됩니다. 각 라우트에 대해 프레임워크는 다음 우선순위로 설명을 선택합니다: RouterConfig.route_descriptions[label](명시적 오버라이드) → Flow.builtin_route_descriptions[label](프레임워크 기본 문구).
chat() — handle_turn()을 REPL로 감싼 로컬 채팅 루프. exit/quit에서 종료하고, 기본적으로 빈 줄은 건너뛰며, 세션 종료 시 finalize_session_traces()를 호출합니다.

스트리밍

대화형 Flow는 stream_turn()으로 단일 턴을 구조화된 런타임 프레임으로 스트리밍할 수 있습니다. 자세한 내용은 Streaming Runtime Contract를 참고하세요.

가져오기

from crewai import Flow
from crewai.flow import listen, start, router
from crewai.flow import ConversationConfig, ConversationState, ChatState

더 알아보기

CrewAI Flows — 대화형 Flow의 기반
Streaming Runtime Contract — 턴·Flow·LLM 스트리밍 프레임
Conversational Flows 공식 문서

Conversational Flows