File size: 3,452 Bytes
5ef3875 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """Planning: a structured plan the model maintains through the update_plan tool.
The model decides how much planning a task needs (the system prompt tells it
to skip plans for trivial requests). When it does plan, the plan is shown to
the user and kept up to date as steps complete.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable
from ycode.errors import ToolError
from ycode.tools.base import Tool, ToolContext, ToolResult
STATUSES = ("pending", "in_progress", "done", "skipped")
@dataclass
class PlanStep:
text: str
status: str = "pending"
@dataclass
class Plan:
steps: list[PlanStep] = field(default_factory=list)
@property
def is_empty(self) -> bool:
return not self.steps
def render_text(self) -> str:
marks = {"pending": " ", "in_progress": "~", "done": "x", "skipped": "-"}
return "\n".join(f"{i}. [{marks[s.status]}] {s.text}" for i, s in enumerate(self.steps, 1))
def parse_steps(raw: Any) -> list[PlanStep]:
if not isinstance(raw, list) or not raw:
raise ToolError("steps must be a non-empty list")
steps: list[PlanStep] = []
for item in raw:
if isinstance(item, str):
steps.append(PlanStep(item.strip()))
continue
if not isinstance(item, dict) or not isinstance(item.get("step"), str):
raise ToolError("each step must be a string or an object with a 'step' string")
status = item.get("status", "pending")
if status not in STATUSES:
raise ToolError(f"invalid status {status!r}; use one of {', '.join(STATUSES)}")
steps.append(PlanStep(item["step"].strip(), status))
if len(steps) > 20:
raise ToolError("keep plans to 20 steps or fewer")
return steps
class UpdatePlanTool(Tool):
name = "update_plan"
description = (
"Record or update your implementation plan for the current task. Call it before making "
"significant changes (skip it for trivial requests), and again to mark steps in_progress "
"or done. Pass the full list of steps every time."
)
parameters = {
"type": "object",
"properties": {
"steps": {
"type": "array",
"description": "Ordered plan steps.",
"items": {
"type": "object",
"properties": {
"step": {"type": "string", "description": "Short imperative description."},
"status": {"type": "string", "enum": list(STATUSES)},
},
"required": ["step"],
},
},
},
"required": ["steps"],
}
def __init__(self, on_update: Callable[[Plan], None] | None = None) -> None:
self.plan = Plan()
self.on_update = on_update
def describe(self, args: dict[str, Any]) -> str:
return "Updating plan"
def run(self, args: dict[str, Any], ctx: ToolContext) -> ToolResult:
self.plan = Plan(parse_steps(args["steps"]))
if self.on_update:
self.on_update(self.plan)
done = sum(s.status == "done" for s in self.plan.steps)
return ToolResult(
content="Plan recorded:\n" + self.plan.render_text(),
summary=f"Plan: {done}/{len(self.plan.steps)} steps done",
display={"plan": self.plan},
)
|