shared/agent/memory/knowledge_web.py

"""
Knowledge Web — persistent graph-based knowledge store.

Ultra-lightweight knowledge graph that stores agent solutions,
allowing agents to learn from past executions and share knowledge
across sessions. Uses a simple JSON-backed graph structure.
"""
import os
import json
import time
import hashlib
from typing import Optional

_STORE_PATH = os.getenv("ADAM_KNOWLEDGE_PATH", "/tmp/adam_knowledge.json")
_MAX_NODES = int(os.getenv("ADAM_KNOWLEDGE_MAX", "500"))


class KnowledgeWeb:
    """
    Persistent knowledge graph for agent learning.

    Stores:
    - Solutions: goal → result mappings with confidence scores
    - Concepts: extracted named entities and relationships
    - Patterns: reusable execution patterns

    Queries:
    - Similarity search over past solutions
    - Concept relationship traversal
    - Pattern matching for known goal types
    """

    def __init__(self):
        self._solutions: dict[str, dict] = {}
        self._concepts: dict[str, set] = {}  # concept → related goal hashes
        self._load()

    def _load(self):
        """Load knowledge from disk."""
        try:
            if os.path.exists(_STORE_PATH):
                with open(_STORE_PATH, "r") as f:
                    data = json.load(f)
                    self._solutions = data.get("solutions", {})
                    self._concepts = {k: set(v) for k, v in data.get("concepts", {}).items()}
        except Exception:
            pass

    def _save(self):
        """Persist knowledge to disk."""
        try:
            os.makedirs(os.path.dirname(_STORE_PATH) or ".", exist_ok=True)
            # Limit size
            solutions = dict(sorted(
                self._solutions.items(),
                key=lambda x: x[1].get("ts", 0),
                reverse=True
            )[:_MAX_NODES])

            with open(_STORE_PATH, "w") as f:
                json.dump({
                    "solutions": solutions,
                    "concepts": {k: list(v) for k, v in self._concepts.items()},
                }, f)
        except Exception:
            pass

    async def store_solution(self, goal: str, action: str, result: str,
                              confidence: float = 0.8):
        """Store a successful solution for future reuse."""
        goal_hash = hashlib.md5(goal.lower().encode()).hexdigest()[:16]

        self._solutions[goal_hash] = {
            "goal": goal[:200],
            "action": action[:100],
            "result": result[:500],
            "confidence": min(1.0, max(0.0, confidence)),
            "ts": time.time(),
            "use_count": self._solutions.get(goal_hash, {}).get("use_count", 0) + 1,
        }

        # Extract and index concepts
        concepts = self._extract_concepts(goal)
        for concept in concepts:
            if concept not in self._concepts:
                self._concepts[concept] = set()
            self._concepts[concept].add(goal_hash)

        self._save()

    async def query_solution(self, goal: str, context: str = "",
                              min_confidence: float = 0.0) -> Optional[dict]:
        """
        Query for a past solution matching the goal.
        Returns the best match if confidence threshold is met.
        """
        if not self._solutions:
            return None

        goal_lower = goal.lower()
        goal_hash = hashlib.md5(goal_lower.encode()).hexdigest()[:16]

        # Exact match
        if goal_hash in self._solutions:
            entry = self._solutions[goal_hash]
            entry["confidence"] = min(1.0, entry.get("confidence", 1.0) * 0.95 + 0.05)
            return {
                "summary": f"Found past solution (used {entry.get('use_count', 0)} times)",
                "result": entry["result"],
                "confidence": entry["confidence"],
                "action": entry["action"],
            }

        # Concept-based match
        goal_concepts = self._extract_concepts(goal)
        best_match = None
        best_score = 0.0

        for concept in goal_concepts:
            if concept in self._concepts:
                for related_hash in self._concepts[concept]:
                    if related_hash in self._solutions:
                        entry = self._solutions[related_hash]
                        score = entry.get("confidence", 0.5) * 0.8 + 0.2
                        if score > best_score:
                            best_score = score
                            best_match = entry

        if best_match and best_score >= min_confidence:
            return {
                "summary": f"Found related solution (confidence: {best_score:.2f})",
                "result": best_match["result"],
                "confidence": best_score,
                "action": best_match["action"],
            }

        return None

    def _extract_concepts(self, text: str) -> list[str]:
        """Extract key concepts from text for indexing."""
        text_lower = text.lower()
        concepts = set()

        # Common concept patterns
        patterns = [
            "code", "python", "javascript", "function", "script",
            "search", "find", "research", "lookup",
            "file", "data", "database", "api",
            "write", "create", "generate", "build",
            "analyze", "analyze", "compare", "evaluate",
            "debug", "fix", "repair", "error",
            "install", "setup", "configure", "deploy",
            "learn", "explain", "understand", "tutorial",
        ]

        for pattern in patterns:
            if pattern in text_lower:
                concepts.add(pattern)

        # Add significant words (3+ chars)
        words = set(w for w in text_lower.split() if len(w) >= 4 and w.isalpha())
        concepts.update(list(words)[:5])

        return list(concepts)

    async def size(self) -> int:
        """Get the number of stored solutions."""
        return len(self._solutions)

    async def clear(self):
        """Clear all knowledge."""
        self._solutions.clear()
        self._concepts.clear()
        self._save()

    def get_stats(self) -> dict:
        """Get knowledge web statistics."""
        return {
            "solutions": len(self._solutions),
            "concepts": len(self._concepts),
            "top_concepts": sorted(
                [(k, len(v)) for k, v in self._concepts.items()],
                key=lambda x: x[1], reverse=True
            )[:5],
        }