CrewAI 如何用 ConditionalTask 按上一步结果决定跳过或执行任务

CrewAI 如何用 ConditionalTask 按上一步结果决定跳过或执行任务【免费下载链接】crewAIFramework for orchestrating role-playing, autonomous AI agents. By fostering collaborative intelligence, CrewAI empowers agents to work together seamlessly, tackling complex tasks.项目地址: https://gitcode.com/GitHub_Trending/cr/crewAI在一个 CrewAI Crew 里有些步骤只有在上一个任务的结果满足某个条件时才需要运行例如抓取到的数据不足时再去补抓。ConditionalTask是Task的子类通过一个condition函数在运行时评估前一个任务的TaskOutput条件返回False时跳过该任务返回True时执行该任务。本文按官方示例抓取数据 → 按条件补抓数据 → 生成摘要的三步流程走一遍定义条件函数、接入 crew、运行后验证任务是被跳过还是被执行。前提条件环境已安装示例导入所需的crewai与crewai_tools两个包示例的 import 即来自这两者。示例使用SerperDevTool在线抓取数据需要 API key。参考 Tasks 文档 中 Creating a Task with Tools 一节的写法通过环境变量传入import os os.environ[OPENAI_API_KEY] Your Key os.environ[SERPER_API_KEY] Your Key # serper.dev API key上面两处Your Key是文档原有的占位写法替换为你自己的密钥后使用。先明确判定机制与位置约束以下事实来自源码 conditional_task.pyConditionalTask继承自Task新增一个condition字段用途是determines whether the task should be executed based on previous task output根据上一个任务的输出决定本任务是否执行。should_execute(context)方法接收上一个任务的TaskOutput调用condition并把结果转成bool如果没有设置condition会抛出ValueError(No condition function set for conditional task)。类注释明确了两条位置约束ConditionalTask 不能是 crew 中唯一的任务也不能是第一个任务因为判定需要前一个任务的输出作为上下文。运行时行为见 check_conditional_skipcrew 在执行任务前取已完成任务输出列表的最后一个元素task_outputs[-1]作为判定输入。如果条件函数返回Falsecrew 会记录一条 debug 级别的日志Skipping conditional task: {任务 description}然后调用get_skipped_task_output()生成一个占位TaskOutput并写入执行日志。也就是说跳过并不会让流程中断后续任务收到的仍是一个TaskOutput对象只是内容为空raw字符串、output_format为RAWagent记录为该任务 agent 的 role未指定 agent 时为空字符串。第一步编写条件函数条件函数的签名是「接收一个TaskOutput返回bool」官方示例为from crewai.tasks.task_output import TaskOutput # If false, the task will be skipped, if true, then execute the task. def is_data_missing(output: TaskOutput) - bool: return len(output.pydantic.events) 10这里判定的是output.pydantic.events前提是上一个任务必须通过output_pydantic声明了结构化输出模型否则TaskOutput中不会包含pydantic字段Tasks 文档 在 Task Output 一节说明了TaskOutput默认只有raw只有任务配置了output_pydantic或output_json时才会分别包含pydantic、json_dict输出。示例中配套的定义是from typing import List from pydantic import BaseModel class EventOutput(BaseModel): events: List[str]第二步组装完整 crew 并运行完整示例来自 Conditional Tasks 文档接上一步的环境变量设置之后运行from crewai import Agent, Crew from crewai.tasks.conditional_task import ConditionalTask from crewai.task import Task from crewai_tools import SerperDevTool # 条件函数与 EventOutput 定义见上一步 # def is_data_missing(output: TaskOutput) - bool: ... # class EventOutput(BaseModel): events: List[str] data_fetcher_agent Agent( roleData Fetcher, goalFetch data online using Serper tool, backstoryBackstory 1, verboseTrue, tools[SerperDevTool()] ) data_processor_agent Agent( roleData Processor, goalProcess fetched data, backstoryBackstory 2, verboseTrue ) summary_generator_agent Agent( roleSummary Generator, goalGenerate summary from fetched data, backstoryBackstory 3, verboseTrue ) task1 Task( descriptionFetch data about events in San Francisco using Serper tool, expected_outputList of 10 things to do in SF this week, agentdata_fetcher_agent, output_pydanticEventOutput, ) conditional_task ConditionalTask( description Check if data is missing. If we have less than 10 events, fetch more events using Serper tool so that we have a total of 10 events in SF this week.. , expected_outputList of 10 Things to do in SF this week, conditionis_data_missing, agentdata_processor_agent, ) task3 Task( descriptionGenerate summary of events in San Francisco from fetched data, expected_outputA complete report on the customer and their customers and competitors, including their demographics, preferences, market positioning and audience engagement., agentsummary_generator_agent, ) crew Crew( agents[data_fetcher_agent, data_processor_agent, summary_generator_agent], tasks[task1, conditional_task, task3], verboseTrue, planningTrue ) result crew.kickoff() print(results, result)几个关键点conditional_task必须排在task1之后——判定依据就是task1的TaskOutput把它放在第一位会违反不能是第一个任务的约束。task1上的output_pydanticEventOutput是让条件函数能读取output.pydantic.events的依据两者要配套。planningTrue是官方示例中的 crew 参数示例原样保留。如何验证任务被跳过还是被执行运行crew.kickoff()之后可以从三个位置核对结果跳过日志源码中跳过分支记录的日志文本是Skipping conditional task: {description}级别为 debug。看到这条日志或其对应级别输出说明条件函数返回了False。任务输出对象通过conditional_task.output访问该任务的TaskOutput。若任务被跳过得到的是get_skipped_task_output()生成的占位对象raw为空字符串、output_format为RAWagent为Data Processor。若任务被执行raw中为该 agent 的实际产出。crew 最终结果result crew.kickoff()打印的results是最终输出Tasks 文档 说明 crew 的最后一个任务的输出即为 crew 本身的最终输出所以result对应task3的产出而不是被跳过任务的占位对象。在 JSONC 项目中声明条件任务可选新创建的 crew 项目crewai create crew name在crew.jsonc中定义任务。Tasks 文档 说明任务条目支持任意公开的Task字段要声明条件任务在条目中使用type: ConditionalTask并提供condition字段。JSONC 任务条目的其他约束不变每个任务必须包含description和expected_outputagent值需与agents列表中的名字匹配context只能引用前面的任务名。限制condition未设置时should_execute会抛出ValueError不要创建不带条件的ConditionalTask参与运行。ConditionalTask不能作为 crew 中唯一的任务也不能作为第一个任务。跳过分支不会向后续任务传递真实内容后续任务拿到的是空raw、RAW 格式的占位TaskOutput条件函数里依赖后续逻辑时需要把这一点纳入设计。【免费下载链接】crewAIFramework for orchestrating role-playing, autonomous AI agents. By fostering collaborative intelligence, CrewAI empowers agents to work together seamlessly, tackling complex tasks.项目地址: https://gitcode.com/GitHub_Trending/cr/crewAI创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考