File size: 2,973 Bytes
b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 | 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 | """Memory retrieval - search and recall from stored memories."""
import aiofiles
from pathlib import Path
class MemoryRetrieval:
"""Search and retrieve memories using keyword matching."""
def __init__(self, base_path: Path):
self.base_path = Path(base_path)
async def search(self, query: str, max_results: int = 50) -> list[dict]:
"""Search memories by keyword matching across all priority folders.
Args:
query: Search query string. Matches against file content.
max_results: Maximum number of results to return.
Returns:
List of matching memory dicts with path, score, and preview.
"""
query_lower = query.lower()
keywords = query_lower.split()
results = []
# Search across all markdown files
for md_file in sorted(self.base_path.rglob("*.md"), reverse=True):
try:
async with aiofiles.open(md_file, "r", encoding="utf-8") as f:
content = await f.read()
except (OSError, UnicodeDecodeError):
continue
content_lower = content.lower()
score = sum(1 for kw in keywords if kw in content_lower)
if score > 0:
# Determine priority from path
rel = md_file.relative_to(self.base_path)
parts = rel.parts
if "critical" in parts:
priority = "critical"
score += 2 # Boost critical memories
elif "journals" in parts:
priority = "journal"
else:
priority = "normal"
# Extract a relevant snippet around the first match
snippet = self._extract_snippet(content, keywords)
results.append({
"path": str(md_file),
"priority": priority,
"score": score,
"snippet": snippet,
})
# Sort by score descending, then by path (most recent first)
results.sort(key=lambda r: (-r["score"], r["path"]))
return results[:max_results]
@staticmethod
def _extract_snippet(content: str, keywords: list[str], context_chars: int = 200) -> str:
"""Extract a text snippet around the first keyword match."""
content_lower = content.lower()
best_pos = -1
for kw in keywords:
pos = content_lower.find(kw)
if pos >= 0:
best_pos = pos
break
if best_pos < 0:
return content[:context_chars]
start = max(0, best_pos - context_chars // 2)
end = min(len(content), best_pos + context_chars // 2)
snippet = content[start:end].strip()
if start > 0:
snippet = "..." + snippet
if end < len(content):
snippet = snippet + "..."
return snippet
|