Spaces:
Runtime error
Runtime error
| """The file formats of chat memory, as pure functions: render and parse. | |
| Layout, relative to the notebook's chat area (R2 `ws/{workspace}/chats/{notebook}/`): | |
| INDEX.md one line per topic across every chat of the notebook | |
| {chat_id}/TOPICS.md this chat's topic list and the `through` pointer | |
| {chat_id}/{slug}.md one file per topic: summary, then exact quotes | |
| Every parser tolerates a missing or hand edited file by returning something empty, so one bad file never breaks a | |
| chat.""" | |
| import re | |
| import uuid | |
| from dataclasses import dataclass, field | |
| from app.corpus import slugify | |
| TOPICS = "TOPICS.md" | |
| INDEX = "INDEX.md" | |
| MAX_TOPICS = 30 # lines kept in TOPICS.md; the least recently active are dropped from the list, never from the database | |
| MAX_VERBATIM = 10 | |
| MAX_VERBATIM_CHARS = 1200 | |
| MAX_RECENT = 8 # message ids kept per topic: the last four rounds | |
| MAX_KEYWORDS = 20 | |
| SMALL_TALK = re.compile( | |
| r"^\s*(hi|hey|hello|yo|hiya|good (morning|afternoon|evening)|thanks|thank you|thx|ty|cheers|ok(ay)?|" | |
| r"cool|great|nice|awesome|perfect|got it|sounds good|bye|goodbye|see you|lol|haha)[\s!.,:)]*$", | |
| re.IGNORECASE, | |
| ) | |
| _MARKER = re.compile(r"\[(\d+)\]") | |
| _VERBATIM_HEAD = re.compile(r"^\[([0-9a-fA-F-]{36}) (writer|assistant) (\d{4}-\d{2}-\d{2})\]$") | |
| def chat_path(chat_id: uuid.UUID | str, name: str) -> str: | |
| return f"{chat_id}/{name}" | |
| def topic_path(chat_id: uuid.UUID | str, slug: str) -> str: | |
| return chat_path(chat_id, f"{slug}.md") | |
| def topic_slug(label: str) -> str: | |
| return slugify(label, max_len=48) | |
| def one_line(text: str, limit: int = 200) -> str: | |
| """A value that sits on one `a | b | c` line: no newlines, no pipes.""" | |
| return re.sub(r"\s+", " ", text.replace("|", "/")).strip()[:limit] | |
| def squeeze(text: str, limit: int) -> str: | |
| """Shorten a long message by keeping its start and end, cut on word boundaries.""" | |
| text = text.strip() | |
| if len(text) <= limit: | |
| return text | |
| half = limit // 2 | |
| head, tail = text[:half], text[-half:] | |
| head = head.rsplit(" ", 1)[0] if " " in head else head | |
| tail = tail.split(" ", 1)[1] if " " in tail else tail | |
| return f"{head.rstrip()} ... {tail.lstrip()}" | |
| def strip_markers(text: str) -> str: | |
| return re.sub(r"[ \t]+([.,;:!?])", r"\1", _MARKER.sub("", text)) | |
| def clean_keywords(words: list[str]) -> list[str]: | |
| seen: dict[str, None] = {} | |
| for w in words: | |
| w = one_line(str(w), 40).lower().replace(",", " ").strip() | |
| if w and w not in seen: | |
| seen[w] = None | |
| return list(seen)[:MAX_KEYWORDS] | |
| def _front_matter(text: str) -> tuple[dict[str, str], str]: | |
| """(key: value pairs, the rest) of a file that starts with a --- block.""" | |
| lines = text.splitlines() | |
| if not lines or lines[0].strip() != "---": | |
| return {}, text | |
| meta: dict[str, str] = {} | |
| for i, line in enumerate(lines[1:], start=1): | |
| if line.strip() == "---": | |
| return meta, "\n".join(lines[i + 1:]) | |
| if ":" in line: | |
| key, _, value = line.partition(":") | |
| meta[key.strip()] = value.strip() | |
| return {}, text # never closed: treat the file as unreadable front matter | |
| # TOPICS.md | |
| class TopicEntry: | |
| slug: str | |
| label: str | |
| last: str # YYYY-MM-DD of the last round | |
| rounds: int | |
| keywords: str # comma separated | |
| gist: str | |
| class TopicsFile: | |
| chat_id: str = "" | |
| through: str = "" # id of the last message processed, "" before the first run | |
| current: str = "" # slug of the topic the newest filed round went to: what the chat was just talking about | |
| updated: str = "" | |
| entries: list[TopicEntry] = field(default_factory=list) | |
| def get(self, slug: str) -> TopicEntry | None: | |
| return next((e for e in self.entries if e.slug == slug), None) | |
| def slugs(self) -> set[str]: | |
| return {e.slug for e in self.entries} | |
| def render_topics(tf: TopicsFile) -> str: | |
| entries = sorted(tf.entries, key=lambda e: e.last, reverse=True)[:MAX_TOPICS] | |
| lines = ["---", f"chat: {tf.chat_id}", f"through: {tf.through or 'none'}", f"current: {tf.current or 'none'}", | |
| f"updated: {tf.updated}", "---"] | |
| for e in entries: | |
| lines.append(f"- {e.slug} | {one_line(e.label, 80)} | {e.last} | {e.rounds} rounds | " | |
| f"{one_line(e.keywords, 200)} | {one_line(e.gist, 200)}") | |
| return "\n".join(lines) + "\n" | |
| def parse_topics(text: str) -> TopicsFile: | |
| meta, body = _front_matter(text or "") | |
| through = meta.get("through", "") | |
| current = meta.get("current", "") | |
| tf = TopicsFile(chat_id=meta.get("chat", ""), through="" if through in ("", "none") else through, | |
| current="" if current in ("", "none") else current, updated=meta.get("updated", "")) | |
| for line in body.splitlines(): | |
| if not line.startswith("- "): | |
| continue | |
| parts = [p.strip() for p in line[2:].split("|")] | |
| if len(parts) < 6: | |
| continue | |
| rounds = re.match(r"\d+", parts[3]) | |
| tf.entries.append(TopicEntry(slug=parts[0], label=parts[1], last=parts[2], | |
| rounds=int(rounds.group()) if rounds else 0, | |
| keywords=parts[4], gist="|".join(parts[5:]))) | |
| return tf | |
| # {slug}.md | |
| class Quote: | |
| message_id: str | |
| role: str # writer | assistant | |
| date: str | |
| text: str | |
| class Topic: | |
| label: str | |
| slug: str | |
| chat_id: str | |
| first: str = "" | |
| last: str = "" | |
| rounds: int = 0 | |
| keywords: list[str] = field(default_factory=list) | |
| gist: str = "" | |
| recent: list[str] = field(default_factory=list) # message ids, oldest first | |
| last_message: str = "" # the newest message filed here: a rerun after a crash skips rounds up to it | |
| summary: str = "" | |
| verbatim: list[Quote] = field(default_factory=list) | |
| def clean_summary(text: str, limit: int) -> str: | |
| """The summary sits under a `## Summary` heading, so a heading inside it would break the file's sections.""" | |
| text = "\n".join(re.sub(r"^#+\s*", "", ln) for ln in (text or "").strip().splitlines()) | |
| return squeeze(text, limit) if len(text) > limit else text | |
| def render_topic(t: Topic) -> str: | |
| lines = ["---", f"topic: {one_line(t.label, 80)}", f"slug: {t.slug}", f"chat: {t.chat_id}", f"first: {t.first}", | |
| f"last: {t.last}", f"rounds: {t.rounds}", f"keywords: {', '.join(t.keywords)}", | |
| f"gist: {one_line(t.gist, 200)}", f"recent: {', '.join(t.recent)}", f"last_message: {t.last_message}", | |
| "---", "", "## Summary", "", | |
| t.summary.strip(), "", "## Verbatim", ""] | |
| for q in t.verbatim: | |
| lines.append(f"[{q.message_id} {q.role} {q.date}]") | |
| lines += [f"> {ln}" if ln else ">" for ln in q.text.splitlines()] | |
| lines.append("") | |
| return "\n".join(lines).rstrip() + "\n" | |
| def parse_topic(text: str) -> Topic | None: | |
| meta, body = _front_matter(text or "") | |
| if not meta.get("slug"): | |
| return None | |
| t = Topic(label=meta.get("topic", meta["slug"]), slug=meta["slug"], chat_id=meta.get("chat", ""), | |
| first=meta.get("first", ""), last=meta.get("last", ""), | |
| keywords=[k.strip() for k in meta.get("keywords", "").split(",") if k.strip()], | |
| gist=meta.get("gist", ""), recent=[r.strip() for r in meta.get("recent", "").split(",") if r.strip()], | |
| last_message=meta.get("last_message", "")) | |
| rounds = re.match(r"\d+", meta.get("rounds", "")) | |
| t.rounds = int(rounds.group()) if rounds else 0 | |
| section, summary, current = "", [], None | |
| for line in body.splitlines(): | |
| if line.startswith("## "): | |
| section = line[3:].strip().lower() | |
| current = None | |
| continue | |
| if section == "summary": | |
| summary.append(line) | |
| elif section == "verbatim": | |
| head = _VERBATIM_HEAD.match(line) | |
| if head: | |
| current = Quote(head.group(1), head.group(2), head.group(3), "") | |
| t.verbatim.append(current) | |
| elif current is not None and line.startswith(">"): | |
| current.text += ("\n" if current.text else "") + line[1:].removeprefix(" ") | |
| t.summary = "\n".join(summary).strip() | |
| return t | |
| # INDEX.md (notebook level) | |
| def index_line(chat_id: str, entry: TopicEntry) -> str: | |
| return (f"{entry.last} | {topic_path(chat_id, entry.slug)} | {one_line(entry.label, 80)} | " | |
| f"{one_line(entry.keywords, 200)} | {one_line(entry.gist, 200)}") | |
| def render_index(per_chat: dict[str, list[TopicEntry]]) -> str: | |
| """Newest first, one line per topic. Rebuilt from every chat's TOPICS.md, never edited in place.""" | |
| rows = [(e.last, index_line(chat, e)) for chat, entries in per_chat.items() for e in entries] | |
| return "\n".join(line for _, line in sorted(rows, key=lambda r: r[0], reverse=True)) + ("\n" if rows else "") | |