mirror of
https://github.com/SuperClaude-Org/SuperClaude_Framework.git
synced 2025-12-29 16:16:08 +00:00
refactor: restructure core modules into context and memory packages
- Move pm_init components to dedicated packages - context/: PM mode initialization and contracts - memory/: Reflexion memory system - Remove deprecated superclaude/core/pm_init/ Breaking change: Import paths updated - Old: superclaude.core.pm_init.context_contract - New: superclaude.context.contract 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
11
superclaude/memory/__init__.py
Normal file
11
superclaude/memory/__init__.py
Normal file
@@ -0,0 +1,11 @@
|
||||
"""Learning and Memory Systems
|
||||
|
||||
Manages long-term learning from experience:
|
||||
- Reflexion Memory (mistake learning)
|
||||
- Metrics collection
|
||||
- Pattern recognition
|
||||
"""
|
||||
|
||||
from .reflexion import ReflexionMemory, ReflexionEntry
|
||||
|
||||
__all__ = ["ReflexionMemory", "ReflexionEntry"]
|
||||
158
superclaude/memory/reflexion.py
Normal file
158
superclaude/memory/reflexion.py
Normal file
@@ -0,0 +1,158 @@
|
||||
"""Reflexion Memory System
|
||||
|
||||
Manages long-term learning from mistakes:
|
||||
- Loads past failures and solutions
|
||||
- Prevents recurrence of known errors
|
||||
- Enables systematic improvement
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, List, Optional
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class ReflexionEntry:
|
||||
"""Single reflexion (learning) entry"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
task: str,
|
||||
mistake: str,
|
||||
evidence: str,
|
||||
rule: str,
|
||||
fix: str,
|
||||
tests: List[str],
|
||||
status: str = "adopted",
|
||||
timestamp: Optional[str] = None
|
||||
):
|
||||
self.task = task
|
||||
self.mistake = mistake
|
||||
self.evidence = evidence
|
||||
self.rule = rule
|
||||
self.fix = fix
|
||||
self.tests = tests
|
||||
self.status = status
|
||||
self.timestamp = timestamp or datetime.now().isoformat()
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Convert to dictionary for serialization"""
|
||||
return {
|
||||
"ts": self.timestamp,
|
||||
"task": self.task,
|
||||
"mistake": self.mistake,
|
||||
"evidence": self.evidence,
|
||||
"rule": self.rule,
|
||||
"fix": self.fix,
|
||||
"tests": self.tests,
|
||||
"status": self.status
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> "ReflexionEntry":
|
||||
"""Create from dictionary"""
|
||||
return cls(
|
||||
task=data["task"],
|
||||
mistake=data["mistake"],
|
||||
evidence=data["evidence"],
|
||||
rule=data["rule"],
|
||||
fix=data["fix"],
|
||||
tests=data["tests"],
|
||||
status=data.get("status", "adopted"),
|
||||
timestamp=data.get("ts")
|
||||
)
|
||||
|
||||
|
||||
class ReflexionMemory:
|
||||
"""Manages Reflexion Memory (learning from mistakes)"""
|
||||
|
||||
def __init__(self, git_root: Path):
|
||||
self.git_root = git_root
|
||||
self.memory_path = git_root / "docs" / "memory" / "reflexion.jsonl"
|
||||
self.entries: List[ReflexionEntry] = []
|
||||
|
||||
def load(self) -> Dict[str, Any]:
|
||||
"""Load Reflexion Memory from disk"""
|
||||
if not self.memory_path.exists():
|
||||
# Create empty memory file
|
||||
self.memory_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self.memory_path.touch()
|
||||
return {
|
||||
"total_entries": 0,
|
||||
"rules": [],
|
||||
"recent_mistakes": []
|
||||
}
|
||||
|
||||
# Load entries
|
||||
self.entries = []
|
||||
with open(self.memory_path, "r") as f:
|
||||
for line in f:
|
||||
if line.strip():
|
||||
try:
|
||||
data = json.loads(line)
|
||||
self.entries.append(ReflexionEntry.from_dict(data))
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
# Extract rules and recent mistakes
|
||||
rules = list(set(entry.rule for entry in self.entries if entry.status == "adopted"))
|
||||
recent_mistakes = [
|
||||
{
|
||||
"task": entry.task,
|
||||
"mistake": entry.mistake,
|
||||
"fix": entry.fix
|
||||
}
|
||||
for entry in sorted(self.entries, key=lambda e: e.timestamp, reverse=True)[:5]
|
||||
]
|
||||
|
||||
return {
|
||||
"total_entries": len(self.entries),
|
||||
"rules": rules,
|
||||
"recent_mistakes": recent_mistakes
|
||||
}
|
||||
|
||||
def add_entry(self, entry: ReflexionEntry) -> None:
|
||||
"""Add new reflexion entry"""
|
||||
self.entries.append(entry)
|
||||
|
||||
# Ensure directory exists
|
||||
self.memory_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Append to JSONL file
|
||||
with open(self.memory_path, "a") as f:
|
||||
f.write(json.dumps(entry.to_dict()) + "\n")
|
||||
|
||||
def search_similar_mistakes(self, error_message: str) -> List[ReflexionEntry]:
|
||||
"""Search for similar past mistakes"""
|
||||
# Simple keyword-based search (can be enhanced with semantic search)
|
||||
keywords = set(error_message.lower().split())
|
||||
similar = []
|
||||
|
||||
for entry in self.entries:
|
||||
entry_keywords = set(entry.mistake.lower().split())
|
||||
union = keywords | entry_keywords
|
||||
# Avoid division by zero
|
||||
if len(union) == 0:
|
||||
continue
|
||||
# If >30% keyword overlap, consider similar (lowered threshold for better recall)
|
||||
overlap = len(keywords & entry_keywords) / len(union)
|
||||
if overlap > 0.3:
|
||||
similar.append(entry)
|
||||
|
||||
return sorted(similar, key=lambda e: e.timestamp, reverse=True)
|
||||
|
||||
def get_rules(self) -> List[str]:
|
||||
"""Get all adopted rules"""
|
||||
return list(set(
|
||||
entry.rule
|
||||
for entry in self.entries
|
||||
if entry.status == "adopted"
|
||||
))
|
||||
|
||||
def get_stats(self) -> Dict[str, Any]:
|
||||
"""Get memory statistics"""
|
||||
return {
|
||||
"total_entries": len(self.entries),
|
||||
"adopted_rules": len(self.get_rules()),
|
||||
"total_tasks": len(set(entry.task for entry in self.entries))
|
||||
}
|
||||
Reference in New Issue
Block a user