scripts/distill_memory_to_wiki.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
Distill conversation memory into wiki topic pages.
Reads recent memory/YYYY-MM-DD.md files, extracts insights worth
long-term preservation (decisions, workflows, principles), and either
stages them for review or applies approved entries to wiki topics.
Usage:
python3 distill_memory_to_wiki.py --dry-run [--days 3]
python3 distill_memory_to_wiki.py --stage [--days 2]
python3 distill_memory_to_wiki.py --apply --staging-file wiki/_staging/2026-04-06-candidates.md
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import xkb_paths
WORKSPACE = xkb_paths.WORKSPACE
_SKILL_DIR = Path(__file__).resolve().parent.parent
WIKI_DIR = xkb_paths.WIKI_DIR
# ── Unified LLM helper ────────────────────────────────────────────────────────
sys.path.insert(0, str(_SKILL_DIR / "scripts"))
from _llm import call as _llm_backend
TOPICS_DIR = WIKI_DIR / "topics"
STAGING_DIR = WIKI_DIR / "_staging"
INDEX_PATH = WIKI_DIR / "index.md"
LOG_PATH = WIKI_DIR / "log.md"
MEMORY_DIR = WORKSPACE / "memory"
def load_env_key() -> str:
return "" # auth handled by _llm.py via openclaw CLI
def llm_call(system: str, user: str, api_key: str = "") -> str:
raw = _llm_backend(system, user)
return re.sub(r"<think>.*?</think>", "", raw, flags=re.DOTALL).strip()
def load_topic_slugs() -> list[str]:
from xkb_review import topic_accepts_candidates
return [p.stem for p in TOPICS_DIR.glob("*.md") if topic_accepts_candidates(p)] if TOPICS_DIR.exists() else []
def append_log(entry: str) -> None:
LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
existing = LOG_PATH.read_text(encoding="utf-8") if LOG_PATH.exists() else "# Wiki Log\n---\n"
LOG_PATH.write_text(existing.rstrip() + "\n" + entry + "\n", encoding="utf-8")
def load_recent_memory(days: int) -> list[tuple[str, str]]:
"""Return [(date_str, content)] for the last N days of memory files."""
entries = []
today = datetime.now(timezone.utc).date()
for i in range(days):
d = today - timedelta(days=i)
fname = MEMORY_DIR / f"{d}.md"
if fname.exists():
entries.append((str(d), fname.read_text(encoding="utf-8")))
return entries
L1_TRACE_DIR = xkb_paths.XKB_DATA_DIR / "runtime" / "l1-traces"
def _trace_message_text(message: object) -> str:
"""Convert an L1 message into compact, distillable conversation text.
L1 traces intentionally retain tool calls and metadata for provenance, but
those are noise for the daily curator. Keep human-readable user/assistant
text and skip tool-call payloads here; the original trace remains on disk.
"""
if not isinstance(message, dict):
return ""
role = str(message.get("role") or "").strip()
if role not in {"user", "assistant"}:
return ""
content = message.get("content", "")
if isinstance(content, str):
text = content.strip()
elif isinstance(content, list):
parts = []
for item in content:
if isinstance(item, dict) and item.get("type") == "text":
value = str(item.get("text") or "").strip()
if value:
parts.append(value)
text = "\n".join(parts)
else:
text = ""
return f"{role}: {text}" if text else ""
def load_recent_l1_traces(days: int) -> list[tuple[str, str]]:
"""Load the newest complete trace per session for the recent date window.
agent_end traces are cumulative snapshots of a session, so loading every
file would repeat the same conversation many times. The newest snapshot per
episode is the canonical input; source files remain immutable evidence.
"""
if not L1_TRACE_DIR.exists():
return []
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
newest_by_episode: dict[str, tuple[datetime, Path, dict]] = {}
for path in L1_TRACE_DIR.glob("*.json"):
try:
trace = json.loads(path.read_text(encoding="utf-8"))
observed = datetime.fromisoformat(str(trace.get("observed_at", "")).replace("Z", "+00:00"))
if observed < cutoff or trace.get("schema") != "xkb-l1-trace.v1":
continue
episode = str(trace.get("episode_id") or trace.get("session_id") or path.name)
previous = newest_by_episode.get(episode)
if previous is None or observed > previous[0]:
newest_by_episode[episode] = (observed, path, trace)
except (OSError, ValueError, json.JSONDecodeError, TypeError):
continue
entries: list[tuple[str, str]] = []
for observed, path, trace in sorted(newest_by_episode.values(), key=lambda item: item[0]):
messages = trace.get("content", [])
if not isinstance(messages, list):
continue
rendered = [line for message in messages if (line := _trace_message_text(message))]
if not rendered:
continue
source_date = observed.date().isoformat()
header = f"[L1 trace: {path.name}; episode: {trace.get('episode_id') or trace.get('session_id')}]"
entries.append((source_date, header + "\n" + "\n".join(rendered)))
return entries
def load_recent_conversations(days: int) -> list[tuple[str, str]]:
"""Recent conversations, read from the store the hook actually writes to.
One entry per session, rendered the way a transcript reads, because a
single turn out of context rarely carries a decision — the reasoning is
usually spread across the exchange either side of it.
"""
if not xkb_paths.SERVICE_DB.exists():
return []
import sqlite3
since = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
try:
with sqlite3.connect(f"file:{xkb_paths.SERVICE_DB}?mode=ro", uri=True) as db:
rows = db.execute(
"""SELECT t.session_id, t.query, t.answer, t.completed_at
FROM turns t
WHERE t.completed_at > ?
AND t.status = 'succeeded'
ORDER BY t.session_id, t.completed_at""",
(since,),
).fetchall()
except sqlite3.Error as err:
print(f"[WARN] could not read conversations: {err}")
return []
sessions: dict[str, list[str]] = {}
dates: dict[str, str] = {}
for session_id, query, answer, completed_at in rows:
lines = sessions.setdefault(session_id, [])
if query:
lines.append(f"user: {query.strip()}")
if answer:
lines.append(f"assistant: {answer.strip()}")
dates.setdefault(session_id, (completed_at or "")[:10])
return [
(dates[session_id], "\n\n".join(lines))
for session_id, lines in sessions.items()
if lines
]
def load_recent_inputs(days: int) -> list[tuple[str, str]]:
"""Daily memory, plus the recent conversations, from wherever they are.
The knowledge service is the live store. The trace directory is the older
inbox — still read when there is no service on this host, because an
external writer may be filling it.
"""
entries = load_recent_memory(days)
conversations = load_recent_conversations(days)
if conversations:
print(f"Loaded {len(conversations)} recent conversation(s)")
entries.extend(conversations)
return entries
traces = load_recent_l1_traces(days)
if traces:
print(f"Loaded {len(traces)} recent L1 trace snapshot(s)")
entries.extend(traces)
return entries
EXTRACT_SYSTEM = """You are a wiki knowledge curator for an AI agent system.
Extract insights worth long-term preservation from conversation memory logs.
Inclusion criteria (ALL must be met):
1. Contains a technical decision with reasoning, reusable workflow/command/rule, or confirmed principle/conclusion
2. Adds real new dimension to existing knowledge (not just paraphrasing)
3. Useful 6 months from now (exclude time-sensitive news, cron logs, daily chores)
Exclusion list:
- Daily activity logs
- Cron / heartbeat execution records
- Unresolved discussions
- Emotional reactions or temporary statements
- Anything only meaningful today
Output ONLY valid JSON, no markdown fences, no explanation."""
EXTRACT_USER_TMPL = """Memory file date: {date}
Content:
{content}
Existing wiki topics:
{topics}
Extract wiki-worthy insights as JSON:
{{
"insights": [
{{
"topic_slug": "most relevant existing topic slug (from list above), or null if none fits",
"topic_suggestion": "if topic_slug is null, suggest new topic name in kebab-case",
"section": "target section: 核心概念 / 做法-Workflow / 案例 / 矛盾與未解問題",
"content": "synthesized insight (1-3 sentences, not a log entry)",
"confidence": "high / medium / low",
"source_date": "{date}"
}}
]
}}
If nothing qualifies, output: {{"insights": []}}"""
def load_todays_staged_content(date_str: str) -> str:
"""Return already-staged insight summaries for today (for dedup context)."""
if not STAGING_DIR.exists():
return ""
parts = []
for f in sorted(STAGING_DIR.glob(f"{date_str}-*-candidates.md")):
# Extract just the content lines from each candidate block
text = f.read_text(encoding="utf-8")
# Grab lines after "Status:" lines (the actual insight text)
in_content = False
for line in text.splitlines():
if line.startswith("- **Status:**"):
in_content = True
continue
if in_content:
if line.startswith("- **") or line.startswith("---") or line.startswith("##"):
in_content = False
elif line.strip():
parts.append(line.strip())
return "\n".join(parts)
def extract_conversation_content(content: str) -> str:
"""
Strip dreaming metadata from memory files, keeping only the actual conversation
content lines (Candidate: User/Assistant) plus any non-dreaming plain text.
Removes: confidence, evidence, recalls, status lines after each Candidate entry.
"""
lines = content.splitlines()
result = []
skip_next_metadata = False
for line in lines:
stripped = line.strip()
# Skip dreaming metadata lines that follow a Candidate entry
if skip_next_metadata:
if stripped.startswith("- confidence:") or \
stripped.startswith("- evidence:") or \
stripped.startswith("- recalls:") or \
stripped.startswith("- status:") or \
stripped.startswith("- note:"):
continue
else:
skip_next_metadata = False
# Keep Candidate lines but strip the "- Candidate: " prefix
if stripped.startswith("- Candidate:"):
text = stripped[len("- Candidate:"):].strip()
if text:
result.append(text)
skip_next_metadata = True
continue
# Skip dreaming block markers and low-value headers
if stripped.startswith("<!-- openclaw:dreaming") or \
stripped in ("## Light Sleep", "## REM Sleep", "### Reflections", "### Possible Lasting Truths"):
continue
# Keep everything else (plain text sections, non-dreaming headers)
result.append(line)
return "\n".join(result).strip()
CHUNK_SIZE = 4000 # chars per LLM call
def extract_insights(entries: list[tuple[str, str]], api_key: str, verbose: bool = False,
already_staged: str = "") -> list[dict]:
existing_topics = load_topic_slugs()
topics_str = "\n".join(f"- {s}" for s in existing_topics) if existing_topics else "(no topics yet)"
all_insights = []
for date, content in entries:
cleaned = extract_conversation_content(content)
if verbose:
print(f" [LLM] scanning {date} ({len(content)} chars raw → {len(cleaned)} chars cleaned)...")
# Chunk into CHUNK_SIZE segments so the full file is always processed
chunks = [cleaned[i:i + CHUNK_SIZE] for i in range(0, max(len(cleaned), 1), CHUNK_SIZE)]
seen_contents: set[str] = set()
for chunk_i, chunk in enumerate(chunks):
if not chunk.strip():
continue
user_prompt = EXTRACT_USER_TMPL.format(date=date, content=chunk, topics=topics_str)
if already_staged:
user_prompt += f"\n\nAlready staged today (SKIP these, do not re-surface):\n{already_staged[:800]}"
try:
raw = llm_call(EXTRACT_SYSTEM, user_prompt, api_key)
raw = re.sub(r"^```(?:json)?\s*", "", raw).rstrip("` \n")
parsed = json.loads(raw)
insights = parsed.get("insights", [])
# Dedup by content to avoid the same insight surfacing across chunks
for ins in insights:
key = ins.get("content", "")[:120]
if key and key not in seen_contents:
seen_contents.add(key)
all_insights.append(ins)
if verbose:
print(f" [LLM] {date} chunk {chunk_i + 1}/{len(chunks)}: {len(insights)} insight(s)")
except Exception as e:
print(f" [WARN] LLM failed for {date} chunk {chunk_i + 1}: {e}")
return all_insights
def write_staging(insights: list[dict], date_str: str, label: str = "") -> Path:
"""Write staging file. label (e.g. 'morning', 'evening') prevents same-day overwrites."""
STAGING_DIR.mkdir(parents=True, exist_ok=True)
suffix = f"-{label}" if label else ""
out_path = STAGING_DIR / f"{date_str}{suffix}-candidates.md"
lines = [
f"# Memory Distillation Candidates — {date_str}",
"",
f"Generated: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}",
f"Insights found: {len(insights)}",
"",
"---",
"",
"## Review Instructions",
"- Mark each item: `[x] approve` or leave `[ ] skip`",
"- Run: `python3 distill_memory_to_wiki.py --apply --staging-file <this file>`",
"",
]
existing_topics = set(load_topic_slugs())
for i, ins in enumerate(insights, 1):
proposed = str(ins.get("topic_slug") or ins.get("topic_suggestion") or "?").strip()
slug = proposed if proposed in existing_topics else f"[NEW: {proposed}]"
lines += [
f"## Candidate {i}",
f"- **Topic:** {slug}",
f"- **Section:** {ins.get('section', '核心概念')}",
f"- **Confidence:** {ins.get('confidence', 'medium')}",
f"- **Source date:** {ins.get('source_date', '')}",
f"- **Status:** [ ] approve [ ] skip",
"",
ins.get("content", ""),
"",
"---",
"",
]
out_path.write_text("\n".join(lines), encoding="utf-8")
return out_path
def apply_staging_file(
staging_path: Path,
approve_all: bool = False,
auto_approve_high: bool = False,
) -> tuple[int, int, list[str]]:
"""
Apply approved candidates from staging file to wiki topics.
Returns (applied, skipped, updated_slugs).
- approve_all: treat every candidate as approved regardless of status/confidence
- auto_approve_high: automatically approve candidates with confidence=high;
others still require manual [x] approve in the staging file
"""
content = staging_path.read_text(encoding="utf-8")
from xkb_review import (_split_candidates, stable_candidate_id, _body, _field,
load_topic_resolutions, resolved_topic)
from xkb_provenance import candidate_marker
blocks = _split_candidates(content)
try:
source_file = staging_path.resolve().relative_to(STAGING_DIR.resolve()).as_posix()
except ValueError:
source_file = staging_path.resolve().as_posix()
resolutions = load_topic_resolutions()
# Validate every override before the first write, using the same identity
# and content guard as governance. Never rewrite the source Topic field.
resolved = {number: resolved_topic(resolutions, stable_candidate_id(source_file, number),
_body(block), _field(block, "Topic"))[0]
for number, block in blocks}
applied = 0
skipped = 0
updated_slugs: list[str] = []
for number, block in blocks:
topic_m = re.search(r"\*\*Topic:\*\* (.+)", block)
section_m = re.search(r"\*\*Section:\*\* (.+)", block)
status_m = re.search(r"\*\*Status:\*\* \[x\] approve", block, re.IGNORECASE)
source_m = re.search(r"\*\*Source date:\*\* (.+)", block)
conf_m = re.search(r"\*\*Confidence:\*\* (\w+)", block)
confidence = conf_m.group(1).strip() if conf_m else "medium"
is_auto_approved = auto_approve_high and confidence == "high"
if not status_m and not approve_all and not is_auto_approved:
skipped += 1
continue
slug = resolved[number]
section = section_m.group(1).strip() if section_m else "核心概念"
source_date = source_m.group(1).strip() if source_m else "unknown"
content_lines = []
in_content = False
for line in block.splitlines():
if line.startswith("- **") or line.startswith("---"):
in_content = False
if in_content and line.strip():
content_lines.append(line)
if line.startswith("- **Status:**"):
in_content = True
entry_text = "\n".join(content_lines).strip()
if not entry_text or not slug or slug.startswith("[NEW:"):
print(f" [SKIP] Missing slug or content")
skipped += 1
continue
marker = candidate_marker(stable_candidate_id(source_file, number))
if not upsert_wiki_section(slug, section, entry_text, source_date, marker=marker):
skipped += 1
continue
applied += 1
if slug not in updated_slugs:
updated_slugs.append(slug)
# Mark staging file as applied
if applied > 0:
applied_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
header_addition = f"\n> ✅ Applied {applied} candidate(s) at {applied_at}\n"
staging_path.write_text(content.replace(
"## Review Instructions",
f"## Apply Status{header_addition}\n## Review Instructions",
1,
), encoding="utf-8")
return applied, skipped, updated_slugs
def upsert_wiki_section(slug: str, section: str, entry: str, source_date: str,
*, marker: str = "") -> bool:
from xkb_review import topic_accepts_candidates
topic_path = TOPICS_DIR / f"{slug}.md"
if not topic_accepts_candidates(topic_path):
print(f" [SKIP] Topic {slug} is missing, archived or a redirect")
return False
content = topic_path.read_text(encoding="utf-8")
if marker and marker in content:
return False
if marker:
entry += f" <!-- {marker} -->"
section_header = f"## {section}"
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
if section_header in content:
content = content.replace(
section_header,
f"{section_header}\n\n- {entry} *(self-derived · memory/{source_date}.md)*",
1,
)
else:
content = content.rstrip() + f"\n\n{section_header}\n\n- {entry} *(self-derived · memory/{source_date}.md)*\n"
content = re.sub(r"(last_updated:\s*)\S+", f"\\g<1>{today}", content)
topic_path.write_text(content, encoding="utf-8")
print(f" [OK] {slug} / {section}")
return True
def main() -> None:
parser = argparse.ArgumentParser(description="Distill memory into wiki")
mode = parser.add_mutually_exclusive_group(required=True)
mode.add_argument("--dry-run", action="store_true")
mode.add_argument("--stage", action="store_true")
mode.add_argument("--apply", action="store_true")
parser.add_argument("--days", type=int, default=2)
parser.add_argument("--label", default="",
help="Staging filename suffix to avoid overwriting (e.g. 'morning', 'evening')")
parser.add_argument("--input", metavar="TEXT",
help="Distill inline text instead of daily memory files")
parser.add_argument("--input-file", metavar="PATH",
help="Distill content from a file instead of daily memory files")
parser.add_argument("--staging-file")
parser.add_argument("--approve-all", action="store_true",
help="Approve all candidates without checking [x] marks")
parser.add_argument("--auto-apply-high", action="store_true",
help="After --stage, automatically apply high-confidence candidates to wiki "
"without manual review. Low/medium candidates remain in staging.")
parser.add_argument("--no-llm", action="store_true")
parser.add_argument("--verbose", action="store_true")
args = parser.parse_args()
if args.apply:
if not args.staging_file:
parser.error("--apply requires --staging-file")
staging_path = Path(args.staging_file)
applied, skipped, slugs = apply_staging_file(
staging_path,
approve_all=args.approve_all,
auto_approve_high=getattr(args, 'auto_approve_high', False),
)
print(f"\nApply results:")
print(f" Applied : {applied}")
print(f" Skipped : {skipped}")
if slugs:
print(f" Topics updated: {', '.join(slugs)}")
if applied > 0:
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M")
append_log(
f"\n## [{now}] apply-staging | {applied} candidate(s) applied"
f" from {staging_path.name} → {', '.join(slugs)}"
)
print(f" Log updated.")
return
# Determine input source: --input TEXT, --input-file PATH, or daily memory files
today_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
if args.input:
entries = [(today_str, args.input)]
print(f"Input mode: inline text ({len(args.input)} chars)")
elif args.input_file:
input_path = Path(args.input_file)
if not input_path.exists():
print(f"[ERROR] --input-file not found: {input_path}")
return
content = input_path.read_text(encoding="utf-8")
entries = [(today_str, content)]
print(f"Input mode: file {input_path.name} ({len(content)} chars)")
else:
entries = load_recent_inputs(args.days)
if not entries:
print(f"No memory files or L1 traces found for last {args.days} days.")
return
print(f"Loaded {len(entries)} input item(s): {[e[0] for e in entries]}")
if args.no_llm:
for date, content in entries:
print(f"\n=== {date} ({len(content)} chars) ===")
print(content[:400] + "..." if len(content) > 400 else content)
return
api_key = "" # auth handled by _llm.py via openclaw CLI
# Load today's already-staged content for dedup (only for memory-based runs, not --input)
already_staged = ""
if not args.input and not args.input_file:
already_staged = load_todays_staged_content(today_str)
if already_staged:
print(f"Dedup context: found existing staging for {today_str}, will skip duplicates")
insights = extract_insights(entries, api_key, verbose=args.verbose, already_staged=already_staged)
high = [i for i in insights if i.get("confidence") == "high"]
print(f"\nInsights found: {len(insights)} ({len(high)} high-confidence)")
if args.dry_run:
for ins in insights:
slug = ins.get("topic_slug") or f"[NEW: {ins.get('topic_suggestion')}]"
print(f"\n [{ins.get('confidence','?')}] {slug} / {ins.get('section','?')}")
print(f" {ins.get('content','')}")
return
if args.stage:
today_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
# Auto-label inline/file input so it doesn't collide with memory-based staging files
label = args.label
if not label and (args.input or args.input_file):
label = "input"
out = write_staging(insights, today_str, label=label)
print(f"\nStaged to: {out}")
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M")
source_desc = "inline input" if (args.input or args.input_file) else f"last {args.days}d"
append_log(f"\n## [{now}] ingest-memory | staged {len(insights)} candidates from {source_desc} → {out.name}")
# Auto-apply high-confidence candidates immediately after staging
if getattr(args, 'auto_apply_high', False):
high_count = sum(1 for ins in insights if ins.get("confidence") == "high")
print(f"\nAuto-applying {high_count} high-confidence candidate(s)...")
applied, skipped, slugs = apply_staging_file(out, auto_approve_high=True)
print(f" Applied : {applied}")
print(f" Skipped : {skipped} (medium/low — review manually)")
if slugs:
print(f" Topics updated: {', '.join(slugs)}")
if applied > 0:
append_log(
f"\n## [{now}] auto-apply-high | {applied} high-confidence candidate(s) "
f"applied from {out.name} → {', '.join(slugs)}"
)
else:
print(f"Run: python3 distill_memory_to_wiki.py --apply --staging-file {out}")
import xkb_usage # noqa: E402 — 量測誰在跑，見 scripts/xkb_usage.py
if __name__ == "__main__":
xkb_usage.record(__file__)
main()