58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from core.models import Scene
|
|
from runtime.context import TaskContext
|
|
|
|
if TYPE_CHECKING:
|
|
from world.models import WorldState
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlannedStep:
|
|
action: str
|
|
description: str
|
|
args: dict[str, Any] = field(default_factory=dict)
|
|
expected_text: str | None = None
|
|
# Structured action semantics, populated by AI planner tool calls and
|
|
# retained independently from executable arguments for later reuse.
|
|
purpose: str | None = None
|
|
expected_outcome: str | None = None
|
|
# The actual prompt sent to the LLM for this step (AI planners only).
|
|
# ``None`` for non-LLM planners; TaskRunner falls back to the task goal.
|
|
prompt: str | None = None
|
|
# Pre-tool text block emitted by the model before the tool call.
|
|
# None when the model omits a text block or for non-LLM planners.
|
|
rationale: str | None = None
|
|
# Extended thinking / reasoning content from the model.
|
|
# None when not enabled or not present.
|
|
thinking: str | None = None
|
|
|
|
|
|
class Planner:
|
|
def plan(
|
|
self,
|
|
*,
|
|
goal: str,
|
|
scene: Scene,
|
|
context: TaskContext,
|
|
world: "WorldState | None" = None,
|
|
screenshot: bytes | None = None,
|
|
) -> list[PlannedStep]:
|
|
if context.step_results:
|
|
return []
|
|
return [
|
|
PlannedStep(
|
|
action="describe_screen",
|
|
description=f"Observe current screen for goal: {goal}",
|
|
args={},
|
|
)
|
|
]
|
|
|
|
def goal_reached(self, *, goal: str, scene: Scene, context: TaskContext) -> bool:
|
|
return bool(context.step_results) and all(
|
|
result.success for result in context.step_results
|
|
)
|