334 lines
10 KiB
Python
334 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""MemPalace after_turn hook for LibreFang.
|
|
|
|
Filters conversation turns for relevant memories and saves them to MemPalace.
|
|
Skips: tool calls, short exchanges, noise, and turns where the agent already
|
|
used mcp_mempalace tools explicitly (deduplication).
|
|
|
|
Input (stdin): {"type": "after_turn", "agent_id": "...", "messages": [...]}
|
|
Output (stdout): {"status": "..."} (fire-and-forget)
|
|
|
|
Install: librefang plugin install mempalace-indexer && librefang plugin requirements mempalace-indexer
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from langdetect import detect as _langdetect, LangDetectException
|
|
_LANGDETECT_AVAILABLE = True
|
|
except ImportError:
|
|
_LANGDETECT_AVAILABLE = False
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Configuration: read from LIBREFANG_PLUGIN_CONFIG (written by the runtime),
|
|
# fall back to individual environment variables for direct invocation.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _load_config():
|
|
cfg_path = os.environ.get("LIBREFANG_PLUGIN_CONFIG")
|
|
if cfg_path:
|
|
try:
|
|
with open(cfg_path) as f:
|
|
return json.load(f)
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
return {}
|
|
|
|
_cfg = _load_config()
|
|
|
|
|
|
def _cfg_str(key, env_key, default):
|
|
if key in _cfg:
|
|
return str(_cfg[key])
|
|
return os.environ.get(env_key, default)
|
|
|
|
|
|
def _cfg_int(key, env_key, default):
|
|
if key in _cfg:
|
|
try:
|
|
return int(_cfg[key])
|
|
except (TypeError, ValueError):
|
|
pass
|
|
try:
|
|
return int(os.environ.get(env_key, str(default)))
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def _cfg_bool(key, env_key, default):
|
|
if key in _cfg:
|
|
v = _cfg[key]
|
|
if isinstance(v, bool):
|
|
return v
|
|
return str(v).lower() not in ("0", "false", "no", "off")
|
|
raw = os.environ.get(env_key)
|
|
if raw is None:
|
|
return default
|
|
return raw != "0"
|
|
|
|
|
|
PALACE_PATH = _cfg_str("palace_path", "MEMPALACE_PALACE_PATH",
|
|
os.path.expanduser("~/.mempalace/palace"))
|
|
# Minimum character length of extracted text to be worth saving.
|
|
MIN_CONTENT_LENGTH = _cfg_int("min_chars", "MEMPALACE_MIN_CHARS", 80)
|
|
# How many recent messages to consider (sliding window).
|
|
WINDOW_SIZE = _cfg_int("window_size", "MEMPALACE_WINDOW_SIZE", 6)
|
|
# Max content hashes to keep in the dedup store (rolling, oldest dropped first).
|
|
DEDUP_MAX = _cfg_int("dedup_max", "MEMPALACE_DEDUP_MAX", 500)
|
|
# When langdetect is available, RELEVANCE_RE is only applied to English text.
|
|
# Non-English text passes on length + dedup alone. Set to false/0 to disable.
|
|
LANG_DETECT_ENABLED = _cfg_bool("lang_detect", "MEMPALACE_LANG_DETECT", True)
|
|
|
|
# Room classification: all matching rules win (multi-room).
|
|
# Falls back to ("default", "sessions") when nothing matches.
|
|
ROOM_RULES: list[tuple[re.Pattern, tuple[str, str]]] = [
|
|
(
|
|
re.compile(
|
|
r"\b(contact|phone|email|address|family|wife|husband"
|
|
r"|son|daughter|parent|colleague|coworker)\b"
|
|
r"|\S+@\S+\.\w+", # bare email address pattern
|
|
re.IGNORECASE,
|
|
),
|
|
("people", "contacts"),
|
|
),
|
|
(
|
|
re.compile(
|
|
r"\b(appointment|deadline|birthday|event|meeting|schedule"
|
|
r"|remind me|reminder|calendar|due date|due on)\b",
|
|
re.IGNORECASE,
|
|
),
|
|
("time", "calendar"),
|
|
),
|
|
(
|
|
re.compile(
|
|
r"\b(budget|expense|transaction|payment|bill|salary|invoice"
|
|
r"|cost|price|paid|spending|refund)\b",
|
|
re.IGNORECASE,
|
|
),
|
|
("finance", "transactions"),
|
|
),
|
|
(
|
|
re.compile(
|
|
r"\b(package|order|shipment|delivery|tracking|shipped|arrived)\b",
|
|
re.IGNORECASE,
|
|
),
|
|
("logistics", "orders"),
|
|
),
|
|
(
|
|
re.compile(
|
|
r"\b(decision|decided|prefer|from now on|going forward"
|
|
r"|we.ll use|i.ll use|switching to|chosen|agreed)\b",
|
|
re.IGNORECASE,
|
|
),
|
|
("knowledge", "decisions"),
|
|
),
|
|
]
|
|
|
|
RELEVANCE_RE = re.compile(
|
|
r"\b(decision|decided|prefer|from now on|going forward|remember that|note that"
|
|
r"|remind me|don.t forget|important|urgent|critical|keep in mind"
|
|
r"|appointment|deadline|birthday|event|meeting|schedule|due date"
|
|
r"|budget|expense|transaction|payment|bill|salary|invoice|cost|price"
|
|
r"|package|order|shipment|delivery|tracking"
|
|
r"|contact|phone|email|address"
|
|
r"|like|dislike|preference|habit|allergy"
|
|
r"|family|wife|husband|son|daughter|parent|colleague"
|
|
r"|work|client|project|we.ll use|i.ll use|switching to)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
# Matches fenced code blocks — stripped from content before relevance checks.
|
|
CODE_BLOCK_RE = re.compile(r"```.*?```", re.DOTALL)
|
|
|
|
# Residual noise patterns after code block stripping.
|
|
# Patterns are anchored or specific to avoid matching normal prose
|
|
# ("with the exception of", "no traceback available" in casual writing).
|
|
NOISE_RE = re.compile(
|
|
r"\[tool_call\]|\[tool_result\]|\"type\":\s*\"tool"
|
|
r"|Traceback \(most recent call last\)" # Python traceback header
|
|
r"|^\s*(Exception|Error|Warning):", # exception/error line starts
|
|
re.IGNORECASE | re.MULTILINE,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def emit(obj: dict) -> None:
|
|
json.dump(obj, sys.stdout)
|
|
sys.stdout.write("\n")
|
|
|
|
|
|
def _detect_language(text: str) -> str:
|
|
"""Return ISO 639-1 language code, or 'unknown' on failure."""
|
|
if not _LANGDETECT_AVAILABLE or not LANG_DETECT_ENABLED:
|
|
return "unknown"
|
|
try:
|
|
return _langdetect(text[:400])
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
|
|
def _is_english(lang: str) -> bool:
|
|
return lang in ("en", "unknown")
|
|
|
|
|
|
def _classify_rooms(text: str) -> list[tuple[str, str]]:
|
|
"""Return all matching (wing, room) destinations. Falls back to default."""
|
|
matches = [dest for pattern, dest in ROOM_RULES if pattern.search(text)]
|
|
return matches if matches else [("default", "sessions")]
|
|
|
|
|
|
def _strip_code_blocks(text: str) -> str:
|
|
"""Replace fenced code blocks with a placeholder, preserving surrounding context."""
|
|
return CODE_BLOCK_RE.sub("[code]", text).strip()
|
|
|
|
|
|
def _content_hash(text: str) -> str:
|
|
"""SHA-256 of the first 500 chars — stable fingerprint for near-duplicate detection."""
|
|
return hashlib.sha256(text[:500].encode()).hexdigest()
|
|
|
|
|
|
def _dedup_path(agent_id: str) -> Path:
|
|
# Per-agent store: prevents one agent's memories from blocking another's
|
|
# when multiple agents share the same palace.
|
|
safe_id = re.sub(r"[^\w-]", "_", agent_id)[:64]
|
|
return Path(PALACE_PATH) / f".after_turn_seen_{safe_id}.json"
|
|
|
|
|
|
def _load_seen(agent_id: str) -> list:
|
|
try:
|
|
return json.loads(_dedup_path(agent_id).read_text())
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return []
|
|
|
|
|
|
def _save_seen(agent_id: str, hashes: list) -> None:
|
|
path = _dedup_path(agent_id)
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(hashes[-DEDUP_MAX:]))
|
|
except OSError:
|
|
pass # dedup is best-effort; don't block indexing
|
|
|
|
|
|
def _is_duplicate(text: str, agent_id: str) -> bool:
|
|
h = _content_hash(text)
|
|
seen = _load_seen(agent_id)
|
|
if h in seen:
|
|
return True
|
|
seen.append(h)
|
|
_save_seen(agent_id, seen)
|
|
return False
|
|
|
|
|
|
def extract_text(messages): # list[dict] -> tuple[str, bool]
|
|
"""Extract user+assistant text; detect if agent already saved to mempalace."""
|
|
recent = messages[-WINDOW_SIZE:]
|
|
parts = []
|
|
agent_used_mempalace = False
|
|
|
|
for msg in recent:
|
|
role = msg.get("role", "")
|
|
content = msg.get("content") or ""
|
|
|
|
if role in ("tool", "assistant") and "mcp_mempalace" in str(content):
|
|
agent_used_mempalace = True
|
|
|
|
if role not in ("user", "assistant"):
|
|
continue
|
|
|
|
if isinstance(content, list):
|
|
content = "\n".join(
|
|
b.get("text", "") for b in content
|
|
if isinstance(b, dict) and b.get("type") == "text"
|
|
)
|
|
|
|
if not content:
|
|
continue
|
|
|
|
content = _strip_code_blocks(content)
|
|
|
|
if not content or NOISE_RE.search(content):
|
|
continue
|
|
|
|
parts.append(f"[{role}] {content}")
|
|
|
|
return "\n".join(parts), agent_used_mempalace
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main() -> None:
|
|
try:
|
|
data = json.load(sys.stdin)
|
|
except (json.JSONDecodeError, EOFError):
|
|
emit({"status": "skip", "reason": "bad input"})
|
|
return
|
|
|
|
messages = data.get("messages", [])
|
|
agent_id = data.get("agent_id", "unknown")
|
|
|
|
if not messages:
|
|
emit({"status": "skip", "reason": "no messages"})
|
|
return
|
|
|
|
text, already_saved = extract_text(messages)
|
|
|
|
if already_saved:
|
|
emit({"status": "skip", "reason": "agent used mcp_mempalace"})
|
|
return
|
|
|
|
if len(text) < MIN_CONTENT_LENGTH:
|
|
emit({"status": "skip", "reason": "too short"})
|
|
return
|
|
|
|
lang = _detect_language(text)
|
|
# RELEVANCE_RE is English-only; skip it for non-English to avoid false negatives.
|
|
if _is_english(lang) and not RELEVANCE_RE.search(text):
|
|
emit({"status": "skip", "reason": "not relevant"})
|
|
return
|
|
|
|
if _is_duplicate(text, agent_id):
|
|
emit({"status": "skip", "reason": "duplicate"})
|
|
return
|
|
|
|
try:
|
|
from mempalace.miner import get_collection, add_drawer
|
|
|
|
collection = get_collection(PALACE_PATH)
|
|
source = f"auto-{agent_id}-{datetime.now().strftime('%Y%m%d-%H%M%S%f')}"
|
|
rooms = _classify_rooms(text)
|
|
|
|
for wing, room in rooms:
|
|
add_drawer(
|
|
collection=collection,
|
|
wing=wing,
|
|
room=room,
|
|
content=text,
|
|
source_file=source,
|
|
chunk_index=0,
|
|
agent="mempalace-indexer",
|
|
)
|
|
|
|
emit({
|
|
"status": "indexed",
|
|
"chars": len(text),
|
|
"lang": lang,
|
|
"rooms": [{"wing": w, "room": r} for w, r in rooms],
|
|
})
|
|
except Exception as e:
|
|
emit({"status": "error", "error": str(e)})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|