Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

update openssa.l2.agent.abstract docstrings and comments #171

Merged
merged 3 commits into from
Apr 25, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 31 additions & 7 deletions openssa/l2/agent/abstract.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""Abstract agent with planning, reasoning & informational resources."""
"""Abstract Agent with Planning, Reasoning & Informational Resources."""


from __future__ import annotations
Expand All @@ -24,10 +24,15 @@

@dataclass
class AbstractAgent(ABC):
"""Abstract agent with planning, reasoning & informational resources."""
"""Abstract Agent with Planning, Reasoning & Informational Resources."""

# Planner for decomposing tasks into executable solution Plans
planner: APlanner | None = None

# Reasoner for working through individual Tasks to either conclude or make partial progress on them
reasoner: AReasoner = field(default_factory=BaseReasoner)

# set of Informational Resources for answering information-querying questions
resources: set[AResource] = field(default_factory=set,
init=True,
repr=True,
Expand All @@ -38,34 +43,49 @@ class AbstractAgent(ABC):

@property
def resource_overviews(self) -> dict[str, str]:
"""Overview available Informational Resources."""
return {r.unique_name: r.overview for r in self.resources}

def solve(self, problem: str, plan: APlan | None = None, dynamic: bool = True) -> str:
"""Solve problem, with an automatically generated plan (default) or explicitly specified plan."""
"""Solve posed Problem.

Solution Plan can optionally be explicitly provided,
or would be automatically generated by default if Planner is given.

Solution Plan, whether explicitly provided or automatically generated,
can be executed Dynamically (i.e., with as-needed further task decomposition)
or Statically (i.e., literally per the Plan).
"""
match (plan, self.planner, dynamic):

# NO PLAN
case (None, None, _):
# if neither Plan nor Planner is given, directly use Reasoner
result: str = self.reasoner.reason(task=Task(ask=problem, resources=self.resources))

# AUTOMATED STATIC PLAN
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)
# then execute such static Plan
plan: APlan = self.planner.plan(problem=problem, resources=self.resources)
pprint(plan)
result: str = plan.execute(reasoner=self.reasoner)

# AUTOMATED DYNAMIC PLAN
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)

# EXPERT-SPECIFIED STATIC PLAN
case (_, None, _) if plan:
# if Plan is given but no Planner is, then execute Plan statically
result: str = plan.execute(reasoner=self.reasoner)

# EXPERT-SPECIFIED STATIC PLAN, with Resource updating
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,
Expand All @@ -74,6 +94,7 @@ def solve(self, problem: str, plan: APlan | None = None, dynamic: bool = True) -
pprint(plan)
result: str = plan.execute(reasoner=self.reasoner)

# EXPERT-GUIDED DYNAMIC PLAN
case (_, _, True) if (plan and self.planner):
# if both Plan and Planner are given, and if solving dynamically,
# TODO: dynamic solution
Expand All @@ -85,12 +106,15 @@ def solve(self, problem: str, plan: APlan | None = None, dynamic: bool = True) -
return result

def solve_dynamically(self, problem: str, planner: APlanner = None, other_results: list[AskAnsPair] | None = None) -> str:
"""Solve task dynamically.
"""Solve posed Problem dynamically.

When first-pass result from Reasoner is unsatisfactory, decompose problem and recursively solve decomposed plan.
When first-pass result from Reasoner is unsatisfactory, decompose Problem and recursively solve decomposed Plan.
"""
# attempt direct solution with Reasoner
self.reasoner.reason(task := Task(ask=problem, resources=self.resources))

# if Reasoner's result is unsatisfactory, and if Planner has not run out of allowed depth,
# decompose Problem into 1-level Plan, and recursively solve such Plan with remaining allowed Planner depth
if (task.status == TaskStatus.NEEDING_DECOMPOSITION) and (planner := planner or self.planner).max_depth:
sub_planner: APlanner = planner.one_fewer_level_deep()

Expand Down