reset
This commit is contained in:
Space-Banane
@@ -1,11 +0,0 @@
|
||||
"""Utility helpers for the Clickthrough agent skill."""
|
||||
|
||||
from .agent_runner import AgentRunResult, ClickthroughAgentRunner
|
||||
from .clickthrough_skill import ActionPlan, ClickthroughSkill
|
||||
|
||||
__all__ = [
|
||||
"ClickthroughSkill",
|
||||
"ActionPlan",
|
||||
"ClickthroughAgentRunner",
|
||||
"AgentRunResult",
|
||||
]
|
||||
@@ -1,60 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict
|
||||
|
||||
from .clickthrough_skill import ActionPlan, ClickthroughSkill
|
||||
|
||||
|
||||
@dataclass
|
||||
class AgentRunResult:
|
||||
summary: Dict[str, Any]
|
||||
action: Dict[str, Any]
|
||||
history: Dict[str, Any]
|
||||
grid: Dict[str, Any]
|
||||
plan_preview: Dict[str, Any]
|
||||
|
||||
|
||||
class ClickthroughAgentRunner:
|
||||
def __init__(self, skill: ClickthroughSkill) -> None:
|
||||
self.skill = skill
|
||||
|
||||
def run_once(
|
||||
self,
|
||||
screenshot_base64: str,
|
||||
width: int,
|
||||
height: int,
|
||||
rows: int = 4,
|
||||
columns: int = 4,
|
||||
preferred_label: str | None = None,
|
||||
action: str = "click",
|
||||
text: str | None = None,
|
||||
) -> AgentRunResult:
|
||||
grid = self.skill.describe_grid(
|
||||
screenshot_base64=screenshot_base64,
|
||||
width=width,
|
||||
height=height,
|
||||
rows=rows,
|
||||
columns=columns,
|
||||
)
|
||||
plan_response = self.skill.plan_with_planner(
|
||||
grid_id=grid["grid_id"],
|
||||
preferred_label=preferred_label,
|
||||
action=action,
|
||||
text=text,
|
||||
)
|
||||
plan_payload = plan_response["plan"]
|
||||
plan = ActionPlan(
|
||||
grid_id=plan_payload["grid_id"],
|
||||
target_cell=plan_payload.get("target_cell"),
|
||||
action=plan_payload["action"],
|
||||
text=plan_payload.get("text"),
|
||||
)
|
||||
action_result = self.skill.plan_action(plan)
|
||||
summary = self.skill.grid_summary(grid["grid_id"])
|
||||
history = self.skill.grid_history(grid["grid_id"])
|
||||
return AgentRunResult(
|
||||
summary=summary,
|
||||
action=action_result,
|
||||
history=history,
|
||||
grid=grid,
|
||||
plan_preview=plan_response,
|
||||
)
|
||||
@@ -1,98 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
@dataclass
|
||||
class ActionPlan:
|
||||
grid_id: str
|
||||
target_cell: str | None
|
||||
action: str
|
||||
text: str | None = None
|
||||
|
||||
|
||||
class ClickthroughSkill:
|
||||
"""Lightweight wrapper around the Clickthrough HTTP API."""
|
||||
|
||||
def __init__(self, server_url: str = "http://localhost:8000") -> None:
|
||||
self._client = httpx.Client(base_url=server_url, timeout=10)
|
||||
|
||||
def describe_grid(
|
||||
self,
|
||||
screenshot_base64: str,
|
||||
width: int,
|
||||
height: int,
|
||||
rows: int = 4,
|
||||
columns: int = 4,
|
||||
) -> Dict[str, Any]:
|
||||
payload = {
|
||||
"width": width,
|
||||
"height": height,
|
||||
"rows": rows,
|
||||
"columns": columns,
|
||||
"screenshot_base64": screenshot_base64,
|
||||
"memo": "agent-powered grid",
|
||||
}
|
||||
response = self._client.post("/grid/init", json=payload)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def plan_action(self, plan: ActionPlan) -> Dict[str, Any]:
|
||||
payload = {
|
||||
"grid_id": plan.grid_id,
|
||||
"action": plan.action,
|
||||
"target_cell": plan.target_cell,
|
||||
"text": plan.text,
|
||||
"comment": "skill-generated plan",
|
||||
}
|
||||
response = self._client.post("/grid/action", json=payload)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def grid_summary(self, grid_id: str) -> Dict[str, Any]:
|
||||
response = self._client.get(f"/grid/{grid_id}/summary")
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def grid_history(self, grid_id: str) -> Dict[str, Any]:
|
||||
response = self._client.get(f"/grid/{grid_id}/history")
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def plan_with_planner(
|
||||
self,
|
||||
grid_id: str,
|
||||
preferred_label: str | None = None,
|
||||
action: str = "click",
|
||||
text: str | None = None,
|
||||
comment: str | None = None,
|
||||
) -> Dict[str, Any]:
|
||||
payload = {
|
||||
"preferred_label": preferred_label,
|
||||
"action": action,
|
||||
"text": text,
|
||||
"comment": comment or "planner-generated",
|
||||
}
|
||||
response = self._client.post(f"/grid/{grid_id}/plan", json=payload)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def refresh_grid(self, grid_id: str, screenshot_base64: str, memo: str | None = None) -> Dict[str, Any]:
|
||||
payload = {"screenshot_base64": screenshot_base64, "memo": memo}
|
||||
response = self._client.post(f"/grid/{grid_id}/refresh", json=payload)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import base64
|
||||
|
||||
dummy = base64.b64encode(b"fake-screenshot").decode()
|
||||
skill = ClickthroughSkill()
|
||||
grid = skill.describe_grid(dummy, width=800, height=600)
|
||||
print("Grid cells:", len(grid.get("cells", [])))
|
||||
if grid.get("cells"):
|
||||
first_cell = grid["cells"][0]["cell_id"]
|
||||
result = skill.plan_action(ActionPlan(grid_id=grid["grid_id"], target_cell=first_cell, action="click"))
|
||||
print("Action result:", result)
|
||||
Reference in New Issue
Block a user