""" ModalitySkillEngine — Load and execute persona-intrinsic modality skills. Architecture: Claude Skill pattern (prompt-driven, not function calling). SKILL.md body → injected as LLM instructions LLM outputs structured JSON → engine parses Engine calls tools via ToolRegistry Works with ANY LLM provider — no function calling support required. Lifecycle: L1 build_prompt() → inject descriptions into Express prompt L2 activate() → load SKILL.md body on first use L3 execute() → prompt LLM → parse JSON → call tools """ from __future__ import annotations import importlib import json import re from pathlib import Path from typing import Optional, List 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 ModalitySkillEngine: """Persona-intrinsic skill engine for modality-triggered skills.""" 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 (L1) ------------------------------------------------------- def load_all(self) -> dict[str, Skill]: """Load L1 metadata for trigger=modality 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 == "modality" and skill.modality: self._skills[skill.skill_id] = skill except Exception as e: print(f"[modality-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 -------------------------------------------------------------- @property def modality_skills(self) -> dict[str, str]: return {s.modality: s.skill_id for s in self._skills.values()} def get_by_modality(self, modality: str) -> Optional[Skill]: skill_id = self.modality_skills.get(modality) return self._skills.get(skill_id) if skill_id else None def build_prompt(self) -> str: skills = list(self._skills.values()) if not skills: return "" parts = ["# 技能指南"] for skill in skills: if skill.description: parts.append(f"\n## {skill.name}\n{skill.description}") return "\n".join(parts) # -- Plan & Execute (L3 — VERA-inspired) ----------------------------------- async def plan_and_execute( self, raw_modality: str, raw_output: str, persona, llm, chat_history: list = None, ) -> List[SkillExecutionResult]: """LLM-driven multi-skill planning and execution. Inspired by VERA's activate_skill pattern: 1. LLM sees all available skill summaries + raw_modality 2. LLM returns an ordered execution plan (JSON array) 3. Engine executes each skill in order, merging results Returns list of SkillExecutionResult (one per executed skill). """ from providers.llm.base import ChatMessage if not self._skills: return [] # Build skill catalog for the planning prompt skill_catalog = [] for skill in self._skills.values(): entry = { "modality": skill.modality, "name": skill.name, "description": skill.description, "tools": skill.tools, } skill_catalog.append(entry) catalog_json = json.dumps(skill_catalog, ensure_ascii=False, indent=2) system_prompt = ( "你是一个 SKILL 调度器。根据 Express 输出的【表达方式】，决定需要执行哪些技能、按什么顺序执行。\n\n" f"## 可用技能\n```json\n{catalog_json}\n```\n\n" "## 规则\n" "1. 只选择【表达方式】中明确提到的技能\n" "2. 如果【表达方式】中同时包含语音和多条拆分，只选择 modality=语音，忽略多条拆分\n" "3. 内容生成类技能（照片、语音）排在前面，投递方式类技能（多条拆分）排在后面\n" "4. 如果没有匹配任何技能，返回空数组 []\n" "5. 每个技能条目必须包含 modality 和 params\n\n" "## 输出格式\n" "返回 JSON 数组，按执行顺序排列：\n" '```json\n[{"modality": "照片", "params": {...}}, {"modality": "多条拆分", "params": {...}}]\n```\n\n' "对于每个技能，params 的格式参考该技能的 SKILL.md 文档（会在激活时提供）。\n" "现在只需要返回 modality 列表，params 设为空对象 {} 即可。" ) user_prompt = f"【表达方式】原文：{raw_modality}" messages = [ ChatMessage("system", system_prompt), ChatMessage("user", user_prompt), ] try: response = await llm.chat(messages, temperature=0.1) plan = self._extract_json(response.content) except Exception as e: print(f" [skill-plan] ❌ Planning failed: {e}") plan = None # Fallback: if planning fails, try simple keyword matching if not plan or not isinstance(plan, list): plan = [] for skill in self._skills.values(): if skill.modality and skill.modality in raw_modality: plan.append({"modality": skill.modality, "params": {}}) if plan: print(f" [skill-plan] ⚠ LLM plan failed, fallback to keyword matching: {[p['modality'] for p in plan]}") if not plan: return [] # Apply excludes rules declared in SKILL.md frontmatter plan_modalities = {p.get("modality") for p in plan} for skill in self._skills.values(): if skill.excludes and skill.modality in plan_modalities: plan = [p for p in plan if p.get("modality") not in skill.excludes] print(f" [skill-plan] 📋 Plan: {[p.get('modality') for p in plan]}") # Execute each skill in plan order results: List[SkillExecutionResult] = [] for step in plan: modality = step.get("modality", "") skill = self.get_by_modality(modality) if not skill: print(f" [skill-plan] ⚠ Unknown modality '{modality}', skipping") continue print(f" [skill] 🎯 modality='{modality}' (from plan)") result = await self.execute(modality, raw_output, persona, llm, chat_history=chat_history) if result: result.output["_modality"] = modality # inject plan's modality results.append(result) return results # -- Single-Skill Execution (L3) ------------------------------------------- async def execute( self, modality: str, raw_output: str, persona, llm, chat_history: list = None, ) -> Optional[SkillExecutionResult]: """Execute a modality skill — prompt-driven, no function calling. Flow: 1. Inject SKILL.md body as LLM instruction 2. LLM outputs structured JSON 3. Engine parses JSON 4. Engine executes tools via ToolRegistry Fast path: split_messages skips LLM (pure text processing). """ skill = self.get_by_modality(modality) if not skill: return None if not skill.is_activated: self.activate(skill.skill_id) # Route: prompt-driven tool-use vs legacy handler if skill.tools and self.tool_registry: return await self._execute_via_prompt(skill, raw_output, persona, llm, chat_history=chat_history) elif skill.handler_fn: return await self._execute_via_handler(skill, raw_output, persona, llm) else: print(f" [modality-skill] ⚠ {skill.name}: no tools or handler_fn") return None # -- Prompt-driven execution (Claude Skill pattern) ---------------------- @staticmethod def _build_chat_summary(chat_history, persona_name: str, max_turns: int = 6, max_chars: int = 600) -> str: """Build a concise chat summary for skill context injection.""" if not chat_history: return "（无历史对话）" recent = chat_history[-max_turns:] lines = [] for m in recent: role = "用户" if m.role == "user" else persona_name lines.append(f"{role}: {m.content[:100]}") return "\n".join(lines)[:max_chars] async def _execute_via_prompt( self, skill: Skill, raw_output: str, persona, llm, chat_history: list = None, ) -> SkillExecutionResult: """Execute skill via prompt-driven structured output. 1. Build prompt from SKILL.md body + context 2. LLM outputs JSON (following SKILL.md format instructions) 3. Engine parses JSON 4. Engine executes tools based on parsed parameters """ from providers.llm.base import ChatMessage try: # Build prompt — SKILL.md body IS the instruction # Conditionally inject chat history (only if skill declares needs_chat_history) chat_block = "" if skill.needs_chat_history and chat_history: chat_summary = self._build_chat_summary(chat_history, persona.name) chat_block = f"## 最近对话\n{chat_summary}\n\n" system_prompt = ( f"{skill.body}\n\n" f"---\n" f"## 当前上下文\n\n" f"角色名：{persona.name}\n" f"角色ID：{persona.persona_id}\n\n" f"{chat_block}" f"角色回复（JSON）：\n{raw_output}\n\n" f"---\n" f"请根据上述技能文档和角色回复上下文，直接输出 JSON。只输出 JSON，不要其他内容。" ) # Pre-inject voice_preset for voice skills if "synthesize_voice" in skill.tools: voice_preset = self._resolve_voice_preset(persona) system_prompt += f"\n\n（系统预设 voice_preset: {voice_preset}）" messages = [ ChatMessage("system", system_prompt), ChatMessage("user", "请输出 JSON。"), ] # Call LLM — NO tools parameter, pure text output response = await llm.chat(messages, temperature=0.3) # Parse JSON from LLM response params = self._extract_json(response.content) if not params: print(f" [modality-skill] ⚠ Failed to parse JSON from LLM output") print(f" [modality-skill] raw: {response.content[:200]}") return SkillExecutionResult( skill_id=skill.skill_id, success=False, status=ExecutionStatus.FAILED, output={"error": "Failed to parse JSON from LLM", "raw": response.content[:500]}, ) print(f" [modality-skill] 📋 LLM params: {json.dumps(params, ensure_ascii=False)[:200]}") # Execute tools based on skill type and parsed params return await self._dispatch_tools(skill, params, persona) except Exception as e: print(f" [modality-skill] ❌ Prompt-driven execution failed: {e}") return SkillExecutionResult( skill_id=skill.skill_id, success=False, status=ExecutionStatus.FAILED, output={"error": str(e)}, ) async def _dispatch_tools( self, skill: Skill, params: dict, persona, ) -> SkillExecutionResult: """Dispatch tool calls based on parsed params. The engine knows the tool orchestration logic for each skill — this is deterministic, not LLM-decided. """ output = {} # ── Photo skill: get_reference_image → generate_photo ── if "generate_photo" in skill.tools: # Step 1: Collect reference images (supports list, fallback to single) ref_types = params.get("reference_types") or [] if not ref_types: single = params.get("reference_type") if single and single != "null": ref_types = [single] reference_images = [] if ref_types and self.tool_registry.has("get_reference_image"): for rt in ref_types: ref_result = await self.tool_registry.execute("get_reference_image", { "persona_id": persona.persona_id, "reference_type": rt, }) output.update(ref_result) if ref_result.get("available"): reference_images.append(ref_result["image_path"]) else: print(f" [modality-skill] ⚠ {rt} not available, skipping") # Step 2: Generate photo (with 2x silent retry) gen_params = { "prompt": params.get("prompt", ""), "persona_id": persona.persona_id, "aspect_ratio": params.get("aspect_ratio", "9:16"), } if reference_images: gen_params["reference_images"] = reference_images gen_result = await self._retry_tool("generate_photo", gen_params) output.update(gen_result) # ── Voice skill: synthesize_voice ── elif "synthesize_voice" in skill.tools: voice_preset = self._resolve_voice_preset(persona) voice_result = await self._retry_tool("synthesize_voice", { "text": params.get("text", ""), "voice_preset": voice_preset, "emotion_instruction": params.get("emotion_instruction", ""), }) output.update(voice_result) # ── Split skill: split_messages ── elif "split_messages" in skill.tools: split_params = {"text": params.get("text", "")} if params.get("delays_ms"): split_params["delays_ms"] = params["delays_ms"] split_result = await self.tool_registry.execute("split_messages", split_params) output.update(split_result) # Determine success — check tool's own flag first, then known output keys success = output.pop("success", False) if not success: success = bool(output.get("image_path") or output.get("audio_path") or output.get("segments")) status_str = "✅" if success else "❌" print(f" [modality-skill] {status_str} {skill.name} {'completed' if success else 'failed'}") return SkillExecutionResult( skill_id=skill.skill_id, success=success, status=ExecutionStatus.COMPLETED if success else ExecutionStatus.FAILED, output=output, ) # -- Helpers --------------------------------------------------------------- async def _retry_tool(self, tool_name: str, params: dict, max_retries: int = 2) -> dict: """Execute a tool with silent retries for transient errors. Retries the same call up to max_retries times. Only the final failure propagates up to the engine. """ import asyncio last_result = {} for attempt in range(1, max_retries + 2): # 1 initial + max_retries result = await self.tool_registry.execute(tool_name, params) success = result.get("success", False) if not success: success = bool(result.get("image_path") or result.get("audio_path")) if success: return result last_result = result if attempt <= max_retries: print(f" [tool] 🔄 {tool_name} retry {attempt}/{max_retries}") await asyncio.sleep(1) # brief pause before retry return last_result def _extract_json(self, text: str): """Extract JSON (object or array) from LLM text output.""" text = text.strip() # Try direct parse (object or array) if text.startswith(("{", "[")): try: return json.loads(text) except json.JSONDecodeError: pass # Strip markdown ```json ... ``` fence and try direct parse stripped = re.sub(r"^```(?:json)?\s*\n?", "", text) stripped = re.sub(r"\n?\s*```\s*$", "", stripped).strip() if stripped != text and stripped.startswith(("{", "[")): try: return json.loads(stripped) except json.JSONDecodeError: # Try sanitizing Chinese curly quotes inside JSON string values sanitized = stripped.replace('\u201c', '\\"').replace('\u201d', '\\"') try: return json.loads(sanitized) except json.JSONDecodeError: pass # Try extracting from ```json ... ``` block (object or array) m = re.search(r"```(?:json)?\s*\n?([{\[].*?[}\]])\s*\n?```", text, re.DOTALL) if m: try: return json.loads(m.group(1)) except json.JSONDecodeError: pass # Try finding [...] block (for plan arrays) m = re.search(r"\[\s*\{.*?\}\s*\]", text, re.DOTALL) if m: try: return json.loads(m.group(0)) except json.JSONDecodeError: pass # Try finding {...} block m = re.search(r"\{[^{}]*\}", text, re.DOTALL) if m: try: return json.loads(m.group(0)) except json.JSONDecodeError: pass return None def _resolve_voice_preset(self, persona) -> str: """Pre-resolve voice_preset from api.yaml voice_map.""" try: from providers.config import _load as _load_config _tts_cfg = _load_config().get("tts", {}) _voice_map = _tts_cfg.get("voice_map", {}) _default_voice = _tts_cfg.get("providers", {}).get( _tts_cfg.get("provider", ""), {} ).get("default_voice", "Cherry") return _voice_map.get(persona.persona_id, _default_voice) except Exception: return "Cherry" # -- Legacy path (fallback) ----------------------------------------------- async def _execute_via_handler( self, skill: Skill, raw_output: str, persona, llm, ) -> Optional[SkillExecutionResult]: """Legacy handler_fn path. Kept during migration.""" from providers.llm.base import ChatMessage if not skill.handler_fn: return None try: system_msg = ChatMessage("system", f"根据以下技能文档和角色回复上下文，生成该技能的结构化输出。\n" f"只输出结构化内容，不要多余解释。\n\n{skill.body}" ) user_msg = ChatMessage("user", f"角色回复：{raw_output}\n角色名：{persona.name}" ) prompt_resp = await llm.chat([system_msg, user_msg], temperature=0.3) module_path, fn_name = skill.handler_fn.rsplit('.', 1) mod = importlib.import_module(module_path) handler = getattr(mod, fn_name) voice_preset = self._resolve_voice_preset(persona) result = await handler( persona_id=persona.persona_id, raw_output=prompt_resp.content, persona_name=persona.name, voice_preset=voice_preset, base_instructions=getattr(persona.voice, 'description', '') or '', ) success = result.get("success", False) return SkillExecutionResult( skill_id=skill.skill_id, success=success, status=ExecutionStatus.COMPLETED if success else ExecutionStatus.FAILED, output=result, ) except Exception as e: print(f" [modality-skill] ❌ handler error: {e}") return SkillExecutionResult( skill_id=skill.skill_id, success=False, status=ExecutionStatus.FAILED, output={"error": str(e)}, )