使用过Dify的同学都知道，你可以在上面拖动方框和箭头来编排大模型的逻辑，如下图所示。
这种拖动框图编排工作流的方式，确实非常简单方便，以至于不会代码的人也可以用来编排大模型Agent。但你有没有考虑过一个问题——你作为一个工程师，有没有可能通过写代码的形式来编排工作流？否则你和不懂代码的人相比有什么竞争力？
CrewAI是一个Agent开发框架，通过它可以非常方便地开发Agent。它提供的Flow功能，可以用来以编程的方式构建工作流。我向来推崇重器轻用的原则，虽然CrewAI是用来做Agent开发的，但它的Flow功能也可以用在不含AI的任何工程代码中。
我们来看一个例子。现在你要从硬盘中读取doc.txt文件，把里面的所有字母转换为大写。然后保存为doc_upper.txt。按常规的写法，我们把这个任务分为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就会方便很多。