linexcel.aidoc

linexcel.aidoc

AI-generated documentation for Excel calculations.

Vendor-neutral by construction: no provider is named in the code and none is chosen for you. There are exactly two ways in:

• base_url= — any OpenAI-compatible endpoint (a local Ollama, vLLM or LM Studio runtime; a hosted gateway such as OpenRouter; OpenAI itself; anything else that speaks /chat/completions)
• provider= — your own callable or :class:LLMProvider object, for an API that speaks something else entirely

The model doesn't guess: each node is presented with its deterministic dossier from the graph (exact formula, step-by-step evaluation, precedents and their values, dependents, stretched group extent, VBA links). The system prompt enforces citing only these facts, making the documentation "provable": every claim traces back to a formula or a workbook value.

TokenUsage dataclass

Tokens consumed by one or more documentation requests.

estimated is True as soon as any request in the tally had to be approximated by :func:estimate_tokens instead of being reported by the provider — treat such a total as an order of magnitude, not a bill.

Source code in src/linexcel/aidoc.py

@dataclass
class TokenUsage:
    """Tokens consumed by one or more documentation requests.

    ``estimated`` is ``True`` as soon as any request in the tally had to be
    approximated by :func:`estimate_tokens` instead of being reported by the
    provider — treat such a total as an order of magnitude, not a bill.
    """

    input_tokens: int = 0
    output_tokens: int = 0
    requests: int = 0
    estimated: bool = False
    model: str = ""
    provider: str = ""

    @property
    def total(self) -> int:
        return self.input_tokens + self.output_tokens

    def add(self, other: TokenUsage) -> None:
        """Accumulate ``other`` in place, keeping the model/provider labels."""
        self.input_tokens += other.input_tokens
        self.output_tokens += other.output_tokens
        self.requests += other.requests
        self.estimated = self.estimated or other.estimated
        self.model = self.model or other.model
        self.provider = self.provider or other.provider

    def __str__(self) -> str:
        about = "~" if self.estimated else ""
        where = f" [{self.provider}/{self.model}]" if self.provider else ""
        return (
            f"{about}{self.total:,} tokens "
            f"({about}{self.input_tokens:,} in + {about}{self.output_tokens:,} out) "
            f"over {self.requests} request(s){where}"
        )

add

add(other: TokenUsage) -> None

Accumulate other in place, keeping the model/provider labels.

Source code in src/linexcel/aidoc.py

def add(self, other: TokenUsage) -> None:
    """Accumulate ``other`` in place, keeping the model/provider labels."""
    self.input_tokens += other.input_tokens
    self.output_tokens += other.output_tokens
    self.requests += other.requests
    self.estimated = self.estimated or other.estimated
    self.model = self.model or other.model
    self.provider = self.provider or other.provider

LLMProvider

Bases: Protocol

Minimal protocol: system + user prompt → text response.

Source code in src/linexcel/aidoc.py

@runtime_checkable
class LLMProvider(Protocol):
    """Minimal protocol: system + user prompt → text response."""

    def generate(
        self,
        system_prompt: str,
        user_prompt: str,
        *,
        temperature: float = 0.2,
        max_tokens: int | None = None,
    ) -> str: ...

UsageReportingProvider

Bases: Protocol

A provider that also reports what the call consumed.

Optional: the built-in OpenAI-compatible client implements it so that token counts come from the API rather than from an approximation. Custom providers only need :class:LLMProvider.

Source code in src/linexcel/aidoc.py

@runtime_checkable
class UsageReportingProvider(Protocol):
    """A provider that also reports what the call consumed.

    Optional: the built-in OpenAI-compatible client implements it so that token
    counts come from the API rather than from an approximation. Custom
    providers only need :class:`LLMProvider`.
    """

    def generate_with_usage(
        self,
        system_prompt: str,
        user_prompt: str,
        *,
        temperature: float = 0.2,
        max_tokens: int | None = None,
    ) -> tuple[str, TokenUsage]: ...

estimate_tokens

estimate_tokens(text: str) -> int

Approximate the token count of text.

Only a fallback: :class:TokenUsage prefers the counts the provider reports. Latin script is counted as words × 4/3 (the usual 1 token ≈ 0.75 words ratio); CJK characters are counted individually, because a Japanese or Chinese sentence carries no spaces and would otherwise register as a single word.

estimate_tokens("the quick brown fox jumps") 6 estimate_tokens("") 0

Source code in src/linexcel/aidoc.py

def estimate_tokens(text: str) -> int:
    """Approximate the token count of ``text``.

    Only a fallback: :class:`TokenUsage` prefers the counts the provider
    reports. Latin script is counted as words × 4/3 (the usual 1 token ≈ 0.75
    words ratio); CJK characters are counted individually, because a Japanese
    or Chinese sentence carries no spaces and would otherwise register as a
    single word.

    >>> estimate_tokens("the quick brown fox jumps")
    6
    >>> estimate_tokens("")
    0
    """
    if not text:
        return 0
    cjk = len(_CJK_RE.findall(text))
    latin_words = len(_WORD_RE.findall(_CJK_RE.sub(" ", text)))
    return cjk + latin_words * 4 // 3

build_dossier

build_dossier(graph: dict[str, Any], node_id: str) -> dict[str, Any] | None

Deterministic dossier for a node: everything the AI is allowed to use.

Source code in src/linexcel/aidoc.py

def build_dossier(graph: dict[str, Any], node_id: str) -> dict[str, Any] | None:
    """
    Deterministic dossier for a node: everything the AI is allowed to use.
    """
    nodes = {n["id"]: n for n in graph["nodes"]}
    node = nodes.get(node_id)
    if node is None:
        return None
    precedents, dependents = [], []
    for e in graph["edges"]:
        if e["target"] == node_id:
            src = nodes.get(e["source"], {})
            precedents.append(_neighbor(src, e))
        elif e["source"] == node_id:
            dst = nodes.get(e["target"], {})
            dependents.append(_neighbor(dst, e))
    dossier = {
        "node_id": node_id,
        "kind": node.get("kind"),
        "sheet": node.get("sheet"),
        "address": node.get("addr"),
        "formula": node.get("formula"),
        "r1c1_form": node.get("r1c1"),
        "group_cells": node.get("count"),
        "extent": node.get("bbox"),
        "computed_value": node.get("value"),
        "value_samples": node.get("samples"),
        "decomposition": _compact_steps(node.get("steps")),
        "precedents": precedents[:30],
        "dependents": dependents[:30],
    }
    if node.get("kind") == "vba":
        dossier["vba"] = {
            "module": node.get("module"),
            "procedure": node.get("proc"),
            "type": node.get("procKind"),
            "code": (node.get("code") or "")[:2500],
        }
    return dossier

build_workbook_dossier

build_workbook_dossier(graph: dict[str, Any], *, context: dict[str, Any] | None = None) -> dict[str, Any]

Return a compact, deterministic dossier for a whole-workbook overview.

context is a :attr:linexcel.LineageResult.workbook_context mapping. The graph alone describes how a workbook computes; it says nothing about what a reader sees on opening it — titles sitting above a table, the labels in the first column, cell comments, hidden columns, frozen panes. Those cues are exactly what the sheet screenshots show, and merging them into each sheet entry lets a text-only model describe the file as it looks without any image ever leaving the machine.

Both parts stay deterministic: every value is read from the workbook, so the "cite only the dossier" rule of the system prompt still holds.

Source code in src/linexcel/aidoc.py

def build_workbook_dossier(
    graph: dict[str, Any], *, context: dict[str, Any] | None = None
) -> dict[str, Any]:
    """Return a compact, deterministic dossier for a whole-workbook overview.

    ``context`` is a :attr:`linexcel.LineageResult.workbook_context` mapping.
    The graph alone describes how a workbook *computes*; it says nothing about
    what a reader sees on opening it — titles sitting above a table, the labels
    in the first column, cell comments, hidden columns, frozen panes. Those cues
    are exactly what the sheet screenshots show, and merging them into each
    sheet entry lets a text-only model describe the file as it looks without any
    image ever leaving the machine.

    Both parts stay deterministic: every value is read from the workbook, so the
    "cite only the dossier" rule of the system prompt still holds.
    """
    nodes = graph.get("nodes", [])
    meta = graph.get("meta", {})
    stats = meta.get("stats", {})
    sheet_stats = stats.get("sheets", [])
    nodes_by_sheet: dict[str, dict[str, int]] = {}
    for node in nodes:
        sheet = node.get("sheet")
        if not sheet:
            continue
        kinds = nodes_by_sheet.setdefault(sheet, {})
        kind = node.get("kind", "unknown")
        kinds[kind] = kinds.get(kind, 0) + 1

    sheets = [
        {
            "name": sheet.get("name"),
            "dimensions": {"rows": sheet.get("rows"), "columns": sheet.get("cols")},
            "formula_cells": sheet.get("formulaCells", 0),
            "lineage_nodes": nodes_by_sheet.get(sheet.get("name"), {}),
        }
        for sheet in sheet_stats
    ]
    if context:
        sheets = _merge_presentation(sheets, context)
    formula_patterns = sorted(
        (
            {
                "sheet": node.get("sheet"),
                "address": node.get("addr"),
                "formula": node.get("formula"),
                "cells": node.get("count", 1),
                "extent": node.get("bbox"),
            }
            for node in nodes
            if node.get("kind") in {"cell", "group"}
        ),
        key=lambda item: item["cells"],
        reverse=True,
    )[:20]
    defined_names = [
        {"name": node.get("label"), "targets": node.get("targets", [])}
        for node in nodes
        if node.get("kind") == "name"
    ]
    vba = [
        {
            "module": node.get("module"),
            "procedure": node.get("proc"),
            "type": node.get("procKind"),
        }
        for node in nodes
        if node.get("kind") == "vba"
    ]
    opaque_references = [
        node.get("label") for node in nodes if node.get("kind") == "opaque"
    ]
    return {
        "filename": meta.get("filename"),
        "analysis": {
            "formula_cells": stats.get("totalFormulas", 0),
            "lineage_nodes": stats.get("totalNodes", 0),
            "lineage_edges": stats.get("totalEdges", 0),
            "grouped_patterns": stats.get("groupedPatterns", 0),
        },
        "sheets": sheets,
        "formula_patterns": formula_patterns,
        "defined_names": defined_names,
        "vba_procedures": vba,
        "external_or_unresolved_references": opaque_references,
        "warnings": meta.get("warnings", []),
    }

document_workbook

document_workbook(graph: dict[str, Any], *, model: str | None = None, api_key: str | None = None, base_url: str | None = None, provider: ProviderLike | None = None, language: str = 'en', usage: TokenUsage | None = None, max_tokens: int | None = None, token_budget: int | None = None, context: dict[str, Any] | None = None) -> str

Generate a Markdown overview grounded in the workbook dossier.

Provider resolution (first match wins; no implicit default): 1. provider — custom LLMProvider instance or callable 2. base_url + model (or LINEXCEL_AI_BASE_URL + LINEXCEL_AI_MODEL) — any OpenAI-compatible endpoint

context is the workbook presentation context — the sheet previews, comments, merged cells, frozen panes and hidden columns a reader sees when opening the file. Pass it to describe the workbook as it looks, not only as it computes; see :func:build_workbook_dossier.

If a :class:TokenUsage is passed as usage, what the call consumed is accumulated into it. token_budget caps cumulative spend across that accumulator: an already-exhausted budget raises before anything is sent.

Source code in src/linexcel/aidoc.py

def document_workbook(
    graph: dict[str, Any],
    *,
    model: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    provider: ProviderLike | None = None,
    language: str = "en",
    usage: TokenUsage | None = None,
    max_tokens: int | None = None,
    token_budget: int | None = None,
    context: dict[str, Any] | None = None,
) -> str:
    """Generate a Markdown overview grounded in the workbook dossier.

    Provider resolution (first match wins; no implicit default):
    1. `provider` — custom LLMProvider instance or callable
    2. `base_url` + `model` (or `LINEXCEL_AI_BASE_URL` + `LINEXCEL_AI_MODEL`) —
       any OpenAI-compatible endpoint

    ``context`` is the workbook presentation context — the sheet previews,
    comments, merged cells, frozen panes and hidden columns a reader sees when
    opening the file. Pass it to describe the workbook as it looks, not only as
    it computes; see :func:`build_workbook_dossier`.

    If a :class:`TokenUsage` is passed as ``usage``, what the call consumed is
    accumulated into it. ``token_budget`` caps cumulative spend across that
    accumulator: an already-exhausted budget raises before anything is sent.
    """
    if language not in _LANGUAGES:
        raise ValueError(f"Unsupported language: {language!r}. Use one of {_LANGUAGES}")
    _check_budget(token_budget, usage)
    dossier = build_workbook_dossier(graph, context=context)
    blob = _fit_workbook_dossier(dossier)
    llm = _resolve_provider(
        provider=provider, model=model, api_key=api_key, base_url=base_url
    )
    system = _WORKBOOK_SYSTEM[language]
    user = "Workbook dossier (deterministic, extracted from workbook):\n" + blob
    try:
        text, call_usage = _generate(llm, system, user, max_tokens=max_tokens)
    except AiDocError:
        raise
    except Exception as exc:
        raise AiDocError(f"AI documentation failed: {exc}") from exc
    if usage is not None:
        usage.add(call_usage)
    return text

document_nodes

document_nodes(graph: dict[str, Any], node_ids: list[str], *, model: str | None = None, api_key: str | None = None, base_url: str | None = None, provider: ProviderLike | None = None, language: str = 'en', max_workers: int = 4, usage: TokenUsage | None = None, max_tokens: int | None = None, token_budget: int | None = None) -> dict[str, str]

Document the requested nodes, returns {node_id: markdown}.

Provider resolution is the same as :func:document_workbook (no implicit default; see :func:_resolve_provider).

Nodes are documented concurrently (max_workers in-flight requests; raise it if the provider's rate limits allow). Documenting a large workbook is a long, often billed operation, so a node that fails does not discard the ones that succeeded: the successful cards are returned and a :class:UserWarning reports how many nodes were dropped. :class:AiDocError is raised only when every node failed.

If a :class:TokenUsage is passed as usage, every successful call is accumulated into it — including those of a run that later fails, since tokens already spent are still billed.

token_budget is a ceiling on the total tokens the run may spend, input and output together, counted against usage so several calls sharing one accumulator share one ceiling. It is enforced between requests, the only point at which a cost is known: nodes still queued when the tally reaches the budget are never sent, and a :class:UserWarning reports how many were left undocumented. Requests already in flight are allowed to finish, so the final tally can exceed the budget by up to max_workers responses — set it as an order of magnitude, not to the token. Use max_tokens to bound each individual response instead.

Source code in src/linexcel/aidoc.py

def document_nodes(
    graph: dict[str, Any],
    node_ids: list[str],
    *,
    model: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    provider: ProviderLike | None = None,
    language: str = "en",
    max_workers: int = 4,
    usage: TokenUsage | None = None,
    max_tokens: int | None = None,
    token_budget: int | None = None,
) -> dict[str, str]:
    """Document the requested nodes, returns {node_id: markdown}.

    Provider resolution is the same as :func:`document_workbook` (no implicit
    default; see :func:`_resolve_provider`).

    Nodes are documented concurrently (``max_workers`` in-flight requests;
    raise it if the provider's rate limits allow). Documenting a large
    workbook is a long, often billed operation, so a node that fails does not
    discard the ones that succeeded: the successful cards are returned and a
    :class:`UserWarning` reports how many nodes were dropped.
    :class:`AiDocError` is raised only when *every* node failed.

    If a :class:`TokenUsage` is passed as ``usage``, every successful call is
    accumulated into it — including those of a run that later fails, since
    tokens already spent are still billed.

    ``token_budget`` is a ceiling on the **total** tokens the run may spend,
    input and output together, counted against ``usage`` so several calls
    sharing one accumulator share one ceiling. It is enforced between requests,
    the only point at which a cost is known: nodes still queued when the tally
    reaches the budget are never sent, and a :class:`UserWarning` reports how
    many were left undocumented. Requests already in flight are allowed to
    finish, so the final tally can exceed the budget by up to ``max_workers``
    responses — set it as an order of magnitude, not to the token. Use
    ``max_tokens`` to bound each individual response instead.
    """
    if language not in _LANGUAGES:
        raise ValueError(f"Unsupported language: {language!r}. Use one of {_LANGUAGES}")
    if max_workers < 1:
        raise ValueError("max_workers must be >= 1")
    _check_budget(token_budget, usage)
    llm = _resolve_provider(
        provider=provider, model=model, api_key=api_key, base_url=base_url
    )
    system = _SYSTEM[language]
    docs: dict[str, str] = {}
    dossiers = []
    for nid in node_ids:
        d = build_dossier(graph, nid)
        if d is not None:
            blob = json.dumps(d, ensure_ascii=False, default=str)
            if len(blob) > MAX_DOSSIER_CHARS:
                d["decomposition"] = "truncated (very long formula)"
                blob = json.dumps(d, ensure_ascii=False, default=str)
            dossiers.append((nid, blob))
    if not dossiers:
        return docs

    def _doc_one(nid_blob: tuple[str, str]) -> tuple[str, str, TokenUsage]:
        nid, blob = nid_blob
        user = "Lineage dossier (deterministic, extracted from workbook):\n" + blob
        text, call_usage = _generate(llm, system, user, max_tokens=max_tokens)
        return nid, text or "(AI returned empty response)", call_usage

    # The tally drives the budget, so it must exist even when the caller wants
    # no accumulator of their own; when they do pass one, it *is* the tally.
    tally = usage if usage is not None else TokenUsage()
    failures: list[tuple[str, Exception]] = []
    queue = iter(dossiers)
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures: dict[Future[tuple[str, str, TokenUsage]], str] = {}

        def _submit_next() -> bool:
            """Queue one more node unless the budget is spent or none is left."""
            if token_budget is not None and tally.total >= token_budget:
                return False
            item = next(queue, None)
            if item is None:
                return False
            futures[pool.submit(_doc_one, item)] = item[0]
            return True

        # Nodes are submitted a few at a time rather than all at once: a budget
        # can only stop work that has not been handed to the pool yet.
        for _ in range(max_workers):
            if not _submit_next():
                break
        # Accumulating here rather than inside the workers keeps `tally` free of
        # races: this loop is the single consumer of the pool's results.
        while futures:
            done, _pending = wait(list(futures), return_when=FIRST_COMPLETED)
            for fut in done:
                node_id = futures.pop(fut)
                try:
                    nid, text, call_usage = fut.result()
                except Exception as exc:
                    failures.append((node_id, exc))
                    continue
                docs[nid] = text
                tally.add(call_usage)
            for _ in range(len(done)):
                if not _submit_next():
                    break

    # Only an exhausted budget can leave the queue undrained.
    unsent = sum(1 for _ in queue) if token_budget is not None else 0
    if unsent:
        warnings.warn(
            f"Token budget of {token_budget:,} tokens reached after "
            f"{tally.total:,}: {len(docs)} of {len(dossiers)} nodes documented, "
            f"{unsent} never sent. Raise token_budget to document the rest.",
            UserWarning,
            stacklevel=2,
        )

    if failures and not docs:
        node_id, exc = failures[0]
        raise AiDocError(
            f"AI documentation failed for all {len(failures)} nodes "
            f"(first error, node {node_id}): {exc}"
        ) from exc
    if failures:
        node_id, exc = failures[0]
        warnings.warn(
            f"AI documentation failed for {len(failures)} of {len(dossiers)} "
            f"nodes; {len(docs)} cards returned. First error (node {node_id}): "
            f"{exc}",
            UserWarning,
            stacklevel=2,
        )
    return docs