Spaces:
Sleeping
Sleeping
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 | |
from crewai.utilities import I18N | |
class Task(BaseModel): | |
"""Class that represent a task to be executed.""" | |
__hash__ = object.__hash__ # type: ignore | |
i18n: I18N = I18N() | |
description: str = Field(description="Description of the actual task.") | |
callback: Optional[Any] = Field( | |
description="Callback to be executed after the task is completed.", default=None | |
) | |
agent: Optional[Agent] = Field( | |
description="Agent responsible for executiong the task.", default=None | |
) | |
expected_output: Optional[str] = Field( | |
description="Clear definition of expected output for the task.", | |
default=None, | |
) | |
output: Optional[TaskOutput] = Field( | |
description="Task output, it's final result after being executed", default=None | |
) | |
tools: List[Any] = Field( | |
default_factory=list, | |
description="Tools the agent is limited to use for this task.", | |
) | |
id: UUID4 = Field( | |
default_factory=uuid.uuid4, | |
frozen=True, | |
description="Unique identifier for the object, not set by user.", | |
) | |
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.", {} | |
) | |
def check_tools(self): | |
"""Check if the tools are set.""" | |
if not self.tools and self.agent and self.agent.tools: | |
self.tools.extend(self.agent.tools) | |
return self | |
def execute(self, context: Optional[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._prompt(), context=context, tools=self.tools | |
) | |
self.output = TaskOutput(description=self.description, result=result) | |
self.callback(self.output) if self.callback else None | |
return result | |
def _prompt(self) -> str: | |
"""Prompt the task. | |
Returns: | |
Prompt of the task. | |
""" | |
tasks_slices = [self.description] | |
if self.expected_output: | |
output = self.i18n.slice("expected_output").format( | |
expected_output=self.expected_output | |
) | |
tasks_slices = [self.description, output] | |
return "\n".join(tasks_slices) |