대화형 Flow

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 링크

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

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

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

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

def kickoff() -> None:
    SupportFlow().chat()

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",
)

@listen("INTERNET_SEARCH")
def handle_internet_search(self) -> str:
    """Fresh web research, current news, real-time lookups."""
    ...

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

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

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

flow = SupportFlow()
flow.chat()

flow.chat(
    session_id="demo-session",
    prompt="You: ",
    assistant_prefix="Assistant: ",
    exit_commands=("exit", "quit", "bye"),
)

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)

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

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

flow.chat(session_id=session_id)

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

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

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

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"],
)

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