splitbit_llm/skills/skills.py

"""Skill Creation System — automatic skill abstraction from conversations.

Adapted from inc_llm_v1's SkillManager + SkillFactory pattern.

After every conversation (voice or text):
1. Extract the pattern (what was asked → what worked)
2. Create a reusable skill with trigger conditions
3. Store in the skill library for future use

Skill types: conversation, code, speed, voice, tool
Bayesian effectiveness scoring: skills that work well get higher priority.
Meta-skills: after 10+ interactions in a category, summarize best practices.
"""

from __future__ import annotations

import hashlib
import logging
import time
from collections import defaultdict, deque
from dataclasses import dataclass, field
from typing import Any

import numpy as np

logger = logging.getLogger(__name__)


@dataclass
class Skill:
    """A reusable skill extracted from conversations."""
    id: str
    name: str
    description: str
    content: str
    category: str  # conversation, code, speed, voice, tool
    trigger_conditions: list[str] = field(default_factory=list)
    created_at: float = field(default_factory=time.time)
    last_used: float = field(default_factory=time.time)
    use_count: int = 0
    success_count: int = 0
    failure_count: int = 0
    effectiveness_score: float = 0.5  # Bayesian prior
    confidence: float = 0.0
    version: int = 1


class SkillFactory:
    """Creates skills from conversation patterns.

    Analyzes conversations and extracts reusable patterns.
    """

    def __init__(self) -> None:
        self._interaction_buffer: deque[dict[str, Any]] = deque(maxlen=500)
        self._category_counts: dict[str, int] = defaultdict(int)

    def record_interaction(
        self,
        user_message: str,
        assistant_response: str,
        channel: str = "cli",
        success: bool = True,
    ) -> None:
        """Record a conversation interaction for skill extraction."""
        self._interaction_buffer.append({
            "user_message": user_message,
            "assistant_response": assistant_response,
            "channel": channel,
            "success": success,
            "timestamp": time.time(),
        })

        # Categorize
        category = self._categorize(user_message, channel)
        self._category_counts[category] += 1

    def _categorize(self, message: str, channel: str) -> str:
        """Categorize an interaction."""
        msg_lower = message.lower()
        if channel == "voice" or channel == "jarvis":
            return "voice"
        if any(kw in msg_lower for kw in ["code", "function", "bug", "error", "debug", "python", "javascript"]):
            return "code"
        if any(kw in msg_lower for kw in ["tool", "search", "run", "execute", "file"]):
            return "tool"
        if len(message) < 20:
            return "speed"
        return "conversation"

    def extract_skill(self) -> Skill | None:
        """Extract a skill from recent interactions.

        Looks for patterns in recent interactions and creates a skill
        if a clear pattern emerges.
        """
        if len(self._interaction_buffer) < 3:
            return None

        # Group recent interactions by category
        recent = list(self._interaction_buffer)[-20:]
        categories: dict[str, list[dict[str, Any]]] = defaultdict(list)
        for interaction in recent:
            cat = self._categorize(interaction["user_message"], interaction["channel"])
            categories[cat].append(interaction)

        # Find the category with most interactions
        best_cat = max(categories.items(), key=lambda x: len(x[1]))
        if len(best_cat[1]) < 3:
            return None

        category, interactions = best_cat

        # Extract pattern
        success_rate = sum(1 for i in interactions if i["success"]) / len(interactions)
        if success_rate < 0.5:
            return None

        # Create skill content
        skill_id = hashlib.sha256(
            f"{category}:{time.time()}".encode()
        ).hexdigest()[:16]

        examples = "\n".join(
            f"  Q: {i['user_message'][:80]}\n  A: {i['assistant_response'][:80]}"
            for i in interactions[:5]
        )

        content = (
            f"Skill: {category.title()} Interaction Pattern\n"
            f"Success rate: {success_rate:.0%}\n"
            f"Examples:\n{examples}\n"
            f"Best practice: Be concise and direct for {category} interactions."
        )

        triggers = self._extract_triggers(interactions, category)

        skill = Skill(
            id=skill_id,
            name=f"{category}-pattern-{skill_id[:8]}",
            description=f"Learned {category} interaction pattern ({success_rate:.0%} success)",
            content=content,
            category=category,
            trigger_conditions=triggers,
            effectiveness_score=success_rate,
            confidence=min(1.0, len(interactions) / 10.0),
        )

        return skill

    def _extract_triggers(self, interactions: list[dict[str, Any]], category: str) -> list[str]:
        """Extract trigger conditions from interactions."""
        triggers = [category]
        # Find common keywords
        word_freq: dict[str, int] = defaultdict(int)
        for i in interactions:
            for word in i["user_message"].lower().split():
                if len(word) > 3:
                    word_freq[word] += 1
        # Top 5 common words as triggers
        top_words = sorted(word_freq.items(), key=lambda x: -x[1])[:5]
        triggers.extend(w for w, _ in top_words)
        return triggers

    def maybe_create_meta_skill(self, skill_manager: "SkillManager") -> Skill | None:
        """Create a meta-skill after enough interactions in a category."""
        for category, count in self._category_counts.items():
            if count >= 10:
                existing = skill_manager.read(f"{category}-meta")
                if not existing:
                    meta_id = hashlib.sha256(
                        f"meta:{category}:{time.time()}".encode()
                    ).hexdigest()[:16]
                    skill = Skill(
                        id=meta_id,
                        name=f"{category}-meta",
                        description=f"Meta-skill for {category} — learned patterns across all {category} interactions",
                        content=(
                            f"Meta-Skill: {category.title()}\n"
                            f"Total interactions: {count}\n"
                            f"Best practices:\n"
                            f"- Be concise and direct\n"
                            f"- Match the user's tone\n"
                            f"- Provide actionable responses\n"
                        ),
                        category=f"{category}_meta",
                        trigger_conditions=[category, "meta"],
                        effectiveness_score=0.7,
                        confidence=min(1.0, count / 20.0),
                    )
                    return skill
        return None

    def get_stats(self) -> dict[str, Any]:
        return {
            "interactions_buffered": len(self._interaction_buffer),
            "category_counts": dict(self._category_counts),
        }


class SkillManager:
    """Manages skills — storage, retrieval, scoring, and lifecycle.

    Uses Bayesian effectiveness scoring. Skills that work well get
    higher priority. Skills that fail get deprecated.
    """

    def __init__(self, storage: Any = None) -> None:
        self.storage = storage
        self._skills: dict[str, Skill] = {}
        self._trigger_index: dict[str, set[str]] = defaultdict(set)
        self._stats = {
            "skills_created": 0,
            "skills_used": 0,
            "skills_deprecated": 0,
            "meta_skills_created": 0,
        }

    def create(self, skill: Skill) -> bool:
        """Create a new skill."""
        if skill.id in self._skills:
            return False
        self._skills[skill.id] = skill
        for trigger in skill.trigger_conditions:
            self._trigger_index[trigger.lower()].add(skill.id)
        self._stats["skills_created"] += 1
        if "meta" in skill.category:
            self._stats["meta_skills_created"] += 1

        if self.storage:
            self.storage.save_skill(skill)

        logger.info("Created skill: %s (category=%s, score=%.2f)",
                     skill.name, skill.category, skill.effectiveness_score)
        return True

    def read(self, name: str) -> Skill | None:
        """Read a skill by name."""
        for skill in self._skills.values():
            if skill.name == name:
                return skill
        return None

    def find_by_triggers(self, message: str, max_results: int = 3) -> list[Skill]:
        """Find skills that match trigger conditions in the message."""
        msg_lower = message.lower()
        matched: dict[str, float] = defaultdict(float)

        for trigger, skill_ids in self._trigger_index.items():
            if trigger in msg_lower:
                for sid in skill_ids:
                    skill = self._skills.get(sid)
                    if skill:
                        matched[sid] += skill.effectiveness_score * skill.confidence

        sorted_ids = sorted(matched.items(), key=lambda x: -x[1])[:max_results]
        return [self._skills[sid] for sid, _ in sorted_ids if sid in self._skills]

    def record_use(self, skill_id: str, success: bool) -> None:
        """Record a skill use and update effectiveness score."""
        skill = self._skills.get(skill_id)
        if not skill:
            return

        skill.use_count += 1
        skill.last_used = time.time()
        if success:
            skill.success_count += 1
        else:
            skill.failure_count += 1

        # Bayesian update
        total = skill.success_count + skill.failure_count
        if total > 0:
            success_rate = skill.success_count / total
            # Bayesian: posterior = (prior * w + observed * n) / (w + n)
            w = 2.0  # prior weight
            skill.effectiveness_score = (0.5 * w + success_rate * total) / (w + total)
            skill.confidence = min(1.0, total / 10.0)

        # Deprecate low-scoring skills
        if skill.effectiveness_score < 0.2 and skill.use_count > 5:
            self._stats["skills_deprecated"] += 1
            logger.info("Deprecated skill: %s (score=%.2f)", skill.name, skill.effectiveness_score)

        self._stats["skills_used"] += 1

    def get_relevant_skills(self, message: str, channel: str = "cli") -> list[Skill]:
        """Get skills relevant to a message and channel."""
        skills = self.find_by_triggers(message, max_results=5)
        # Filter by channel
        if channel in ("voice", "jarvis"):
            voice_skills = [s for s in skills if s.category in ("voice", "speed")]
            if voice_skills:
                return voice_skills
        return skills

    def get_skill_context(self, message: str, channel: str = "cli") -> str:
        """Get skill context to inject into the prompt."""
        skills = self.get_relevant_skills(message, channel)
        if not skills:
            return ""
        parts = [s.content[:200] for s in skills[:3]]
        return " | ".join(parts)

    def get_stats(self) -> dict[str, Any]:
        return {
            **self._stats,
            "total_skills": len(self._skills),
            "active_skills": sum(1 for s in self._skills.values() if s.effectiveness_score > 0.2),
            "trigger_index_size": len(self._trigger_index),
        }