|
import uuid |
|
from typing import Any, List, Optional |
|
|
|
from pydantic import UUID4, BaseModel, Field, field_validator, model_validator |
|
from pydantic_core import PydanticCustomError |
|
|
|
from crewai.agent import Agent |
|
from crewai.tasks.task_output import TaskOutput |
|
|
|
|
|
class Task(BaseModel): |
|
"""Class that represent a task to be executed.""" |
|
|
|
__hash__ = object.__hash__ |
|
description: str = Field(description="Description of the actual task.") |
|
agent: Optional[Agent] = Field( |
|
description="Agent responsible for the task.", default=None |
|
) |
|
tools: List[Any] = Field( |
|
default_factory=list, |
|
description="Tools the agent are limited to use for this task.", |
|
) |
|
output: Optional[TaskOutput] = Field( |
|
description="Task output, it's final result.", default=None |
|
) |
|
id: UUID4 = Field( |
|
default_factory=uuid.uuid4, |
|
frozen=True, |
|
description="Unique identifier for the object, not set by user.", |
|
) |
|
|
|
@field_validator("id", mode="before") |
|
@classmethod |
|
def _deny_user_set_id(cls, v: Optional[UUID4]) -> None: |
|
if v: |
|
raise PydanticCustomError( |
|
"may_not_set_field", "This field is not to be set by the user.", {} |
|
) |
|
|
|
@model_validator(mode="after") |
|
def check_tools(self): |
|
if not self.tools and (self.agent and self.agent.tools): |
|
self.tools.extend(self.agent.tools) |
|
return self |
|
|
|
def execute(self, context: str = None) -> str: |
|
"""Execute the task. |
|
|
|
Returns: |
|
Output of the task. |
|
""" |
|
if not self.agent: |
|
raise Exception( |
|
f"The task '{self.description}' has no agent assigned, therefore it can't be executed directly and should be executed in a Crew using a specific process that support that, either consensual or hierarchical." |
|
) |
|
result = self.agent.execute_task( |
|
task=self.description, context=context, tools=self.tools |
|
) |
|
|
|
self.output = TaskOutput(description=self.description, result=result) |
|
return result |
