Merge pull request #156 from aitomatic/planning/abstract

update openssa.l2.planning.abstract.AbstractPlan & AbstractPlanner

openssa/l2/agent/abstract.py

@@ -1,20 +1,22 @@
 """Abstract agent with planning, reasoning & informational resources."""
 
 
+from __future__ import annotations
+
 from abc import ABC
 from dataclasses import dataclass, field
 from pprint import pprint
 from typing import TYPE_CHECKING
 
-from openssa.l2.planning.abstract.plan import APlan
-from openssa.l2.planning.abstract.planner import APlanner
-from openssa.l2.reasoning.abstract import AReasoner
 from openssa.l2.reasoning.base import BaseReasoner
-from openssa.l2.resource.abstract import AResource
 from openssa.l2.task.status import TaskStatus
 from openssa.l2.task.task import Task
 
 if TYPE_CHECKING:
+    from openssa.l2.planning.abstract.plan import APlan
+    from openssa.l2.planning.abstract.planner import APlanner
+    from openssa.l2.reasoning.abstract import AReasoner
+    from openssa.l2.resource.abstract import AResource
     from openssa.l2.task.abstract import ATask
 
 

openssa/l2/planning/abstract/plan.py

@@ -4,8 +4,8 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, TypeVar
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Self, TypeVar
 
 from openssa.l2.reasoning.base import BaseReasoner
 
@@ -17,8 +17,17 @@
 @dataclass
 class AbstractPlan(ABC):
     """Abstract plan."""
+
     task: ATask
 
+    sub_plans: list[Self] = field(default_factory=list,
+                                  init=True,
+                                  repr=True,
+                                  hash=False,  # mutable
+                                  compare=True,
+                                  metadata=None,
+                                  kw_only=True)
+
     @abstractmethod
     def execute(self, reasoner: AReasoner = BaseReasoner()) -> str:
         """Execute and return result, using specified reasoner to reason through involved tasks."""

openssa/l2/planning/hierarchical/__init__.py

@@ -37,14 +37,6 @@ class HTPDict(TypedDict, total=False):
 class HTP(AbstractPlan):
     """Hierarchical task plan (HTP)."""
 
-    sub_plans: list[HTP] = field(default_factory=list,
-                                 init=True,
-                                 repr=True,
-                                 hash=False,  # mutable
-                                 compare=True,
-                                 metadata=None,
-                                 kw_only=True)
-
     @classmethod
     def from_dict(cls, htp_dict: HTPDict, /) -> HTP:
         """Create HTP from dictionary representation."""

openssa/l2/task/abstract.py

@@ -7,6 +7,7 @@
 from dataclasses import dataclass, asdict, field
 from typing import TYPE_CHECKING, Self, TypedDict, Required, NotRequired, TypeVar
 
+from openssa.l2.planning.abstract.planner import AbstractPlanner
 from openssa.l2.resource._global import GLOBAL_RESOURCES
 
 from .status import TaskStatus
@@ -77,8 +78,14 @@ def from_dict_or_str(cls, dict_or_str: TaskDict | str, /) -> Self:
 
     def decompose(self) -> APlan:
         """Decompose task into modular plan."""
-        assert self.dynamic_decomposer, '*** MISSING DYNAMIC DECOMPOSER ***'
-        return self.dynamic_decomposer.plan(problem=self.ask, resources=self.resources)
+        assert isinstance(self.dynamic_decomposer, AbstractPlanner), '*** Dynamic Decomposer must be Planner instance ***'
+        assert isinstance(self.dynamic_decomposer.max_depth), '*** Dynamic Decomposer must have positive Max Depth ***'
+
+        for sub_plan in (plan := self.dynamic_decomposer.plan(problem=self.ask, resources=self.resources)):
+            (sub_task := sub_plan.task).resources: set[AResource] = self.resources
+            sub_task.dynamic_decomposer: APlanner = self.dynamic_decomposer.one_fewer_level_deep()
+
+        return plan
 
 
 ATask: TypeVar = TypeVar('ATask', bound=AbstractTask, covariant=False, contravariant=False)