Blame for backend/app/pipeline/report_themes.py

"""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."""


@dataclass
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


@dataclass
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

    @property
    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)