build-flow

构建 CrewAI Flow

介绍如何编写一个 Flow，使其连接正确、保留类型化状态、支持持久化与恢复，并能调用 crews 和 agents，同时使用当前 API。

已于 2026-10-01 根据 crewai 1.15.22 和 1.15.23 验证。 已于 2026-10-01 在 CrewAI AMP 和真实大模型上进行实时测试。 先运行 crewai version；如果主版本号或次版本号与 1.15 不同，请先使用 ask-docs 技能重新验证对版本敏感的行，再依赖这些内容。

---

1. 搭建并运行 Flow 项目

crewai create flow lead-router      # the package is still lead_router
cd lead_router
crewai install                      # uv sync; writes uv.lock
crewai run                          # runs the `kickoff` script in pyproject.toml
crewai flow plot                    # runs the `plot` script

脚手架生成的内容：

文件	内容
pyproject.toml	[project.scripts] 中的 kickoff、run_crew（两者均指向 <pkg>.main:kickoff）、plot、run_with_trigger；[tool.crewai] type = "flow"
src/<pkg>/main.py	一个 Flow[ContentState] 子类，以及 kickoff() 和 plot() 函数；from crewai.flow import Flow, listen, start
src/<pkg>/crews/content_crew/	一个经典的 @CrewBase crew，包含 config/agents.yaml 和 config/tasks.yaml；其中的 agents 未设置 llm，因此使用默认的 OpenAI 模型
.env	OPENAI_API_KEY=YOUR_API_KEY——一个占位符；使用它运行会失败，并报错 401 ... Incorrect API key provided。在 shell 中导出的 OPENAI_API_KEY 优先级高于 .env：导出有效密钥后，未修改的脚手架通过 crewai run 在 gpt-4.1-mini 上运行了它的三个 agents，并写入了 output/post.md
AGENTS.md、CLAUDE.md、GEMINI.md（以及从 1.15.23 开始提供的 CURSOR.md）、README.md、tests/	编码助手说明和一个空的测试文件夹
.git/	crewai create 会运行 git init（不创建提交，也不设置远程仓库）

要使用其他提供商，请在 agents.yaml 中为每个 agent 设置 llm: anthropic/claude-haiku-4-5（例如），运行 uv add "crewai[anthropic]"，并在 .env 中填写 ANTHROPIC_API_KEY=<your-key>——即使 shell 中没有导出该变量，crewai run 也能从 .env 读取密钥。

脚手架所隐含的规则：

• 保留 [tool.crewai] type = "flow"。CLI 会据此判断 crewai run 表示 uv run kickoff。
• 模板的 plan_content() 会将 self.state.topic 覆盖为 "AI Agents"，除非存在 trigger payload，因此 kickoff(inputs={"topic": ...})（以及已部署 kickoff 的 inputs）会被忽略。在依赖输入之前，请删除该赋值。
• 模板的 plot() 使用默认参数 show=True 调用 plot()，因此 crewai flow plot 会打开浏览器。若要无头运行，请将其改为 plot("<name>.html", show=False)。
• 使用绝对包路径导入（如模板所示：from lead_router.crews.content_crew.content_crew import ContentCrew）。
• main.py 中的 kickoff() 必须返回 None。生成的控制台脚本会执行 sys.exit(kickoff())，因此如果返回 Flow 结果（例如 dict），进程会以状态码 1 退出，并将该值打印到 stderr；同时 crewai run 会打印 "An error occurred while running the flow"——但 crewai run 本身仍以状态码 0 退出，因此只检查退出码的 CI 步骤会通过。
• 对 Python Flow 使用 crewai run --inputs '{...}' 会被拒绝：Error: --inputs requires a declarative flow definition ([tool.crewai].definition) or --definition。请在代码中传入输入（flow.kickoff(inputs=...)），或在 kickoff() 中从 sys.argv 读取输入，再调用 uv run kickoff '<json>'。
• crewai flow kickoff 仍可使用，但会打印 "deprecated. Use 'crewai run' instead."
• 使用占位符或无效密钥会导致 Flow 失败，但 crewai run 仍以状态码 0 退出（它会打印 An error occurred while running the flow: Command '['uv', 'run', 'kickoff']' returned non-zero exit status 1.）。请检查是否出现该行，而不是检查退出码。

---

2. 状态：结构化或字典

	结构化（推荐）	非结构化
声明	class MyFlow(Flow[MyState])，并使用 MyState(BaseModel)	class MyFlow(Flow)
访问	self.state.topic	self.state["topic"]
id	自动为你添加（self.state.id，一个 UUID 字符串）	self.state["id"]
验证	kickoff(inputs=...) 的值会进行类型检查	无

from pydantic import BaseModel
from crewai.flow import Flow, listen, start


class ReportState(BaseModel):
    topic: str = ""          # give every field a default
    draft: str = ""
    notes: list[str] = []


class ReportFlow(Flow[ReportState]):
    @start()
    def plan(self):
        self.state.notes.append(f"planning {self.state.topic}")
        return self.state.topic.upper()

    @listen(plan)
    def write(self, plan_output):          # receives plan()'s return value
        self.state.draft = f"Draft about {plan_output}"
        return self.state.draft


flow = ReportFlow()
result = flow.kickoff(inputs={"topic": "bees"})   # inputs set state BEFORE @start runs
print(result)           # "Draft about BEES" - kickoff returns the last method's output
print(flow.state.id)    # UUID string, generated per run

状态规则（每条规则都能避免真实故障）：

规则	违反后的结果
不要在模型中声明 id；Flow[MyState] 会添加它	声明的 id 默认值会被生成的 UUID 静默替换
使用 Flow[MyState]，而不是 initial_state = MyState	使用普通 BaseModel 作为 initial_state：实例化时出现 ValidationError ... Flow state model must have an 'id' field
为每个字段提供默认值	必填字段：在流实例化时、调用 kickoff 之前出现 ValidationError
对结构化状态使用属性访问	self.state["topic"]：TypeError: 'StateWithId' object is not subscriptable
inputs 的键名必须与模型字段完全一致	结构化状态中的未知键会被静默丢弃
传入类型正确的 inputs	inputs={"topic": 5}：ValueError: Invalid inputs for structured state
切勿手动设置 self.state.id	它是持久化键；请使用 restore_from_state_id（第 6 节）

在已部署的流中（在 CrewAI AMP 上于 2026-10-01 测试），GET /inputs 会列出状态模型中声明的字段（不会列出自动生成的 id；模型本身声明了 id 时也会将其列出），而 POST /kickoff 的 inputs 对象会像 kickoff(inputs=...) 一样应用到状态中。不带 inputs 包装的请求体（{"text": "..."}）仍会被接受（返回 200 和一个 kickoff_id），且运行成功，但使用的是默认状态——该值从未传递到流中。运行时的 state.id 与其 kickoff_id 相同。调用方式：call-deployed-crew 技能。

---

3. 连线：@start、@listen、@router

from typing import Literal
from pydantic import BaseModel
from crewai.flow import Flow, listen, router, start


class TicketState(BaseModel):
    text: str = ""
    queue: str = ""


class TicketFlow(Flow[TicketState]):
    @start()
    def receive(self):
        return self.state.text

    @router(receive)
    def triage(self) -> Literal["urgent", "normal"]:   # labels, not method names
        return "urgent" if "outage" in self.state.text else "normal"

    @listen("urgent")
    def page_oncall(self):
        self.state.queue = "oncall"

    @listen("normal")
    def file_ticket(self):
        self.state.queue = "backlog"


flow = TicketFlow()
flow.kickoff(inputs={"text": "outage in eu-west"})
print(flow.state.queue)   # oncall

规则	原因
路由器标签不能与监听该标签的方法名相同	@listen("go") def go 会在实例化时引发错误：Invalid flow definition for ...: methods.go.listen listen condition 'go' references the handler name 'go'. A listener triggered by its own completion creates an infinite loop. 标签和方法名共用同一个命名空间
选择与所有方法名都不相同的标签	标签只是事件名，每个方法名也一样。路由器返回触发它的方法名时，该分支会不断重新运行，直到出现 RecursionError: Method 'r' has been called 100 times in this flow execution
确保路由器可能返回的每个标签都有对应的监听器	未匹配的标签会静默结束流；kickoff() 会返回标签字符串本身
使用 Literal[...] 返回类型或 @router(x, emit=[...]) 声明标签	这只用于 plot() 边和静态检查；运行时不会强制执行 emit=（路由器返回 emit 之外的标签时仍会运行）
使用 @listen(method_ref) 比 @listen("method_name") 更安全	字符串中的拼写错误只会成为一个无人发出的事件

---

4. or_ 和 and_

from crewai.flow import Flow, and_, listen, or_, start

calls = []


class FanIn(Flow):
    @start()
    def fetch_a(self):
        return "A"

    @start()
    def fetch_b(self):
        return "B"

    @listen(or_(fetch_a, fetch_b))
    def first_result(self, value):
        calls.append(("or", value))

    @listen(and_(fetch_a, fetch_b))
    def all_results(self, value):
        calls.append(("and", value))


FanIn().kickoff()
print(calls)   # e.g. [('or', 'A'), ('and', 'B')] - the start methods run concurrently, so the values (and order) can vary

• or_ 每次启动只触发一次，在第一个完成的条件满足时触发。第二个条件完成时不会再次运行，因此无需设置“已处理”标记。
• 路由器再次发出某个标签时，or_ 会重新准备触发。这正是修订循环能够运行的原因（@listen(or_(write_draft, "needs_revision")) 会在初稿完成时运行一次，并在每次出现 needs_revision 时再次运行；请参阅 routing-patterns）。
• 不要让单独的修订步骤监听与评审步骤相同的标签：两个 @listen("needs_revision") 方法会在该标签触发时并行运行，因此评审会重新检查尚未修订的初稿（在 1.15.22 和 1.15.23 上验证：三次评审都看到了 v1）。应让修订步骤成为一个返回新标签的路由器（@router("needs_revision") def revise(...): ...; return "revised"），并让评审监听 or_(write_draft, "revised")。
• and_ 会在所有条件都完成后触发一次，并接收最后完成条件的输出。其他条件的输出可从 self.state 中读取。
• 路由器之后常见的汇合方式：@listen(or_(page_oncall, file_ticket)) 会恰好运行一次，无论运行的是哪个分支。

---

5. 在 Flow 方法中调用 Crew 和 Agent

from crewai import Agent, Crew, Task
from crewai.flow import Flow, listen, start


class ResearchFlow(Flow):
    @start()
    def research(self):
        researcher = Agent(role="Researcher", goal="Research {topic}",
                           backstory="You find facts.", llm="openai/gpt-4o-mini")
        task = Task(description="List three facts about {topic}.",
                    expected_output="Three bullets.", agent=researcher)
        out = Crew(agents=[researcher], tasks=[task]).kickoff(inputs={"topic": "bees"})
        return out.raw                       # CrewOutput: .raw .pydantic .json_dict

    @listen(research)
    def summarize(self, facts):
        writer = Agent(role="Writer", goal="Summarize", backstory="Concise.",
                       llm="openai/gpt-4o-mini")
        return writer.kickoff(f"One sentence summary: {facts}").raw   # LiteAgentOutput

方法类型	调用方式	错误形式
def	crew.kickoff(inputs=...)、agent.kickoff(...)	-
async def	await crew.akickoff(inputs=...)	在 async def 中调用 crew.kickoff()：RuntimeError: Agent execution was invoked synchronously from within a running event loop

• 在 await crew.akickoff() 下，Agent 的 mcps=["https://..."] 字符串引用会解析为零个工具，且不会报错（这些引用需要启动各自的事件循环）。在异步方法中使用 crewai.mcp 中的 MCPServerHTTP(url=...)；详情请参阅 connect-tools-and-mcp 技能。
• Crew 的 {placeholders} 会从 Crew 自身的 kickoff(inputs=...) 获取值，而不是从 Flow 状态中获取。未传入输入而启动的 Crew 会将字面量 {topic} 发送给 LLM。请显式传入所需的状态值。
• 在方法内部构建 Crew（如果是 @CrewBase 类，则每次调用 MyCrew().crew()），这样每次运行都会获得全新的对象。
• Flow 方法是普通方法：可以在单元测试中直接调用 ReportFlow().plan()。

---

6. 使用 @persist 持久化

from pydantic import BaseModel
from crewai.flow import Flow, persist, start   # or: from crewai.flow.persistence import persist


class CounterState(BaseModel):
    count: int = 0


@persist()                      # class level: state saved after every method
class CounterFlow(Flow[CounterState]):
    @start()
    def bump(self):
        self.state.count += 1
        return self.state.count


first = CounterFlow()
first.kickoff()                                        # count == 1
second = CounterFlow()
second.kickoff(restore_from_state_id=first.state.id)   # count == 2
print(second.state.id != first.state.id)               # True: a fork with a new id

你想要	应该这样做	不要这样做
导入	from crewai.flow import persist 或 from crewai.flow.persistence import persist	from crewai.flow.flow import persist（ImportError）
添加装饰器	@persist()——始终带括号	在类上裸用 @persist：类会被替换为函数，实例化时会引发 TypeError: persist.<locals>.decorator() missing 1 required positional argument: 'target'。在方法上裸用 @persist：完全不会报错——方法体永远不会运行，kickoff() 返回 flow 对象，也不会保存任何内容
从保存的状态启动新运行	kickoff(restore_from_state_id=sid)——先恢复状态，再使用新的 state.id 保存	kickoff(inputs={"id": sid})——已弃用；会使用同一 id 恢复，且不会发出警告
仅在值得创建检查点的步骤保存	在该方法上使用 @persist()（将它放在 @listen 上方）	只需要最终状态时，在类级别使用
更改恢复运行中的某些字段	使用 restore_from_state_id，并在 inputs 中传入这些字段；状态恢复后会应用这些输入	编辑 SQLite 文件

状态的存储位置及静默情况：

• 默认后端：SQLiteFlowPersistence，文件 flow_states.db 位于 appdirs.user_data_dir(<CREWAI_STORAGE_DIR or current dir name>, "CrewAI") 中；在 macOS 上为 ~/Library/Application Support/<name>/，在 Linux 上为 ~/.local/share/<name>/。将 CREWAI_STORAGE_DIR 设置为绝对路径后，文件会直接写入该目录。也可以传入路径：@persist(SQLiteFlowPersistence("/abs/path/flows.db"))。
• 未知的 restore_from_state_id 不会引发错误：它会记录日志 "No flow state found ... proceeding without hydration"，并从默认状态运行。
• 对没有使用 @persist 的 flow 设置 restore_from_state_id 会被静默忽略。
• 本地的 @persist 使用 SQLite 文件。若要控制持久化状态，请针对你自己的数据库实现 FlowPersistence（persistence-and-checkpoints）。

在 CrewAI AMP 上（于 2026-10-01 使用通过 ZIP 上传部署的类级 @persist() flow 进行测试），在 POST /kickoff 的 inputs 旁边，将 id 作为顶层 restoreFromStateId 发送：

Kickoff 请求体	观察结果
{"inputs": {...}, "restoreFromStateId": "<state_id>"}	状态已恢复（计数器和列表得以保留，inputs 中未包含的字段也得以保留），在此基础上应用 inputs，生成新的 state.id（即新的 kickoff_id）
相同请求体，在两次 crewai deploy push 重新部署后发送	仍能恢复——保存的状态在重新部署后依然存在
没有匹配项的 restoreFromStateId	不会报错；运行从默认状态开始
{"inputs": {"id": "<state_id>", ...}}（已弃用）	已恢复并保留旧的 state.id，因此它与 kickoff_id 不同

从最后一个方法返回 self.state.id（或读取 kickoff_id），以便调用方获得可用于恢复的 id。

---

7. 检查点（崩溃后恢复）

检查点与 @persist 相互独立：它们会将运行状态（状态、已完成的方法、输出）保存到文件，以便失败的运行可以继续。

from crewai import CheckpointConfig
from crewai.flow import Flow, listen, start


class EtlFlow(Flow):
    @start()
    def extract(self):
        return "rows"

    @listen(extract)
    def load(self, rows):
        return f"loaded {rows}"


flow = EtlFlow(checkpoint=CheckpointConfig(
    location="./.checkpoints",
    on_events=["method_execution_finished"],   # required for flows, see below
))
flow.kickoff()

恢复：EtlFlow().kickoff(from_checkpoint=CheckpointConfig(restore_from="./.checkpoints/<branch>/<file>.json"))。已完成的方法会跳过。若要在继续前更改状态（例如清除导致失败的标记），请使用 flow = EtlFlow.from_checkpoint(CheckpointConfig(restore_from=path))，编辑 flow.state，然后调用 flow.kickoff()。

• 在 Flow 上设置 on_events=["method_execution_finished"]。 默认值 ["task_completed"]（以及 checkpoint=True）不会为 Flow 写入任何检查点，即使其方法运行了一个 crew 也是如此。
• 文件会写入 location/<branch>/，命名为 <timestamp>_<id>_p-<parent>.json，每个已完成的方法对应一个文件。时间戳精确到秒，id 为随机值，因此当多个方法在同一秒内完成时，按文件名排序无法找到最新文件（排序后的列表中，被选中的是倒数第二个）。请使用 max(files, key=os.path.getmtime)，或运行 crewai checkpoint list <location>（按时间从新到旧）。
• 恢复不会继续执行路由器之后的步骤。 在 @router（或带有 emit 的 @human_feedback 方法）完成后创建的检查点，恢复时不会运行任何内容，并会返回路由器的标签（或 None），且不报错——该标签对应的监听器永远不会运行。因此，路由器下游任何位置发生的崩溃都无法从最新检查点恢复；只有在路由器完成前创建的检查点，恢复后才会继续经过路由器。已在 1.15.22 和 1.15.23 上验证。请将容易引发崩溃的工作安排在路由器上游，或确保下游步骤幂等并重新运行 Flow。
• kickoff(from_checkpoint=..., restore_from_state_id=...) 会引发 ValueError: Cannot combine ...。两种系统只能选其一。

---

8. 人工反馈（@human_feedback）

from crewai.flow import Flow, HumanFeedbackResult, human_feedback, listen, start


class ReviewFlow(Flow):
    @start()
    @human_feedback(
        message="Approve this draft?",
        emit=["needs_revision", "approved"],
        llm="openai/gpt-4o-mini",            # maps free-text feedback to one label
        default_outcome="needs_revision",    # used when feedback is empty
    )
    def draft(self):
        return "Draft v1"

    @listen("approved")
    def publish(self, result: HumanFeedbackResult):
        return f"published: {result.output} ({result.feedback})"

    @listen("needs_revision")
    def revise(self, result: HumanFeedbackResult):
        return f"revise: {result.feedback}"

• HumanFeedbackResult 的字段：output、feedback、outcome、timestamp、method_name、metadata。文本内容在 result.feedback 中（feedback_text 会引发 AttributeError）。Flow 上也有：self.last_human_feedback、self.human_feedback_history。
• 反馈为空时，使用 default_outcome；若未设置默认值，则使用 emit[0]。请将安全的结果放在首位，并将其设为 default_outcome，这样按 Enter 键绝不会表示批准。
• 非空反馈若无法由 LLM 映射到某个标签，会引发 HumanFeedbackCollapseError，不会回退到其他结果。
• 非空反馈会由 llm 进行映射：使用 anthropic/claude-haiku-4-5 时，"Looks good to me, ship it." 被映射为 approved，而 "No, the tone is wrong - please rewrite it more politely." 被映射为 needs_revision；原始文本仍保留在 result.feedback 中。
• 未设置 emit 却设置了 default_outcome，会引发 ValueError: default_outcome requires emit to be specified.
• 未设置 emit 时，下一个监听器会收到 HumanFeedbackResult，且不会发生路由。
• 默认提供程序会读取控制台。测试或使用 UI 时，请传入 provider=（一个包含 request_feedback(context, flow) -> str 方法的对象）；若要暂停，请从该对象中引发 HumanFeedbackPending，之后可调用 MyFlow.from_pending(flow_id, persistence).resume(feedback) 继续（见 persistence-and-checkpoints）。没有 stdin 时（CI、< /dev/null、后台作业），控制台提供程序会引发 EOFError: EOF when reading a line，导致 Flow 失败。切勿在已部署的 Flow 中使用 input()。

---

9. plot()

path = TicketFlow().plot("ticket_flow.html", show=False)
print(path)   # /tmp/.../crewai_flow_xxxx/ticket_flow.html

• plot(filename="crewai_flow.html", show=True) 会将交互式 HTML（以及一个 .js 和一个 .css 文件）写入新的临时目录，并返回绝对路径。不会向当前目录写入任何文件，也不会生成 PNG。如需保留这些文件，请复制它们。
• 文件名会按给定值使用：plot("my_flow") 会写入名为 my_flow、不带 .html 扩展名的文件。
• show=True 会打开浏览器；在 CI 和无头运行中使用 show=False。
• 没有 Literal[...] 返回类型或 emit=[...] 的路由器会记录“Router events for 'r' are dynamic or not statically inferable”，图中可能会省略其边。图数据位于 .js 文件中，而非 .html 文件中。

---

10. 对话流

对于多轮聊天，请从 crewai.flow 导入（ConversationConfig、ConversationState、RouterConfig）；crewai.experimental.conversational 是指向同一模块的已弃用别名。使用 flow.handle_turn(message, session_id=...) 推进对话轮次，不要使用 kickoff(user_message=...)（会引发 TypeError）。没有 ChatSession 类。详情和经过测试的示例：conversational-flows。

---

11. 常见错误

症状	原因	修复方法
ImportError: cannot import name 'persist'	from crewai.flow.flow import persist	from crewai.flow import persist
TypeError: persist.<locals>.decorator() missing 1 required positional argument: 'target'，或某个方法始终不运行	使用不带括号的 @persist	@persist()
listen condition 'x' references the handler name 'x'	路由器标签与监听方法同名	重命名标签或方法
RecursionError: Method 'r' has been called 100 times	路由器返回了与方法名相同的标签（通常是其自身的触发器）	使用非方法名的标签
路由器之后流程停止，且没有错误	路由器返回了没有对应 @listen 的标签	监听每个标签；为路由器指定类型 -> Literal[...]
所有上游都已完成，但监听器只运行一次	这是 or_ 的行为	使用 and_ 等待所有上游完成
输入值被忽略	键名拼写错误；结构化状态会丢弃未知键	精确匹配字段名称
'StateWithId' object is not subscriptable	对结构化状态使用了字典访问方式	self.state.field
Flow state model must have an 'id' field	initial_state = MyModel	class F(Flow[MyModel])
crewai run 提示流程出错，但输出看起来正确	main.py 中的 kickoff() 返回了值	返回 None
--inputs requires a declarative flow definition	在 Python 流程中使用了 crewai run --inputs	在代码中传入输入，或通过 uv run kickoff '<json>' 传入
invoked synchronously from within a running event loop	在 async def 中调用了 crew.kickoff()	使用 await crew.akickoff()，或将该方法改为同步方法
恢复的运行从默认值开始	id 未知，或类上没有 @persist	检查 id 和装饰器；这两种情况都不会报错
多次运行复用了相同的执行 id	inputs={"id": ...}	使用 restore_from_state_id=
流程没有检查点文件	默认值为 on_events=["task_completed"]	使用 on_events=["method_execution_finished"]
检查点恢复返回了标签（或 None），且没有执行任何操作	检查点是在路由器完成后创建的	从路由器执行前的检查点恢复，或重新运行
@human_feedback 中出现 EOFError: EOF when reading a line	控制台提供程序没有 stdin	传入 provider=
脚手架生成的流程忽略了 topic 输入	模板中的 plan_content() 将 "AI Agents" 硬编码	删除该赋值
AttributeError: ... 'feedback_text'	字段名已过时	result.feedback
HumanFeedbackCollapseError	反馈没有匹配任何 emit 标签	清除标签，使用能力足够的 llm，并捕获该错误
项目中找不到 plot() 文件	文件写入了临时目录	使用返回的路径

---

12. 检查清单

• 状态为 Flow[MyState]，且每个字段都有默认值；不要手动设置 id
• 每个路由器标签都不同于所有方法名，且都有监听器
• 路由器返回类型为 Literal[...]（或使用 emit=[...]），这样 plot() 才会显示边
• 有意选择使用 or_（先满足者获胜，只触发一次）或 and_（全部满足）进行汇合
• async def 方法要 await crew.akickoff()；同步方法调用 kickoff()
• 从 crewai.flow 导入 persist，并写作 @persist()；恢复时使用 restore_from_state_id（HTTP 中使用 restoreFromStateId）
• 将 CREWAI_STORAGE_DIR 设置为绝对路径，以便之后能再次找到状态
• 使用检查点的工作流设置 on_events=["method_execution_finished"]；根据 mtime 选择最新文件；容易导致崩溃的步骤应放在路由器之前
• @human_feedback 将安全结果放在首位，并设为 default_outcome；代码读取 .feedback
• main.kickoff() 返回 None；存在 [tool.crewai] type = "flow"
• 运行了 crewai flow plot（或 plot(show=False)）并检查了图

---

参考资料

• 路由模式 - 使用 or_ + 路由器的修订循环、嵌套条件、多个路由器，以及完整的 router/or_/persist 示例工作流
• 持久化与检查点 - 存储路径、分叉与恢复、自定义 FlowPersistence、恢复并编辑检查点、暂停的人工反馈
• 对话工作流 - handle_turn、ConversationConfig、对话轮次的路由

公开文档：https://docs.crewai.com/en/concepts/flows, https://docs.crewai.com/en/concepts/checkpointing, https://docs.crewai.com/en/learn/human-feedback-in-flows, https://docs.crewai.com/en/guides/flows/inputs-id-deprecation

相关技能：

• getting-started - 在 crew 和 flow 之间进行选择、项目布局
• check-crewai-api - 当前 Agent/Task/Crew/LLM API 与记忆中的 0.x 形式
• design-task - flow 调用的 crew 所用的任务描述、结构化输出和防护措施
• test-crewai-project - 使用桩 LLM 离线运行 flow，并对状态进行断言
• deploy-to-amp - type = "flow" 的托管构建所需的项目结构和入口点
• call-deployed-crew - 通过 HTTP 启动已部署的 flow
• ask-docs - 当此处的某一项与当前版本不符时，查询实时文档