""" TaskSkillEngine — VERA-inspired ReAct loop for task-oriented skills. Architecture: Prompt-driven ReAct (no function calling dependency). L1 build_catalog() → inject skill metadata into ReAct prompt L2 activate() → JIT inject SKILL.md body on LLM request L3 react_loop() → ReAct cycle: Thought → Action → Observation Execution paths: executor=sandbox → LLM generates shell command → execute_shell() executor=handler → LLM generates params → ToolRegistry.execute() Supports parallel (multiple actions per round) and serial (multi-round) chaining. """ from __future__ import annotations import asyncio import json import re from pathlib import Path from typing import Optional import frontmatter from agent.skills.skill_types import ( SKILL_FILENAME, ExecutionStatus, Skill, SkillExecutionResult, load_skill, ) from typing import TYPE_CHECKING if TYPE_CHECKING: from agent.skills.tool_registry import ToolRegistry class TaskSkillEngine: """Task-oriented skill engine with ReAct loop.""" def __init__(self, skills_dir: str, tool_registry: "Optional[ToolRegistry]" = None): self.skills_dir = Path(skills_dir) self.tool_registry = tool_registry self._skills: dict[str, Skill] = {} # -- Loading --------------------------------------------------------------- def load_all(self) -> dict[str, Skill]: """Load L1 metadata for trigger=tool skills only.""" self._skills.clear() if not self.skills_dir.exists(): return {} for entry in sorted(self.skills_dir.iterdir()): if entry.is_dir(): skill_file = entry / SKILL_FILENAME if skill_file.exists(): try: skill = load_skill(entry) if skill.trigger == "tool": self._skills[skill.skill_id] = skill except Exception as e: print(f"[task-skill] Failed to load {entry.name}: {e}") return self._skills # -- L2 activation --------------------------------------------------------- def activate(self, skill_id: str) -> None: """Load L2 body (SKILL.md content) for a skill. Idempotent.""" skill = self._skills.get(skill_id) if not skill or skill.is_activated: return post = frontmatter.load(str(Path(skill.base_dir) / SKILL_FILENAME)) skill.body = post.content.strip() # -- Queries --------------------------------------------------------------- def get(self, skill_id: str) -> Optional[Skill]: if not self._skills: self.load_all() return self._skills.get(skill_id) @property def tool_skills(self) -> list[Skill]: """List of trigger:tool skills.""" return [s for s in self._skills.values() if s.trigger == "tool"] def get_cron_skills(self) -> list[Skill]: """Get all skills with cron triggers.""" if not self._skills: self.load_all() return [s for s in self._skills.values() if s.trigger == "cron" and s.cron_schedule] # -- L1 Catalog (Progressive Disclosure) ----------------------------------- def build_catalog(self) -> str: """Build L1 skill catalog text for ReAct prompt injection. Returns a concise description of available skills (metadata only). """ if not self._skills: self.load_all() if not self.tool_skills: return "" lines = ["可用工具技能:"] for skill in self.tool_skills: lines.append(f"- {skill.skill_id}: {skill.description}") return "\n".join(lines) # -- ReAct Loop ------------------------------------------------------------ async def react_loop( self, user_message: str, llm, max_rounds: int = 3, ) -> Optional[str]: """Run a pre-engine ReAct loop for task skill detection + execution. Pure prompt-driven — no function calling dependency. Flow: Round 1: LLM sees skill catalog (L1) + user message → outputs nothing (no skill needed) or {"activate": "skill_id"} Round 2+: Engine JIT injects SKILL.md body (L2) → LLM outputs {"actions": [...]} or {"done": true} → Engine executes actions (sandbox or ToolRegistry) → Observations fed back for next round Returns: Merged observation text to inject into user_message, or None. """ from providers.llm.base import ChatMessage if not self._skills: self.load_all() if not self.tool_skills: return None catalog = self.build_catalog() if not catalog: return None # Build ReAct system prompt system_prompt = ( "你是一个工具调度器。判断用户消息是否需要调用工具。\n\n" f"## {catalog}\n\n" "## 协议\n" "- 如果用户消息**直接、明确**地请求了某个技能的能力,输出 JSON:\n" ' {"activate": "skill_id"}\n\n' "- 如果已有技能文档,需要执行动作:\n" ' {"actions": [{"tool": "execute_shell", "params": {"command": "..."}}]}\n\n' "- **其他所有情况**,什么都不要输出,返回空。\n\n" "## 严格规则\n" "- 99% 的消息都不需要工具,默认返回空\n" "- 聊天、闲聊、提问、情感表达、讨论话题 → 返回空\n" "- 不要联想、不要推测用户可能需要什么工具\n" "- 用户没有明说要用工具,就不要激活\n" ) messages = [ChatMessage("system", system_prompt)] messages.append(ChatMessage("user", user_message)) all_observations: list[str] = [] active_skill: Optional[Skill] = None for round_idx in range(max_rounds): try: response = await llm.chat(messages, temperature=0.1, max_tokens=500) raw = response.content.strip() except Exception as e: print(f" [react] ❌ Round {round_idx + 1} LLM error: {e}") break # Empty output = LLM decided no skill needed → silent return if not raw: break parsed = self._extract_json(raw) if not parsed: # LLM output non-JSON (e.g. "不需要") = no skill needed break # done = no skill needed (backward compat) if parsed.get("done"): break # activate_skill — JIT inject SKILL.md body (L2) if "activate" in parsed: skill_id = parsed["activate"].lower() skill = self._skills.get(skill_id) if not skill: print(f" [react] ⚠ Unknown skill: {skill_id}") break print(f" [react] 🎯 Activate: {skill_id} (round {round_idx + 1})") self.activate(skill_id) active_skill = skill # JIT inject SKILL.md body into context skill_injection = ( f"技能 [{skill.name}] 已激活。以下是技能文档:\n\n" f"{skill.body}\n\n" f"请根据技能文档和用户请求,生成具体的执行动作。" ) messages.append(ChatMessage("assistant", response.content)) messages.append(ChatMessage("user", skill_injection)) continue # actions — execute via sandbox or ToolRegistry actions = parsed.get("actions", []) if not actions: break thought = parsed.get("thought", "") print(f" [react] 🔧 Actions (round {round_idx + 1}): " f"{len(actions)} action(s), thought: {thought[:60]}") # Parallel execution via asyncio.gather tasks = [self._execute_action(a, active_skill) for a in actions] results = await asyncio.gather(*tasks, return_exceptions=True) # Collect observations round_observations = [] for i, result in enumerate(results): if isinstance(result, Exception): obs = f"[错误] {result}" elif isinstance(result, str): obs = result else: obs = str(result) round_observations.append(obs) all_observations.append(obs) # Feed observations back for next round obs_text = "\n".join(f"[Observation {i+1}] {o}" for i, o in enumerate(round_observations)) messages.append(ChatMessage("assistant", response.content)) messages.append(ChatMessage("user", f"执行结果:\n{obs_text}\n\n" f"根据结果,是否需要更多操作?如果完成,返回 {{\"done\": true}}。" )) if not all_observations: return None # P4 fix: per-observation limit + truncation marker MAX_PER_OBS = 300 trimmed = [] for obs in all_observations: if len(obs) > MAX_PER_OBS: trimmed.append(obs[:MAX_PER_OBS] + "…(已截断)") else: trimmed.append(obs) merged = "\n".join(trimmed) print(f" [react] 📋 Total observations: {len(all_observations)}, {len(merged)} chars") return merged # -- Action Execution ------------------------------------------------------ async def _execute_action( self, action: dict, active_skill: Optional[Skill], ) -> str: """Execute a single action from the ReAct output. Routes to sandbox (execute_shell) or ToolRegistry based on action type. """ tool_name = action.get("tool", "execute_shell") params = action.get("params", {}) # Sandbox path if tool_name == "execute_shell": command = params.get("command", "") if not command: return "[错误] 空命令" # Clean markdown wrapping command = re.sub(r'^```\w*\n?', '', command) command = re.sub(r'\n?```$', '', command) command = command.strip() from agent.skills.sandbox_executor import execute_shell result = await execute_shell(command) stdout = result.get("stdout", "").strip() stderr = result.get("stderr", "").strip() if result["success"]: return stdout or "[执行成功,无输出]" else: return f"[执行失败] {stderr or stdout or '未知错误'}" # ToolRegistry path if self.tool_registry and self.tool_registry.has(tool_name): try: result = await self.tool_registry.execute(tool_name, params) return json.dumps(result, ensure_ascii=False)[:500] except Exception as e: return f"[工具错误] {tool_name}: {e}" return f"[未知工具] {tool_name}" async def _execute_with_skill( self, skill: Skill, user_message: str, llm, ) -> Optional[str]: """Fallback: execute a skill directly (keyword match path). Used when JSON parsing fails but keyword matching finds a skill. """ from providers.llm.base import ChatMessage if not skill.body: return None system_msg = ChatMessage("system", f"根据以下技能文档,为用户请求生成一条可执行的 shell 命令。\n" f"只输出命令本身,不要解释,不要 markdown 格式。\n\n" f"## 技能文档\n{skill.body}" ) user_msg = ChatMessage("user", user_message) resp = await llm.chat([system_msg, user_msg], temperature=0.1) content = resp.content.strip() content = re.sub(r'^```\w*\n?', '', content) content = re.sub(r'\n?```$', '', content) command = content.strip() if not command: return None from agent.skills.sandbox_executor import execute_shell result = await execute_shell(command) stdout = result.get("stdout", "").strip() if result["success"] and stdout: return stdout return None # -- Keyword Fallback ------------------------------------------------------ def _keyword_match(self, user_message: str) -> Optional[Skill]: """Simple keyword matching fallback when LLM JSON fails.""" msg_lower = user_message.lower() for skill in self.tool_skills: # Check skill name and description keywords triggers = [skill.skill_id, skill.name] desc_words = skill.description.split() triggers.extend(w for w in desc_words if len(w) >= 5) for trigger in triggers: if trigger.lower() in msg_lower: return skill return None # -- JSON Extraction ------------------------------------------------------- def _extract_json(self, text: str) -> Optional[dict]: """Extract JSON object from LLM text output.""" text = text.strip() # Direct parse if text.startswith("{"): try: return json.loads(text) except json.JSONDecodeError: pass # Strip markdown ```json ... ``` fence stripped = re.sub(r"^```(?:json)?\s*\n?", "", text) stripped = re.sub(r"\n?\s*```\s*$", "", stripped).strip() if stripped.startswith("{"): try: return json.loads(stripped) except json.JSONDecodeError: pass # P1 fix: bracket-counting extraction (replaces greedy regex) obj_str = self._find_first_json_object(text) if obj_str: try: return json.loads(obj_str) except json.JSONDecodeError: pass return None @staticmethod def _find_first_json_object(text: str) -> Optional[str]: """Find the first balanced {...} block using bracket counting.""" start = text.find("{") if start == -1: return None depth = 0 in_string = False escape = False for i in range(start, len(text)): c = text[i] if escape: escape = False continue if c == "\\" and in_string: escape = True continue if c == '"' and not escape: in_string = not in_string continue if in_string: continue if c == "{": depth += 1 elif c == "}": depth -= 1 if depth == 0: return text[start:i + 1] return None # -- Legacy compat (kept for tests referencing execute()) ------------------ async def execute(self, skill_id: str, user_intent: str, llm) -> SkillExecutionResult: """Execute a task skill directly. Legacy path, prefer react_loop(). Args: skill_id: ID of the skill to execute. user_intent: Original user message. llm: LLMClient instance for command generation. """ from providers.llm.base import ChatMessage skill_id = skill_id.lower() skill = self._skills.get(skill_id) if not skill: return SkillExecutionResult( skill_id=skill_id, success=False, status=ExecutionStatus.FAILED, output={"error": f"Unknown skill: {skill_id}"}, ) if not skill.is_activated: self.activate(skill_id) if not skill.body: return SkillExecutionResult( skill_id=skill_id, success=False, status=ExecutionStatus.FAILED, output={"error": "Skill body is empty", "stdout": "", "stderr": "", "returncode": -1}, ) # LLM generates shell command from body + user intent system_msg = ChatMessage("system", f"根据以下技能文档,为用户请求生成一条可执行的 shell 命令。\n" f"只输出命令本身,不要解释,不要 markdown 格式。\n\n" f"## 技能文档\n{skill.body}" ) user_msg = ChatMessage("user", user_intent) resp = await llm.chat([system_msg, user_msg], temperature=0.1) content = resp.content.strip() content = re.sub(r'^```\w*\n?', '', content) content = re.sub(r'\n?```$', '', content) command = content.strip() if not command: return SkillExecutionResult( skill_id=skill_id, success=False, status=ExecutionStatus.FAILED, output={"error": "LLM generated empty command", "stdout": "", "stderr": "", "returncode": -1}, ) from agent.skills.sandbox_executor import execute_shell result = await execute_shell(command) return SkillExecutionResult( skill_id=skill_id, success=result["success"], status=ExecutionStatus.COMPLETED if result["success"] else ExecutionStatus.FAILED, output={**result, "command": command}, )