对话式 Flows

概述

对话式应用会把每一行用户输入视为一次新的 flow 运行，但使用相同的 session id。CrewAI 为消息历史、可选的意图路由、延迟 tracing、UI 桥接，以及本地 flow.chat() REPL 提供了辅助能力，方便你构建对话式 flows。

概念	实现
Session id	handle_turn(..., session_id=...) → kickoff(inputs={"id": ...}) → state.id
用户输入	handle_turn(message) 会在图运行前把消息追加到 state.messages
一轮完成	只表示本次运行的 FlowFinished；聊天会在下一次 handle_turn 继续
全会话 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= 关键字参数。对于对话式 flows，handle_turn() 会存储待处理消息，并在重置每轮执行状态后，内部调用 kickoff(inputs={"id": session_id})。

API	用途
handle_turn(message, session_id=...)	面向对话式 Flow 的便捷单轮包装器
chat()	对话式 Flow 的本地终端 REPL
kickoff(inputs={...})	不处理对话轮次的高级 flow 执行
ask()	在单个步骤内部使用的阻塞式提示（向导、澄清）
@human_feedback	审批/拒绝步骤输出 - 不是下一条聊天消息
ChatSession.handle_turn(...)	基于 handle_turn 的传输层（SSE / WebSocket）

快速开始

from uuid import uuid4

from crewai import Flow
from crewai.flow import listen
from crewai.experimental.conversational import (
    ConversationConfig,
    ConversationState,
)


@ConversationConfig(defer_trace_finalization=True)
class SupportFlow(Flow[ConversationState]):
    conversational = True

    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()  # 为整个聊天保留一条 trace 链接

每轮生命周期

每次 handle_turn 都会执行以下流程：

1. 轮次准备 - 存储待处理的用户消息、解析 session 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. 图执行 - conversation_start → route_conversation → 被选中的 @listen 处理器。
6. 运行结束 - 当启用延迟时，会跳过每轮的 flow_finished 和 trace 最终化；嵌套的 Agent.kickoff() / crews 也不会提前关闭父 batch。

处理器应该调用 append_assistant_message(reply)，这样下一轮的 conversation_messages 才能包含助手文本。用户消息已经由 handle_turn 存储了 - 不要在处理器里再追加一遍。

ConversationConfig（类级默认值）

用 ConversationConfig 装饰你的对话式 Flow 子类。

字段	默认值	作用
system_prompt	Framework 默认值	内置 converse_turn 使用的系统消息。
llm	None	对话 LLM，供 converse_turn 和路由 fallback 使用。
router	None	用于 LLM 驱动路由的 RouterConfig。
intent_llm	None	用于 intents= / default_intents 预分类的 LLM。
default_intents	None	预分类的结果标签。
defer_trace_finalization	True	在多次 handle_turn() 调用之间保持一个 trace batch 打开。

你可以通过 handle_turn(..., intents=..., intent_llm=...) 覆盖每轮的预分类。

底层 ChatState 辅助能力

ChatState、ConversationalConfig 和 crewai.flow.conversation 辅助函数仍然可以被导入，用于高级编排、测试或自定义包装器。它们不会为 Flow.kickoff() 添加 user_message= 或 session_id= 关键字参数。

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	一次性的启动标志（权限、缓存等）

ConversationalInputs 是一个 TypedDict，用于 kickoff(inputs={...}) 的常规键：id、user_message、last_intent。

Flow 的对话式 API

handle_turn 参数

参数	作用
message	本轮的文本
session_id	Conversation UUID → inputs["id"] / state.id
intents	预 kickoff 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。当你需要原始 flow 执行时，传入 inputs={"id": session_id}；但当调用代表聊天消息时，应使用 handle_turn()。

实例属性

属性	作用
conversational	设为 True 以启用对话图和 handle_turn()
defer_trace_finalization	实例级标志；在 handle_turn() 时会根据配置自动设置
suppress_flow_events	隐藏控制台 flow 面板；tracing 仍然会记录方法/flow 事件
stream	启用流式输出；与 ChatSession.handle_turn(..., stream=True) 配合使用

方法与属性

名称	描述
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 并最终化会话 trace batch
_should_defer_trace_finalization()	当前 flow 是否延迟每轮 trace 最终化
input_history	ask() 提示及回复的审计轨迹

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

可供测试或自定义编排导入：

函数	描述
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=...)	用于自定义包装器的低层轮次 hydration
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() 中读取 self.state.last_intent。

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

当 flow 结束但用户继续聊天

FlowFinished 表示这次图运行已经完成。对话会在下一个 handle_turn() 中继续，并沿用同一个 session_id。@persist 会恢复 messages、标志位和上下文。

持久化模式： 更推荐把 @persist 放在单个终止步骤上（例如 finalize），而不是整个 Flow 类上。类级持久化会在每个方法后保存；load_state 读取的是最新的一条记录，而那条记录可能是运行中的中间快照（例如 bootstrap 刚结束后），从而错过同一轮中的处理器更新。

除非有人工必须批准某个具体步骤输出后才能展示，否则不要把 @human_feedback 用于后续聊天消息。

对话式 Flow（实验性）

Warning

这是一个实验性功能。 对话式 Flow 接口 (conversational = True、handle_turn、ConversationConfig、 RouterConfig、ConversationState、内置图和辅助函数) 位于 crewai.experimental 下，形态在正式发布前可能会继续变化。 如果你依赖特定行为，请锁定 CrewAI 版本，并关注 changelog 中的破坏性更新。欢迎提交问题与反馈。

通过在 Flow 子类上设置 conversational = True，即可启用对话图。基础 Flow 会提供内置的 @start / @router / converse_turn / end_conversation 图，管理 state.messages，可以驱动 router LLM，并在多轮之间保持 trace batch 打开。你负责编写自定义路由，其余部分由框架承担。

当你想要一个带路由器和按路由处理器的多轮聊天，但不想自己编写生命周期时，请使用这个模式。需要完全控制时，请使用上面更底层的 Flow[ChatState] 模式。

快速示例

from crewai import Flow
from crewai.flow import listen
from crewai.experimental.conversational import (
    ConversationConfig,
    ConversationState,
)


@ConversationConfig(defer_trace_finalization=True)
class SupportFlow(Flow[ConversationState]):
    conversational = True

    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:
        """新鲜的网页调研、最新新闻、实时查询。"""
        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:
        """查找 CrewAI 文档，以回答框架/API 问题。"""
        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?")              # 路由到 converse
    flow.handle_turn("Search the web for AI news.")   # 路由到 INTERNET_SEARCH
    flow.handle_turn("Check the CrewAI docs.")        # 路由到 CREWAI_DOCS
finally:
    flow.finalize_session_traces()

对于本地终端聊天，请使用 chat()：

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

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

ConversationConfig

类装饰器，用于附加每类聊天默认值。

字段	默认值	作用
system_prompt	来自 i18n 的 slices.conversational_system_prompt	内置 converse_turn 使用的系统消息。传入 "" 可完全关闭。
llm	None	对话 LLM（供 converse_turn 和路由 fallback 使用）。
router	None	用于 LLM 驱动路由的 RouterConfig。没有它时，flow 总会落到 converse。
answer_from_history_prompt	Framework 默认值	可选 answer_from_history 路由使用的系统消息。
answer_from_history_llm	None	设定后启用 answer_from_history 快速通道。
intent_llm	None	用于旧版 intents=/default_intents 预分类的 LLM。
default_intents	None	旧版预分类的结果标签。
visible_agent_outputs	None	"all"，或一个 agent 名称列表；这些 agent 的 append_agent_result() 会被提升为公开助手消息。
defer_trace_finalization	True	在多次 handle_turn() 调用之间保持一个 trace batch 打开。

RouterConfig 与自动生成的路由目录

from typing import Literal

from pydantic import BaseModel

from crewai import LLM
from crewai.experimental.conversational import RouterConfig


class MyRoute(BaseModel):
    intent: Literal["INTERNET_SEARCH", "CREWAI_DOCS", "converse"]


ROUTER_LLM = LLM(model="gpt-4o-mini")

router_config = RouterConfig(
    prompt="可选的领域框架（策略、语气、人格）。",
    response_format=MyRoute,        # 可选；否则自动生成
    llm=ROUTER_LLM,                  # 回退到 ConversationConfig.llm
    routes=["INTERNET_SEARCH", "CREWAI_DOCS"],   # 可选；可从监听器推断
    route_descriptions={
        "INTERNET_SEARCH": "覆盖这个单一路由的 docstring。",
    },
    default_intent="converse",      # LLM 调用失败或不可用时使用
    fallback_intent="converse",     # LLM 返回无效路由时使用
    intent_field="intent",
)

发送给 LLM 的 router prompt 会自动构建。对于每个路由，框架会按以下优先级选择描述：

1. RouterConfig.route_descriptions[label] - 显式覆盖。
2. Flow.builtin_route_descriptions[label] - 框架为 converse、end、answer_from_history 提供的内置文本（已改写以适配 router LLM）。
3. @listen(label) 处理器 docstring 的第一行非空内容。
4. 空白（该路由会被列出，但没有描述）。

因此，实际上新增一个路由就是 @listen("X") + 一行 docstring：

from crewai.flow import listen


@listen("INTERNET_SEARCH")
def handle_internet_search(self) -> str:
    """新鲜的网页调研、最新新闻、实时查询。"""
    ...

而 router LLM 会看到：

Routes:
- CREWAI_DOCS: 查找 CrewAI 文档，以回答框架/API 问题。
- INTERNET_SEARCH: 新鲜的网页调研、最新新闻、实时查询。
- converse: 普通聊天、追问、总结、澄清...
- end: 用户表明对话结束（再见、退出、完成）。

RouterConfig.prompt 用于领域框架（assistant persona、业务规则、语气）。路由目录是自动构建的 - 不要在 prompt 里手动列出 routes；一旦你添加了处理器，它们就会过时。

内置路由

路由	处理器	作用
converse	converse_turn	默认聊天处理器。使用 ConversationConfig.llm，结合系统提示和规范化消息历史。
end	end_conversation	将 state.ended = True 并发出终止回复。
answer_from_history	answer_from_history_turn	可选。启用 ConversationConfig.answer_from_history_llm 时，如果消息可以从已有历史中回答，就会路由到这里。

你可以通过在子类中定义同名处理器来覆盖这些内置路由。

handle_turn() 语义

flow.handle_turn(message) 运行一轮：

1. 重置每次执行的跟踪（_completed_methods、_method_outputs），以便图重新运行 - 否则对同一个 flow 实例重复调用 kickoff 时，第二轮之后会短路，因为 Flow.kickoff_async 会把 inputs={"id": ...} 当作 checkpoint 恢复。
2. 把用户消息追加到 state.messages，设置 current_user_message / last_user_message。last_intent 会保留上一轮的值，这样 router LLM 可以把它当作信号使用。
3. 运行 conversation_start → route_conversation → 选中的 @listen 处理器。
4. router 会把决策存入 state.last_intent（下一轮 router 上下文可见）。
5. 如果你的处理器返回了字符串，并且没有先调用 append_assistant_message，handle_turn 会替你追加它。

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

面向本地 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 块中最终化延迟的会话 traces。

你可以用可注入的 I/O 自定义终端行为：

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

对于 Web 应用、后台 worker、测试和自定义传输，请继续直接使用 handle_turn()。

自定义 router 行为

如果你想在每次路由决策时执行副作用（事件总线初始化、遥测等），可以覆盖 route_turn：

from typing import Any

from crewai import Flow
from crewai.experimental.conversational 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 返回字符串；返回 None 则会回退到 _route_with_config(...)。

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 可以在全局范围内把特定 agent 的私有结果提升为公开结果（"all" 或一个 agent 名称列表）。

跨轮 tracing

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

• 整个聊天会话只有 一个 trace batch。
• flow_started 只在第一轮出现；flow_finished 只会在 finalize_session_traces() 中发出一次。
• 每轮的 kickoff 不会打印“Trace batch finalized”。
• 嵌套工作（Agent.kickoff()、crews、Exa tools）会追加到父 batch；内部的 AgentExecutor flows 不会过早关闭会话 batch。

flow.chat(session_id=session_id)

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

suppress_flow_events=True 只会隐藏 Rich 控制台面板；trace 和 method 事件仍会发出，用于可观测性。

对话式 Flow 的 trace 生命周期

实验性的 对话式 Flow 使用相同的 tracing 生命周期：defer_trace_finalization 默认仍然是 True，因此每个 handle_turn() 都会保持会话 trace 打开。一定要在会话结束时最终化它 - 把你的 REPL / 循环放在 try/finally 中，并在退出时调用 flow.finalize_session_traces()。否则，trace batch 会一直保持打开，最终对话可能永远不会导出。

流式输出

把 Flow 类上的 stream 设为 True。这样 kickoff(...) 就会通过标准事件总线发出 assistant_delta（以及相关）事件。

导入

from crewai.flow import (
    ChatState,
    ConversationalConfig,
    ConversationalInputs,
    Flow,
    listen,
    persist,
    router,
    start,
)

另见

• 精通 Flow 状态管理 - 持久化、Pydantic state、@persist
• 构建你的第一个 Flow - Flow 基础
• Demo: lib/crewai/runner_conversational_flow_simple.py - 具有 RESEARCH + Exa agent 的最小 REPL