File size: 6,427 Bytes
043a495 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 | """
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],
}
|