> ## Documentation Index
> Fetch the complete documentation index at: https://docs.crewai.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 대화형 Flow

> 턴별 handle_turn, 메시지 기록, 의도 라우팅, 트레이싱, 구조화된 스트리밍으로 멀티턴 채팅 앱을 만듭니다.

## 개요

대화형 앱은 각 사용자 입력을 **동일한 세션 id**로 **새 flow 실행**으로 처리합니다. CrewAI는 메시지 기록, 선택적 의도 라우팅, 지연 트레이싱, 구조화된 턴 스트리밍, 로컬 `flow.chat()` REPL을 위한 헬퍼를 제공합니다.

| 개념 | 구현 |
| - | - |
| 세션 id | `handle_turn(..., session_id=...)` → `kickoff(inputs={"id": ...})` → `state.id` |
| 사용자 입력 | `handle_turn(message)`가 그래프 실행 전 `state.messages`에 추가 |
| 턴 완료 | `conversation_turn_completed`; 기본 trace 지연을 사용하면 `FlowFinished`는 `finalize_session_traces()`까지 대기 |
| 세션 전체 트레이스 | `ConversationConfig(defer_trace_finalization=True)` + `finalize_session_traces()` |

## 턴 API

REST, WebSocket, 테스트, 커스텀 UI에서 오는 모든 사용자 메시지에는 \*\*`flow.handle_turn(message, session_id=...)`\*\*를 사용하세요. 대화형 `Flow`를 로컬 터미널 채팅 루프로 실행하고 싶을 때는 \*\*`flow.chat()`\*\*을 사용하세요.

`Flow.kickoff()`는 `user_message=` 또는 `session_id=` 키워드 인자를 받지 않습니다. 대화형 flow에서는 `handle_turn()`이 보류 중인 메시지를 저장하고 턴별 실행 상태를 초기화한 뒤 내부적으로 `kickoff(inputs={"id": session_id})`를 호출합니다.

| API | 용도 |
| - | - |
| `handle_turn(message, session_id=...)` | 대화형 `Flow`용 한 턴 편의 래퍼 |
| `stream_turn(message, session_id=...)` | 대화형 한 턴을 순서가 보장된 런타임 frame으로 스트리밍 |
| `chat()` | 대화형 `Flow`용 로컬 터미널 REPL |
| `kickoff(inputs={...})` | 대화형 턴 처리 없이 flow를 직접 실행하는 고급 용도 |
| `ask()` | 한 스텝 **내부** 블로킹 프롬프트 (마법사, 확인) |
| `@human_feedback` | **스텝 출력** 승인/거부 — 다음 채팅 줄이 아님 |

대화형 모드가 활성화되지 않으면 `handle_turn()`, `stream_turn()`, `chat()`은 `ValueError`를 발생시킵니다. `@ConversationConfig(...)`를 적용하면 자동으로 활성화되며, 그렇지 않으면 `conversational = True`로 설정하세요.

## 빠른 시작

```python theme={null}
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 ""
        if "주문" in message or "order" in message.lower():
            return "order"
        if "안녕" in message or "goodbye" in message.lower():
            return "goodbye"
        return "help"

    @listen("order")
    def handle_order(self):
        reply = "주문이 배송 중입니다."
        self.append_assistant_message(reply)
        return reply

    @listen("help")
    def handle_help(self):
        reply = "무엇을 도와드릴까요?"
        self.append_assistant_message(reply)
        return reply

    @listen("goodbye")
    def handle_goodbye(self):
        reply = "안녕히 가세요!"
        self.append_assistant_message(reply)
        return reply


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

try:
    flow.handle_turn("주문 어디까지 왔나요?", session_id=session_id)
    flow.handle_turn("반품은 어떻게 하나요?", session_id=session_id)
finally:
    flow.finalize_session_traces()  # 전체 대화에 대한 단일 trace 링크
```

## 턴 스트리밍

UI나 런타임에서 한 채팅 턴의 구조화된 이벤트가 필요하면 `stream_turn()`을 사용하세요. Flow 라우팅, LLM chunk, tool 활동, 대화 메시지를 순서가 보장된 frame으로 제공하는 stream session을 반환합니다.

```python theme={null}
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
```

전체 frame 계약과 channel 목록은 [스트리밍 런타임 계약](/edge/ko/learn/streaming-runtime-contract)을 참고하세요.

## 턴 생명주기

각 `handle_turn`은 다음 파이프라인을 실행합니다:

1. **턴 설정** — 보류 중인 사용자 메시지를 저장하고 세션 id를 결정하며 턴별 실행 추적을 초기화한 뒤 `kickoff(inputs={"id": session_id})`를 호출.
2. **상태 복원** — `inputs["id"]`가 있고 `@persist`가 설정되면 최신 스냅샷 로드.
3. **`FlowStarted`** — 지연 세션의 첫 턴에서만 발생.
4. **보류 중인 턴 수화** — 사용자 메시지를 `state.messages`에 추가하고 `current_user_message` / `last_user_message`를 설정하며, `intents` / `default_intents` + `intent_llm` 설정 시 선택적으로 분류.
5. **그래프 실행** — 사용자 정의 `@start` 메서드(있는 경우) → `route_conversation`(내장 start/router) → 선택된 `@listen` 핸들러. `route_conversation`은 재정의 가능한 `conversation_start()` 헬퍼도 호출합니다.
6. **실행 종료** — 지연 활성화 시 턴별 `flow_finished` 및 trace 종료 **건너뜀**; 중첩 `Agent.kickoff()` / crew도 부모 batch를 닫지 않음.

핸들러는 보이는 응답이 반환값과 다를 때, 또는 히스토리를 자를 때 \*\*`append_assistant_message(reply)`\*\*를 호출하세요. public 문자열 반환값도 assistant로 기록되며 `@persist` 스냅샷에 포함되므로, 새 Flow 인스턴스에서도 복원됩니다. 사용자 입력은 `handle_turn`이 이미 저장합니다 — 핸들러에서 다시 추가하지 마세요.

## 설정 개요

`Flow` 서브클래스에 `ConversationConfig`를 데코레이터로 적용하면 채팅 기본값이 부착되고 대화형 모드도 활성화됩니다. 아래의 [전체 필드 레퍼런스](#conversationconfig)를 참고하세요. 턴마다 `handle_turn(..., intents=..., intent_llm=...)`로 사전 분류 설정을 재정의할 수 있습니다.

## 하위 수준 `ChatState` 헬퍼

`ChatState`, 레거시 `ConversationalConfig`, `crewai.flow.conversation` 헬퍼는 고급 오케스트레이션, 테스트, 커스텀 래퍼에서 계속 import할 수 있습니다. 이들은 `ConversationState` / `ConversationConfig` API와 별개이며 `Flow.kickoff()`에 `user_message=` 또는 `session_id=` 키워드 인자를 추가하지 않습니다.

```python theme={null}
from crewai.flow import ChatState


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

| 필드 | 역할 |
| - | - |
| `id` | 세션 UUID (`inputs["id"]`와 동일) |
| `messages` | LLM 기록용 `{role, content}` 리스트 |
| `last_user_message` | 이번 턴의 최신 사용자 입력 |
| `last_intent` | 분류 후 라우트 라벨 (사용 시) |
| `session_ready` | 일회성 bootstrap 플래그 |

`ConversationalInputs`는 `kickoff(inputs={...})`용 `TypedDict`: `id`, `user_message`, `last_intent`.

`ConversationState`는 `messages`를 `ConversationMessage` 객체로 저장하며 `current_user_message`, `ended`, `events`, `agent_threads`도 제공합니다. 정식 기록을 LLM에 전달할 때는 `conversation_messages`를 사용하세요.

## `Flow` 대화 API

### `handle_turn` 파라미터

| 파라미터 | 목적 |
| - | - |
| `message` | 이번 턴의 텍스트 |
| `session_id` | 대화 UUID → `inputs["id"]` / `state.id` |
| `intents` | kickoff 전 `classify_intent`용 결과 라벨 |
| `intent_llm` | 분류 LLM (`intents`와 함께 필수) |
| `**kickoff_kwargs` | `input_files`, `from_checkpoint`, `restore_from_state_id` 같은 옵션을 `kickoff()`로 전달 |

### `kickoff` 파라미터

`Flow.kickoff()`는 `inputs`, `input_files`, `from_checkpoint`, `restore_from_state_id`를 받습니다. 원시 flow 실행이 필요하면 `inputs={"id": session_id}`를 전달할 수 있지만, 채팅 메시지를 나타내는 호출에는 `handle_turn()`을 사용하세요.

### 인스턴스 속성

| 속성 | 목적 |
| - | - |
| `conversational` | 대화형 그래프와 `handle_turn()`을 활성화하려면 `True`로 설정 |
| `defer_trace_finalization` | 선택적 인스턴스 재정의. 없으면 `_should_defer_trace_finalization()`이 `ConversationConfig.defer_trace_finalization`을 읽음 |
| `suppress_flow_events` | 콘솔 flow 패널과 메서드 실행 이벤트를 숨김. flow start/finish 이벤트는 계속 발생 |
| `stream` | 일반 Flow 스트리밍 플래그. 대화형 턴에서는 이 플래그와 `handle_turn()`을 함께 쓰지 말고 `stream_turn()` 사용 |

### 메서드 및 프로퍼티

| 이름 | 설명 |
| - | - |
| `append_assistant_message(content)` | 사용자에게 보이는 어시스턴트 응답을 `state.messages`에 추가 |
| `append_message(role, content, **extra)` | `state.messages`에 추가 |
| `conversation_messages` | LLM 호출용 읽기 전용 기록 |
| `classify_intent(text, outcomes, *, llm, context=None)` | outcome 매핑 (`@human_feedback`와 동일 collapse) |
| `receive_user_message(text, *, outcomes=None, llm=None)` | 사용자 메시지 추가; 선택적 `last_intent` |
| `finalize_session_traces()` | 지연 `flow_finished` 발생 및 세션 trace batch 종료 |
| `_should_defer_trace_finalization()` | 턴별 trace 종료 지연 여부를 결정하는 고급/내부 hook |
| `input_history` | `ask()` 프롬프트/응답 감사 기록 |

### 모듈 헬퍼 (`crewai.flow.conversation`)

테스트 또는 커스텀 오케스트레이션을 위해 `crewai.flow.conversation`에서 import할 수 있습니다. 이 헬퍼들은 레거시 `ConversationalConfig` 형태를 사용합니다. 또한 `prepare_conversational_turn()`은 `last_intent`를 지우지만, `handle_turn()`은 router 컨텍스트로 보존합니다.

| 함수 | 설명 |
| - | - |
| `normalize_kickoff_inputs(inputs, user_message=..., session_id=...)` | 대화 kwargs를 `inputs`에 병합 |
| `get_conversation_messages(flow)` | 상태 또는 내부 버퍼에서 메시지 읽기 |
| `append_message(flow, role, content, **extra)` | 인스턴스 메서드와 동일 |
| `prepare_conversational_turn(flow, user_message=..., intents=..., intent_llm=..., config=...)` | 커스텀 래퍼용 하위 수준 턴 수화 |
| `receive_user_message(flow, text, ...)` | 인스턴스 메서드와 동일 |
| `set_state_field(flow, name, value)` | dict 또는 Pydantic 상태 필드 설정 |
| `get_conversational_config(flow)` | 클래스 `conversational_config` 읽기 |
| `input_history_to_messages(entries)` | `input_history`를 LLM 메시지 형식으로 |

## 의도 라우팅 패턴

### A. `ConversationConfig`로 사전 분류 (가장 단순)

`default_intents`와 `intent_llm`을 설정하세요. 각 `handle_turn()`이 현재 메시지를 사전 분류합니다. 커스텀 `route_turn()`이 반환한 비어 있지 않은 결과가 우선하며, 그렇지 않으면 `route_conversation`이 현재 턴의 분류된 intent를 사용합니다.

### B. `route_turn` 내부에서 분류 (풍부한 프롬프트)

`default_intents=None`으로 설정하면 `handle_turn()`은 사용자 메시지만 추가합니다. `route_turn()`에서 커스텀 프롬프트나 설명과 함께 `classify_intent`를 호출하세요:

```python theme={null}
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
```

웹 리서치나 다단계 tool이 필요하면 **`@listen("RESEARCH")`** 등에서 `Agent.kickoff()`와 tool 사용 — 단순 `LLM.call()` 대신.

## flow가 끝났지만 사용자는 계속 대화할 때

각 `handle_turn()`은 하나의 그래프 실행을 완료하며, 같은 `session_id`로 다음 `handle_turn()`을 호출해 대화를 이어갑니다. 기본 지연 trace 수명 주기에서는 해당 실행이 `conversation_turn_completed`를 발생시키고, `finalize_session_traces()`가 세션을 닫을 때 `FlowFinished`가 한 번 발생합니다. `@persist`는 `messages`, 플래그, 컨텍스트를 복원합니다.

**Persist 패턴:** 전체 `Flow` 클래스보다 **단일 종료 스텝**(예: `finalize`)에 `@persist`를 두는 것이 좋습니다. 클래스 수준 persist는 매 메서드 후 저장하며, `load_state`는 최신 행을 사용해 같은 턴의 핸들러 업데이트를 놓칠 수 있습니다.

후속 채팅 줄에 `@human_feedback`를 쓰지 마세요. 특정 스텝 출력을 사람이 승인해야 할 때만 사용하세요.

## 대화형 `Flow`

`Flow` 서브클래스에 `conversational = True`를 지정하거나 `@ConversationConfig(...)`를 적용하면 대화형 채팅 그래프가 활성화됩니다. 베이스 `Flow`는 내장 start/router인 `route_conversation`과 `converse_turn`, `end_conversation` 리스너를 제공합니다. 사용 중단된 `answer_from_history_turn` 리스너는 호환성을 위해 계속 제공됩니다. 또한 `state.messages`를 관리하고 router LLM을 구동할 수 있으며 턴 간 trace batch를 열린 상태로 유지합니다. 여러분은 **커스텀 라우트**를 작성하고 나머지는 프레임워크에 맡기면 됩니다.

LLM 기반 라우터와 라우트별 핸들러로 멀티턴 챗을 만들고 싶지만 라이프사이클을 직접 배선하고 싶지 않을 때 사용하세요. 완전한 제어가 필요하면 위의 `Flow[ChatState]`로 내려가세요.

### 빠른 예제

```python theme={null}
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()
```

로컬 터미널 채팅에는 `chat()`을 사용하세요:

```python theme={null}
def kickoff() -> None:
    SupportFlow().chat()
```

`chat()`은 `handle_turn()`을 REPL로 감싸고, `exit` / `quit`에서 종료하며, 기본적으로 빈 줄을 건너뛰고, 세션이 끝날 때 `finalize_session_traces()`를 호출합니다.

### `ConversationConfig`

클래스 단위의 챗 기본값을 부착하는 클래스 데코레이터입니다.

| 필드 | 기본값 | 목적 |
| - | - | - |
| `system_prompt` | i18n `slices.conversational_system_prompt` | 빌트인 `converse_turn`이 사용하는 system 메시지. 빈 문자열(`""`)을 전달하면 system 메시지를 끕니다. |
| `llm` | `None` | 대화용 LLM (빌트인 `converse_turn`이 사용하고 router 폴백도 됨). |
| `router` | `None` | 선택적 `RouterConfig` 재정의. 커스텀 listener와 결정 가능한 LLM이 있으면 생략해도 라우팅이 자동 활성화됩니다. |
| `answer_from_history_prompt` | 프레임워크 기본값 | **사용 중단됨.** `converse` system prompt를 사용하거나 `converse_turn()`을 재정의하세요. |
| `answer_from_history_llm` | `None` | **사용 중단됨.** `llm`을 사용하세요. `converse`는 이미 정식 기록을 전달받습니다. |
| `intent_llm` | `None` | 레거시 `intents=`/`default_intents` 사전 분류용 LLM. |
| `default_intents` | `None` | 레거시 사전 분류용 outcome 레이블. |
| `visible_agent_outputs` | `None` | `"all"` 또는 `append_agent_result()` 결과를 사용자에게 공개로 승격할 에이전트 이름 목록. |
| `defer_trace_finalization` | `True` | `handle_turn()` 호출들 사이에서 하나의 trace 배치를 열어 둡니다. |

<Warning>
  `answer_from_history_prompt`, `answer_from_history_llm`, `answer_from_history`
  라우트는 사용 중단되었으며 향후 릴리스에서 제거될 예정입니다. 이들은 이미
  정식 기록을 처리하는 `converse`와 기능이 중복되고, 답변 가능 여부를 판단하는
  LLM 호출을 추가하며, 일반 auto-router가 라우트를 반환하면 우회됩니다. 기존
  설정은 계속 작동하며 `DeprecationWarning`을 발생시킵니다.
</Warning>

커스텀 라우트가 없으면 턴은 `converse`로 이어집니다. 커스텀 라우트와 대화/router LLM이 있으면 프레임워크가 기본 `RouterConfig`를 합성합니다. prompt, 라우트 목록, 설명, fallback 동작을 바꿔야 할 때만 명시적으로 제공하세요. `default_intents`를 설정하면 레거시 사전 분류 경로를 사용합니다.

대화 LLM을 설정하지 않으면 내장 `converse_turn`은 답변을 생성하는 대신 설정 안내 placeholder를 반환합니다.

### `RouterConfig`와 자동 생성되는 라우트 카탈로그

```python theme={null}
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",
)
```

router에 전달되는 프롬프트는 자동으로 만들어집니다. 각 라우트의 설명은 다음 우선순위로 결정됩니다:

1. `RouterConfig.route_descriptions[label]` — 명시적 오버라이드.
2. `Flow.builtin_route_descriptions[label]` — `converse`, `end`, 사용 중단된 `answer_from_history` 호환 라우트용 프레임워크 기본 텍스트 (router LLM용으로 다듬어진 문구).
3. 메서드에 선언된 `description` — 선언적 flow와 DSL projection에서 사용.
4. `@listen(label)` 핸들러 docstring의 첫 번째 비어 있지 않은 줄.
5. 빈 문자열 — 설명 없이 라우트만 표시.

실제 사용에서 **새 라우트를 추가하는 방법은 `@listen("X")` + 한 줄짜리 docstring**입니다:

```python theme={null}
@listen("INTERNET_SEARCH")
def handle_internet_search(self) -> str:
    """Fresh web research, current news, real-time lookups."""
    ...
```

### 핸들러 이름 짓기

`@listen("…")`의 문자열은 Python 메서드 이름이 아니라 **router 라우트 레이블**(이벤트 이름)입니다. 라우트 레이블과 메서드 완료 이벤트는 하나의 트리거 namespace를 공유하므로, 핸들러 이름을 라우트와 같게 지정하면 핸들러가 자기 자신을 반복해서 다시 실행합니다.

서로 다른 메서드 이름을 사용하세요. 문서 예제에서는 `handle_*` 접두사를 사용합니다:

```python theme={null}
@listen("create_video")
def handle_create_video(self) -> str:
    """User wants a new video."""
    ...
```

메서드 이름을 라우트 레이블과 같게 만들지 마세요:

```python theme={null}
@listen("create_video")
def create_video(self) -> str:  # rejected at flow instantiation
    ...
```

…그러면 router LLM은 다음을 봅니다:

```
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`는 **도메인 프레이밍** (어시스턴트 페르소나, 비즈니스 규칙, 톤)을 위한 자리입니다. 라우트 카탈로그는 자동 생성되니 `prompt` 안에 라우트 목록을 넣지 마세요. 핸들러를 추가하는 순간 동기화가 깨집니다.

### 빌트인 라우트

| 라우트 | 핸들러 | 목적 |
| - | - | - |
| `converse` | `converse_turn` | 기본 챗 핸들러. system prompt + 정식 메시지 히스토리와 함께 `ConversationConfig.llm`을 호출합니다. |
| `end` | `end_conversation` | `state.ended = True`로 설정하고 종료 응답을 보냅니다. |
| `answer_from_history` | `answer_from_history_turn` | **사용 중단된 호환 라우트.** 이미 정식 기록을 전달받는 `converse`를 사용하세요. |

서브클래스에 같은 이름의 핸들러를 정의하면 어떤 것이든 오버라이드할 수 있습니다.

### `handle_turn()` 시맨틱

`flow.handle_turn(message)`는 한 턴을 실행합니다:

1. 그래프가 다시 실행되도록 턴 단위 실행 추적(`_completed_methods`, `_method_outputs`)을 초기화합니다 — 이게 없으면 동일 인스턴스에서 반복 `kickoff` 호출 시 `Flow.kickoff_async`가 `inputs={"id": ...}`를 체크포인트 복원으로 간주해 2번째 턴부터 단락 회로가 발생합니다.
2. 사용자 메시지를 `state.messages`에 추가하고 `current_user_message` / `last_user_message`를 설정합니다. `last_intent`는 **이전 턴 값이 유지**되어 router LLM이 신호로 활용할 수 있습니다.
3. 사용자 정의 `@start` 메서드(있는 경우)를 실행한 다음 내장 start/router인 `route_conversation`을 거쳐 선택된 `@listen` 핸들러를 실행합니다. `route_conversation`은 재정의 가능한 `conversation_start()` 헬퍼를 호출합니다.
4. router는 결정을 `state.last_intent`에 저장합니다 (다음 턴의 router 컨텍스트에서 보입니다).
5. 핸들러가 문자열을 반환했지만 `append_assistant_message`를 직접 호출하지 않았다면, `handle_turn`이 대신 추가한 뒤 갱신된 `state.messages`를 persist합니다. `@persist` 복원 시 assistant 턴이 포함됩니다.

채팅 메시지에는 `handle_turn()`을 호출하세요. `kickoff(inputs={"id": ...})`를 직접 호출하면 대화형 턴 래퍼 없이 flow 그래프가 실행됩니다.

### 로컬 REPL용 `chat()`

`flow.chat()`은 `handle_turn()` 위에 얹은 바로 쓸 수 있는 터미널 래퍼입니다:

```python theme={null}
flow = SupportFlow()
flow.chat()
```

일반적인 로컬 루프를 처리합니다:

1. 사용자 메시지를 입력받습니다.
2. `exit` / `quit`, `EOFError`, `KeyboardInterrupt`에서 멈춥니다.
3. `handle_turn(message, session_id=...)`를 호출합니다.
4. 어시스턴트 결과를 출력합니다.
5. `finally` 블록에서 지연된 세션 trace를 finalize합니다.

`chat(defer_trace_finalization=True)`는 REPL 동안 인스턴스의 지연 플래그를 임시로 활성화하고 종료할 때 이전 값으로 복원합니다.

주입 가능한 I/O로 터미널 동작을 커스터마이즈할 수 있습니다:

```python theme={null}
flow.chat(
    session_id="demo-session",
    prompt="You: ",
    assistant_prefix="Assistant: ",
    exit_commands=("exit", "quit", "bye"),
)
```

웹 앱, 백그라운드 worker, 테스트, 커스텀 transport에서는 계속 `handle_turn()`을 직접 사용하세요.

### 커스텀 router 동작

매 라우팅 결정마다 사이드 이펙트(이벤트 버스 셋업, 텔레메트리)를 실행하려면 `route_turn`을 오버라이드하세요:

```python theme={null}
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)
```

LLM router를 완전히 우회하고 프로그램 방식으로 라우트를 선택하려면 `route_turn`에서 비어 있지 않은 문자열을 반환하세요. falsy 값을 반환해도 오버라이드에서 `_route_with_config()`가 호출되지는 않습니다. 대신 현재 턴의 사전 분류된 intent, 설정된 경우 사용 중단된 `answer_from_history` 호환 경로, 마지막으로 `converse` 순으로 fallback합니다. 이전 턴의 `last_intent`는 router 컨텍스트에서 사용할 수 있지만 fallback으로 다시 실행되지는 않습니다.

### `append_assistant_message`와 `append_agent_result`

`@listen(label)` 핸들러 안에서 두 가지 중 선택하세요:

* `self.append_assistant_message(text)` — 사용자에게 보이는 어시스턴트 턴을 `state.messages`에 추가합니다. 다음 턴의 `converse_turn`이 이 내용을 보게 됩니다.
* `self.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 결과를 전역적으로 public으로 승격할 수 있습니다 (`"all"` 또는 이름 리스트).

## JSON/YAML로 대화형 플로우 선언하기

[선언적 Flow](/edge/ko/concepts/cli)도 대화형으로 만들 수 있습니다. 최상위 `conversational` 블록을 추가하고 라우트 레이블을 `listen`하는 메서드로 자체 라우트를 선언하세요:

```yaml theme={null}
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}"
```

블록 선언 자체가 opt-in이며 `enabled`의 기본값은 `true`입니다. 설정은 유지하면서 채팅을 끄려면 `enabled: false`로 지정하세요. 이 경우 내장 메서드 합성도 비활성화되므로 선언에 일반 비대화형 그래프를 제공해야 합니다.

세 가지가 자동으로 제공됩니다:

| 제공 항목 | 설명 |
| - | - |
| 내장 그래프 | `route_conversation`, `converse_turn`, `end_conversation`이 자동으로 추가됩니다. 사용 중단된 `answer_from_history_turn`은 호환성을 위해 유지됩니다. 같은 이름 중 하나로 메서드를 선언하면 재정의됩니다. |
| 대화 상태 | `state` 블록이 없으면 `ConversationState`가 사용됩니다. Pydantic `ref` 또는 `json_schema` state는 대화형 필드와 자동으로 합성되며 `ConversationState`를 상속할 필요가 없습니다. |
| 라우트 카탈로그 | 내부 라우트를 제외하고 `listen` 레이블이 있는 비-router 메서드에서 추론됩니다. 설명에는 위 우선순위가 적용되며 명시적인 `router.routes`로 선택지를 제한할 수 있습니다. |

선언적 `llm`, `router.llm`, `intent_llm` 필드는 모델 id 또는 `{model: openai/gpt-4o-mini, max_tokens: 512}` 같은 설정 mapping을 받습니다. `conversational` 블록은 `default_intents`, `visible_agent_outputs`, `defer_trace_finalization`과 위에 나온 `RouterConfig` 필드도 지원합니다. 사용 중단된 `answer_from_history_prompt` / `answer_from_history_llm` 선언은 호환성을 위해 계속 허용됩니다.

클래스 기반 대화형 플로우와 동일한 턴 API로 Python에서 실행합니다:

```python theme={null}
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()
```

### 라우트 이름 짓기

라우트 레이블과 메서드 이름은 하나의 트리거 네임스페이스를 공유하므로, 핸들러 이름이 자신이 listen하는 라우트와 같으면 안 됩니다 — `create_video`가 `create_video`를 listen하면 플로우 생성 시 거부됩니다. `handle_*` 접두사를 사용하세요.

### 선언으로 표현할 수 없는 것

| 표현 불가 | 대신 사용 |
| - | - |
| 살아 있는 `LLM` 인스턴스나 커스텀 `BaseLLM` | 모델 id 문자열 또는 정적 설정 mapping |
| 살아 있는 모델 클래스로서의 `router.response_format` | python ref로 클래스를 지정하세요: `response_format: {python: my_project.schemas.ConversationRoute}`. 생략하면 프레임워크가 생성합니다 |
| `route_turn()` 재정의 | Flow를 Python으로 작성하거나 선언적 `route_conversation` 메서드를 `call: code` / expression action으로 교체 |
| `can_answer_from_history()` 재정의 | 사용 중단됨. `converse`를 사용하거나 Python에서 `converse_turn()`을 재정의하세요. |

`crewai run`은 선언적 대화형 Flow에 대해 Python 대화형 Flow와 같은 채팅 TUI를 엽니다. 채팅 루프에는 터미널이 필요하므로 headless 실행은 단일 턴을 실행하는 대신 안내와 함께 0이 아닌 코드로 종료됩니다. 이런 환경에서는 Python의 `handle_turn()` 또는 `stream_turn()`으로 실행하세요. `human_feedback:` 블록이 있는 선언적 메서드(Python: `@human_feedback`)는 터미널 REPL에서 실행됩니다. 런타임이 TUI가 처리할 수 없는 블로킹 prompt로 feedback을 수집하기 때문입니다. 대화형 Flow에서는 `--inputs`를 받지 않습니다. 각 턴의 입력은 사용자가 입력하는 메시지이며 id로 세션을 재개하는 기능은 아직 CLI에 연결되지 않았습니다. 필요하면 Python에서 `flow.handle_turn(message, session_id=...)`을 사용하세요.

## 턴 간 트레이싱

`defer_trace_finalization=True` (`ConversationConfig` 기본값):

* 채팅 세션 전체에 **하나의 trace batch**.
* 첫 턴에만 **`flow_started`**; `finalize_session_traces()`에서 **`flow_finished`** 한 번.
* 턴별 `kickoff`는 “Trace batch finalized”를 출력하지 않음.
* **중첩 작업** (`Agent.kickoff()`, crew, Exa tool)은 **부모** batch에 추가; 내부 `AgentExecutor` flow가 세션 batch를 조기 종료하지 않음.

```python theme={null}
flow.chat(session_id=session_id)
```

`flow.chat()`이 `finalize_session_traces()`를 대신 호출합니다. `handle_turn()`으로 직접 루프를 소유하는 경우 세션이 끝날 때 `finalize_session_traces()`를 호출하세요.

`suppress_flow_events=True`는 Rich 콘솔 패널을 숨기고 메서드 실행 이벤트를 억제합니다. Flow start/finish 이벤트는 계속 발생하므로 바깥쪽 Flow 수명 주기는 추적할 수 있지만 개별 메서드 span은 생략됩니다.

### 대화형 `Flow` trace 수명 주기

[대화형 `Flow`](#대화형-flow)는 동일한 tracing 수명 주기를 따릅니다. `defer_trace_finalization` 기본값이 `True`이므로 각 `handle_turn()`은 세션 trace를 열린 상태로 유지합니다. 지연된 턴은 턴별 `flow_failed`도 억제합니다. 턴 오류나 세션 중단이 발생하면 세션을 명시적으로 finalize하세요. 그러면 턴별 `FlowFailed` 이벤트 대신 세션 수준 `FlowFinished` 이벤트로 batch가 닫힙니다. REPL/루프는 항상 `try/finally`로 감싸고 종료 시 `flow.finalize_session_traces()`를 호출하세요. 호출하지 않으면 trace batch가 열린 채 남아 최종 대화가 export되지 않을 수 있습니다.

## 스트리밍

대화형 UI에서는 `stream_turn()`을 사용하고 순서가 보장된 `StreamFrame` 객체를 순회하세요:

```python theme={null}
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
```

비대화형 Flow에서는 `stream = True`로 설정하면 `kickoff()`가 `StreamSession`을 반환합니다. `handle_turn()`을 사용할 때 `flow.stream = True`로 설정하지 마세요. 대화형 스트리밍 수명 주기는 `stream_turn()`이 관리합니다.

## import

```python theme={null}
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,
)
```

## 참고

* [Flow 상태 관리 마스터하기](/ko/guides/flows/mastering-flow-state)
* [첫 Flow 만들기](/ko/guides/flows/first-flow)

## 실험적 Jobs: 대화 턴에 걸친 실행

이 API는 실험적이며 변경될 수 있습니다. `crewai.experimental.flow_jobs`에서 명시적으로 import합니다. 안정적인 `crewai.flow` API에 포함되지 않습니다.

`JobRecord`, `JobState`, `JobUpdate`, `JobWorkState`, `JobWorkFlow`, `JobRunner`는 선택적으로 사용하는 메모리 기반 job 계약을 제공합니다. 기존 conversational Flow의 동작은 유지됩니다. Job은 독립적인 식별자, 시작 턴, revision, 실행 attempt, 상태, 현재 단계, 완료된 단계와 오류를 가집니다. 애플리케이션은 타입이 지정된 입력과 출력을 추가하며, 별도의 artifact 클래스는 필요하지 않습니다.

쓰기 가능한 결과 필드는 `output_fields`에 선언합니다. Worker는 출력 payload를 통해 입력, 소유권 또는 lifecycle 필드를 변경할 수 없습니다. Commit 전에 전체 후보 record를 검증하며, 잘못된 형식, 다른 소유자, 오래된 revision/attempt, 중복 또는 종료 후 update는 수락된 상태를 바꾸지 않습니다.

```python theme={null}
import asyncio
from typing import ClassVar
from pydantic import Field
from crewai.flow import ConversationState, start
from crewai.experimental.flow_jobs import (
    JobRecord, JobRunner, JobState, JobWorkFlow, JobWorkState, add_job,
)

class ReportJob(JobRecord):
    output_fields: ClassVar[frozenset[str]] = frozenset({"notes"})
    question: str
    notes: list[str] = Field(default_factory=list)
    stage: str = "collect"

class WorkChatState(ConversationState, JobState[ReportJob]):
    pass

class ReportWork(JobWorkFlow[JobWorkState[ReportJob]]):
    @start()
    def collect(self):
        self.check_open()
        self.publish_update("started")
        self.publish_update(
            "stage_completed", stage="collect", outputs={"notes": ["Example fact"]},
        )
        self.publish_update("completed")

async def run_job():
    state = WorkChatState()
    changes = asyncio.Queue()

    def make_work(job, inputs, publish):
        return ReportWork(
            initial_state=JobWorkState[ReportJob](job=job),
            publish=publish, suppress_flow_events=True,
        )

    runner = JobRunner(state, make_work, on_update=changes.put)
    try:
        async with runner.state_lock:
            job = ReportJob(session_id=state.id, question="Example report")
            if not add_job(state, job):
                raise RuntimeError("Job registration rejected")
            runner.submit(job, {})
        while True:
            snapshot = await changes.get()
            if snapshot["jobs"][0]["status"] in {"completed", "failed"}:
                return snapshot
    finally:
        await runner.aclose()

# Run in a script: asyncio.run(run_job())

```

실행 중인 asyncio loop에서 `JobRunner(state, make_work, on_update=publish_snapshot)`를 생성합니다. `make_work(job, inputs, publish)`는 독립적인 typed state, 복사된 job/입력, `publish=publish`를 사용하는 non-streaming `JobWorkFlow`를 반환합니다. `on_update`는 JSON 호환 snapshot (`session_id`, `seq`, `jobs`)을 받는 async callback이며 기존 transport에 연결합니다. `format_error`는 provider의 민감한 세부 정보를 제거할 수 있고, `on_worker_event`는 worker 시작과 종료를 관찰합니다.

**모든** foreground turn, job 등록 및 부모 상태 변경에는 `runner.state_lock`을 사용합니다. `add_job(state, job)`으로 등록한 뒤 `runner.submit(job, inputs)`를 호출합니다. Lock을 유지한 상태에서 동기 turn을 worker thread로 실행하며 하나의 Flow에서 두 turn을 동시에 실행하지 않습니다. Worker의 `publish_update()`도 runner의 event loop 밖에서 실행되며 다음 단계 전에 commit 수락을 기다립니다. `stage_started` update는 이전 단계의 commit을 요구하며, `stage_completed`는 성공적으로 종료하기 전에 typed 출력을 보존합니다. 이후 실패해도 이전 출력은 유지됩니다.

`build_router_context()`와 `build_agent_context()`를 통해 관련 job 상태와 출력을 제공합니다. Snapshot은 자동으로 공개 chat 메시지나 음성 응답이 되지 않습니다. Domain 해석과 작업 수용 한도는 애플리케이션 정책입니다.

Session 종료 시 foreground 실행을 정리하기 전에 소유 loop에서 `runner.request_close()`를 호출하고 이후 `await runner.aclose()`를 실행합니다. 새 작업 등록과 publication을 닫고 각 worker에 신호를 보내며 대기 중인 publisher를 해제하고 소유 task를 정리합니다. Work method는 단계 경계에서 `check_open()`을 호출해야 합니다. 진행 중인 provider 작업은 강제로 중단되지 않습니다. Subscriber 실패 시 publisher가 무한히 기다리지 않도록 runner가 닫힙니다. 음성 재생만 중지할 때는 runner를 닫지 않습니다.

초기 lifecycle은 `queued → running → completed/failed`입니다. Pause/resume, replacement, 자동 응답 전달과 분산 recovery는 제공하지 않습니다. 기존 state 기능으로 record를 직렬화할 수 있지만, 복원은 live worker를 다시 만들거나 외부 작업의 exactly-once 실행을 보장하지 않습니다.

## 실험적 턴 및 공개 응답 식별자

`TurnRecord`, `ReplyRecord`, `TurnState`, `ReplyEvent`, `TurnRunner`는 `crewai.experimental.flow_turns`에서 명시적으로 가져옵니다. 이 API는 선택적으로 사용하며 변경될 수 있습니다. 기존 conversational Flow를 감싸므로 안정적인 `handle_turn()`/`stream_turn()` API, 반환값, 실험적 job 계약은 바뀌지 않습니다.

`TurnState`는 `ConversationState`에 직렬화 가능한 `turns`와 `replies`를 추가합니다. `TurnRunner.stream_turn()` 호출 하나는 확정된 입력 하나와 기본 공개 응답을 추적합니다. 입력마다 세션 내에서 한 번만 사용하는 `turn_id`가 있으며, 새 발화나 정정에는 새 턴 ID를 사용합니다. `input_revision`은 양의 정수(기본값 `1`)이며 잠정 전사를 자동으로 교체하지 않습니다. `delivery_id`가 유일한 응답 ID이고 별도의 `reply_id`는 없습니다. 응답 종류는 `acknowledgment`, `answer`, `progress`입니다.

```python theme={null}
from crewai.flow import Flow, listen
from crewai.experimental.flow_turns import TurnRunner, TurnState

class Chat(Flow[TurnState]):
    conversational = True

    def route_turn(self, context):
        return "acknowledge"

    @listen("acknowledge")
    def handle_acknowledge(self):
        return "Request received."

flow = Chat(suppress_flow_events=True)
runner = TurnRunner(flow)
events = runner.stream_turn("Hello", turn_id="turn-1", kind="acknowledgment")
try:
    for event in events:
        if runner.accept_event(event):
            print(event.model_dump_json())
finally:
    events.close()

snapshot = runner.snapshot()
reply = snapshot["replies"][0]
runner.interrupt(
    session_id=reply["session_id"], turn_id=reply["turn_id"],
    input_revision=reply["input_revision"], delivery_id=reply["delivery_id"],
)
```

고정 응답과 최종 결과만 반환하는 provider도 스트리밍 응답과 같은 공개 `text`/`completed` 계약을 사용합니다. 기존 Flow가 정식 대화 기록을 관리합니다. `seq`는 delivery 내 이벤트 순서이고 `segment_id`는 1부터 시작하는 텍스트 델타 번호입니다. 델타가 완전한 음성 문장이라는 보장은 없습니다. 라우팅과 생성 관측 이벤트도 같은 ID를 포함합니다. `completed`는 텍스트 생성 완료이며, **오디오 재생이나 청취 완료가 아닙니다**.

전용 공개 응답 메서드(기본값 `converse_turn`, `answer_from_history_turn`) 또는 명시적으로 선택한 공개 Agent ID만 실시간 모델 텍스트를 허용합니다. `public_methods`와 `public_agent_ids` callback을 신중하게 설정하고 비공개 모델 작업과 공개 응답을 섞는 메서드를 선택하지 마세요. router 스트림, 추론, tool-call chunk, tools가 활성화된 호출, 무관한 Agent는 제외합니다. 비공개 Agent 결과는 응답으로 전환되지 않습니다. 관측 가능한 델타가 없는 provider는 정식 최종 메시지만 공개하며 모델 타이밍을 만들어내지 않습니다.

`runner.accept_event(event)`는 전송 직전에 runner가 발행한 이벤트 하나를 승인합니다. 다른 소유자의 ID, 중복되거나 순서가 뒤바뀐 이벤트, 중단 또는 실패로 무효화된 출력을 거부합니다. 애플리케이션 queue나 지연 이후 실제 전달 직전에 호출하세요. 신뢰할 수 없는 클라이언트 재생 receipt를 검증하는 API는 아닙니다. 이미 전달한 출력은 adapter에서 ID를 확인하고 오디오를 중단해야 합니다.

`runner.interrupt(session_id=..., turn_id=..., input_revision=..., delivery_id=...)`는 응답 하나만 대상으로 하며 올바른 반복 요청은 idempotent합니다. 생성 중에는 턴과 응답을 interrupted로 표시합니다. 생성 후에는 완료된 텍스트를 보존하고 job 성공 상태를 바꾸지 않은 채 응답 중단을 표시합니다. `on_interrupt`는 기존 협력적 중단 검사에 신호를 보내고 `should_interrupt`는 애플리케이션의 기존 중단 신호를 관찰할 수 있습니다. callback은 임의의 provider 호출을 강제로 중단하지 않습니다. 같은 Flow에서 다음 턴을 실행하기 전에 stream이 종료되어야 합니다. iterator를 닫으면 중단을 요청하고 기존 실행을 끝까지 기다리므로 진행 중인 provider 작업을 기다릴 수 있습니다.

백그라운드 작업과 조합하려면 `class WorkChatState(TurnState, JobState[MyJob]): ...`를 선언합니다. **전체** foreground 반복과 모든 부모 job 변경 동안 `JobRunner.state_lock`을 유지하세요. TurnRunner는 자체 기록과 중단 제어를 별도로 보호합니다. wrapper와 기본 턴 API를 동시에 실행하지 마세요. job을 성공적으로 등록한 뒤 `runner.associate_job(job)`을 호출하면 세션, 원래 턴, revision/attempt를 활성 응답에 연결합니다. 이 메서드는 작업을 등록, 제출 또는 취소하지 않습니다. 새 대화 턴은 독립적인 연구를 무효화하지 않습니다.

`elapsed_ms`는 추적 턴 시작부터 server monotonic clock을 사용합니다. 관측 가능한 `model_first_text_ms`는 provider 요청 시작과 첫 공개 텍스트의 **이벤트 timestamp 차이**입니다. provider/SDK/네트워크 작업을 포함하며 추론 엔진만의 TTFT가 아닙니다. 브라우저와 서버 timestamp를 서로 빼지 마세요. TTS 첫 chunk, 캐시 오디오, 실제 가청 시작, 재생 완료는 adapter 측정입니다.

이 단계는 백그라운드 응답 queue, 발언권 스케줄링, 재생 receipt, 정확한 단어 정렬, speculative generation, job pause/resume를 제공하지 않습니다. snapshot에는 기록만 있고 lock, 취소 신호, 실행 중인 작업은 없습니다. 작업을 재개하거나 delivery를 다시 재생하지 않습니다. 새 추적 live turn 전에 상태를 복원하세요. wrapper는 `from_checkpoint` 및 `restore_from_state_id`를 의도적으로 거부합니다. 영속적인 조정과 최종 추적 기록 저장은 별도 작업입니다. 기존 checkpoint 워크플로에는 안정적인 Flow API를 계속 사용하세요.

추적 실행 중에는 현재 복원된 상태가 기준이 됩니다. 자동 세션 재로드를 억제하여 등록된 turn/reply 기록이나 승인된 백그라운드 job 업데이트가 사라지지 않도록 합니다. 기본 Flow 턴 API의 기존 영속 상태 재로드 동작은 유지됩니다. 종료 이벤트 이후 iterator를 닫아도 완료된 응답을 중단하지 않습니다. 실패 이후 중단 요청이 도착해도 실패 이벤트는 유효한 종료 수명 주기 신호로 처리됩니다.

## 실험적 백그라운드 응답 큐

`crewai.experimental.flow_replies`는 턴/응답 계약 위에 `ReplyQueueState`, `ReplyQueue`, `DeliveryRecord`, `DeliveryFloor`, `CoveredUpdate`, `ClientActivity`를 추가합니다. 상태에 `ReplyQueueState`와 `JobState[MyJob]`를 조합하고 live Flow마다 `TurnRunner` **하나**와 `ReplyQueue` **하나**를 만드세요. 기존 Flow는 선택적으로 사용하며 job 실행기와 안정적인 대화 API는 바뀌지 않습니다.

성공한 job 결과는 안내가 기다리는 동안에도 즉시 상태에 있습니다. `enqueue(job)`은 현재 커밋된 `completed` job과 기존 원래 턴만 허용합니다. 대기 delivery는 job의 revision, attempt, 승인된 업데이트 순서를 저장하고 중복 알림을 같은 `delivery_id`로 합칩니다. 이 ID가 이후 공개 `ReplyRecord`의 ID이기도 합니다. job 완료마다 출처를 추적할 수 있는 안내 하나를 만들며 중간 단계 진행은 발화 큐를 만들지 않습니다.

```python theme={null}
from crewai.experimental.flow_jobs import JobState
from crewai.experimental.flow_turns import TurnRunner
from crewai.experimental.flow_replies import (
    ClientActivity, CoveredUpdate, ReplyQueue, ReplyQueueState,
)

class WorkChatState(ReplyQueueState, JobState[MyJob]):
    pass

# Your conversational Flow uses WorkChatState; jobs is its JobRunner.
turns = TurnRunner(flow)
replies = ReplyQueue(turns, is_relevant=lambda job: job.job_id not in cancelled_job_ids)

async def on_job_snapshot(snapshot):
    async with jobs.state_lock:
        for job in flow.state.jobs.values():
            if job.status == "completed":
                replies.enqueue(job)
    # Publish snapshots and wake your application-owned delivery task.

async def try_delivery():
    async with jobs.state_lock:
        delivery = replies.claim()
        if delivery is None:
            return
        events = replies.prepare(
            delivery.delivery_id,
            lambda current_jobs: "\n\n".join(job.answer for job in current_jobs),
        )
    for event in events:
        async with jobs.state_lock:
            if not replies.accepts_output(delivery.delivery_id):
                replies.settle(delivery.delivery_id, "skipped")
                return
            if turns.accept_event(event):
                await send_public_event(event)
    # Generation has completed. The speaking floor is still leased.
    # Validate whole-message adapter feedback before calling:
    # replies.settle(delivery.delivery_id, "completed")
```

enqueue, claim, 준비, coverage 변경, 출력 전달을 foreground 턴 및 job 커밋과 같은 `JobRunner.state_lock`으로 직렬화하세요. `ReplyQueue`는 자체 제어도 lock으로 보호하므로 두 claim이 동시에 발언권을 얻을 수 없습니다. `prepare()`는 enqueue 당시 저장한 텍스트가 아니라 복사한 **현재 승인된 결과**를 읽고 원래 턴과 job ID를 포함한 공통 `started`/`text`/`completed` 이벤트를 만듭니다. LLM 실행, history 삽입, TTS 예약은 하지 않습니다. 공개 텍스트, history, delivery task, transport는 애플리케이션이 선택합니다.

`set_foreground(True)`는 대기 중인 사용자 요청을 포함하여 foreground 생성과 종료 대기 동안 gate를 유지합니다. 작업이 끝난 뒤 해제하세요. `observe_client(ClientActivity(...))`는 소유 세션과 증가하는 client 순서, 엄격한 boolean `recording`, `playback`, `muted`만 허용합니다. ID가 있는 playback은 이 Flow 소유여야 하며 다른 응답의 오래된 stop은 새 playback을 해제할 수 없습니다. null `delivery_id`는 로컬 캐시 acknowledgment를 허용합니다. 녹음, playback, mute, 실행 중인 턴, 세션 종료, 이미 활성인 delivery는 `claim()`을 막습니다. 안내 중 녹음/mute 또는 새 사용자 요청이 오면 adapter가 활성 delivery를 중단하고 멈춘 뒤 activity를 갱신하고 스케줄러를 깨워야 합니다.

`claim()`은 적합한 delivery 하나를 원자적으로 `pending`에서 `scheduled`로 바꿉니다. 준비 전과 **모든** 전달 전에 `accepts_output(delivery_id)`로 relevance를 확인하고 `turns.accept_event(event)`로 텍스트 이벤트를 승인하세요. ID/버전 변경, 사라진 job, 성공하지 않은 작업, 끝난 세션, 이미 coverage된 업데이트, 애플리케이션 `is_relevant(job)` 정책은 안내를 무효화합니다. callback은 부수 효과가 없어야 합니다. 기존 job 계약 밖의 애플리케이션 취소/교체 정책에 사용하세요. activity 또는 relevance가 바뀌면 delivery task를 깨우세요. 건너뛰어도 보존한 결과는 삭제하지 않습니다.

요청한 요약이 실제 전달된 뒤 `mark_covered(CoveredUpdate.from_job(job))`은 승인된 버전을 기록하고 일치하는 pending 안내를 건너뜁니다. 요약 요청만으로 delivery가 되지 않습니다. 이후 job 버전은 계속 적합합니다. 상태 응답이나 관련 없는 대화로 context의 모든 job을 coverage 처리하지 마세요.

`settle(delivery_id, status, reason=...)`은 활성 delivery만 한 번 해제합니다. 최종 상태는 `completed`, `interrupted`, `skipped`, `failed`입니다. 생성 완료만으로 delivery가 완료되지는 **않습니다**. adapter 경계에서 feedback의 세션, 턴, input revision, delivery ID를 검증하고 출력 생성이 끝난 뒤에만 completion을 받으세요. 전체 메시지 adapter completion은 스케줄링 관측이며 들린 단어의 증거가 아닙니다. 중단, 실패, 무효화 시 로컬 playback과 진행 중인 adapter delivery를 멈추고 남은 공개 출력을 차단하며 결과를 보존하세요. 중단/실패 안내는 자동 재생하지 않고 사용자가 새 요약을 요청할 수 있습니다. TTS 실패는 job 성공을 바꾸지 않습니다.

애플리케이션 delivery task를 취소하고 기다리기 전에 `close()`를 호출하세요. pending/active delivery를 차단하고 job 결과를 보존합니다. snapshot에는 발언권 관측과 delivery 기록이 있고 timer, media 연결, 재개 가능한 lease는 없습니다. 복원은 delivery를 재시작하지 않습니다. 활성 기록은 애플리케이션 recovery 정책으로 조정하세요. segment receipt, 정확한 spoken context, 자동 job 취소/교체, 분산 스케줄링, playback timeout은 별도 기능입니다. completion을 보내지 않는 client는 보수적으로 중단 또는 종료까지 발언권을 유지합니다. 애플리케이션이 명시적인 timeout 정책을 추가할 수 있습니다.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.