企业官网建设流程全解析

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.py：

• ConditionalTask继承自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_skip：crew 在执行任务前取已完成任务输出列表的最后一个元素（task_outputs[-1]）作为判定输入。如果条件函数返回False，crew 会记录一条 debug 级别的日志Skipping conditional task: {任务 description}，然后调用get_skipped_task_output()生成一个占位TaskOutput并写入执行日志。也就是说，跳过并不会让流程中断，后续任务收到的仍是一个TaskOutput对象，只是内容为空raw字符串、output_format为RAW，agent记录为该任务 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( role="Data Fetcher", goal="Fetch data online using Serper tool", backstory="Backstory 1", verbose=True, tools=[SerperDevTool()] ) data_processor_agent = Agent( role="Data Processor", goal="Process fetched data", backstory="Backstory 2", verbose=True ) summary_generator_agent = Agent( role="Summary Generator", goal="Generate summary from fetched data", backstory="Backstory 3", verbose=True ) task1 = Task( description="Fetch data about events in San Francisco using Serper tool", expected_output="List of 10 things to do in SF this week", agent=data_fetcher_agent, output_pydantic=EventOutput, ) 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_output="List of 10 Things to do in SF this week", condition=is_data_missing, agent=data_processor_agent, ) task3 = Task( description="Generate summary of events in San Francisco from fetched data", expected_output="A complete report on the customer and their customers and competitors, including their demographics, preferences, market positioning and audience engagement.", agent=summary_generator_agent, ) crew = Crew( agents=[data_fetcher_agent, data_processor_agent, summary_generator_agent], tasks=[task1, conditional_task, task3], verbose=True, planning=True ) result = crew.kickoff() print("results", result)

几个关键点：

• conditional_task必须排在task1之后——判定依据就是task1的TaskOutput，把它放在第一位会违反"不能是第一个任务"的约束。
• task1上的output_pydantic=EventOutput是让条件函数能读取output.pydantic.events的依据，两者要配套。
• planning=True是官方示例中的 crew 参数，示例原样保留。

如何验证任务被跳过还是被执行

运行crew.kickoff()之后，可以从三个位置核对结果：

1. 跳过日志：源码中跳过分支记录的日志文本是Skipping conditional task: {description}，级别为 debug。看到这条日志（或其对应级别输出）说明条件函数返回了False。
2. 任务输出对象：通过conditional_task.output访问该任务的TaskOutput。
   • 若任务被跳过，得到的是get_skipped_task_output()生成的占位对象：raw为空字符串、output_format为RAW，agent为Data Processor。
   • 若任务被执行，raw中为该 agent 的实际产出。
3. 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_output，agent值需与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