chore: enhance typing and documentation in tasks module (#3467)

- Disable E501 line length linting rule
- Add Google-style docstrings to tasks leaf file
- Modernize typing and docs in task_output.py
- Improve typing and documentation in conditional_task.py
This commit is contained in:
Greyson LaLonde
2025-09-08 11:42:23 -04:00
committed by GitHub
parent 37c5e88d02
commit f936e0f69b
4 changed files with 87 additions and 24 deletions

View File

@@ -1,4 +1,7 @@
from typing import Any, Callable
"""Conditional task execution based on previous task output."""
from collections.abc import Callable
from typing import Any
from pydantic import Field
@@ -8,37 +11,54 @@ from crewai.tasks.task_output import TaskOutput
class ConditionalTask(Task):
"""
A task that can be conditionally executed based on the output of another task.
Note: This cannot be the only task you have in your crew and cannot be the first since its needs context from the previous task.
"""A task that can be conditionally executed based on the output of another task.
This task type allows for dynamic workflow execution based on the results of
previous tasks in the crew execution chain.
Attributes:
condition: Function that evaluates previous task output to determine execution.
Notes:
- Cannot be the only task in your crew
- Cannot be the first task since it needs context from the previous task
"""
condition: Callable[[TaskOutput], bool] = Field(
condition: Callable[[TaskOutput], bool] | None = Field(
default=None,
description="Maximum number of retries for an agent to execute a task when an error occurs.",
description="Function that determines whether the task should be executed based on previous task output.",
)
def __init__(
self,
condition: Callable[[Any], bool],
condition: Callable[[Any], bool] | None = None,
**kwargs,
):
super().__init__(**kwargs)
self.condition = condition
def should_execute(self, context: TaskOutput) -> bool:
"""
Determines whether the conditional task should be executed based on the provided context.
"""Determines whether the conditional task should be executed based on the provided context.
Args:
context (Any): The context or output from the previous task that will be evaluated by the condition.
context: The output from the previous task that will be evaluated by the condition.
Returns:
bool: True if the task should be executed, False otherwise.
True if the task should be executed, False otherwise.
Raises:
ValueError: If no condition function is set.
"""
if self.condition is None:
raise ValueError("No condition function set for conditional task")
return self.condition(context)
def get_skipped_task_output(self):
def get_skipped_task_output(self) -> TaskOutput:
"""Generate a TaskOutput for when the conditional task is skipped.
Returns:
Empty TaskOutput with RAW format indicating the task was skipped.
"""
return TaskOutput(
description=self.description,
raw="",