结构化智能：Flow 简介

结构化智能：Flows介绍

虽然Crew在编排任务方面功能强大，但实际应用通常需要的不仅仅是AI响应链。你可能需要运行代码、等待事件、执行条件逻辑或在不同的AI操作之间管理状态。这正是Flows发挥作用的地方。

Flows代表了在CrewAI中构建AI应用程序的现代、事件驱动方式。它们允许你将结构化编程逻辑与AI Agent的灵活性结合起来。

状态管理

在一个Flow中，状态是你应用程序的内存。它允许工作流的不同步骤共享数据。与简单的函数链不同，一个Flow维护一个状态对象，该对象可供该Flow中的所有方法访问。这对于将一个Agent的输出作为Python逻辑的输入，反之亦然，至关重要。

定义Flow

要创建一个Flow，你需要从Flow类继承，并使用装饰器定义你的逻辑。两个最关键的装饰器是：

• @start(): 标记启动Flow的方法。当你运行.kickoff()时，此方法首先执行。
• @listen(method_name): 标记一个等待另一个方法完成的方法。它接收其监听方法的输出。

构建你的第一个Flow

让我们看一个简单的例子，我们生成一个城市名称，然后生成一个关于它的趣事。请注意，我们不仅仅使用Agent；如果需要，我们也可以混合使用标准Python代码。

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

class CityFunFactFlow(Flow):

    @start()
    def generate_city(self):
        # 选择城市的逻辑（可以是Agent或简单代码）
        print("步骤 1: 正在选择城市...")
        return "Tokyo"

    @listen(generate_city)
    def generate_fact(self, city_name):
        # 此方法等待 generate_city 完成
        print(f"步骤 2: 正在为 {city_name} 生成趣事...")

        # 我们可以在Flow步骤中使用Crew！
        agent = Agent(role='Traveler', goal='Share facts', backstory='...')
        task = Task(
            description=f'Tell me a fun fact about {city_name}',
            expected_output='A one sentence fact.',
            agent=agent
        )
        crew = Crew(agents=[agent], tasks=[task])
        result = crew.kickoff()
        return result.raw

# 运行Flow
flow = CityFunFactFlow()
final_output = flow.kickoff()
print(f"最终结果: {final_output}")

这种架构允许你构建复杂的、非线性工作流（DAGs），其中多个方法可以监听相同的启动事件，并行运行，并在稍后重新汇聚。它弥合了传统软件工程模式与AI编排之间的差距。