update openssa.l2.reasoning.base.BaseReasoner

examples/Planning-and-Reasoning.ipynb

@@ -64,7 +64,7 @@
     "\n",
     "from openssa import (Agent,\n",
     "                     HTP, AutoHTPlanner,\n",
-    "                     OodaReasoner,\n",
+    "                     BaseReasoner, OodaReasoner,\n",
     "                     FileResource)"
    ]
   },
@@ -97,6 +97,62 @@
     "display(Markdown(resource.overview))"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Base Reasoning"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "base_reasoning_agent = Agent(planner=None,\n",
+    "                             reasoner=BaseReasoner(),\n",
+    "                             resources={resource})\n",
+    "pprint(base_reasoning_agent)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "display(Markdown(base_reasoning_agent.solve(PROBLEM)))"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# OODA Reasoning"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "ooda_reasoning_agent = Agent(planner=None,\n",
+    "                             reasoner=OodaReasoner(),\n",
+    "                             resources={resource})\n",
+    "pprint(ooda_reasoning_agent)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "display(Markdown(ooda_reasoning_agent.solve(PROBLEM)))"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},
@@ -139,7 +195,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "solution_1 = agent.solve(PROBLEM)\n",
+    "solution_1 = agent.solve(PROBLEM, dynamic=False)\n",
     "display(Markdown(solution_1))"
    ]
   },
@@ -195,7 +251,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "solution_2 = agent.solve(PROBLEM, plan=expert_plan)\n",
+    "solution_2 = agent.solve(PROBLEM, plan=expert_plan, dynamic=False)\n",
     "display(Markdown(solution_2))"
    ]
   }
@@ -216,7 +272,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.12.2"
+   "version": "3.12.3"
   }
  },
  "nbformat": 4,

openssa/l2/agent/abstract.py

@@ -4,12 +4,17 @@
 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.task import Task
+
+if TYPE_CHECKING:
+    from openssa.l2.task.abstract import ATask
 
 
 @dataclass
@@ -30,11 +35,54 @@
     def resource_overviews(self) -> dict[str, str]:
         return {r.unique_name: r.overview for r in self.resources}
 
-    def solve(self, problem: str, plan: APlan | None = None) -> str:
+    def solve(self, problem: str, plan: APlan | None = None, dynamic: bool = True) -> str | None:
         """Solve problem, with an automatically generated plan (default) or explicitly specified plan."""
-        plan: APlan = (self.planner.update_plan_resources(plan, resources=self.resources)
-                       if plan
-                       else self.planner.plan(problem, resources=self.resources))
-        pprint(plan)
+        match (plan, self.planner, dynamic):
+            case (None, None, _):
+                # if neither Plan nor Planner is given, directly use Reasoner
+                result: str | None = self.reasoner.reason(task=Task(ask=problem, resources=self.resources))
+
+            case (None, _, False) if self.planner:
+                # if no Plan is given but Planner is, and if solving statically,
+                # then use Planner to generate static Plan,
+                # then execute such static plan
+                plan: APlan = self.planner.plan(problem, resources=self.resources)
+                pprint(plan)
+                result: str = plan.execute(reasoner=self.reasoner)
+
+            case (None, _, True) if self.planner:
+                # if no Plan is given but Planner is, and if solving dynamically,
+                # then first directly use Reasoner,
+                # and if that does not work, then use Planner to decompose 1 level more deeply,
+                # and recurse until reaching confident solution or running out of depth
+                result: str = self.solve_dynamically(problem=problem)
+
+            case (_, None, _) if plan:
+                # if Plan is given but no Planner is, then execute Plan statically
+                result: str = plan.execute(reasoner=self.reasoner)
+
+            case (_, _, False) if (plan and self.planner):
+                # if both Plan and Planner are given, and if solving statically,
+                # then use Planner to update Plan's resources,
+                # then execute such updated static Plan
+                plan: APlan = self.planner.update_plan_resources(plan, resources=self.resources)
+                pprint(plan)
+                result: str = plan.execute(reasoner=self.reasoner)
+
+            case (_, _, True) if (plan and self.planner):
+                # if both Plan and Planner are given, and if solving dynamically,
+                # TODO: dynamic solution
+                raise NotImplementedError('Dynamic execution of given Plan and Planner not yet implemented')
+
+            case _:
+                raise ValueError('*** Invalid Plan-Planner-Dynamism Combination ***')
+
+        return result
 
-        return plan.execute(reasoner=self.reasoner)
+    def solve_dynamically(self, problem: str) -> str | None:
+        task: ATask = Task(ask=problem, resources=self.resources)
+        if (result := self.reasoner.reason(task)) is None:
+            planner_1_level_deep: APlanner = self.planner.one_level_deep()
+            plan: APlan = planner_1_level_deep.plan(problem=problem)
+            planner_1_level_fewer_deep: APlanner = self.planner.one_fewer_level_deep()
+            ...

openssa/l2/agent/agent.py

@@ -5,6 +5,8 @@
 
 from openssa.l2.planning.abstract.planner import APlanner
 from openssa.l2.planning.hierarchical import AutoHTPlanner
+from openssa.l2.reasoning.abstract import AReasoner
+from openssa.l2.reasoning.ooda import OodaReasoner
 
 from .abstract import AbstractAgent
 
@@ -14,3 +16,4 @@ class Agent(AbstractAgent):
     """Agent with planning, reasoning & informational resources."""
 
     planner: APlanner = field(default_factory=AutoHTPlanner)
+    reasoner: AReasoner = field(default_factory=OodaReasoner)

openssa/l2/planning/abstract/planner.py

@@ -5,7 +5,7 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, TypeVar
+from typing import TYPE_CHECKING, Self, TypeVar
 
 from openssa.utils.llms import AnLLM, OpenAILLM
 
@@ -20,6 +20,17 @@ class AbstractPlanner(ABC):
 
     lm: AnLLM = field(default_factory=OpenAILLM.get_gpt_4_1106_preview)
 
+    max_depth: int = 3
+    max_subtasks_per_decomp: int = 3
+
+    @abstractmethod
+    def one_level_deep(self) -> Self:
+        """Make 1-level-deep planner."""
+
+    @abstractmethod
+    def one_fewer_level_deep(self) -> Self:
+        """Make 1-fewer-level-deep planner."""
+
     @abstractmethod
     def plan(self, problem: str, resources: set[AResource] | None = None) -> APlan:
         """Make plan for solving problem based on informational resources."""

openssa/l2/planning/hierarchical/__init__.py

@@ -3,7 +3,7 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass, asdict, field
+from dataclasses import dataclass, field
 import json
 from typing import TYPE_CHECKING, TypedDict, Required, NotRequired
 
@@ -53,14 +53,14 @@ def from_dict(cls, htp_dict: HTPDict, /) -> HTP:
 
     def to_dict(self) -> HTPDict:
         """Return dictionary representation of HTP."""
-        return {'task': asdict(self.task),
+        return {'task': self.task.to_json_dict(),
                 'sub-plans': [p.to_dict() for p in self.sub_plans]}
 
     def fix_missing_resources(self):
         """Fix missing resources in HTP."""
         for p in self.sub_plans:
             if not p.task.resources:
-                p.task.resources: set[AResource] | None = self.task.resources
+                p.task.resources: set[AResource] = self.task.resources
             p.fix_missing_resources()
 
     def execute(self, reasoner: AReasoner = BaseReasoner(), other_results: list[AskAnsPair] | None = None) -> str:
@@ -103,8 +103,17 @@ def execute(self, reasoner: AReasoner = BaseReasoner(), other_results: list[AskA
 class AutoHTPlanner(AbstractPlanner):
     """Automated (generative) hierarchical task planner."""
 
-    max_depth: int = 3
-    max_subtasks_per_decomp: int = 3
+    def one_level_deep(self) -> AutoHTPlanner:
+        """Make 1-level-deep planner."""
+        return AutoHTPlanner(lm=self.lm,
+                             max_depth=1,
+                             max_subtasks_per_decomp=self.max_subtasks_per_decomp)
+
+    def one_fewer_level_deep(self) -> AutoHTPlanner:
+        """Make 1-fewer-level-deep planner."""
+        return AutoHTPlanner(lm=self.lm,
+                             max_depth=self.max_depth - 1,
+                             max_subtasks_per_decomp=self.max_subtasks_per_decomp)
 
     def plan(self, problem: str, resources: set[AResource] | None = None) -> HTP:
         """Make HTP for solving problem."""

openssa/l2/planning/hierarchical/_prompts.py

@@ -1,15 +1,15 @@
 HTP_JSON_TEMPLATE: str = """
 {{
-    "task": "(textual description of problem/task to solve)",
+    "task": "(textual description of question/problem/task to answer/solve)",
     "sub-plans": [
         {{
-            "task": "(textual description of 1st sub-problem/sub-task to solve)",
+            "task": "(textual description of 1st sub-question/problem/task to answer/solve)",
             "sub-plans": [
                 (... nested sub-plans ...)
             ]
         }},
         {{
-            "task": "(textual description of 2nd sub-problem/sub-task to solve)",
+            "task": "(textual description of 2nd sub-question/problem/task to answer/solve)",
             "sub-plans": [
                 (... nested sub-plans ...)
             ]
@@ -22,22 +22,26 @@
 HTP_WITH_RESOURCES_JSON_TEMPLATE: str = """
 {{
     "task": {{
-        "ask": "(textual description of problem/task to solve)"
+        "ask": "(textual description of question/problem/task to answer/solve)"
     }},
     "sub-plans": [
         {{
             "task": {{
-                "ask": "(textual description of 1st sub-problem/sub-task to solve)",
-                "resource": "(unique name of most relevant informational resource, IF ANY)" OR null
+                "ask": "(textual description of 1st sub-question/problem/task to answer/solve)",
+                "resources": [
+                    (... unique names of most relevant informational resources, if any ...)
+                ]
             }},
             "sub-plans": [
                 (... nested sub-plans ...)
             ]
         }},
         {{
             "task": {{
-                "ask": "(textual description of 2nd sub-problem/sub-task to solve)",
-                "resource": "(unique name of most relevant informational resource, IF ANY)" OR null
+                "ask": "(textual description of 2nd sub-question/problem/task to answer/solve)",
+                "resources": [
+                    (... unique names of most relevant informational resources, if any ...)
+                ]
             }},
             "sub-plans": [
                 (... nested sub-plans ...)
@@ -87,8 +91,9 @@ def htp_prompt_template(with_resources: bool) -> str:
 HTP_UPDATE_RESOURCES_PROMPT_TEMPLATE: str = (
 RESOURCE_OVERVIEW_PROMPT_SECTION +  # noqa: E122
 """please return an updated version of the following JSON hierarchical task plan
-by appropriately replacing `"resource": null` with `"resource": "(unique name of most relevant informational resource)"`
-for any case in which such a relevant informational resource can be identified for the corresponding problem/task:
+by appropriately replacing `"resources": null` or `"resources": []`
+with `"resources": [(... unique names of most relevant informational resources ...)]`
+for any case in which such relevant informational resource(s) can be identified for the corresponding question/problem/task:
 
 ```json
 {htp_json}

openssa/l2/reasoning/abstract.py

@@ -16,7 +16,7 @@ class AbstractReasoner(ABC):
     lm: AnLLM = field(default_factory=OpenAILLM.get_gpt_4_1106_preview)
 
     @abstractmethod
-    def reason(self, task: ATask, n_words: int = 300) -> str:
+    def reason(self, task: ATask, n_words: int = 1000) -> str:
         """Reason through task and return conclusion."""
 
 

openssa/l2/reasoning/base/__init__.py

@@ -5,14 +5,36 @@
 
 from openssa.l2.reasoning.abstract import AbstractReasoner
 from openssa.l2.task.abstract import ATask
+from openssa.l2.task.status import TaskStatus
+
+from ._prompts import RESOURCE_QA_CONSO_PROMPT_TEMPLATE
 
 
 @dataclass
 class BaseReasoner(AbstractReasoner):
     """Base reasoner."""
 
-    def reason(self, task: ATask, n_words: int = 300) -> str:
+    def reason(self, task: ATask, n_words: int = 1000) -> str:
         """Reason through task and return conclusion."""
-        return (task.resource.answer(question=task.ask, n_words=n_words)
-                if task.resource
-                else self.lm.get_response(prompt=f'`[WITHIN {n_words:,} WORDS:]`\n{task.ask}'))
+        task.result: str = ((self.lm.get_response(
+                                prompt=RESOURCE_QA_CONSO_PROMPT_TEMPLATE.format(
+                                    question=task.ask, n_words=n_words,
+                                    resources_and_answers='\n\n'.join(
+                                        (f'INFORMATIONAL RESOURCE #{i + 1} (name: "{r.name}"):\n'
+                                         '\n'
+                                         f'INFORMATIONAL RESOURCE #{i + 1} OVERVIEW:\n{r.overview}\n'
+                                         '\n'
+                                         f'ANSWER/SOLUTION #{i + 1}:\n{r.answer(question=task.ask, n_words=n_words)}\n')
+                                        for i, r in enumerate(task.resources))))
+
+                             if len(task.resources) > 1
+
+                             else next(iter(task.resources)).answer(question=task.ask, n_words=n_words))
+
+                            if task.resources
+
+                            else self.lm.get_response(prompt=f'`[WITHIN {n_words:,} WORDS:]`\n{task.ask}'))
+
+        task.status: TaskStatus = TaskStatus.DONE
+
+        return task.result

openssa/l2/reasoning/base/_prompts.py

@@ -0,0 +1,15 @@
+RESOURCE_QA_CONSO_PROMPT_TEMPLATE: str = \
+"""Assuming that the following question/problem/task is posed
+
+```
+{question}
+```
+
+and you have received various answers/solutions from different informational resources as detailed below,
+please consolidate within {n_words:,} words a final answer/solution that you believe is the most correct.
+
+
+```
+{resources_and_answers}
+```
+"""  # noqa: E122