流程

概览

CrewAI Flows 是一个强大的功能，旨在简化 AI 工作流的创建与管理。Flows 允许开发者高效地组合和协调编码任务与 Crews，从而为构建复杂的 AI 自动化提供稳健框架。

Flows 让你可以创建结构化、事件驱动的工作流。它们提供了一种无缝方式来连接多个任务、管理状态，并控制 AI 应用中的执行流程。借助 Flows，你可以轻松设计和实现发挥 CrewAI 全部能力的多步骤流程。

简化工作流创建：轻松将多个 Crew 和任务串联起来，构建复杂的 AI 工作流。
状态管理：Flows 让你在工作流的不同任务之间轻松管理和共享状态。
事件驱动架构：基于事件驱动模型构建，可实现动态且响应迅速的工作流。
灵活的控制流：在工作流中实现条件逻辑、循环和分支。

快速开始

让我们创建一个简单的 Flow：先用 OpenAI 在一个任务中生成随机城市，再用这个城市在另一个任务中生成一个有趣事实。

from crewai.flow.flow import Flow, listen, start
from dotenv import load_dotenv
from litellm import completion

load_dotenv()

class ExampleFlow(Flow):
    model = "gpt-4o-mini"

    @start()
    def generate_city(self):
        print("Starting flow")
        # Each flow state automatically gets a unique ID
        print(f"Flow State ID: {self.state['id']}")

        response = completion(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": "Return the name of a random city in the world.",
                },
            ],
        )

        random_city = response["choices"][0]["message"]["content"]
        # Store the city in our state
        self.state["city"] = random_city
        print(f"Random City: {random_city}")

        return random_city

    @listen(generate_city)
    def generate_fun_fact(self, random_city):
        response = completion(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": f"Tell me a fun fact about {random_city}",
                },
            ],
        )

        fun_fact = response["choices"][0]["message"]["content"]
        # Store the fun fact in our state
        self.state["fun_fact"] = fun_fact
        return fun_fact



flow = ExampleFlow()
flow.plot()
result = flow.kickoff()

print(f"Generated fun fact: {result}")

Flow 视觉图在上面的示例中，我们创建了一个简单的 Flow：先使用 OpenAI 生成随机城市，然后基于该城市生成一个有趣事实。这个 Flow 由两个任务组成：generate_city 和 generate_fun_fact。generate_city 任务是 Flow 的起点，而 generate_fun_fact 任务会监听 generate_city 任务的输出。

每个 Flow 实例在其状态中都会自动获得一个唯一标识符（UUID），这有助于跟踪和管理 flow 的执行。状态还可以保存其他数据（例如生成的城市和有趣事实），并在整个 flow 执行期间持续保留。

运行 Flow 时，它会：

为 flow 状态生成唯一 ID
生成随机城市并将其存入状态
基于该城市生成有趣事实并将其存入状态
将结果打印到控制台

状态的唯一 ID 和存储的数据对于跟踪 flow 执行以及在任务之间维持上下文都很有用。

注意： 请确保你已经在 .env 文件中设置了 OPENAI_API_KEY。此密钥是向 OpenAI API 发起认证请求所必需的。

`@start()`

@start() 装饰器用于标记 Flow 的入口点。你可以：

声明多个无条件的起点：@start()
通过先前的方法或 router 标签控制起点：@start("method_or_label")
提供一个可调用条件来控制起点是否触发

当 Flow 开始或恢复时，所有满足条件的 @start() 方法都会执行，通常会并行运行。

`@listen()`

@listen() 装饰器用于将某个方法标记为监听 Flow 中另一个任务输出的监听器。被 @listen() 装饰的方法会在指定任务发出输出时执行。该方法可以将其监听任务的输出作为参数接收。

用法

@listen() 可以通过几种方式使用：

按方法名监听：你可以以字符串形式传入想要监听的方法名。当该方法完成时，监听器方法会被触发。
```
@listen("generate_city")
def generate_fun_fact(self, random_city):
    # Implementation
```
直接监听方法：你可以直接传入方法本身。当该方法完成时，监听器方法会被触发。
```
@listen(generate_city)
def generate_fun_fact(self, random_city):
    # Implementation
```

Flow 输出

访问和处理 Flow 的输出对于将 AI 工作流集成到更大的应用或系统中至关重要。CrewAI Flows 提供了直接的方法来获取最终输出、访问中间结果，以及管理 Flow 的整体状态。

获取最终输出

当你运行一个 Flow 时，最终输出由最后完成的方法决定。kickoff() 方法会返回这个最终方法的输出。

你可以这样访问最终输出：

Code

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

class OutputExampleFlow(Flow):
    @start()
    def first_method(self):
        return "Output from first_method"

    @listen(first_method)
    def second_method(self, first_output):
        return f"Second method received: {first_output}"


flow = OutputExampleFlow()
flow.plot("my_flow_plot")
final_output = flow.kickoff()

print("---- Final Output ----")
print(final_output)

---- Final Output ----
Second method received: Output from first_method

Flow 视觉图

在这个示例中，second_method 是最后完成的方法，因此它的输出将成为 Flow 的最终输出。 kickoff() 方法会返回最终输出，然后将其打印到控制台。plot() 方法会生成 HTML 文件，帮助你理解这个 flow。

访问和更新状态

除了获取最终输出之外，你还可以在 Flow 内访问和更新状态。状态可用于在 Flow 的不同方法之间存储和共享数据。Flow 运行完成后，你可以访问状态以检索执行期间添加或更新的任何信息。

下面是一个更新和访问状态的示例：

Code

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

class ExampleState(BaseModel):
    counter: int = 0
    message: str = ""

class StateExampleFlow(Flow[ExampleState]):

    @start()
    def first_method(self):
        self.state.message = "Hello from first_method"
        self.state.counter += 1

    @listen(first_method)
    def second_method(self):
        self.state.message += " - updated by second_method"
        self.state.counter += 1
        return self.state.message

flow = StateExampleFlow()
flow.plot("my_flow_plot")
final_output = flow.kickoff()
print(f"Final Output: {final_output}")
print("Final State:")
print(flow.state)

Final Output: Hello from first_method - updated by second_method
Final State:
counter=2 message='Hello from first_method - updated by second_method'

在这个示例中，状态同时被 first_method 和 second_method 更新。 Flow 运行结束后，你可以访问最终状态，查看这些方法所做的更新。

通过确保返回最终方法的输出并提供状态访问能力，CrewAI Flows 让你很容易将 AI 工作流的结果集成到更大的应用或系统中，同时还能在整个 Flow 执行期间维护并访问状态。

Flow 使用指标

Flow 执行完成后，你可以访问 usage_metrics 属性，查看本次运行中每一次 LLM 调用的聚合 token 使用情况 - 包括 Flow 编排的每个 Crew 中的调用、Agent 工具中的调用，以及 Flow 方法中直接的 LLM.call(...) 调用。这是 SDK 侧对应 CrewAI Enterprise UI 中总计数的实现。

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

class UsageMetricsFlow(Flow):
    @start()
    def run_first_crew(self):
        self.state.first_result = FirstCrew().crew().kickoff()

    @listen(run_first_crew)
    def call_llm_directly(self):
        # Bare LLM call — still counted by flow.usage_metrics
        llm = LLM(model="openai/gpt-4o-mini")
        self.state.summary = llm.call("Summarize the key takeaways.")

    @listen(call_llm_directly)
    def run_second_crew(self):
        self.state.second_result = SecondCrew().crew().kickoff()

flow = UsageMetricsFlow()
flow.kickoff()

print(flow.usage_metrics)
# UsageMetrics(total_tokens=8579, prompt_tokens=6210, completion_tokens=2369,
#              cached_prompt_tokens=0, reasoning_tokens=0,
#              cache_creation_tokens=0, successful_requests=5)

返回的 UsageMetrics 中每一项都是单次 flow.kickoff() 调用内所有 LLM 调用的总和。计数器会在下一次 kickoff() 调用时重置（或在每次 kickoff_for_each 迭代时重置），因此连续运行不会重复计数。只要 kickoff() 完成后，随时读取该属性都是安全的；在执行过程中读取则会返回截至当前已累计的部分总值。

Flow 状态管理

有效管理状态对于构建可靠且易维护的 AI 工作流至关重要。CrewAI Flows 为非结构化和结构化状态管理都提供了稳健机制，让开发者能够选择最适合自己应用需求的方法。

非结构化状态管理

在非结构化状态管理中，所有状态都存储在 Flow 类的 state 属性中。这种方式灵活，允许开发者在不定义严格模式的情况下随时添加或修改状态属性。即使是非结构化状态，CrewAI Flows 也会自动为每个状态实例生成并维护一个唯一标识符（UUID）。

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

class UnstructuredExampleFlow(Flow):

    @start()
    def first_method(self):
        # The state automatically includes an 'id' field
        print(f"State ID: {self.state['id']}")
        self.state['counter'] = 0
        self.state['message'] = "Hello from structured flow"

    @listen(first_method)
    def second_method(self):
        self.state['counter'] += 1
        self.state['message'] += " - updated"

    @listen(second_method)
    def third_method(self):
        self.state['counter'] += 1
        self.state['message'] += " - updated again"

        print(f"State after third_method: {self.state}")


flow = UnstructuredExampleFlow()
flow.plot("my_flow_plot")
flow.kickoff()

Flow 视觉图

注意： id 字段会在 flow 执行期间自动生成并保持不变。你无需手动管理或设置它，即使使用新数据更新状态，它也会一直被保留。

要点：

灵活性： 你可以不预先定义约束，动态地向 self.state 添加属性。
简单性： 适合状态结构较少或变化较大的直接型工作流。

结构化状态管理

结构化状态管理利用预定义模式来确保整个工作流的一致性和类型安全。通过使用 Pydantic 的 BaseModel 等模型，开发者可以准确定义状态的形状，从而在开发环境中获得更好的校验和自动补全。

CrewAI Flows 中的每个状态都会自动获得一个唯一标识符（UUID），以便跟踪和管理状态实例。这个 ID 由 Flow 系统自动生成并管理。

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


class ExampleState(BaseModel):
    # Note: 'id' field is automatically added to all states
    counter: int = 0
    message: str = ""


class StructuredExampleFlow(Flow[ExampleState]):

    @start()
    def first_method(self):
        # Access the auto-generated ID if needed
        print(f"State ID: {self.state.id}")
        self.state.message = "Hello from structured flow"

    @listen(first_method)
    def second_method(self):
        self.state.counter += 1
        self.state.message += " - updated"

    @listen(second_method)
    def third_method(self):
        self.state.counter += 1
        self.state.message += " - updated again"

        print(f"State after third_method: {self.state}")


flow = StructuredExampleFlow()
flow.kickoff()

Flow 视觉图

要点：

明确的模式： ExampleState 清楚地定义了状态结构，提高了代码可读性和可维护性。
类型安全： 借助 Pydantic，状态属性会遵循指定类型，从而减少运行时错误。
自动补全： IDE 可以基于已定义的状态模型提供更好的自动补全和错误检查。

在非结构化和结构化状态管理之间选择

在以下情况下使用非结构化状态管理：
- 工作流的状态很简单或高度动态。
- 你更看重灵活性，而不是严格的状态定义。
- 需要快速原型开发，而不想承担定义模式的额外开销。
在以下情况下使用结构化状态管理：
- 工作流需要清晰且一致的状态结构。
- 类型安全和校验对应用可靠性很重要。
- 你希望利用 IDE 的自动补全和类型检查来获得更好的开发体验。

通过同时提供非结构化和结构化两种状态管理方式，CrewAI Flows 让开发者能够构建既灵活又稳健的 AI 工作流，以满足各种应用需求。

Flow 持久化

@persist 装饰器可为 CrewAI Flows 启用自动状态持久化，让你在重启或不同工作流执行之间维持 flow 状态。该装饰器既可以应用在类级别，也可以应用在方法级别，为状态持久化的管理方式提供灵活性。

类级持久化

当应用在类级别时，@persist 装饰器会自动持久化所有 flow 方法的状态：

@persist  # Using SQLiteFlowPersistence by default
class MyFlow(Flow[MyState]):
    @start()
    def initialize_flow(self):
        # This method will automatically have its state persisted
        self.state.counter = 1
        print("Initialized flow. State ID:", self.state.id)

    @listen(initialize_flow)
    def next_step(self):
        # The state (including self.state.id) is automatically reloaded
        self.state.counter += 1
        print("Flow state is persisted. Counter:", self.state.counter)

方法级持久化

如果想要更细粒度的控制，你可以将 @persist 应用于特定方法：

class AnotherFlow(Flow[dict]):
    @persist  # Persists only this method's state
    @start()
    def begin(self):
        if "runs" not in self.state:
            self.state["runs"] = 0
        self.state["runs"] += 1
        print("Method-level persisted runs:", self.state["runs"])

持久化状态分叉

@persist 在 kickoff / kickoff_async 上支持两种不同的加载模式：

kickoff(inputs={"id": <uuid>}) - 恢复：加载所提供 UUID 的最新快照，并在相同 flow_uuid 下继续写入。历史会延续。
kickoff(restore_from_state_id=<uuid>) - 分叉：加载所提供 UUID 的最新快照，从中为新运行加载状态，并分配一个新的 state.id（自动生成；如果 inputs["id"] 被固定则使用它）。新运行中的 @persist 写入会落到新的 state.id 下；源 flow 的历史会被保留。

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

class CounterState(BaseModel):
    id: str = ""
    counter: int = 0

@persist
class CounterFlow(Flow[CounterState]):
    @start()
    def step(self):
        self.state.counter += 1
        print(f"[id={self.state.id}] counter={self.state.counter}")

# Run 1: fresh state, counter 0 -> 1, persisted under flow_1.state.id
flow_1 = CounterFlow()
flow_1.kickoff()

# Fork: hydrate from flow_1's latest snapshot, but use a NEW state.id
flow_2 = CounterFlow()
flow_2.kickoff(restore_from_state_id=flow_1.state.id)
# flow_2.state.counter starts at 1 (hydrated), then step() bumps it to 2.
# flow_2.state.id != flow_1.state.id; flow_1's history is unchanged.

如果提供的 restore_from_state_id 与任何已持久化状态都不匹配，kickoff 会静默回退 - 与现有 inputs["id"] 恢复未命中的行为一致。将 restore_from_state_id 与 from_checkpoint 组合会抛出 ValueError；请二选一作为加载来源。固定 inputs["id"] 进行分叉会与另一个 flow 共享持久化键 - 通常你只需要 restore_from_state_id。

工作原理

唯一状态标识
- 每个 flow 状态都会自动获得唯一 UUID
- 该 ID 会在状态更新和方法调用间保持不变
- 同时支持结构化（Pydantic BaseModel）和非结构化（字典）状态
默认 SQLite 后端
- SQLiteFlowPersistence 是默认存储后端
- 状态会自动保存到本地 SQLite 数据库
- 若数据库操作失败，稳健的错误处理会提供清晰信息
错误处理
- 数据库操作提供完整的错误信息
- 在保存和加载期间自动进行状态校验
- 当持久化操作遇到问题时提供明确反馈

重要注意事项

状态类型：同时支持结构化（Pydantic BaseModel）和非结构化（字典）状态
自动 ID：如果 id 字段不存在，会自动添加
状态恢复：失败或重启的 flow 可以自动重新加载之前的状态
自定义实现：你可以提供自己的 FlowPersistence 实现，以满足专门的存储需求

技术优势

通过底层访问实现精确控制
- 为高级用例提供对持久化操作的直接访问
- 通过方法级持久化装饰器实现细粒度控制
- 内置状态检查与调试能力
- 对状态变更和持久化操作有完整可见性
增强可靠性
- 系统故障或重启后自动恢复状态
- 基于事务的状态更新，保证数据完整性
- 提供清晰错误信息的完整错误处理
- 在状态保存与加载操作中具备稳健校验
可扩展架构
- 通过 FlowPersistence 接口提供可定制的持久化后端
- 支持 SQLite 之外的专用存储方案
- 同时兼容结构化（Pydantic）和非结构化（dict）状态
- 与现有 CrewAI flow 模式无缝集成

持久化系统的架构强调技术精确性与可定制选项，让开发者在受益于内置可靠性特性的同时，仍能完全掌控状态管理。

Flow 控制

条件逻辑：`or`

Flows 中的 or_ 函数允许你监听多个方法，并在任一指定方法发出输出时触发监听器方法。

Code

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

class OrExampleFlow(Flow):

    @start()
    def start_method(self):
        return "Hello from the start method"

    @listen(start_method)
    def second_method(self):
        return "Hello from the second method"

    @listen(or_(start_method, second_method))
    def logger(self, result):
        print(f"Logger: {result}")



flow = OrExampleFlow()
flow.plot("my_flow_plot")
flow.kickoff()

Logger: Hello from the start method
Logger: Hello from the second method

Flow 视觉图

当你运行这个 Flow 时，logger 方法会在 start_method 或 second_method 的输出出现时被触发。 or_ 函数用于监听多个方法，并在任一指定方法发出输出时触发监听器方法。

条件逻辑：`and`

Flows 中的 and_ 函数允许你监听多个方法，并且只有当所有指定方法都发出输出时，才触发监听器方法。

Code

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

class AndExampleFlow(Flow):

    @start()
    def start_method(self):
        self.state["greeting"] = "Hello from the start method"

    @listen(start_method)
    def second_method(self):
        self.state["joke"] = "What do computers eat? Microchips."

    @listen(and_(start_method, second_method))
    def logger(self):
        print("---- Logger ----")
        print(self.state)

flow = AndExampleFlow()
flow.plot()
flow.kickoff()

---- Logger ----
{'greeting': 'Hello from the start method', 'joke': 'What do computers eat? Microchips.'}

Flow 视觉图

当你运行这个 Flow 时，只有当 start_method 和 second_method 都发出输出时，logger 方法才会被触发。 and_ 函数用于监听多个方法，并且只有当所有指定方法都发出输出时才触发监听器方法。

Router

Flows 中的 @router() 装饰器允许你基于某个方法的输出定义条件路由逻辑。你可以根据该方法的输出指定不同路由，从而动态控制执行流程。

Code

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

class ExampleState(BaseModel):
    success_flag: bool = False

class RouterFlow(Flow[ExampleState]):

    @start()
    def start_method(self):
        print("Starting the structured flow")
        random_boolean = random.choice([True, False])
        self.state.success_flag = random_boolean

    @router(start_method)
    def second_method(self):
        if self.state.success_flag:
            return "success"
        else:
            return "failed"

    @listen("success")
    def third_method(self):
        print("Third method running")

    @listen("failed")
    def fourth_method(self):
        print("Fourth method running")


flow = RouterFlow()
flow.plot("my_flow_plot")
flow.kickoff()

Starting the structured flow
Third method running
Fourth method running

Flow 视觉图

在上面的示例中，start_method 会生成一个随机布尔值并将其写入状态。 second_method 使用 @router() 装饰器，根据该布尔值定义条件路由逻辑。如果布尔值为 True，方法返回 "success"；如果为 False，则返回 "failed"。 third_method 和 fourth_method 会监听 second_method 的输出，并根据返回值执行。

运行这个 Flow 时，输出会随着 start_method 生成的随机布尔值而变化。

人在回路中（human feedback）

@human_feedback 装饰器通过暂停 flow 执行来收集来自人的反馈，从而实现人在回路中的工作流。这对于审批门、质量审查以及需要人工判断的决策点非常有用。

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

class ReviewFlow(Flow):
    @start()
    @human_feedback(
        message="Do you approve this content?",
        emit=["approved", "rejected", "needs_revision"],
        llm="gpt-4o-mini",
        default_outcome="needs_revision",
    )
    def generate_content(self):
        return "Content to be reviewed..."

    @listen("approved")
    def on_approval(self, result: HumanFeedbackResult):
        print(f"Approved! Feedback: {result.feedback}")

    @listen("rejected")
    def on_rejection(self, result: HumanFeedbackResult):
        print(f"Rejected. Reason: {result.feedback}")

当指定了 emit 时，人的自由形式反馈会被 LLM 解释并归纳为所列出的某个结果，然后触发对应的 @listen 装饰器。

你也可以在不做路由的情况下使用 @human_feedback，只收集反馈：

@start()
@human_feedback(message="Any comments on this output?")
def my_method(self):
    return "Output for review"

@listen(my_method)
def next_step(self, result: HumanFeedbackResult):
    # Access feedback via result.feedback
    # Access original output via result.output
    pass

可以通过 self.last_human_feedback（最近一次）或 self.human_feedback_history（全部反馈列表）访问在 flow 中收集到的所有反馈。

如需完整了解 flow 中的人类反馈，包括使用自定义提供方（Slack、webhook 等）进行异步/非阻塞反馈，请参阅 Flow 中的人类反馈。

向 Flows 添加 Agents

Agents 可以无缝集成到你的 flows 中，在你需要更简单、聚焦的任务执行时，提供一种比完整 Crews 更轻量的替代方案。下面是一个在 flow 中使用 Agent 进行市场研究的示例：

import asyncio
from typing import Any, Dict, List

from crewai_tools import SerperDevTool
from pydantic import BaseModel, Field

from crewai.agent import Agent
from crewai.flow.flow import Flow, listen, start


# Define a structured output format
class MarketAnalysis(BaseModel):
    key_trends: List[str] = Field(description="List of identified market trends")
    market_size: str = Field(description="Estimated market size")
    competitors: List[str] = Field(description="Major competitors in the space")


# Define flow state
class MarketResearchState(BaseModel):
    product: str = ""
    analysis: MarketAnalysis | None = None


# Create a flow class
class MarketResearchFlow(Flow[MarketResearchState]):
    @start()
    def initialize_research(self) -> Dict[str, Any]:
        print(f"Starting market research for {self.state.product}")
        return {"product": self.state.product}

    @listen(initialize_research)
    async def analyze_market(self) -> Dict[str, Any]:
        # Create an Agent for market research
        analyst = Agent(
            role="Market Research Analyst",
            goal=f"Analyze the market for {self.state.product}",
            backstory="You are an experienced market analyst with expertise in "
            "identifying market trends and opportunities.",
            tools=[SerperDevTool()],
            verbose=True,
        )

        # Define the research query
        query = f"""
        Research the market for {self.state.product}. Include:
        1. Key market trends
        2. Market size
        3. Major competitors

        Format your response according to the specified structure.
        """

        # Execute the analysis with structured output format
        result = await analyst.kickoff_async(query, response_format=MarketAnalysis)
        if result.pydantic:
            print("result", result.pydantic)
        else:
            print("result", result)

        # Return the analysis to update the state
        return {"analysis": result.pydantic}

    @listen(analyze_market)
    def present_results(self, analysis) -> None:
        print("\nMarket Analysis Results")
        print("=====================")

        if isinstance(analysis, dict):
            # If we got a dict with 'analysis' key, extract the actual analysis object
            market_analysis = analysis.get("analysis")
        else:
            market_analysis = analysis

        if market_analysis and isinstance(market_analysis, MarketAnalysis):
            print("\nKey Market Trends:")
            for trend in market_analysis.key_trends:
                print(f"- {trend}")

            print(f"\nMarket Size: {market_analysis.market_size}")

            print("\nMajor Competitors:")
            for competitor in market_analysis.competitors:
                print(f"- {competitor}")
        else:
            print("No structured analysis data available.")
            print("Raw analysis:", analysis)


# Usage example
async def run_flow():
    flow = MarketResearchFlow()
    flow.plot("MarketResearchFlowPlot")
    result = await flow.kickoff_async(inputs={"product": "AI-powered chatbots"})
    return result


# Run the flow
if __name__ == "__main__":
    asyncio.run(run_flow())

Flow 视觉图

这个示例展示了在 flows 中使用 Agents 的几个关键特性：

结构化输出：使用 Pydantic 模型定义预期输出格式（MarketAnalysis）可确保整个 flow 中的类型安全和结构化数据。
状态管理：flow 状态（MarketResearchState）在步骤之间维护上下文，并存储输入与输出。
工具集成：Agents 可以使用工具（例如 WebsiteSearchTool）来增强自身能力。

向 Flows 添加 Crews

在 CrewAI 中创建一个包含多个 crews 的 flow 非常直接。

你可以通过运行以下命令生成一个新的 CrewAI 项目，其中会包含创建多 crew flow 所需的全部脚手架：

crewai create flow name_of_flow

该命令会生成一个带有所需文件夹结构的新 CrewAI 项目。生成的项目包含一个已经可用的预建 crew，名为 poem_crew。内置的起始 crew 使用传统的 Python/YAML 布局；通过 crewai create crew 创建的新独立 crew 使用 JSON-first 布局。

文件夹结构

运行 crewai create flow name_of_flow 命令后，你会看到类似下面的文件夹结构：

Directory/File	Description
`name_of_flow/`	Flow 的根目录。
├── `crews/`	包含各个特定 crew 的目录。
│ └── `poem_crew/`	用于 “poem_crew” 的目录，包含其配置和脚本。
│ ├── `config/`	”poem_crew” 的配置文件目录。
│ │ ├── `agents.yaml`	为 “poem_crew” 定义 agents 的 YAML 文件。
│ │ └── `tasks.yaml`	为 “poem_crew” 定义 tasks 的 YAML 文件。
│ ├── `poem_crew.py`	”poem_crew” 功能脚本。
├── `tools/`	flow 中使用的附加工具目录。
│ └── `custom_tool.py`	自定义工具实现。
├── `main.py`	运行 flow 的主脚本。
├── `README.md`	项目说明和使用指南。
├── `pyproject.toml`	项目依赖和设置的配置文件。
└── `.gitignore`	指定要在版本控制中忽略的文件和目录。

构建你的 Crews

在 crews 文件夹中，你可以定义多个 crews。生成的 poem_crew 使用经典的嵌入式 crew 结构：

config/agents.yaml：定义该 crew 的 agents。
config/tasks.yaml：定义该 crew 的 tasks。
poem_crew.py：包含 crew 的定义，包括 agents、tasks 以及 crew 本身。

你可以复制、粘贴并编辑 poem_crew，以创建其他经典嵌入式 crews。

对于 JSON-first 的嵌入式 crews，请改用包含 crew.jsonc 和 agents/*.jsonc 的文件夹：

crews/
└── research_crew/
    ├── agents/
    │   └── researcher.jsonc
    └── crew.jsonc

然后在 Flow 步骤中加载它：

from pathlib import Path
from crewai.project import load_crew

crew, default_inputs = load_crew(
    Path(__file__).parent / "crews" / "research_crew" / "crew.jsonc"
)
result = crew.kickoff(inputs={**default_inputs, "topic": "AI Agents"})

在 `main.py` 中连接 Crews

main.py 文件是你创建 flow 并将 crews 连接在一起的地方。你可以通过使用 Flow 类以及 @start 和 @listen 装饰器来定义执行流程。

下面是一个在 main.py 文件中连接 poem_crew 的示例：

#!/usr/bin/env python
from random import randint

from pydantic import BaseModel
from crewai.flow.flow import Flow, listen, start
from .crews.poem_crew.poem_crew import PoemCrew

class PoemState(BaseModel):
    sentence_count: int = 1
    poem: str = ""

class PoemFlow(Flow[PoemState]):

    @start()
    def generate_sentence_count(self):
        print("Generating sentence count")
        self.state.sentence_count = randint(1, 5)

    @listen(generate_sentence_count)
    def generate_poem(self):
        print("Generating poem")
        result = PoemCrew().crew().kickoff(inputs={"sentence_count": self.state.sentence_count})

        print("Poem generated", result.raw)
        self.state.poem = result.raw

    @listen(generate_poem)
    def save_poem(self):
        print("Saving poem")
        with open("poem.txt", "w") as f:
            f.write(self.state.poem)

def kickoff():
    poem_flow = PoemFlow()
    poem_flow.kickoff()


def plot():
    poem_flow = PoemFlow()
    poem_flow.plot("PoemFlowPlot")

if __name__ == "__main__":
    kickoff()
    plot()

在这个示例中，PoemFlow 类定义了一个 flow，它会生成句子数量，使用 PoemCrew 生成一首诗，然后将这首诗保存到文件中。通过调用 kickoff() 方法启动这个 flow。PoemFlowPlot 会由 plot() 方法生成。

Flow 视觉图

运行 Flow

（可选）在运行 flow 之前，你可以通过执行以下命令安装依赖：

crewai install

安装完所有依赖后，你需要通过以下命令激活虚拟环境：

source .venv/bin/activate

激活虚拟环境后，你可以使用 CrewAI CLI 运行 flow：

crewai run

你也可以直接运行项目脚本：

uv run kickoff

flow 会执行，你应该能在控制台看到输出。

绘制 Flows

将 AI 工作流可视化可以为 flows 的结构和执行路径提供重要洞察。CrewAI 提供了一个强大的可视化工具，可用于生成 flows 的交互式图表，让你更容易理解和优化 AI 工作流。

什么是图表？

CrewAI 中的图表是 AI 工作流的图形化表示。它们展示了各种任务、它们之间的连接，以及数据在它们之间的流动方式。此可视化有助于理解操作顺序、识别瓶颈，并确保工作流逻辑符合预期。

如何生成图表

CrewAI 提供了两种便捷方式来生成 flow 图表：

选项 1：使用 `plot()` 方法

如果你直接使用某个 flow 实例，可以通过在 flow 对象上调用 plot() 方法来生成图表。此方法会创建一个包含 flow 交互式图表的 HTML 文件。

# Assuming you have a flow instance
flow.plot("my_flow_plot")

这会在当前目录下生成名为 my_flow_plot.html 的文件。你可以在浏览器中打开该文件，查看交互式图表。

选项 2：使用命令行

如果你在一个结构化的 CrewAI 项目中工作，也可以通过命令行生成图表。这对于希望可视化整个 flow 设置的大型项目尤其有用。

crewai flow plot

该命令会生成一个带有 flow 图表的 HTML 文件，效果与 plot() 方法类似。文件会保存在你的项目目录中，你可以在浏览器中打开它来浏览 flow。

理解图表

生成的图表会显示表示 flow 中任务的节点，以及指示执行流程的有向边。图表是交互式的，你可以放大、缩小，并悬停在节点上查看更多细节。

通过可视化你的 flows，你可以更清楚地理解工作流结构，从而更容易进行调试、优化，并向他人传达你的 AI 流程。

结论

绘制 flows 是 CrewAI 的一个强大功能，它增强了你设计和管理复杂 AI 工作流的能力。无论你选择使用 plot() 方法还是命令行，生成图表都会为你的工作流提供可视化表示，帮助开发与展示。

下一步

如果你有兴趣探索更多 flow 示例，我们在 examples 仓库中准备了多种推荐。下面是四个具体的 flow 示例，每个都展示了独特的用例，帮助你将当前的问题类型与合适的示例对应起来：

邮件自动回复 Flow：此示例展示了一个无限循环，后台任务会持续运行以自动化邮件回复。对于需要重复执行、无需人工干预的任务，这是很好的用例。查看示例
线索评分 Flow：此 flow 展示了如何添加人在回路中的反馈，以及如何使用 router 处理不同的条件分支。这是一个很好的示例，说明如何将动态决策和人工监督整合到工作流中。查看示例
写书 Flow：此示例擅长将多个 crew 串联在一起，其中一个 crew 的输出会被另一个 crew 使用。具体来说，一个 crew 负责勾勒整本书的大纲，另一个 crew 根据该大纲生成章节。最终一切连接起来，产出完整书籍。这个 flow 非常适合需要多个任务协同的复杂多步骤流程。查看示例
会议助手 Flow：此 flow 展示了如何广播一个事件以触发多个后续动作。例如，会议结束后，flow 可以更新 Trello 看板、发送 Slack 消息并保存结果。这是处理单个事件触发多个结果的好例子，非常适合全面的任务管理和通知系统。查看示例

通过探索这些示例，你可以了解如何将 CrewAI Flows 用于各种场景，从自动化重复任务到管理复杂的多步骤流程，再到动态决策和人类反馈。

另外，也可以查看下面这段关于如何在 CrewAI 中使用 flows 的 YouTube 视频！

运行 Flows

运行 flow 有两种方式：

使用 Flow API

你可以通过创建 flow 类的实例并调用 kickoff() 方法来以编程方式运行 flow：

flow = ExampleFlow()
result = flow.kickoff()

流式执行 Flow

如果你希望实时查看 flow 的执行过程，可以启用 streaming，让输出在生成时就被接收：

class StreamingFlow(Flow):
    stream = True  # Enable streaming

    @start()
    def research(self):
        # Your flow implementation
        pass

# Iterate over streaming output
flow = StreamingFlow()
streaming = flow.kickoff()
for chunk in streaming:
    print(chunk.content, end="", flush=True)

# Access final result
result = streaming.result

在流式 Flow 执行指南中了解更多有关 streaming 的内容。

Flows 中的 Memory

每个 Flow 都可以自动访问 CrewAI 的统一 Memory 系统。你可以直接在任意 flow 方法中使用三个内置便捷方法来存储、召回和提取 memory。

内置方法

方法	描述
`self.remember(content, **kwargs)`	将内容存入 memory。可选接受 `scope`、`categories`、`metadata`、`importance`。
`self.recall(query, **kwargs)`	检索相关 memories。可选接受 `scope`、`categories`、`limit`、`depth`。
`self.extract_memories(content)`	将原始文本拆分为离散、自包含的 memory 语句。

当 Flow 初始化时，会自动创建一个默认的 Memory() 实例。你也可以传入自定义实例：

from crewai.flow.flow import Flow
from crewai import Memory

custom_memory = Memory(
    recency_weight=0.5,
    recency_half_life_days=7,
    embedder={"provider": "ollama", "config": {"model_name": "mxbai-embed-large"}},
)

flow = MyFlow(memory=custom_memory)

示例：研究与分析 Flow

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


class ResearchAnalysisFlow(Flow):
    @start()
    def gather_data(self):
        # Simulate research findings
        findings = (
            "PostgreSQL handles 10k concurrent connections with connection pooling. "
            "MySQL caps at around 5k. MongoDB scales horizontally but adds complexity."
        )

        # Extract atomic facts and remember each one
        memories = self.extract_memories(findings)
        for mem in memories:
            self.remember(mem, scope="/research/databases")

        return findings

    @listen(gather_data)
    def analyze(self, raw_findings):
        # Recall relevant past research (from this run or previous runs)
        past = self.recall("database performance and scaling", limit=10, depth="shallow")

        context_lines = [f"- {m.record.content}" for m in past]
        context = "\n".join(context_lines) if context_lines else "No prior context."

        return {
            "new_findings": raw_findings,
            "prior_context": context,
            "total_memories": len(past),
        }


flow = ResearchAnalysisFlow()
result = flow.kickoff()
print(result)

由于 memory 会跨运行持久化（由磁盘上的 LanceDB 提供支持），analyze 步骤也会召回之前执行中留下的 findings - 这让 flows 能够随着时间积累知识并不断学习。

有关 scopes、slices、复合评分、embedder 配置等详细信息，请参阅 Memory 文档。

使用 CLI

从 0.103.0 版本开始，你可以使用 crewai run 命令来运行 flows：

crewai run

该命令会自动检测你的项目是否是 flow（基于 pyproject.toml 中的 type = "flow" 设置），并据此运行。对于从命令行运行 flows，这是推荐方式。

旧的 crewai flow kickoff 命令已经弃用。请为 crews 和 flows 都使用 crewai run。

流程

概览

快速开始

@start()

@listen()

用法

Flow 输出

获取最终输出

访问和更新状态

Flow 使用指标

Flow 状态管理

非结构化状态管理

结构化状态管理

在非结构化和结构化状态管理之间选择

Flow 持久化

类级持久化

方法级持久化

持久化状态分叉

工作原理

重要注意事项

技术优势

Flow 控制

条件逻辑：or

条件逻辑：and

Router

人在回路中（human feedback）

向 Flows 添加 Agents

向 Flows 添加 Crews

文件夹结构

构建你的 Crews

在 main.py 中连接 Crews

运行 Flow

绘制 Flows

什么是图表？

如何生成图表

选项 1：使用 plot() 方法

选项 2：使用命令行

理解图表

结论

下一步

运行 Flows

使用 Flow API

流式执行 Flow

Flows 中的 Memory

内置方法

示例：研究与分析 Flow

使用 CLI

`@start()`

`@listen()`

条件逻辑：`or`

条件逻辑：`and`

在 `main.py` 中连接 Crews

选项 1：使用 `plot()` 方法