CrewAI 基础 05-FLow

Crew 组建了团队，团队内共享知识，解决特定的问题，Crew 团队内的 Agent 是以任务顺序或者分层的思想去执行，实际工作会有更加复杂的流程，因此 Crew 定义了工作流 (Flow)，进一步明确规划 Agent 的执行过程

特性	Crew	Flow
定义	一组协作完成特定任务的 AI 代理	用于创建和管理 AI 工作流的高级功能
执行方式	顺序（Sequential）或层级（Hierarchical）	提供复杂的工作流管理，包括状态共享和事件驱动执行
复杂度	专注于特定任务的代理协作	提供更高层次的结构，用于组织和连接多个 Crew 或任务
主要功能	代理间协作和任务分配	简化工作流创建、状态管理、事件驱动架构、灵活控制流
应用场景	适用于需要多个 AI 代理协作的单一任务或相关任务集	适用于创建多步骤流程、链接不同 Crew 的输出，实现复杂逻辑
状态管理	基本的任务间状态传递	提供强大的结构化和非结构化状态管理机制

# 获取简历->筛选简历(财务部，人事部)->确认面试人员
from crewai.flow.flow import Flow,listen,start
import nest_asyncio
nest_asyncio.apply()

from pydantic import BaseModel
class curriculum(BaseModel):
    name: str
    person_info: str
    pay: str
    year: str

class curriculum_list(BaseModel):
    curriculums: list[curriculum]

# 获取简历->筛选简历(财务部)->确认面试人员
class RecruitmentFLow(Flow):
    from ollama import Client
    client=Client(host='http://localhost:11434')
    
    @start()
    def generate_curriculums(self):
        print("流程开始...")

        messages=[
            {'role':'user','content':"""
            生成5份简历，包含2名计算机视觉工程师，其他是销售、人事、管理岗位的
            只包含人名，个人情况、要求薪资、工作年限，
            只输出简历文本，不输出其他任何内容"""}
        ]

        response=self.client.chat(
            model='qwen2.5:latest',
            messages=messages,
            format=curriculum_list.schema()
        )

        curriculums_json=response['message']['content']
        print('curriculums_json:',curriculums_json)

        return curriculums_json

    @listen(generate_curriculums)
    def filter_curriculum_MinistryofPersonnel(self,curriculums_json):
        print('人事部筛选简历...')

        messages=[
            {'role':'user','content':f"""
            按照要求筛选以下简历，要求：招收一名有5年以上工作经验得计算机视觉工程师，
            薪资不能超过25000一个月；简历{curriculums_json}
            """}
        ]

        response=self.client.chat(
            model='qwen2.5:latest',
            messages=messages,
            format=curriculum_list.schema()
        )

        filter_curriculums_json=response['message']['content']
        # print('filter_curriculums_json:',filter_curriculums_json)

        return filter_curriculums_json

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

流程开始...
curriculums_json: {
    "curriculums": [
        {
            "name": "张伟",
            "person_info": "计算机视觉工程师，5 年相关工作经验，熟悉深度学习与图像处理技术。",
            "pay": "12000-15000 元/月",
            "year": "5"
        },
        {
            "name": "李丽",
            "person_info": "计算机视觉工程师，3 年相关工作经验，具有较强的算法研发能力。",
            "pay": "8000-10000 元/月",
            "year": "3"
        },
        {
            "name": "王明",
            "person_info": "销售经理，6 年销售经验，擅长市场开发与客户关系管理。",
            "pay": "20000-25000 元/月",
            "year": "6"
        },
        {
            "name": "赵芳",
            "person_info": "人事专员，4 年人力资源工作经验，熟悉招聘、培训及员工关系管理流程。",
            "pay": "7000-9000 元/月",
            "year": "4"
        },
        {
            "name": "陈涛",
            "person_info": "项目管理，2 年项目管理经验，擅长项目规划与执行管控。",
            "pay": "10000-13000 元/月",
            "year": "2"
        }
    ]
}
人事部筛选简历...
{
    "curriculums": [
        {
            "name": "张伟",
            "person_info": "计算机视觉工程师，5 年相关工作经验，熟悉深度学习与图像处理技术。",
            "pay": "12000-15000 元/月",
            "year": "5"
        }
    ]
}

在构建 FLow，有两个关键的装饰器：start、listen

• start：用于将方法标记为 Flow 的起点。当 Flow 启动时，所有修饰的方法都将并行执行。一个 Flow 中可以有多个 start 方法，它们都将在 Flow 启动时执行
• listen: 用于将方法标记为 Flow 中另一个任务输出的侦听器。当指定的 task 发出 output 时，将执行 with 修饰的方法。该方法可以访问它正在侦听的任务的输出作为参数

# 将工作流程绘制到html
flow.plot("my_flow_plot")

Plot saved as my_flow_plot. Html