duffyduckandClaude Opus 4.8
5992a7e441
Neues Feld scope=system|personal auf jedem Memory-Punkt. Bootstrap-Export getrennt: System-Regeln (generisch, teilbar) vs. Persönliches (Name, Zugangsdaten, Projekte). Import ist scope-sicher — ein System-Import löscht NICHT die persönlichen pinned Memories. seed_rules + AGENT.md/TOOLING.md → system, USER.md-Präferenzen → personal. Backfill für Bestand (57 system / 617 personal). Diagnostic: zwei Export-Buttons, scope-Badge (SYS/PRIV) + Umschalter pro Memory. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
"""Einmaliger Backfill: weist bestehenden Memory-Punkten ein `scope`
|
|
(system | personal) zu. Sicher & reversibel — Stefan kann pro Eintrag in der
|
|
Diagnostic-UI umschalten. Idempotent: laeuft mehrfach ohne Schaden.
|
|
|
|
Heuristik (datengetrieben aus dem realen Bestand):
|
|
- type=preference / fact / conversation / reminder -> personal
|
|
- source in (seed, auto-feedback) -> system
|
|
- type=identity -> system
|
|
- type in (rule, tool, skill) und category in SYSTEM_CATS -> system
|
|
- sonst -> personal (sicher: nichts leakt)
|
|
|
|
Aufruf im Brain-Container:
|
|
docker exec aria-brain python3 /app/backfill_scope.py # dry-run
|
|
docker exec aria-brain python3 /app/backfill_scope.py --apply # schreibt
|
|
"""
|
|
import os
|
|
import sys
|
|
from collections import Counter
|
|
|
|
from qdrant_client import QdrantClient
|
|
from qdrant_client.http import models as qm
|
|
|
|
COLLECTION = "aria_memory"
|
|
SYSTEM_CATS = {
|
|
"sicherheit", "arbeitsweise", "architektur", "ehrlichkeit", "verhalten",
|
|
"voice", "skills", "freigaben", "infrastruktur", "persoenlichkeit",
|
|
"pentest", "ausgabe",
|
|
}
|
|
|
|
|
|
def compute_scope(pl: dict) -> str:
|
|
typ = pl.get("type")
|
|
src = pl.get("source")
|
|
cat = (pl.get("category") or "").lower()
|
|
if typ == "preference":
|
|
return "personal"
|
|
if typ in ("fact", "conversation", "reminder"):
|
|
return "personal"
|
|
if src in ("seed", "auto-feedback"):
|
|
return "system"
|
|
if typ == "identity":
|
|
return "system"
|
|
if typ in ("rule", "tool", "skill") and cat in SYSTEM_CATS:
|
|
return "system"
|
|
return "personal"
|
|
|
|
|
|
def main():
|
|
apply = "--apply" in sys.argv
|
|
force = "--force" in sys.argv # auch schon gesetzte scopes ueberschreiben
|
|
c = QdrantClient(
|
|
host=os.environ.get("QDRANT_HOST", "aria-qdrant"),
|
|
port=int(os.environ.get("QDRANT_PORT", "6333")),
|
|
)
|
|
pts, _ = c.scroll(collection_name=COLLECTION, limit=5000,
|
|
with_payload=True, with_vectors=False)
|
|
|
|
per_scope: dict[str, list] = {"system": [], "personal": []}
|
|
pinned_examples = Counter()
|
|
skipped = 0
|
|
for p in pts:
|
|
pl = p.payload or {}
|
|
if pl.get("scope") in ("system", "personal") and not force:
|
|
skipped += 1
|
|
continue
|
|
scope = compute_scope(pl)
|
|
per_scope[scope].append(p.id)
|
|
if pl.get("pinned"):
|
|
pinned_examples[(scope, pl.get("source"), pl.get("type"),
|
|
pl.get("category"))] += 1
|
|
|
|
print(f"total={len(pts)} skipped(already set)={skipped}")
|
|
print(f"-> system={len(per_scope['system'])} personal={len(per_scope['personal'])}")
|
|
print("pinned split (scope, source, type, category):")
|
|
for k, v in sorted(pinned_examples.items()):
|
|
print(" ", k, v)
|
|
|
|
if not apply:
|
|
print("\nDRY-RUN — nichts geschrieben. Mit --apply ausfuehren.")
|
|
return
|
|
|
|
for scope, ids in per_scope.items():
|
|
if not ids:
|
|
continue
|
|
c.set_payload(collection_name=COLLECTION, payload={"scope": scope}, points=ids)
|
|
print(f"\nAPPLIED: system={len(per_scope['system'])} personal={len(per_scope['personal'])}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|