registry

Class info

Classes

Name	Children	Inherits
BaseRegistry llmling_agent.utils.baseregistry Base class for registries providing item storage and change notifications.	ConnectionRegistry SkillsRegistry VFSRegistry TaskRegistry AgentPool	MutableMapping ABC Generic
Job llmling_agent_config.task A task is a piece of work that can be executed by an agent.		Schema Generic
JobRegistrationError llmling_agent.tasks.exceptions Task could not get registered.		JobError
TaskRegistry llmling_agent.tasks.registry Registry for managing tasks.		BaseRegistry

🛈 DocStrings

Task definition and registry for agents.

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]:
        from llmling_agent_config.task import Job

        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) -> None:
        """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) -> None

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) -> None:
    """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)