364 lines
13 KiB
Python
364 lines
13 KiB
Python
# tool/planning.py
|
|
from typing import Dict, List, Literal, Optional
|
|
|
|
from app.exceptions import ToolError
|
|
from app.tool.base import BaseTool, ToolResult
|
|
|
|
|
|
_PLANNING_TOOL_DESCRIPTION = """
|
|
A planning tool that allows the agent to create and manage plans for solving complex tasks.
|
|
The tool provides functionality for creating plans, updating plan steps, and tracking progress.
|
|
"""
|
|
|
|
|
|
class PlanningTool(BaseTool):
|
|
"""
|
|
A planning tool that allows the agent to create and manage plans for solving complex tasks.
|
|
The tool provides functionality for creating plans, updating plan steps, and tracking progress.
|
|
"""
|
|
|
|
name: str = "planning"
|
|
description: str = _PLANNING_TOOL_DESCRIPTION
|
|
parameters: dict = {
|
|
"type": "object",
|
|
"properties": {
|
|
"command": {
|
|
"description": "The command to execute. Available commands: create, update, list, get, set_active, mark_step, delete.",
|
|
"enum": [
|
|
"create",
|
|
"update",
|
|
"list",
|
|
"get",
|
|
"set_active",
|
|
"mark_step",
|
|
"delete",
|
|
],
|
|
"type": "string",
|
|
},
|
|
"plan_id": {
|
|
"description": "Unique identifier for the plan. Required for create, update, set_active, and delete commands. Optional for get and mark_step (uses active plan if not specified).",
|
|
"type": "string",
|
|
},
|
|
"title": {
|
|
"description": "Title for the plan. Required for create command, optional for update command.",
|
|
"type": "string",
|
|
},
|
|
"steps": {
|
|
"description": "List of plan steps. Required for create command, optional for update command.",
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
},
|
|
"step_index": {
|
|
"description": "Index of the step to update (0-based). Required for mark_step command.",
|
|
"type": "integer",
|
|
},
|
|
"step_status": {
|
|
"description": "Status to set for a step. Used with mark_step command.",
|
|
"enum": ["not_started", "in_progress", "completed", "blocked"],
|
|
"type": "string",
|
|
},
|
|
"step_notes": {
|
|
"description": "Additional notes for a step. Optional for mark_step command.",
|
|
"type": "string",
|
|
},
|
|
},
|
|
"required": ["command"],
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
plans: dict = {} # Dictionary to store plans by plan_id
|
|
_current_plan_id: Optional[str] = None # Track the current active plan
|
|
|
|
async def execute(
|
|
self,
|
|
*,
|
|
command: Literal[
|
|
"create", "update", "list", "get", "set_active", "mark_step", "delete"
|
|
],
|
|
plan_id: Optional[str] = None,
|
|
title: Optional[str] = None,
|
|
steps: Optional[List[str]] = None,
|
|
step_index: Optional[int] = None,
|
|
step_status: Optional[
|
|
Literal["not_started", "in_progress", "completed", "blocked"]
|
|
] = None,
|
|
step_notes: Optional[str] = None,
|
|
**kwargs,
|
|
):
|
|
"""
|
|
Execute the planning tool with the given command and parameters.
|
|
|
|
Parameters:
|
|
- command: The operation to perform
|
|
- plan_id: Unique identifier for the plan
|
|
- title: Title for the plan (used with create command)
|
|
- steps: List of steps for the plan (used with create command)
|
|
- step_index: Index of the step to update (used with mark_step command)
|
|
- step_status: Status to set for a step (used with mark_step command)
|
|
- step_notes: Additional notes for a step (used with mark_step command)
|
|
"""
|
|
|
|
if command == "create":
|
|
return self._create_plan(plan_id, title, steps)
|
|
elif command == "update":
|
|
return self._update_plan(plan_id, title, steps)
|
|
elif command == "list":
|
|
return self._list_plans()
|
|
elif command == "get":
|
|
return self._get_plan(plan_id)
|
|
elif command == "set_active":
|
|
return self._set_active_plan(plan_id)
|
|
elif command == "mark_step":
|
|
return self._mark_step(plan_id, step_index, step_status, step_notes)
|
|
elif command == "delete":
|
|
return self._delete_plan(plan_id)
|
|
else:
|
|
raise ToolError(
|
|
f"Unrecognized command: {command}. Allowed commands are: create, update, list, get, set_active, mark_step, delete"
|
|
)
|
|
|
|
def _create_plan(
|
|
self, plan_id: Optional[str], title: Optional[str], steps: Optional[List[str]]
|
|
) -> ToolResult:
|
|
"""Create a new plan with the given ID, title, and steps."""
|
|
if not plan_id:
|
|
raise ToolError("Parameter `plan_id` is required for command: create")
|
|
|
|
if plan_id in self.plans:
|
|
raise ToolError(
|
|
f"A plan with ID '{plan_id}' already exists. Use 'update' to modify existing plans."
|
|
)
|
|
|
|
if not title:
|
|
raise ToolError("Parameter `title` is required for command: create")
|
|
|
|
if (
|
|
not steps
|
|
or not isinstance(steps, list)
|
|
or not all(isinstance(step, str) for step in steps)
|
|
):
|
|
raise ToolError(
|
|
"Parameter `steps` must be a non-empty list of strings for command: create"
|
|
)
|
|
|
|
# Create a new plan with initialized step statuses
|
|
plan = {
|
|
"plan_id": plan_id,
|
|
"title": title,
|
|
"steps": steps,
|
|
"step_statuses": ["not_started"] * len(steps),
|
|
"step_notes": [""] * len(steps),
|
|
}
|
|
|
|
self.plans[plan_id] = plan
|
|
self._current_plan_id = plan_id # Set as active plan
|
|
|
|
return ToolResult(
|
|
output=f"Plan created successfully with ID: {plan_id}\n\n{self._format_plan(plan)}"
|
|
)
|
|
|
|
def _update_plan(
|
|
self, plan_id: Optional[str], title: Optional[str], steps: Optional[List[str]]
|
|
) -> ToolResult:
|
|
"""Update an existing plan with new title or steps."""
|
|
if not plan_id:
|
|
raise ToolError("Parameter `plan_id` is required for command: update")
|
|
|
|
if plan_id not in self.plans:
|
|
raise ToolError(f"No plan found with ID: {plan_id}")
|
|
|
|
plan = self.plans[plan_id]
|
|
|
|
if title:
|
|
plan["title"] = title
|
|
|
|
if steps:
|
|
if not isinstance(steps, list) or not all(
|
|
isinstance(step, str) for step in steps
|
|
):
|
|
raise ToolError(
|
|
"Parameter `steps` must be a list of strings for command: update"
|
|
)
|
|
|
|
# Preserve existing step statuses for unchanged steps
|
|
old_steps = plan["steps"]
|
|
old_statuses = plan["step_statuses"]
|
|
old_notes = plan["step_notes"]
|
|
|
|
# Create new step statuses and notes
|
|
new_statuses = []
|
|
new_notes = []
|
|
|
|
for i, step in enumerate(steps):
|
|
# If the step exists at the same position in old steps, preserve status and notes
|
|
if i < len(old_steps) and step == old_steps[i]:
|
|
new_statuses.append(old_statuses[i])
|
|
new_notes.append(old_notes[i])
|
|
else:
|
|
new_statuses.append("not_started")
|
|
new_notes.append("")
|
|
|
|
plan["steps"] = steps
|
|
plan["step_statuses"] = new_statuses
|
|
plan["step_notes"] = new_notes
|
|
|
|
return ToolResult(
|
|
output=f"Plan updated successfully: {plan_id}\n\n{self._format_plan(plan)}"
|
|
)
|
|
|
|
def _list_plans(self) -> ToolResult:
|
|
"""List all available plans."""
|
|
if not self.plans:
|
|
return ToolResult(
|
|
output="No plans available. Create a plan with the 'create' command."
|
|
)
|
|
|
|
output = "Available plans:\n"
|
|
for plan_id, plan in self.plans.items():
|
|
current_marker = " (active)" if plan_id == self._current_plan_id else ""
|
|
completed = sum(
|
|
1 for status in plan["step_statuses"] if status == "completed"
|
|
)
|
|
total = len(plan["steps"])
|
|
progress = f"{completed}/{total} steps completed"
|
|
output += f"• {plan_id}{current_marker}: {plan['title']} - {progress}\n"
|
|
|
|
return ToolResult(output=output)
|
|
|
|
def _get_plan(self, plan_id: Optional[str]) -> ToolResult:
|
|
"""Get details of a specific plan."""
|
|
if not plan_id:
|
|
# If no plan_id is provided, use the current active plan
|
|
if not self._current_plan_id:
|
|
raise ToolError(
|
|
"No active plan. Please specify a plan_id or set an active plan."
|
|
)
|
|
plan_id = self._current_plan_id
|
|
|
|
if plan_id not in self.plans:
|
|
raise ToolError(f"No plan found with ID: {plan_id}")
|
|
|
|
plan = self.plans[plan_id]
|
|
return ToolResult(output=self._format_plan(plan))
|
|
|
|
def _set_active_plan(self, plan_id: Optional[str]) -> ToolResult:
|
|
"""Set a plan as the active plan."""
|
|
if not plan_id:
|
|
raise ToolError("Parameter `plan_id` is required for command: set_active")
|
|
|
|
if plan_id not in self.plans:
|
|
raise ToolError(f"No plan found with ID: {plan_id}")
|
|
|
|
self._current_plan_id = plan_id
|
|
return ToolResult(
|
|
output=f"Plan '{plan_id}' is now the active plan.\n\n{self._format_plan(self.plans[plan_id])}"
|
|
)
|
|
|
|
def _mark_step(
|
|
self,
|
|
plan_id: Optional[str],
|
|
step_index: Optional[int],
|
|
step_status: Optional[str],
|
|
step_notes: Optional[str],
|
|
) -> ToolResult:
|
|
"""Mark a step with a specific status and optional notes."""
|
|
if not plan_id:
|
|
# If no plan_id is provided, use the current active plan
|
|
if not self._current_plan_id:
|
|
raise ToolError(
|
|
"No active plan. Please specify a plan_id or set an active plan."
|
|
)
|
|
plan_id = self._current_plan_id
|
|
|
|
if plan_id not in self.plans:
|
|
raise ToolError(f"No plan found with ID: {plan_id}")
|
|
|
|
if step_index is None:
|
|
raise ToolError("Parameter `step_index` is required for command: mark_step")
|
|
|
|
plan = self.plans[plan_id]
|
|
|
|
if step_index < 0 or step_index >= len(plan["steps"]):
|
|
raise ToolError(
|
|
f"Invalid step_index: {step_index}. Valid indices range from 0 to {len(plan['steps'])-1}."
|
|
)
|
|
|
|
if step_status and step_status not in [
|
|
"not_started",
|
|
"in_progress",
|
|
"completed",
|
|
"blocked",
|
|
]:
|
|
raise ToolError(
|
|
f"Invalid step_status: {step_status}. Valid statuses are: not_started, in_progress, completed, blocked"
|
|
)
|
|
|
|
if step_status:
|
|
plan["step_statuses"][step_index] = step_status
|
|
|
|
if step_notes:
|
|
plan["step_notes"][step_index] = step_notes
|
|
|
|
return ToolResult(
|
|
output=f"Step {step_index} updated in plan '{plan_id}'.\n\n{self._format_plan(plan)}"
|
|
)
|
|
|
|
def _delete_plan(self, plan_id: Optional[str]) -> ToolResult:
|
|
"""Delete a plan."""
|
|
if not plan_id:
|
|
raise ToolError("Parameter `plan_id` is required for command: delete")
|
|
|
|
if plan_id not in self.plans:
|
|
raise ToolError(f"No plan found with ID: {plan_id}")
|
|
|
|
del self.plans[plan_id]
|
|
|
|
# If the deleted plan was the active plan, clear the active plan
|
|
if self._current_plan_id == plan_id:
|
|
self._current_plan_id = None
|
|
|
|
return ToolResult(output=f"Plan '{plan_id}' has been deleted.")
|
|
|
|
def _format_plan(self, plan: Dict) -> str:
|
|
"""Format a plan for display."""
|
|
output = f"Plan: {plan['title']} (ID: {plan['plan_id']})\n"
|
|
output += "=" * len(output) + "\n\n"
|
|
|
|
# Calculate progress statistics
|
|
total_steps = len(plan["steps"])
|
|
completed = sum(1 for status in plan["step_statuses"] if status == "completed")
|
|
in_progress = sum(
|
|
1 for status in plan["step_statuses"] if status == "in_progress"
|
|
)
|
|
blocked = sum(1 for status in plan["step_statuses"] if status == "blocked")
|
|
not_started = sum(
|
|
1 for status in plan["step_statuses"] if status == "not_started"
|
|
)
|
|
|
|
output += f"Progress: {completed}/{total_steps} steps completed "
|
|
if total_steps > 0:
|
|
percentage = (completed / total_steps) * 100
|
|
output += f"({percentage:.1f}%)\n"
|
|
else:
|
|
output += "(0%)\n"
|
|
|
|
output += f"Status: {completed} completed, {in_progress} in progress, {blocked} blocked, {not_started} not started\n\n"
|
|
output += "Steps:\n"
|
|
|
|
# Add each step with its status and notes
|
|
for i, (step, status, notes) in enumerate(
|
|
zip(plan["steps"], plan["step_statuses"], plan["step_notes"])
|
|
):
|
|
status_symbol = {
|
|
"not_started": "[ ]",
|
|
"in_progress": "[→]",
|
|
"completed": "[✓]",
|
|
"blocked": "[!]",
|
|
}.get(status, "[ ]")
|
|
|
|
output += f"{i}. {status_symbol} {step}\n"
|
|
if notes:
|
|
output += f" Notes: {notes}\n"
|
|
|
|
return output
|