update openssa.l2.planning.abstract interfaces

openssa/l2/agent/abstract.py

@@ -8,6 +8,8 @@
 from pprint import pprint
 from typing import TYPE_CHECKING
 
+from tqdm import tqdm
+
 from openssa.l2.reasoning.base import BaseReasoner
 from openssa.l2.task.status import TaskStatus
 from openssa.l2.task.task import Task
@@ -82,13 +84,23 @@ def solve(self, problem: str, plan: APlan | None = None, dynamic: bool = True) -
 
         return result
 
-    def solve_dynamically(self, problem: str) -> str:
+    def solve_dynamically(self, problem: str, planner: APlanner = None) -> str:
+        """Solve task dynamically.
+
+        When first-pass result from Reasoner is unsatisfactory, decompose problem and recursively solve decomposed plan.
+        """
         self.reasoner.reason(task := Task(ask=problem, resources=self.resources))
 
-        if task.status == TaskStatus.NEEDING_DECOMPOSITION:
-            task.dynamic_decomposer: APlanner = self.planner.one_level_deep()
+        if (task.status == TaskStatus.NEEDING_DECOMPOSITION) and (planner := planner or self.planner).max_depth:
+            sub_planner: APlanner = planner.one_fewer_level_deep()
+
+            for sub_plan in tqdm(plan_1_level_deep := (planner.one_level_deep()
+                                                       .plan(problem=problem, resources=self.resources)).sub_plans):
+                sub_task: ATask = sub_plan.task
+                sub_task.result: str = self.solve_dynamically(problem=sub_task.ask, planner=sub_planner)
+                sub_task.status: TaskStatus = TaskStatus.DONE
 
-            for sub_plan in (plan_1_level_deep := task.decompose()).sub_plans:
-                sub_plan.task.dynamic_decomposer: APlanner = self.planner.one_fewer_level_deep()
+            task.result: str = plan_1_level_deep.execute(reasoner=self.reasoner)
+            task.status: TaskStatus = TaskStatus.DONE
 
-            plan_1_level_deep.execute(reasoner=self.reasoner)
+        return task.result

openssa/l2/planning/abstract/plan.py

@@ -14,6 +14,9 @@
     from openssa.l2.task.abstract import ATask
 
 
+type AskAnsPair = tuple[str, str]
+
+
 @dataclass
 class AbstractPlan(ABC):
     """Abstract plan."""
@@ -29,7 +32,7 @@ class AbstractPlan(ABC):
                                   kw_only=True)
 
     @abstractmethod
-    def execute(self, reasoner: AReasoner = BaseReasoner()) -> str:
+    def execute(self, reasoner: AReasoner = BaseReasoner(), other_results: list[AskAnsPair] | None = None) -> str:
         """Execute and return result, using specified reasoner to reason through involved tasks."""
 
 

openssa/l2/planning/abstract/planner.py

@@ -20,7 +20,7 @@ class AbstractPlanner(ABC):
 
     lm: AnLLM = field(default_factory=OpenAILLM.get_gpt_4_1106_preview)
 
-    max_depth: int = 3
+    max_depth: int = 2
     max_subtasks_per_decomp: int = 3
 
     @abstractmethod

openssa/l2/planning/hierarchical/__init__.py

@@ -10,7 +10,7 @@
 from loguru import logger
 from tqdm import tqdm
 
-from openssa.l2.planning.abstract.plan import AbstractPlan
+from openssa.l2.planning.abstract.plan import AbstractPlan, AskAnsPair
 from openssa.l2.planning.abstract.planner import AbstractPlanner
 from openssa.l2.reasoning.base import BaseReasoner
 from openssa.l2.task.status import TaskStatus
@@ -30,9 +30,6 @@ class HTPDict(TypedDict, total=False):
     sub_plans: NotRequired[list[HTPDict]]
 
 
-type AskAnsPair = tuple[str, str]
-
-
 @dataclass
 class HTP(AbstractPlan):
     """Hierarchical task plan (HTP)."""

openssa/l2/task/abstract.py

@@ -78,14 +78,10 @@ def from_dict_or_str(cls, dict_or_str: TaskDict | str, /) -> Self:
 
     def decompose(self) -> APlan:
         """Decompose task into modular plan."""
-        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 ***'
+        assert isinstance(self.dynamic_decomposer, AbstractPlanner), \
+            TypeError('*** Dynamic Decomposer must be Planner instance ***')
 
-        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
+        return self.dynamic_decomposer.plan(problem=self.ask, resources=self.resources)
 
 
 ATask: TypeVar = TypeVar('ATask', bound=AbstractTask, covariant=False, contravariant=False)