对话式 Flow

使用 handle_turn 逐轮处理、消息历史、意图路由、追踪和结构化流式传输，构建多轮聊天应用。

概述

对话式应用把每一行用户输入视为一次新的流程运行，但使用相同的会话 id。CrewAI 提供了消息历史、可选的意图路由、延迟追踪、结构化轮次流式传输，以及本地 flow.chat() REPL 的辅助功能。

概念	实现
会话 id	handle_turn(..., session_id=...) → kickoff(inputs=""id": ...") → state.id
用户输入行	handle_turn(message) 在图运行前追加到 state.messages
轮次完成	conversation_turn_completed；在默认的追踪延迟下，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= 关键字参数。对于对话式流程，handle_turn() 会存储待处理消息，并在重置每轮执行状态后，在内部调用 kickoff(inputs=""id": session_id")。

API	用于
handle_turn(message, session_id=...)	对话式 Flow 的易用单轮封装
stream_turn(message, session_id=...)	将单个对话轮次作为有序的运行时帧进行流式输出
chat()	对话式 Flow 的本地终端 REPL
kickoff(inputs="...")	不涉及对话轮次处理的高级流程执行
ask()	在单个步骤内部进行阻塞式提示（向导、澄清）
@human_feedback	批准/拒绝某个步骤的输出——而不是下一行聊天输入

除非启用对话模式，否则 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()  # one trace link for the whole chat

流式输出单个轮次

当 UI 或运行时需要单个聊天轮次的结构化事件时，使用 stream_turn()。它返回一个流会话，其中包含流程路由、LLM 分块、工具活动和对话消息的有序帧。

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

完整的帧契约和通道列表，请参阅流式运行时契约。

轮次生命周期

每次 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（内置的启动/路由）→ 选定的 @listen 处理器。route_conversation 还会调用可覆盖的 conversation_start() 辅助方法。
6. 运行结束 — 启用延迟时，每轮的 flow_finished 和追踪终结会被跳过；嵌套的 Agent.kickoff() / 团队也不会关闭父批次。

当可见回复不是返回值，或者你裁剪历史时，处理器应调用 append_assistant_message(reply)。公开的字符串返回值也会被记录为助手消息并包含在 @persist 快照中，因此新的 Flow 实例可以恢复它。用户输入行已由 handle_turn 存储——不要在处理器中再次追加。

配置概览

用 ConversationConfig 装饰 Flow 子类，既会附加聊天默认值，也会启用对话模式。参见下方的完整字段参考。可通过 handle_turn(..., intents=..., intent_llm=...) 逐轮覆盖预分类。

更底层的 ChatState 辅助方法

ChatState、旧的 ConversationalConfig 以及 crewai.flow.conversation 中的辅助方法仍可导入，用于高级编排、测试或自定义封装。它们与 ConversationState / ConversationConfig API 相互独立，并且不会为 Flow.kickoff() 增加 user_message= 或 session_id= 关键字参数。

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

字段	作用
id	会话 UUID（与 inputs["id"] 相同）
messages	用于 LLM 历史的 {role, content} 的 list
last_user_message	本轮最新的用户输入行
last_intent	分类后的路由标签（如果使用）
session_ready	一次性引导标志（权限、缓存等）

ConversationalInputs 是一个 TypedDict，用于常规的 kickoff(inputs="...") 键：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	用于启动前 classify_intent 的结果标签
intent_llm	用于分类的 LLM（与 intents 一起使用时必填）
**kickoff_kwargs	透传给 kickoff()，用于 input_files、from_checkpoint 和 restore_from_state_id 等选项

kickoff 参数

Flow.kickoff() 接受 inputs、input_files、from_checkpoint 和 restore_from_state_id。当你需要原生流程执行时，传入 inputs=""id": session_id"；但当该调用表示一条聊天消息时，应改用 handle_turn()。

实例属性

属性	用途
conversational	设为 True 以启用会话图与 handle_turn()
defer_trace_finalization	可选的实例级覆盖。否则 _should_defer_trace_finalization() 会读取 ConversationConfig.defer_trace_finalization。
suppress_flow_events	隐藏控制台的流程面板并抑制方法执行事件；流程开始/结束事件仍会发出
stream	通用的 Flow 流式传输标志。对于会话轮次，请使用 stream_turn()，而不要将此标志与 handle_turn() 组合使用。

方法与属性

名称	描述
append_assistant_message(content)	向 state.messages 追加一条用户可见的助手回复
append_message(role, content, **extra)	更底层的向 state.messages 追加
conversation_messages	供 LLM 调用使用的只读历史记录
classify_intent(text, outcomes, *, llm, context=None)	将文本映射到某个结果（与 @human_feedback 相同的归并逻辑）
receive_user_message(text, *, outcomes=None, llm=None)	追加用户消息；可选地设置 last_intent
finalize_session_traces()	发出被延迟的 flow_finished 并完成会话追踪批次的收尾
_should_defer_trace_finalization()	高级/内部钩子，用于判定是否延迟每轮次的追踪收尾
input_history	ask() 提示词与响应的审计记录

模块辅助函数（crewai.flow.conversation）

可从 crewai.flow.conversation 导入，用于测试或自定义编排。这些辅助函数使用旧版的 ConversationalConfig 结构；prepare_conversational_turn() 还会清除 last_intent，而 handle_turn() 会将其保留为路由上下文。

函数	描述
normalize_kickoff_inputs(inputs, user_message=..., session_id=...)	将会话关键字参数合并进 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)	在字典或 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 使用当前轮次分类出的意图。

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")（或类似写法）来运行带工具的 Agent.kickoff()，而不是裸的 LLM.call()。

流程已结束但用户仍在继续聊天时

每次 handle_turn() 完成一次图运行，然后使用相同的 session_id 通过另一次 handle_turn() 继续对话。在默认的延迟追踪生命周期下，该次运行会发出 conversation_turn_completed，而 FlowFinished 只在 finalize_session_traces() 关闭会话时发出一次。@persist 会恢复 messages、标志和上下文。

持久化模式： 优先在单个终结步骤（例如 finalize）上使用 @persist，而不是在整个 Flow 类上使用。类级别的持久化会在每个方法之后保存；load_state 使用最新的一行，而那可能是运行中途的快照（例如紧接在 bootstrap 之后），从而遗漏同一轮次中处理器的更新。

不要将 @human_feedback 用于后续聊天内容，除非必须由人工在某个步骤输出展示之前予以批准。

会话式 Flow

在 Flow 子类上设置 conversational = True 或应用 @ConversationConfig(...)，即可启用会话式聊天图。此时基础 Flow 会提供内置的启动/路由 route_conversation，以及 converse_turn 和 end_conversation 监听器。已弃用的 answer_from_history_turn 监听器为兼容性保留可用。框架负责管理 state.messages、可驱动路由 LLM，并在多个轮次间保持追踪批次开启。你只需编写自定义路由，其余由框架负责。

当你想要一个带路由和按路由处理器的多轮聊天，而不必自己接好生命周期时，请使用此方式。当你需要完全控制时，请使用 Flow[ChatState]（上面那种更底层的模式）。

快速示例

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

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

chat() 在 REPL 中包装 handle_turn()，遇到 exit / quit 时退出，默认跳过空行，并在会话结束时调用 finalize_session_traces()。

ConversationConfig

类装饰器，用于为每个类附加聊天默认配置。

字段	默认值	用途
system_prompt	来自 i18n 的 slices.conversational_system_prompt	内置 converse_turn 使用的系统消息。传入 "" 可完全退出。
llm	None	会话 LLM（供 converse_turn 使用，并作为路由器的回退）。
router	None	可选的 RouterConfig 覆盖项。在存在自定义监听器且 LLM 可解析的情况下，即使省略此项也会自动启用路由。
answer_from_history_prompt	框架默认值	已弃用。 请使用 converse 系统提示词或覆写 converse_turn()。
answer_from_history_llm	None	已弃用。 请使用 llm；converse 已经接收规范历史记录。
intent_llm	None	用于旧版 intents=/default_intents 预分类的 LLM。
default_intents	None	用于旧版预分类的结果标签。
visible_agent_outputs	None	"all"，或一组智能体名称，其 append_agent_result() 调用应被提升为公开的助手消息。
defer_trace_finalization	True	在多次 handle_turn() 调用之间保持一个追踪批次开启。

WARNING

answer_from_history_prompt、answer_from_history_llm 以及 answer_from_history 路由已弃用，将在未来版本中移除。它们与 converse 重复， 增加了一次资格判断的 LLM 调用，并且在常规自动路由返回路由时会被绕过。 现有配置仍可正常工作，并会发出 DeprecationWarning。

在没有自定义路由的情况下，轮次会落到 converse。在存在自定义路由且有会话/路由器 LLM 的情况下，框架会合成一个默认 RouterConfig；只有需要自定义其提示词、路由列表、描述或回退行为时才需显式提供。设置 default_intents 则会改用旧版的预分类路径。

如果未配置会话 LLM，内置的 converse_turn 会返回一个配置占位符，而不是生成答案。

RouterConfig 与自动构建的路由目录

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

发送给 LLM 的路由器提示词会自动构建。对于每条路由，框架按以下优先级选取描述：

1. RouterConfig.route_descriptions[label] —— 显式覆盖。
2. Flow.builtin_route_descriptions[label] —— 框架预置文本，用于 converse、end 以及已弃用的 answer_from_history 兼容路由（措辞面向路由器 LLM）。
3. 方法声明的 description（供声明式流程和 DSL 投影使用）。
4. @listen(label) 处理器文档字符串中第一个非空行。
5. 空（该路由被列出但不带描述）。

因此实际上，新增一条路由就是 @listen("X") + 一行文档字符串：

from crewai.flow import listen

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

处理器命名

@listen("…") 中的字符串是路由器路由标签（一个事件名），而不是 Python 方法名。路由标签和方法完成事件共享同一个触发命名空间，因此若把处理器命名得与其路由相同，就会导致该处理器自我循环触发。

请使用不同的方法名 —— 文档示例使用 handle_* 前缀：

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

……这样路由器 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	默认聊天处理器。使用系统提示词 + 规范消息历史调用 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），使图重新运行 —— 若不这样做，在同一 flow 实例上重复调用 kickoff 会在第 2 轮及以后短路，因为 Flow.kickoff_async 会把 inputs=""id": ..." 视为检查点恢复。
2. 将用户消息追加到 state.messages，设置 current_user_message / last_user_message。last_intent 会从上一轮保留下来，以便路由器 LLM 将其用作信号。
3. 运行用户定义的 @start 方法（如有），然后运行作为内置启动/路由的 route_conversation，再运行选中的 @listen 处理器。route_conversation 会调用可覆写的 conversation_start() 辅助方法。
4. 路由器将其决策存入 state.last_intent（下一轮的路由上下文可见）。
5. 如果你的处理器返回了字符串且未自行调用 append_assistant_message，handle_turn 会替你追加它，并持久化更新后的 state.messages，从而让 @persist 恢复包含该助手轮次。

对于聊天消息，请调用 handle_turn()。直接调用 kickoff(inputs=""id": ...") 会在不应用会话轮次包装的情况下运行流程图。

用于本地 REPL 的 chat()

flow.chat() 是围绕 handle_turn() 的开箱即用终端包装器：

flow = SupportFlow()
flow.chat()

它负责常见的本地循环：

1. 提示输入用户消息。
2. 在 exit / quit、EOFError 或 KeyboardInterrupt 时停止。
3. 调用 handle_turn(message, session_id=...)。
4. 打印助手结果。
5. 在 finally 块中收尾被延迟的会话追踪。

chat(defer_trace_finalization=True) 会为 REPL 临时启用实例延迟标志，并在退出时恢复其先前的值。

通过可注入的 I/O 自定义终端行为：

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

对于 Web 应用、后台工作进程、测试和自定义传输方式，请继续直接使用 handle_turn()。

自定义路由器行为

若要在每次路由决策时执行副作用（事件总线设置、遥测），请覆写 route_turn：

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 路由并以编程方式选择路由，可从 route_turn 返回一个非空字符串。返回假值不会调用你覆写中的 _route_with_config()；路由会回落到本轮预先分类的意图，然后在已配置时回落到已弃用的 answer_from_history 兼容路径，最后回落到 converse。上一轮的 last_intent 在路由上下文中可用，但绝不会作为回退被重放。

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] 中记录一个线程。公开可见性还会替你调用 append_assistant_message。对于不应污染规范历史的草稿性工作，请使用私有结果。

ConversationConfig.visible_agent_outputs 可以在全局范围内将特定智能体的私有结果提升为公开（"all"，或智能体名称列表）。

在 JSON/YAML 中声明对话流程

声明式 Flow 也可以是对话式的。添加一个顶层 conversational 块，并把你的路由声明为 listen 某个路由标签的方法：

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 可保留配置但关闭聊天。这同时也会禁用内置方法合成，因此该声明必须提供一个普通的非对话式图。

以下三项由系统自动提供：

提供项	详情
内置图	route_conversation、converse_turn 和 end_conversation 会自动添加。为兼容性保留了已弃用的 answer_from_history_turn。在其中一个名称下声明方法即可覆写它。
对话状态	当没有 state 块时使用 ConversationState。Pydantic ref 或 json_schema 状态会自动与对话字段组合；它无需继承 ConversationState。
路由目录	由带 listen 标签的非路由方法推断，排除内部路由。描述遵循上述优先级，显式的 router.routes 可以限制可选项。

声明式的 llm、router.llm 和 intent_llm 字段接受模型 id 或配置映射，例如 {model: openai/gpt-4o-mini, max_tokens: 512}。conversational 块还支持 default_intents、visible_agent_outputs、defer_trace_finalization 以及上面展示的 RouterConfig 字段。为兼容性，仍接受已弃用的 answer_from_history_prompt / answer_from_history_llm 声明。

用与基于类的对话 Flow 相同的轮次 API 从 Python 运行它：

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

命名路由

路由标签和方法名共享同一个触发器命名空间，因此处理程序不能以其监听的路由命名 —— create_video 监听 create_video 在构建 Flow 时会被拒绝。请使用 handle_* 前缀。

声明方式无法表达的内容

无法表达	改用
实时的 LLM 实例或自定义 BaseLLM	模型 id 字符串或静态配置映射
将 router.response_format 作为实时模型类	用 python ref 指定类：response_format: {python: my_project.schemas.ConversationRoute}。省略它则由框架合成一个
route_turn() 覆写	用 Python 编写该 Flow，或用 call: code / 表达式动作替换声明式的 route_conversation 方法
can_answer_from_history() 覆写	已弃用。请使用 converse，或在 Python 中覆写 converse_turn()。

crewai run 会为声明式对话 Flow 打开聊天 TUI —— 与 Python 对话 Flow 得到的相同。聊天循环需要终端，因此无头运行会以非零状态退出并给出指引，而不是运行单轮；在这种情况下请从 Python 用 handle_turn() 或 stream_turn() 驱动它。带 human_feedback: 块的声明式方法（Python 中为 @human_feedback）会在终端 REPL 上运行，因为运行时会用 TUI 无法处理的阻塞式提示词收集反馈。对话 Flow 不接受 --inputs —— 每轮的输入就是你输入的消息 —— 而且按 id 恢复会话尚未接入 CLI；请从 Python 使用 flow.handle_turn(message, session_id=...) 实现。

跨轮次追踪

在 defer_trace_finalization=True（ConversationConfig 中的默认值）时：

• 整个聊天会话一个追踪批次。
• flow_started 仅在第一轮触发；flow_finished 在 finalize_session_traces() 中触发一次。
• 每轮的 kickoff 不会打印“Trace batch finalized”。
• 嵌套工作（Agent.kickoff()、团队、Exa 工具）会追加到父批次；内部 AgentExecutor 流程不会提前关闭会话批次。

flow.chat(session_id=session_id)

flow.chat() 会替你调用 finalize_session_traces()。当你用 handle_turn() 自己掌控循环时，请在会话结束时 调用 finalize_session_traces()。

suppress_flow_events=True 会隐藏 Rich 控制台面板并抑制方法执行事件。Flow 启动/结束事件仍会发出，因此外层 Flow 生命周期仍可追踪，但单个方法 span 会被省略。

对话式 Flow 的追踪生命周期

对话式 Flow 使用相同的追踪生命周期：defer_trace_finalization 默认为 True，因此每次 handle_turn() 都会保持会话追踪开启。延迟的轮次也会抑制每轮的 flow_failed；在轮次出错或会话中止时，请显式终结会话。这会以会话级的 FlowFinished 事件关闭批次，而不是每轮的 FlowFailed 事件。始终用 try/finally 包裹你的 REPL/循环，并在退出时调用 flow.finalize_session_traces()。否则追踪批次会保持开启，最终的对话可能永远不会导出。

流式传输

对于对话式 UI，请使用 stream_turn() 并迭代其有序的 StreamFrame 对象：

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() 掌管对话式流式传输生命周期。

导入

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 状态管理 —— 持久化、Pydantic 状态、@persist
• 构建你的第一个 Flow —— Flow 基础

跨对话轮次的实验性作业

该 API 是实验性的，可能发生变化。请显式从 crewai.experimental.flow_jobs 导入；它不属于稳定的 crewai.flow 接口。

JobRecord、JobState、JobUpdate、JobWorkState、JobWorkFlow 和 JobRunner 提供可选的内存作业契约。现有的对话 Flow 保持其行为不变。作业拥有自己的标识、起始轮次、修订版本、执行尝试、状态、当前阶段、已完成阶段和错误。应用自行添加类型化的输入和输出；无需单独的产物类。

在 output_fields 中声明可写的结果字段。工作者更新不能通过输出负载更改输入、归属或生命周期字段。完整候选记录在提交前会被校验；格式错误、非本作业、过期、重复以及终结后的更新都会使已接受的状态保持不变。

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 事件循环上构造 JobRunner(state, make_work, on_update=publish_snapshot)。make_work(job, inputs, publish) 返回一个非流式的 JobWorkFlow，带有自己的类型化状态、复制的 job/inputs 以及 publish=publish。on_update 是一个异步回调，接收 JSON 兼容的快照（session_id、seq、jobs）；将它们适配到你现有的传输方式。format_error 可以脱敏服务商细节，on_worker_event 则观察工作者的启动/结算。

对每一次前台轮次、作业准入以及父状态变更都持有 runner.state_lock。先用 add_job(state, job) 注册，然后调用 runner.submit(job, inputs)。在持有该锁的同时，在工作线程中运行同步轮次；绝不要在一个 Flow 上并发运行两个轮次。工作者的 publish_update() 同样在 runner 的事件循环之外运行，并会等待提交被接受后才进入下一阶段。stage_started 更新要求上一阶段已提交；stage_completed 会在成功完成前保留类型化输出。后续失败会保留先前的输出。

通过 build_router_context() 和 build_agent_context() 投影相关的作业状态与输出。快照不会自动成为公开聊天消息或语音回复。领域解释和准入容量仍由应用策略决定。

在会话拆除时，先在其所属循环上调用 runner.request_close()，然后排空前台执行，再 await runner.aclose()。这会关闭准入/发布、通知每个工作者、释放等待中的发布者并结算其拥有的任务。工作方法必须在阶段边界调用 check_open()。进行中的服务商操作不会被中止。订阅者失败会关闭 runner，而不是让发布者搁浅。仅停止语音不应关闭 runner。

初始生命周期为 queued → running → completed/failed。不提供暂停/恢复、替换、自动回复投递和分布式恢复。记录可以用现有的状态设施序列化，但恢复它们并不会重建活动工作者，也不保证外部效果恰好一次。

实验性的轮次与公开回复标识

请显式从 crewai.experimental.flow_turns 导入 TurnRecord、ReplyRecord、TurnState、ReplyEvent 和 TurnRunner。这一可选 API 可能发生变化。它包装现有的对话 Flow；稳定的 handle_turn()/stream_turn() API、返回值以及实验性作业契约均不变。

TurnState 在 ConversationState 基础上扩展了可序列化的 turns 和 replies。一次 TurnRunner.stream_turn() 调用会跟踪一条已提交的输入及其主要的公开回复。每条被接受的输入都有一个会话范围内、一次性使用的 turn_id；新的语句或纠正会获得新的轮次 ID。input_revision 是一个正整数（默认 1），并非自动的临时转录替换。delivery_id 是唯一的回复标识；不存在第二个 reply_id。回复类型有 acknowledgment、answer 和 progress。

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

硬编码回复和仅返回最终结果的提供方发出与流式响应器相同的公开 text/completed 契约。现有 Flow 仍负责规范历史。seq 在一次投递内对事件排序；segment_id 从 1 开始对文本增量编号。文本增量不一定是一个完整的口语句子。路由和生成观测事件包含相同的标识。completed 表示文本生成结束，并非表示音频已播放或被听到。

只有专用的公开响应方法（默认为 converse_turn 和 answer_from_history_turn），或显式选定的公开智能体 ID，才允许实时模型文本。请审慎配置 public_methods 和 public_agent_ids 回调；不要指定将私有模型工作与公开响应生成混在一起的方法。路由流、推理、工具调用分块、启用工具的调用以及无关智能体都被排除。私有智能体结果不会变成回复。没有可观测增量的提供方仍会发布其规范最终消息；不会虚构模型时序。

runner.accept_event(event) 会在传输交接前认领一个由 runner 发出的事件。它会拒绝外来标识、重复/乱序的序列，以及因中断或失败而失效的输出。在经历任何应用层队列/延迟之后、交接之前立即调用它。它不是用于不可信客户端播放回执的校验器。已交接的输出需要适配器侧的标识围栏和音频中断。

runner.interrupt(session_id=..., turn_id=..., input_revision=..., delivery_id=...) 精确针对一个回复，并且对有效的重复请求是幂等的。在生成期间，它将该轮次/回复标记为已中断；生成之后，它保留已完成的文本并将该回复标记为已中断，而不改变任务成功状态。on_interrupt 可以通知现有的协作式阶段检查；should_interrupt 可以观察应用已有的停止信号。这些回调不会中止任意的服务商调用。流必须先稳定下来，之后另一个受跟踪的轮次才能在同一 Flow 上执行。关闭迭代器会发出中断信号并排空原生执行；它可能需要等待一个在途的服务商调用。

要与后台工作组合，声明 class WorkChatState(TurnState, JobState[MyJob]): ...。在整个前台迭代以及所有父任务变更期间持有 JobRunner.state_lock。TurnRunner 单独保护其记录和中断控制。不要与包装器并发运行原生轮次 API。成功接纳任务后，runner.associate_job(job) 会将其会话/发起轮次以及修订/尝试关联到活动回复；它从不接纳、提交或取消工作。新的对话轮次不会使独立研究失效。

elapsed_ms 使用从受跟踪轮次启动开始的服务器单调时钟。model_first_text_ms（在可用时）是服务商请求开始与首个公开文本事件时间戳之间的差值；它包含服务商/SDK/网络的工作，并不是孤立的推理引擎 TTFT。切勿用浏览器时间戳减去服务器时间戳。TTS 首个数据块、缓存音频、物理可听起始点和播放完成仍属于适配器测量。

本次增量不提供后台回复队列、发言权调度、播放回执、精确词对齐、推测式生成或任务暂停/恢复。快照包含记录，而不是锁、取消信号或运行中的执行。它们不会恢复工作或重放投递。在创建新的受跟踪实时轮次之前恢复状态；from_checkpoint 和 restore_from_state_id 会被此包装器有意拒绝。终态跟踪记录的持久化协调与持久化仍是独立的工作。对于现有的检查点工作流，请继续使用稳定的 Flow API。

在受跟踪执行期间，当前恢复的状态仍是权威来源：自动会话重载被抑制，因此它不会丢弃已接纳的轮次/回复记录或已接受的后台任务更新。原生 Flow 轮次 API 保留其现有的持久化重载行为。在终态事件之后关闭不会中断已完成的回复。如果中断在失败之后到达，失败仍是可接受的终态生命周期信号。

实验性排队的后台回复

crewai.experimental.flow_replies 在轮次/回复契约之上增加了 ReplyQueueState、ReplyQueue、DeliveryRecord、DeliveryFloor、CoveredUpdate 和 ClientActivity。声明同时组合 ReplyQueueState 和 JobState[MyJob] 的状态，然后为该实时 Flow 创建一个 TurnRunner 和一个 ReplyQueue。现有 Flow 仍为可选启用；任务执行器和稳定的对话 API 不变。

成功的任务会立即在状态中可用，即使其播报还在等待。enqueue(job) 只接受具有现有发起轮次的当前已提交 completed 任务。待处理的投递会存储该任务的修订、尝试以及已接受的更新序列；重复通知会合并为同一个 delivery_id。该 ID 也是最终公开的 ReplyRecord 标识。每个任务完成都会获得一条可归属的播报；中间阶段进展不会创建发言队列。

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

使用与前台轮次和任务提交相同的 JobRunner.state_lock 来串行化入队、认领、准备、覆盖变更和输出交接。ReplyQueue 也用锁保护自身的控制，因此两个认领无法同时获得发言权。prepare() 读取复制的当前已接受产物，而不是入队时捕获的缓存文本，并发出共享的 started/text/completed 公开回复事件，附带来源和任务标识。它不执行 LLM、不追加历史，也不调度 TTS。应用负责选择有意公开的文本、历史插入、投递任务和传输。

set_foreground(True) 在前台生成和排空期间保持门控，包括待处理的用户请求；当这些工作稳定后释放它。observe_client(ClientActivity(...)) 只接受所属会话和递增的客户端序列，并带有严格布尔值的 recording、playback 和 muted 观测。带标识的播放必须属于该 Flow；针对不同回复的过期停止不能清除较新的播放。delivery_id 为 null 时可容纳本地缓存的确认。录制、播放、静音、运行中的轮次、会话关闭或已处于活动状态的投递都会阻止 claim()。如果在播报期间出现录制/静音或新的用户请求，适配器必须中断并停止该活动投递，然后更新活动状态并唤醒调度。

claim() 会原子地将一个符合条件的投递从 pending 变为 scheduled。在准备之前以及每次交接之前，用 accepts_output(delivery_id) 重新检查相关性，并用 turns.accept_event(event) 认领文本事件。标识/版本变化、任务缺失、未成功的工作、已结束的会话、已被覆盖的更新以及应用的 is_relevant(job) 策略都会使播报失效。该回调必须无副作用；将其用于现有任务契约之外的应用取消/取代逻辑。当活动状态或相关性变化时，唤醒投递任务。相关性跳过绝不会移除保留的产物。

在请求的摘要实际完成交接后，mark_covered(CoveredUpdate.from_job(job)) 会记录已接受的版本，并跳过匹配的待处理播报。仅仅请求摘要并不等于投递。后续的任务版本仍会符合条件。不要将状态响应或无关对话所覆盖的所有上下文任务都标记为已覆盖。

settle(delivery_id, status, reason=...) 只释放活动投递，且仅一次。其终态为 completed、interrupted、skipped 或 failed；仅生成完成并不能结算投递。在适配器边界验证反馈会话、轮次、输入修订和投递标识，并且只有在输出生成稳定后才接受完成。整条消息的适配器完成只是一个调度观测，并不是可听词语的证明。在中断、失败或失效时，停止本地播放和在途适配器投递，围栏剩余的公开输出，并让结果保持可用。中断/失败的播报不会自动重放；用户可以请求新的摘要。TTS 失败不会改变任务成功状态。

在取消并排空应用拥有的投递任务之前调用 close()。它会围栏待处理/活动投递，同时保留任务输出。快照数据包含发言权观测和投递记录，不包含计时器、媒体连接或可恢复的租约。恢复不会重启投递；通过应用恢复策略协调活动记录。分段级回执、精确的口述上下文、自动任务取消/替换、分布式调度和播放超时仍是独立功能。从不报告完成的客户端会保守地保留发言权，直到被中断或关闭；应用可以添加显式的超时策略。