tasks

Class info

Classes

Name	Children	Inherits
ChainAbortedError llmling_agent.tasks.exceptions Agent chain was aborted by user.		JobError
JobError llmling_agent.tasks.exceptions General task-related exception.	ToolSkippedError RunAbortedError ChainAbortedError JobRegistrationError	LLMLingError
JobRegistrationError llmling_agent.tasks.exceptions Task could not get registered.		JobError
RunAbortedError llmling_agent.tasks.exceptions Run was aborted by user.		JobError
TaskRegistry llmling_agent.tasks.registry Registry for managing tasks.		BaseRegistry
ToolSkippedError llmling_agent.tasks.exceptions Tool execution was skipped by user.		JobError

🛈 DocStrings

Task management.

ChainAbortedError

Bases: JobError

Agent chain was aborted by user.

Source code in src/llmling_agent/tasks/exceptions.py

class ChainAbortedError(JobError):
    """Agent chain was aborted by user."""

JobError

Bases: LLMLingError

General task-related exception.

Source code in src/llmling_agent/tasks/exceptions.py

class JobError(LLMLingError):
    """General task-related exception."""

JobRegistrationError

Bases: JobError

Task could not get registered.

Source code in src/llmling_agent/tasks/exceptions.py

class JobRegistrationError(JobError):
    """Task could not get registered."""

RunAbortedError

Bases: JobError

Run was aborted by user.

Source code in src/llmling_agent/tasks/exceptions.py

class RunAbortedError(JobError):
    """Run was aborted by user."""

TaskRegistry

Bases: BaseRegistry[str, Job[Any, Any]]

Registry for managing tasks.

Source code in src/llmling_agent/tasks/registry.py

class TaskRegistry(BaseRegistry[str, Job[Any, Any]]):
    """Registry for managing tasks."""

    @property
    def _error_class(self) -> type[JobRegistrationError]:
        return JobRegistrationError

    def _validate_item(self, item: Any) -> Job[Any, Any]:
        if not isinstance(item, Job):
            msg = f"Expected Job, got {type(item)}"
            raise self._error_class(msg)
        return item

    def register(self, name: str, task: Job[Any, Any], replace: bool = False):
        """Register a task with name.

        Creates a copy of the task with the name set.
        """
        task_copy = task.model_copy(update={"name": name})
        super().register(name, task_copy, replace)

register

register(name: str, task: Job[Any, Any], replace: bool = False)

Register a task with name.

Creates a copy of the task with the name set.

Source code in src/llmling_agent/tasks/registry.py

def register(self, name: str, task: Job[Any, Any], replace: bool = False):
    """Register a task with name.

    Creates a copy of the task with the name set.
    """
    task_copy = task.model_copy(update={"name": name})
    super().register(name, task_copy, replace)

ToolSkippedError

Bases: JobError

Tool execution was skipped by user.

Source code in src/llmling_agent/tasks/exceptions.py

class ToolSkippedError(JobError):
    """Tool execution was skipped by user."""