使用过Dify的同学都知道，你可以在上面拖动方框和箭头来编排大模型的逻辑，如下图所示。


这种拖动框图编排工作流的方式，确实非常简单方便，以至于不会代码的人也可以用来编排大模型Agent。但你有没有考虑过一个问题——你作为一个工程师，有没有可能通过写代码的形式来编排工作流？否则你和不懂代码的人相比有什么竞争力？

CrewAI是一个Agent开发框架，通过它可以非常方便地开发Agent。它提供的Flow功能，可以用来以编程的方式构建工作流。我向来推崇重器轻用的原则，虽然CrewAI是用来做Agent开发的，但它的Flow功能也可以用在不含AI的任何工程代码中。

我们来看一个例子。现在你要从硬盘中读取doc.txt文件，把里面的所有字母转换为大写。然后保存为doc_upper.txt。按常规的写法，我们把这个任务分为3步：

  1. 读取文件
  2. 转换大小写
  3. 写入文件

那么常规代码可能是这样写的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def step_1_read_file():
    with open('doc.txt') as f:
        content = f.read()
        return content

def step_2_to_upper(content):
    return content.upper()

def step_3_save_file(content):
    with open('doc_upper.txt', 'w') as f:
        f.write(content)

def start():
    content = step_1_read_file()
    content_upper = step_2_to_upper(content)
    step_3_save_file(content_upper)


start()

其中函数start就是用来控制代码的工作流。

现在，我们使用crewAI的flow功能来重构这个代码，那么代码可以写成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from crewai.flow.flow import Flow, listen, start
class UpperTask(Flow):
    @start()
    def step_1_read_file(self):
    with open('doc.txt') as f:
        content = f.read()
        return content

@listen(step_1_read_file)
def step_2_to_upper(self, content):
    return content.upper()

@listen(step_2_to_upper)
def step_3_save_file(self, content):
    with open('doc_upper.txt', 'w') as f:
        f.write(content)


flow = UpperTask()
result = flow.kickoff()

工作流会从@start装饰器装饰的方法开始运行。@listen装饰器用来装饰后续的每一个节点。当@listen参数对应的节点运行完成以后，就会自动触发自身装饰的节点。被listen的节点return的数据，就会作为参数传入当前节点。而kickoff()会返回最后一个被@listen装饰的节点的返回值。

Flow还支持状态管理、条件逻辑和路由控制。详情可以查看官方文档。Flow更方便的地方在于，它可以把你的工作流可视化出来，例如下面这段代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
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('test')

生成的流程图如下图所示：

对于简单的逻辑，可能不好区分使用Flow和常规写法的区别。但当你的代码流程多起来，逻辑复杂起来以后，使用Flow就会方便很多。