Spaces:
Runtime error
Runtime error
| """Themes for a report: what a notebook's posts are about, and how much of it each thing is. | |
| A report no longer reads the posts' text to find out. Every post already has its key ideas stored (documents.metadata_json["ideas"], | |
| written by generate.extract_ideas, each with the lines it came from), so those ideas stand for the posts, however many are picked: | |
| 1. one model call (GroupThemes) names themes and says which ideas belong to which (it sees at most MAX_LISTED_IDEAS of them); | |
| 2. code files every other idea into a theme and counts: how many posts and ideas each theme has gives its weight, so the weights | |
| are true for the whole selection and not for the sample the model saw. A model groups well and counts badly. | |
| The weights decide how much room a theme gets in the report (report_blueprint.py) and how much of the posts' own text it is written | |
| from (report_evidence.py). A theme the reader asked for is moved up a tier whatever its share.""" | |
| import logging | |
| import time | |
| import uuid | |
| from collections import Counter | |
| from dataclasses import dataclass, field | |
| from sqlalchemy.orm import Session | |
| from app.config import settings | |
| from app.corpus import Corpus | |
| from app.llm import run | |
| from app.llm.provider import fast_lm | |
| from app.llm.signatures import GroupThemes | |
| from app.models import Document, Job, Notebook, Workspace | |
| from app.pipeline.generate import extract_ideas, ideas_fresh | |
| from app.pipeline.idea_pool import _tokens | |
| from app.pipeline.passages import _body_start, notebook_docs, workspace_docs | |
| from app.services.jobs import update_job | |
| from app.services.plans import effective_plan | |
| log = logging.getLogger(__name__) | |
| SMALL_SELECTION = 5 # posts or fewer: the posts are read whole, as a report always was | |
| MIN_FRESH_SHARE = 0.5 # fewer posts than this with ideas stored: read the posts' text as before | |
| MAX_LISTED_IDEAS = 300 # ideas the model sees; the rest are filed by code | |
| MAX_THEMES = 12 | |
| MIN_THEMES = 2 | |
| MAJOR_AT, MEDIUM_AT = 0.25, 0.10 # a theme's share of the report: major from, medium from | |
| POST_WEIGHT, IDEA_WEIGHT = 0.6, 0.4 # how many posts mention it counts for more than how many ideas | |
| CATCH_ALL = "More from your posts" | |
| OPENING_LINES = 8 # lines of a post's opening that stand for a post with no ideas stored yet | |
| class ThemesUnavailable(Exception): | |
| """The ideas did not group into themes: the caller reads the posts as before.""" | |
| class Idea: | |
| id: str | |
| doc: Document | |
| label: str | |
| note: str | |
| details: list[dict] = field(default_factory=list) | |
| sources: list[dict] = field(default_factory=list) | |
| bare: bool = False # no ideas stored for this post: this is its title and opening | |
| class Theme: | |
| id: str | |
| name: str | |
| summary: str | |
| ideas: list[Idea] | |
| asked: bool = False # the reader's request asks for it | |
| weight: float = 0.0 # share of the whole, all themes add up to 1 | |
| tier: str = "minor" # major | medium | minor | |
| pinned: bool = False # asked for: always given a deep section | |
| def doc_ids(self) -> set[uuid.UUID]: | |
| return {i.doc.id for i in self.ideas} | |
| # The posts | |
| def select_docs(db: Session, workspace_id: uuid.UUID, notebook_id: uuid.UUID | None, params: dict) -> list[Document]: | |
| """The posts a report is made from: the ones picked, or the notebook's, newest first. Not cut at ARTIFACT_MAX_POSTS: only | |
| what the plan indexes (and REPORT_MAX_POSTS, when set) limits it.""" | |
| picked = [uuid.UUID(str(i)) for i in params.get("document_ids") or []] | |
| nb = db.get(Notebook, notebook_id) if notebook_id else None | |
| if picked: | |
| docs = workspace_docs(db, workspace_id, picked) | |
| elif nb: | |
| docs = notebook_docs(db, nb.id) | |
| else: | |
| docs = [] | |
| cap = effective_plan(db, db.get(Workspace, workspace_id)).indexed_posts | |
| if settings.report_max_posts > 0: | |
| cap = min(cap, settings.report_max_posts) if cap and cap > 0 else settings.report_max_posts | |
| chosen = docs[:cap] if cap and cap > 0 else docs | |
| log.info("report posts: workspace=%s notebook=%s picked=%d found=%d used=%d plan_cap=%s report_max_posts=%s", workspace_id, | |
| notebook_id, len(picked), len(docs), len(chosen), cap, settings.report_max_posts) | |
| if len(chosen) < len(docs): | |
| log.warning("report posts cut by the plan or REPORT_MAX_POSTS: workspace=%s found=%d used=%d", workspace_id, len(docs), | |
| len(chosen)) | |
| return chosen | |
| def ensure_ideas(db: Session, job: Job | None, workspace_id: uuid.UUID, docs: list[Document]) -> None: | |
| """Extract the ideas of posts that have none yet, together, at most REPORT_INLINE_IDEAS of them. The rest are represented | |
| by their opening; the background ideas job reaches them later.""" | |
| stale = [d for d in docs if not ideas_fresh(d)] | |
| if not stale or settings.report_inline_ideas <= 0: | |
| log.info("report ideas: workspace=%s posts=%d stale=%d inline_limit=%d, nothing extracted now", workspace_id, len(docs), | |
| len(stale), settings.report_inline_ideas) | |
| return | |
| started = time.monotonic() | |
| result = extract_ideas(db, job, workspace_id, [str(d.id) for d in stale], parallel=True, limit=settings.report_inline_ideas) | |
| for d in stale: | |
| db.refresh(d) | |
| log.info("report ideas: workspace=%s posts=%d stale=%d extracted=%s skipped=%s left_for_background=%d seconds=%.1f", | |
| workspace_id, len(docs), len(stale), result.get("extracted"), result.get("skipped"), | |
| sum(1 for d in stale if not ideas_fresh(d)), time.monotonic() - started) | |
| def fresh_share(docs: list[Document]) -> float: | |
| return sum(1 for d in docs if ideas_fresh(d)) / max(len(docs), 1) | |
| def _opening_idea(d: Document, di: int, corpus: Corpus) -> Idea: | |
| """A post with no ideas stored: its title, its topics and where its opening is, so it still counts and can be cited.""" | |
| meta = d.metadata_json or {} | |
| tags = ", ".join(t["name"] for t in (meta.get("topics") or [])[:3] if t.get("name")) | |
| try: | |
| lines = corpus.read_lines(d.path) | |
| except Exception: | |
| lines = [] | |
| first = _body_start(lines) + 1 if lines else 1 | |
| sources = [{"path": d.path, "line_start": first, "line_end": min(first + OPENING_LINES, len(lines) or first)}] | |
| return Idea(id=f"p{di}.0", doc=d, label=d.title[:80], bare=True, sources=sources, | |
| note=f"Topics: {tags}" if tags else "Only the opening of this post was read.") | |
| def collect_ideas(docs: list[Document], corpus: Corpus) -> list[Idea]: | |
| """Every post's stored ideas as Ideas with ids like p3.1 (the third post's second idea), or its opening when it has none.""" | |
| out: list[Idea] = [] | |
| for di, d in enumerate(docs): | |
| stored = [i for i in ((d.metadata_json or {}).get("ideas") or []) if (i.get("label") or "").strip()] \ | |
| if ideas_fresh(d) else [] | |
| if not stored: | |
| out.append(_opening_idea(d, di, corpus)) | |
| continue | |
| for ii, i in enumerate(stored): | |
| out.append(Idea(id=f"p{di}.{ii}", doc=d, label=i["label"].strip(), note=(i.get("note") or "").strip(), | |
| details=i.get("details") or [], sources=i.get("sources") or [])) | |
| log.info("report ideas collected: posts=%d ideas=%d posts_with_only_an_opening=%d", len(docs), len(out), | |
| sum(1 for i in out if i.bare)) | |
| return out | |
| # Themes | |
| def listed_ideas(ideas: list[Idea]) -> list[Idea]: | |
| """The ideas the model sees: all of them, or when there are too many, the same number from every post in turn.""" | |
| if len(ideas) <= MAX_LISTED_IDEAS: | |
| return ideas | |
| by_doc: dict[uuid.UUID, list[Idea]] = {} | |
| for i in ideas: | |
| by_doc.setdefault(i.doc.id, []).append(i) | |
| lists, picked, rank = list(by_doc.values()), [], 0 | |
| while len(picked) < MAX_LISTED_IDEAS and any(rank < len(x) for x in lists): | |
| for x in lists: | |
| if rank < len(x) and len(picked) < MAX_LISTED_IDEAS: | |
| picked.append(x[rank]) | |
| rank += 1 | |
| return picked | |
| def _file_leftovers(themes: list[Theme], leftovers: list[Idea]) -> None: | |
| """Every idea the model did not place goes to the theme its post's other ideas are mostly in, else the theme whose words it | |
| shares most, else a catch-all theme: the way the Mind Constellation files what the model leaves out.""" | |
| by_doc: dict[uuid.UUID, Counter] = {} | |
| for ti, t in enumerate(themes): | |
| for i in t.ideas: | |
| by_doc.setdefault(i.doc.id, Counter())[ti] += 1 | |
| theme_words = [_tokens(" ".join([t.name, t.summary] + [i.label for i in t.ideas[:20]])) for t in themes] | |
| # A word in most themes ("idea", "post") says nothing about which one an idea is about. | |
| common = {w for w, n in Counter(w for tw in theme_words for w in tw).items() if n > max(1, len(themes) // 2)} | |
| theme_words = [tw - common for tw in theme_words] | |
| catch: Theme | None = None | |
| for idea in leftovers: | |
| counts = by_doc.get(idea.doc.id) | |
| if counts: | |
| themes[counts.most_common(1)[0][0]].ideas.append(idea) | |
| continue | |
| words = _tokens(f"{idea.label} {idea.note}") - common | |
| best, score = max(((ti, len(words & tw)) for ti, tw in enumerate(theme_words)), key=lambda x: x[1]) | |
| if score > 0: | |
| themes[best].ideas.append(idea) | |
| continue | |
| if catch is None: | |
| catch = Theme(id="", name=CATCH_ALL, summary="Ideas that fit none of the other themes.", ideas=[]) | |
| themes.append(catch) | |
| catch.ideas.append(idea) | |
| log.info("report leftover ideas filed: leftovers=%d catch_all=%d", len(leftovers), len(catch.ideas) if catch else 0) | |
| def weigh(themes: list[Theme], total_posts: int, total_ideas: int) -> None: | |
| """Each theme's share of the whole, from how many posts and ideas it has (not from anything the model said), and its tier. A | |
| theme the reader asked for moves up one tier and is pinned. If nothing reached major, the heaviest theme becomes major.""" | |
| raw = [POST_WEIGHT * len(t.doc_ids) / max(total_posts, 1) + IDEA_WEIGHT * len(t.ideas) / max(total_ideas, 1) for t in themes] | |
| total = sum(raw) or 1.0 | |
| for t, r in zip(themes, raw, strict=True): | |
| t.weight = round(r / total, 4) | |
| t.tier = "major" if t.weight >= MAJOR_AT else "medium" if t.weight >= MEDIUM_AT else "minor" | |
| if themes and not any(t.tier == "major" for t in themes): | |
| max(themes, key=lambda t: t.weight).tier = "major" | |
| for t in themes: | |
| if t.asked: | |
| t.pinned = True | |
| t.tier = {"minor": "medium", "medium": "major"}.get(t.tier, t.tier) | |
| def _relations(raw: list[dict] | None, themes: list[Theme]) -> list[dict]: | |
| by_name = {t.name.lower(): t.id for t in themes} | |
| out = [] | |
| for r in raw or []: | |
| a, b = by_name.get(str(r.get("a", "")).strip().lower()), by_name.get(str(r.get("b", "")).strip().lower()) | |
| relation = (r.get("relation") or "").strip() | |
| if a and b and a != b and relation: | |
| out.append({"a": a, "b": b, "relation": relation[:200]}) | |
| return out[:8] | |
| def group_themes(db: Session, job: Job | None, workspace_id: uuid.UUID, title: str, request: str, | |
| ideas: list[Idea]) -> tuple[list[Theme], list[dict]]: | |
| """Themes with their weights, and how they relate (by theme id). Raises ThemesUnavailable when the answer is not usable.""" | |
| by_id = {i.id: i for i in ideas} | |
| listed = listed_ideas(ideas) | |
| listing = [f"{i.id} | {i.label}: {i.note[:140]}" for i in listed] | |
| log.info("report themes: asking the model workspace=%s ideas=%d listed=%d (cap %d) posts=%d", workspace_id, len(ideas), | |
| len(listed), MAX_LISTED_IDEAS, len({i.doc.id for i in ideas})) | |
| started = time.monotonic() | |
| out = run.predict(GroupThemes, db=db, workspace_id=workspace_id, job=job, lm=fast_lm(), title=title, | |
| request=request, ideas=listing) | |
| themes: list[Theme] = [] | |
| taken: set[str] = set() | |
| dropped = 0 | |
| for t in out.get("themes") or []: | |
| name = (t.get("name") or "").strip() | |
| wanted = list(t.get("idea_ids") or []) | |
| mine = [by_id[i] for i in wanted if i in by_id and i not in taken] | |
| dropped += len(wanted) - len(mine) # unknown ids, or ideas another theme already took | |
| if not name or not mine: | |
| log.info("report theme skipped: name=%r usable_ideas=%d", name, len(mine)) | |
| continue | |
| taken.update(i.id for i in mine) | |
| themes.append(Theme(id="", name=name[:80], summary=(t.get("summary") or "").strip()[:400], ideas=mine, | |
| asked=bool(t.get("matches_request")))) | |
| if len(themes) == MAX_THEMES: | |
| break | |
| log.info("report themes: model answered themes=%d usable=%d ids_dropped=%d seconds=%.1f", len(out.get("themes") or []), | |
| len(themes), dropped, time.monotonic() - started) | |
| if len(themes) < MIN_THEMES: | |
| log.warning("report themes unusable: usable=%d needed=%d", len(themes), MIN_THEMES) | |
| raise ThemesUnavailable("the ideas did not group into themes") | |
| _file_leftovers(themes, [i for i in ideas if i.id not in taken]) | |
| for n, t in enumerate(themes, start=1): | |
| t.id = f"t{n}" | |
| weigh(themes, len({i.doc.id for i in ideas}), len(ideas)) | |
| relations = _relations(out.get("relations"), themes) | |
| for t in themes: | |
| log.info("report theme: id=%s name=%r weight=%.3f tier=%s asked=%s pinned=%s posts=%d ideas=%d", t.id, t.name, t.weight, | |
| t.tier, t.asked, t.pinned, len(t.doc_ids), len(t.ideas)) | |
| log.info("report theme relations: model=%d kept=%d", len(out.get("relations") or []), len(relations)) | |
| return themes, relations | |
| def theme_summary(t: Theme) -> dict: | |
| """What is kept with the report about a theme.""" | |
| return {"id": t.id, "name": t.name, "summary": t.summary, "weight": t.weight, "tier": t.tier, "pinned": t.pinned, | |
| "posts": len(t.doc_ids), "ideas": len(t.ideas)} | |
| def progress(db: Session, job: Job | None, value: float, message: str) -> None: | |
| if job: | |
| update_job(db, job, progress=value, message=message) | |