#!/usr/bin/env python3 """ Task Planner Scans available skills, then uses an LLM to decompose a complex user query into an ordered execution plan. Returns a structured JSON plan for the calling agent to execute step by step. Usage: python task_planner.py --query "Your complex question" [--skills_dir /path/to/skills] """ import argparse import json import re import sys from pathlib import Path from typing import Dict, List, Optional import requests # ── Configuration ────────────────────────────────────────────────────────────── DEFAULT_SKILLS_DIR = "/home/admin/.openclaw/workspace/skills" LLM_MODEL = "qwen-plus" DASHSCOPE_API_KEY = "sk-b36adcdd25d64b7c847e7c941b5c1e03" DASHSCOPE_API_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation" # Skills that should never appear as executable steps in a plan EXCLUDED_SKILLS = {"task-planner", "skill-creator", "skill-orchestrator"} # ── LLM Helper ───────────────────────────────────────────────────────────────── def llm_call(messages: List[dict], temperature: float = 0.01) -> str: """ Call DashScope Qwen via direct HTTP REST API. Mirrors the node-fetch calling style provided in the project spec. """ payload = { "model": LLM_MODEL, "input": { "messages": messages }, "parameters": { "result_format": "message", "temperature": temperature, } } response = requests.post( DASHSCOPE_API_URL, headers={ "Authorization": f"Bearer {DASHSCOPE_API_KEY}", "Content-Type": "application/json", }, json=payload, timeout=60, ) response.raise_for_status() data = response.json() # Check for API-level errors if data.get("code") and data["code"] != "200": raise RuntimeError( f"DashScope API error [{data.get('code')}]: {data.get('message')}" ) return data["output"]["choices"][0]["message"]["content"].strip() # ── Skill Discovery ───────────────────────────────────────────────────────────── def discover_skills(skills_dir: str) -> List[Dict]: """ Scan the skills directory and extract lightweight metadata from each skill's SKILL.md (name + description only). Full SKILL.md content is intentionally NOT loaded here to keep token usage minimal. Returns a list of dicts: [{name, description, skill_dir}, ...] """ skills = [] skills_path = Path(skills_dir) if not skills_path.exists(): return skills for skill_dir in sorted(skills_path.iterdir()): if not skill_dir.is_dir(): continue if skill_dir.name in EXCLUDED_SKILLS: continue skill_md = skill_dir / "SKILL.md" if not skill_md.exists(): continue content = skill_md.read_text(encoding="utf-8") name_match = re.search(r"^name:\s*(.+)$", content, re.MULTILINE) desc_match = re.search(r"^description:\s*(.+)$", content, re.MULTILINE) if not name_match or not desc_match: continue skills.append({ "name": name_match.group(1).strip(), "description": desc_match.group(1).strip(), "skill_dir": str(skill_dir), }) return skills # ── Task Decomposition ────────────────────────────────────────────────────────── SYSTEM_PROMPT = """\ You are a task planning assistant. Given a user query and a list of available skills, decompose the query into an ordered list of steps. Each step must use exactly one skill. Rules: - Only use skills from the provided list. - If a step's output is needed as input for a later step, set "depends_on" to that step's index (0-based). - If a step does NOT depend on a previous step's output, set "depends_on" to null. - Use the placeholder {step_N_output} in "input_query" to reference a previous step's output (N is 0-based index). - Keep each "input_query" concise and focused on what that specific skill needs. - If the query can be answered by a single skill, return a plan with just one step. - Output ONLY a valid JSON array. No markdown fences, no commentary, no extra text. Output format: [ { "step": 0, "skill_name": "skill-name", "input_query": "specific sub-question or instruction for this skill", "depends_on": null, "purpose": "one-sentence explanation of why this step is needed" }, { "step": 1, "skill_name": "another-skill", "input_query": "instruction referencing {step_0_output} if needed", "depends_on": 0, "purpose": "one-sentence explanation" } ]\ """ def decompose(user_query: str, skills: List[Dict]) -> List[Dict]: """ Use Qwen to decompose the user query into an ordered execution plan. Token cost: ~400-800 tokens total. Only skill name + description summaries are sent — NOT full SKILL.md content. """ skills_summary = "\n".join( f"- name: {s['name']}\n description: {s['description']}" for s in skills ) user_message = ( f"User query: {user_query}\n\n" f"Available skills:\n{skills_summary}\n\n" "Decompose the query into an ordered execution plan." ) raw = llm_call( messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_message}, ], temperature=0.01, ) # Strip markdown fences if the model wraps the output raw = re.sub(r"^```(?:json)?\s*", "", raw.strip()) raw = re.sub(r"\s*```$", "", raw.strip()) parsed = json.loads(raw) # Normalise: model may return {"steps": [...]} or {"plan": [...]} or [...] if isinstance(parsed, dict): for v in parsed.values(): if isinstance(v, list): return v return [] return parsed # ── Main ──────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Task Planner — decomposes a complex query into an ordered skill execution plan" ) parser.add_argument("--query", "-q", required=True, help="The user's question or request") parser.add_argument( "--skills_dir", "-s", default=DEFAULT_SKILLS_DIR, help=f"Path to the skills directory (default: {DEFAULT_SKILLS_DIR})", ) args = parser.parse_args() # 1. Discover available skills (lightweight metadata only) skills = discover_skills(args.skills_dir) if not skills: result = { "error": f"No skills found in '{args.skills_dir}'.", "plan": [], } print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(1) # 2. Decompose the query into an execution plan try: plan = decompose(args.query, skills) except Exception as e: result = { "error": f"Task decomposition failed: {str(e)}", "plan": [], } print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(1) # 3. Output the plan as JSON # The calling agent reads this output and executes each step independently. output = { "query": args.query, "total_steps": len(plan), "plan": plan, "instructions": ( "Execute each step in order. For steps with depends_on set, " "replace {step_N_output} in input_query with the actual output " "from that step before calling the skill." ), } print(json.dumps(output, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()