콘텐츠로 이동

대화형 플로우 (Conversational Flows)

여러 차례 주고받는 채팅 앱을 만들 때 턴마다 handle_turn을 돌리고, 메시지 기록·의도 라우팅·트레이싱·WebSocket 브리지를 묶어주는 CrewAI의 플로우 방식입니다.

개념

일반적인 플로우는 한 번 kickoff() 하면 흐름이 끝나요. 그런데 채팅 앱은 사용자가 한 줄을 보낼 때마다 새 턴이 시작되고, 같은 대화가 이어져야 하죠. CrewAI는 대화형 앱에서 사용자의 한 줄을 같은 세션 id를 가진 새로운 플로우 실행으로 취급합니다. 그리고 이 패턴에 필요한 것들을 기본으로 준비해 둬요.

  • 세션 id — handle_turn(..., session_id=...) → kickoff(inputs={"id": ...}) → state.id로 이어져요.
  • 사용자 한 줄 — handle_turn(message)가 그래프를 돌리기 전에 state.messages에 그 줄을 추가해요.
  • 턴 완료 — conversation_turn_completed로 이번 턴만 끝나요. 대화는 다음 handle_turn에서 계속돼요.
  • 세션 전체 트레이스 — 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= 키워드 인자를 받지 않아요. 대화형 플로우에서 handle_turn()은 보류된 메시지를 저장하고, 턴 실행 상태를 리셋한 뒤 내부적으로 kickoff(inputs={"id": session_id})를 호출하는 방식이에요.

API 용도
handle_turn(message, session_id=...) 대화형 Flow의 턴 하나를 감싸는 편한 래퍼
stream_turn(message, session_id=...) 대화 턴 하나를 순서 있는 런타임 프레임으로 스트리밍
chat() 로컬 터미널 REPL
kickoff(inputs={...}) 대화 턴 처리 없이 진행하는 고급 실행
ask() 한 스텝 안에서 막는 프롬프트 (마법사·명확화)
@human_feedback 스텝의 출력을 승인/거부 (다음 채팅 한 줄과는 달라요)
ChatSession.handle_turn(...) handle_turn 위의 전송 계층 (SSE / WebSocket)

handle_turn(), stream_turn(), chat()은 대화 모드가 켜져 있지 않으면 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):
        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()  # 채팅 전체를 하나의 트레이스로 묶어요

실행 흐름을 보면 — route_turn이 메시지를 읽어 라우트 라벨(order, help, goodbye)을 고르고, 해당하는 @listen 핸들러가 응답을 만들어요. 핸들러는 append_assistant_message로 사용자에게 보일 응답을 state.messages에 남겨요. 반환값이 문자열이라면 그것도 어시스턴트 메시지로 기록되니, 두 가지 중 하나만 하면 돼요. 사용자의 줄은 handle_turn이 이미 저장하니까 핸들러에서 다시 추가하면 안 돼요.

턴 파이프라인

handle_turn 한 번은 이런 순서로 돌아가요.

  1. 턴 준비 — 보류된 사용자 메시지를 저장하고, 세션 id를 해석하며, 턴 실행 추적을 리셋하고 kickoff(inputs={"id": session_id})를 호출해요.
  2. 상태 복원 — inputs["id"]가 있고 @persist가 설정되어 있으면 최신 스냅샷을 불러와요.
  3. 발화 — 첫 번째 지연 턴에서만 FlowStarted가 발생해요.
  4. 턴 수화(hydration) — 사용자 메시지를 state.messages에 추가하고 current_user_message / last_user_message를 설정해요. intents / default_intents + intent_llm이 설정되어 있으면 메시지를 분류하기도 해요.
  5. 그래프 실행 — 사용자 정의 @start 메서드(있으면) → route_conversation(내장 시작/라우터) → 선택된 @listen 핸들러 순으로 실행돼요. route_conversation은 오버라이드 가능한 conversation_start() 헬퍼도 호출해요.
  6. 턴 종료 — 지연(deferral)이 켜져 있으면 턴마다의 flow_finished와 트레이스 마무리가 건너뛰어져요. 중첩된 Agent.kickoff() / 크루도 부모 배치를 닫지 않아요.

트레이싱: 세션 전체를 하나로

ConversationConfig의 defer_trace_finalization은 기본값이 True예요. 이 설정이 켜져 있으면:

  • 채팅 세션 전체가 하나의 트레이스 배치로 묶여요.
  • flow_started는 첫 턴에서만, flow_finished는 finalize_session_traces()에서 딱 한 번 나와요.
  • 턴마다의 kickoff는 "Trace batch finalized"를 출력하지 않아요.
  • 중첩 작업(Agent.kickoff(), 크루, Exa 툴)은 부모 배치에 붙고, 내부 AgentExecutor 플로우가 세션 배치를 일찍 닫지 않아요.

flow.chat()은 종료 시 finalize_session_traces()를 알아서 호출해요. 반면에 handle_turn()으로 루프를 직접 소유한다면 세션이 끝날 때 직접 불러줘야 해요.

실무 포인트 — REPL이나 루프를 만들 땐 항상 try/finally로 감싸고 종료 시 flow.finalize_session_traces()를 호출하세요. 지연된 턴은 턴 단위 flow_failed도 억제하니까, 턴 오류나 세션 중단 때도 명시적으로 finalize 해야 해요. 그렇지 않으면 트레이스 배치가 열린 채로 남아 최종 대화가 내보내지지 않을 수 있어요.

스트리밍

UI가 한 턴의 구조화된 이벤트(라우팅, LLM 청크, 툴 활동, 대화 메시지)를 받아야 한다면 stream_turn()을 써요.

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

턴마다 flow.stream을 쓰지 않도록 주의해요. 대화형 턴의 스트리밍 수명주기는 stream_turn()이 소유하니까, handle_turn()과 함께 flow.stream = True를 설정하지 말아요. (비대화형 Flow에선 stream = True가 kickoff()가 StreamSession을 반환하게 해요.)

의도 라우팅 패턴

라우팅을 구성하는 방법은 두 가지가 있어요.

A. ConversationConfig로 미리 분류 (가장 단순) — default_intents와 intent_llm을 설정하면 매 handle_turn()이 현재 메시지를 미리 분류해요. 커스텀 route_turn()이 비어 있지 않은 결과를 반환하면 그게 우선하고, 아니면 route_conversation이 이번 턴의 분류된 의도를 사용해요.

B. route_turn 안에서 분류 (더 풍부한 프롬프트) — default_intents=None으로 두면 handle_turn()은 메시지만 추가해요. route_turn()에서 classify_intent를 커스텀 프롬프트나 설명과 함께 호출해요.

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

웹 리서치나 여러 단계의 툴 사용이 필요하다면, @listen("RESEARCH") 같은 스텝에서 맨 LLM.call() 대신 툴을 단 Agent.kickoff()를 돌리도록 해요.

핸들러 이름 짓기 — 함정 하나

@listen("…")의 문자열은 라우터의 라우트 라벨(이벤트 이름)이지, 파이썬 메서드 이름이 아니에요. 라우트 라벨과 메서드 완료 이벤트가 같은 트리거 네임스페이스를 공유하니까, 핸들러 이름을 라우트와 똑같이 지으면 핸들러가 자기 자신을 다시 트리거하는 무한 루프가 생겨요.

@listen("create_video")
def handle_create_video(self) -> str:   # handle_* 접두사를 써서 구분
    """User wants a new video."""
    ...

이렇게 하지 마세요 — 플로우 인스턴스화 때 거부돼요.

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

대화형 플로우 선언 (JSON/YAML)

데클러러티브(선언형) 플로우도 대화형으로 만들 수 있어요. 최상위에 conversational 블록을 두고, 라우트 라벨을 듣는 메서드로 나만의 라우트를 선언해요.

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}"

블록을 선언하는 것 자체가 옵트인이에요(enabled 기본값은 true). enabled: false로 두면 설정은 유지하면서 채팅만 끄고, 내장 메서드 합성도 꺼지니까 정상적인 비대화형 그래프를 직접 제공해야 해요.

파이썬에서는 클래스 기반 대화형 플로우와 같은 턴 API로 실행해요.

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

crewai run은 데클러러티브 대화형 플로우에 대해 대화 TUI를 열어요. 채팅 루프는 터미널이 필요하므로, 헤드리스 실행은 한 턴을 돌리는 대신 0이 아닌 코드로 나가며 안내를 출력해요. 그때는 파이썬에서 handle_turn() / stream_turn()으로 구동하세요. --inputs는 대화형 플로우에서는 받지 않아요 — 각 턴의 입력은 당신이 치는 메시지니까요.

더 알아보기

  • Flow 상태 관리 익히기 — 영속화, Pydantic 상태, @persist
  • 첫 플로우 만들기 — 플로우 기본기

함께 보면 좋은 세부사항

  • @persist는 클래스 전체보다 마지막 터미널 스텝(예: finalize)에 두는 걸 권장해요. 클래스 레벨 영속화는 매 메서드 후 저장하고, load_state는 가장 최신 행을 쓰기 때문에 턴 중간 스냅샷(예: 부트스트랩 직후)을 잡아 같은 턴의 핸들러 갱신을 놓칠 수 있어요.
  • 팔로업 채팅 한 줄에 @human_feedback을 쓰지 마세요. 특정 스텝 출력을 사용자가 보기 전에 승인해야 할 때만 쓰는 거예요.
  • append_agent_result(agent_name, result, visibility="private") — state.events에 구조화된 이벤트를, state.agent_threads[agent_name]에 스레드를 기록해요. 공개(public) 가시성은 append_assistant_message도 호출하죠. 시험용 스크래치 작업은 캐노니컬 히스토리를 오염하지 않도록 private으로 남겨요.
  • ConversationConfig.visible_agent_outputs — 특정 에이전트의 private 결과를 전역으로 공개 승격할 수 있어요 ("all" 또는 에이전트 이름 목록).
  • RouterConfig.prompt는 도메인 프레이밍(어시스턴트 페르소나, 비즈니스 규칙, 목소리)용이에요. 라우트 카탈로그는 자동 생성되니 프롬프트에 라우트를 나열하지 마세요 — 핸들러를 추가할 때마다 어긋나요.