feat: skill with check list (#512)

* feat: skill with check list

* fix: check list for instruction

Author: zakahan

veadk/agent.py

@@ -156,6 +156,8 @@ class Agent(LlmAgent):
 
     enable_dataset_gen: bool = False
 
+    _skills_with_checklist: Dict[str, Any] = {}
+
     def model_post_init(self, __context: Any) -> None:
         super().model_post_init(None)  # for sub_agents init
 
@@ -300,6 +302,21 @@ def model_post_init(self, __context: Any) -> None:
 
         if self.skills:
             self.load_skills()
+            from veadk.skills.utils import create_init_skill_check_list_callback
+
+            init_callback = create_init_skill_check_list_callback(
+                self._skills_with_checklist
+            )
+            if self.before_tool_callback:
+                if isinstance(self.before_tool_callback, list):
+                    self.before_tool_callback.append(init_callback)
+                else:
+                    self.before_tool_callback = [
+                        self.before_tool_callback,
+                        init_callback,
+                    ]
+            else:
+                self.before_tool_callback = init_callback
 
         if self.example_store:
             from google.adk.tools.example_tool import ExampleTool
@@ -433,12 +450,23 @@ def load_skills(self):
                 for skill in load_skills_from_cloud(item):
                     skills[skill.name] = skill
         if skills:
+            self._skills_with_checklist = skills
+
             self.instruction += "\nYou have the following skills:\n"
 
+            has_checklist = False
             for skill in skills.values():
                 self.instruction += (
                     f"- name: {skill.name}\n- description: {skill.description}\n\n"
                 )
+                if skill.checklist:
+                    has_checklist = True
+
+            if has_checklist:
+                self.instruction += (
+                    "Some skills have a checklist that you must complete step by step. "
+                    "Use the `update_check_list` tool to mark each item as completed.\n\n"
+                )
 
             if self.skills_mode not in [
                 "skills_sandbox",

veadk/skills/skill.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from pydantic import BaseModel
-from typing import Optional
+from typing import Optional, List, Dict
 
 
 class Skill(BaseModel):
@@ -22,3 +22,7 @@ class Skill(BaseModel):
     path: str  # local path or tos path
     skill_space_id: Optional[str] = None
     bucket_name: Optional[str] = None
+    checklist: List[Dict[str, str]] = []
+
+    def get_checklist_items(self) -> List[str]:
+        return [item.get("item", item.get("id", "")) for item in self.checklist]

veadk/skills/utils.py

@@ -17,13 +17,74 @@
 import os
 import frontmatter
 
+from google.adk.tools import BaseTool, ToolContext
+from typing import Any, Dict, Optional, Callable
+
 from veadk.skills.skill import Skill
 from veadk.utils.logger import get_logger
 from veadk.utils.volcengine_sign import ve_request
 
 logger = get_logger(__name__)
 
 
+def update_check_list(
+    tool_context: ToolContext, skill_name: str, check_item: str, state: bool
+):
+    """
+    Update the checklist item state for a specific skill.
+    Use this tool to mark checklist items as completed during skill execution.
+
+    eg:
+    update_check_list(skill_name="skill-creator", check_item="analyze_content", state=True)
+    """
+    agent_name = tool_context.agent_name
+    if agent_name not in tool_context.state:
+        tool_context.state[agent_name] = {}
+    if skill_name not in tool_context.state[agent_name]:
+        tool_context.state[agent_name][skill_name] = {}
+    if "check_list" not in tool_context.state[agent_name][skill_name]:
+        tool_context.state[agent_name][skill_name]["check_list"] = {}
+    tool_context.state[agent_name][skill_name]["check_list"][check_item] = state
+    logger.info(f"Updated agent[{agent_name}] state: {tool_context.state[agent_name]}")
+
+
+def create_init_skill_check_list_callback(
+    skills_with_checklist: Dict[str, Skill],
+) -> Callable[[BaseTool, Dict[str, Any], ToolContext], Optional[Dict]]:
+    """
+    Create a callback function to initialize checklist when a skill is invoked.
+
+    Args:
+        skills_with_checklist: Dictionary mapping skill names to Skill objects
+
+    Returns:
+        A callback function for before_tool_callback
+    """
+
+    def init_skill_check_list(
+        tool: BaseTool, args: Dict[str, Any], tool_context: ToolContext
+    ) -> Optional[Dict]:
+        """Callback to initialize checklist when a skill is invoked."""
+        if tool.name == "skills_tool":
+            skill_name = args.get("command")
+            agent_name = tool_context.agent_name
+            if skill_name in skills_with_checklist:
+                skill = skills_with_checklist[skill_name]
+                check_list_items = skill.get_checklist_items()
+                check_list_state = {item: False for item in check_list_items}
+                if agent_name not in tool_context.state:
+                    tool_context.state[agent_name] = {}
+                tool_context.state[agent_name][skill_name] = {
+                    "check_list": check_list_state
+                }
+                logger.info(
+                    f"Initialized agent[{agent_name}] skill[{skill_name}] check_list: {check_list_state}"
+                )
+        return None
+
+    return init_skill_check_list
+
+
 def load_skill_from_directory(skill_directory: Path) -> Skill:
     logger.info(f"Load skill from {skill_directory}")
     skill_readme = skill_directory / "SKILL.md"
@@ -35,6 +96,7 @@ def load_skill_from_directory(skill_directory: Path) -> Skill:
 
     skill_name = skill.get("name", "")
     skill_description = skill.get("description", "")
+    checklist = skill.get("checklist", [])
 
     if not skill_name or not skill_description:
         logger.error(
@@ -47,10 +109,14 @@ def load_skill_from_directory(skill_directory: Path) -> Skill:
     logger.info(
         f"Successfully loaded skill {skill_name} locally from {skill_readme}, name={skill_name}, description={skill_description}"
     )
+    if checklist:
+        logger.info(f"Skill {skill_name} checklist: {checklist}")
+
     return Skill(
         name=skill_name,  # type: ignore
         description=skill_description,  # type: ignore
         path=str(skill_directory),
+        checklist=checklist,
     )
 
 

veadk/tools/skills_tools/skills_toolset.py

@@ -34,6 +34,7 @@
     bash_tool,
     register_skills_tool,
 )
+from veadk.skills.utils import update_check_list
 from veadk.utils.logger import get_logger
 
 logger = get_logger(__name__)
@@ -73,6 +74,7 @@ def __init__(self, skills: Dict[str, Skill], skills_mode: str) -> None:
             "edit_file": FunctionTool(edit_file_tool),
             "bash": FunctionTool(bash_tool),
             "register_skills": FunctionTool(register_skills_tool),
+            "update_check_list": FunctionTool(update_check_list),
         }
 
     @override