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) |