crew AI笔记[7] - flow特性示例
flow是什么
flow能够创建结构化、事件驱动的工作流，两大核心特点：
1、实现条件、循环、分支的逻辑
2、实现事件驱动（类似于中断机制）
有啥用呢？
不同任务之间共享状态，任务与任务之间实现无缝编排，输入输出高度可控，协作形式、输入输出格式定义都能更精确控制。
核心要素
@start
@listen
你可以监听一个方法或者某个状态
当方法/状态就绪的时候，会触发监听函数被动执行
@router[可选]
可以为一个结果的输出创建非常复杂的处理（路由）机制
状态管理
所有状态全部存入state属性
非结构化状态
用户手动修改状态，提供了高度的灵活性
结构化状态
利用pydantic定义更清晰严格的state属性
listen控制
or、and
在监听过程中，进行布尔逻辑运算，实现更复杂的事件触发逻辑【这个太好理解了，多说一个字都是浪费生命】
语法：
@listen(and_(start_method, second_method))router
@router()装饰器允许我们根据上一个flow的状态设置一个状态路由，你可以自主定义这个路由规则，进而更好控制后续的执行流。
有点绕？
假设上一个flow的过程是吃饭，那么吃完饭我该干嘛呢？我可以设置一个路由规则（你爱怎么定就怎么定）：
吃饱了，去看电视
没吃饱，去喝牛奶
。。。
下文给了一个实际例子，看了就明白了。
绘制流程图
图形化展示AI工作流程，展示各种任务、相互连接、数据流交互细节。
如何生成
方法一：
flow.plot("my_flow_plot")你会得到一个“my_flow_plot.html”
方法二：
$ crewai flow plot应用示例
flow单独使用
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModelclass ExampleState(BaseModel):counter: int = 0message: str = ""class StructuredExampleFlow(Flow[ExampleState]):@start()def first_method(self):self.state.message = "Hello from structured flow"print(f"State after first_method: {self.state}")@listen(first_method)def second_method(self):self.state.counter += 1self.state.message += " - updated"print(f"State after second_method: {self.state}")@listen(second_method)def third_method(self):self.state.counter += 1self.state.message += " - updated again"print(f"State after third_method: {self.state}")flow = StructuredExampleFlow()
flow.kickoff()
flow.plot("test")flow+crew联合使用
总体实现与上面类似，
1、把crew的实现用单独的类实现
2、在flow的listen里面调用crew的kickoff，获取实际输入输出内容
民以食为天，举个吃饭的例子吧 ^_^
test_flow_app
利用flow框架调用crew用一个agent写一首诗，另一个agent判断作者吃饱了没。
利用flow路由机制对不同的case进行不同处理。
from crewai.flow.flow import Flow, listen, start, router
from pydantic import BaseModelfrom test_poetry import poetry_crew
from test_distinguish import distinguish_crewclass ExampleState(BaseModel):counter: int = 0message: str = ""eating_state: str = ""class StructuredExampleFlow(Flow[ExampleState]):def __init__(self, input_str=""):super().__init__(ExampleState())self.input_str = input_str@start()def eating(self): result = poetry_crew.kickoff(inputs={'input_str': self.input_str}) # 以参数形式调用crewprint(result.raw) # 打印实际的诗歌内容self.state.message = result.raw # 只保存实际的文本内容print(f"State after first_method: {self.state}")@router(eating)def next_step(self): result = distinguish_crew.kickoff(inputs={'input_str': self.state.message}) print(f"State after next_step: {self.state}")self.state.eating_state = result.raw.strip() # 获取原始输出并去除空白字符print(result.raw) # 打印判断结果if self.state.eating_state == "full":return "watch_TV"elif self.state.eating_state == "hungry":return "drink milk"else:return "something wrong"@listen("watch_TV")def method_watch_tv(self):self.state.counter += 1self.state.message = "吃饱了撑得慌，去看看电影吧"print(f"State after method_watch_tv: {self.state}")@listen("drink milk")def method_drink_milk(self):self.state.counter += 1self.state.message = "没吃饱咋整？喝点牛奶对付一下好了"print(f"State after method_drink_milk: {self.state}")@listen("something wrong")def method_something_wrong(self):self.state.counter += 1self.state.message = "出了点问题"print(f"State after method_something_wrong: {self.state}")if __name__ == "__main__":input_str = '请就吃饭的场景写一首诗，要突出很美味，很满足'input_str = '请就吃饭的场景写一首诗，要突出很美味，不够吃'flow = StructuredExampleFlow(input_str)flow.kickoff()flow.plot("test")test_poetry
import os
from crewai import Agent, Task, Crew, LLMLLM_QW = LLM(model='dashscope/qwen-plus', base_url='https://dashscope.aliyuncs.com/compatible-mode/v1', api_key=os.environ.get("QWEN_API_KEY"), stream=False, # 设置为同步模式temperature=0.7, # 控制模型输出的随机性request_timeout=240, # 增加请求超时时间到240秒max_retries=2,
)poetry_agent = Agent(role='You are a poet. You can write poetry.',goal='Write beautiful poems based on the given themes',backstory='你是一位精通五言、七言绝句，严格遵循平仄对仗的诗人',llm=LLM_QW,
)poetry_task = Task(name='poetry_task',description='Write a poem based on the following theme: {input_str}',expected_output='一首对仗工整的诗歌',agent=poetry_agent,
)poetry_crew = Crew(agents=[poetry_agent],tasks=[poetry_task],llm=LLM_QW)if __name__ == '__main__':input_str = '请就吃饭的场景写一首诗，要突出我吃得很饱'input_str = '请就吃饭的场景写一首诗，要突出我没吃饱，肚子还很饿'result = poetry_crew.kickoff(inputs={'input_str': input_str})print(result)test_distinguish
import os
from crewai import Agent, Task, Crew, LLMLLM_QW = LLM(model='dashscope/qwen-plus', base_url='https://dashscope.aliyuncs.com/compatible-mode/v1', api_key=os.environ.get("QWEN_API_KEY"), stream=False, # 设置为同步模式temperature=0.7, # 控制模型输出的随机性request_timeout=240, # 增加请求超时时间到240秒max_retries=2,
)poetry_agent = Agent(role='你是一位精通诗歌的文人，你能分辨出诗歌的意境，并抓住要点',goal='明确判断一首诗的真实含义，并给出简明扼要的结论',backstory='你会读出一首诗，并给出对它的理解',llm=LLM_QW,
)poetry_task = Task(name='poetry_task',description='根据输入的诗歌内容: {input_str}，准确判断诗歌实际的意思，并直接输出判断',expected_output='只允许两种输出：full，hungry',agent=poetry_agent,
)distinguish_crew = Crew(agents=[poetry_agent],tasks=[poetry_task],llm=LLM_QW)if __name__ == '__main__':input_str = '碗尽粒无存，腹圆衣带紧。箸停犹抚腹，笑语饱中真。'input_str = "饭尽匙空碗底凉，腹鸣如鼓尚彷徨。盘中残渍犹堪恋，未饱饥肠恨日长。"result = distinguish_crew.kickoff(inputs={'input_str': input_str})print(result)