CLI & debug output

varagity.cli.app

CLI argument parsing, subcommand dispatch, and the terminal Q&A loop.

Subcommands: ingest, chat (the default when no subcommand is given), and eval / eval ocr / eval chat / eval graph. chat follows the spec §13 startup sequence: ingest the corpus first, then loop — prompt → retrieve → show matches → grounded answer — until :quit (or end-of-input, e.g. in a non-interactive container).

Every subcommand runs through the Prefect flows (varagity.pipeline), invoked directly in-process — no worker or deployment (spec §21 #8) — so every ingest stage and every question is a tracked run at the Prefect UI (:4200).

An ingest refused by the corpus lock — a backup restore holds the corpus (ADR-019) — is not a crash: ingest and chat say so in one line and exit 1. They ask the lock before the tracked flow starts, so no failed flow run's traceback lands above that line.

build_parser()

Build the top-level argument parser.

Returns:

Type	Description
ArgumentParser	A parser accepting --verbose/-v (before or after the subcommand)
ArgumentParser	and the registered subcommands; with no subcommand the app runs
ArgumentParser	chat.

Source code in varagity/cli/app.py

def build_parser() -> argparse.ArgumentParser:
    """Build the top-level argument parser.

    Returns:
        A parser accepting ``--verbose/-v`` (before or after the subcommand)
        and the registered subcommands; with no subcommand the app runs
        ``chat``.
    """
    parser = argparse.ArgumentParser(
        prog="varagity",
        description="Varagity — terminal RAG with Contextual Retrieval. "
        "Runs `chat` when no command is given.",
    )
    _add_verbose_option(parser, default=None)
    subparsers = parser.add_subparsers(dest="command", metavar="COMMAND")
    ingest = subparsers.add_parser(
        "ingest",
        help="scan DOCS_PATH and ingest the corpus into both stores (pgvector + BM25)",
        description="Parse, chunk, contextualize, embed, and store every supported document "
        "under DOCS_PATH.",
    )
    _add_verbose_option(ingest, default=argparse.SUPPRESS)
    ingest.add_argument(
        "--reingest",
        action="store_true",
        help="delete and re-process every discovered document. Needed after pipeline-setting "
        "changes (CONTEXTUALIZE, chunk params): those don't change content hashes, so "
        "unchanged files are otherwise skipped",
    )
    chat = subparsers.add_parser(
        "chat",
        help="ingest the corpus, then answer questions from the terminal (the default)",
        description="Ingest DOCS_PATH, then loop: retrieve the top-k chunks per question "
        f"and generate a grounded, cited answer. Type {QUIT_COMMAND} to exit.",
    )
    _add_verbose_option(chat, default=argparse.SUPPRESS)
    evaluate = subparsers.add_parser(
        "eval",
        help="measure retrieval quality (recall@k/pass@k, 7-config matrix) on ephemeral stores",
        description="Run the spec §16 evaluation harness against ephemeral testcontainers "
        "stores (Docker required) and the live GPU services. Without a target, runs the "
        "7-configuration retrieval matrix; `eval ocr` benchmarks the OCR engines; "
        "`eval chat` compares the chat engines on multi-turn conversations; "
        "`eval graph` runs the GraphRAG bake-off (no stores — engines self-store).",
    )
    _add_verbose_option(evaluate, default=argparse.SUPPRESS)
    eval_targets = evaluate.add_subparsers(dest="eval_command", metavar="TARGET")
    eval_ocr = eval_targets.add_parser(
        "ocr",
        help="benchmark the OCR engines (CER/WER, pages/sec, retrieval recall)",
        description="Parse the scanned fixture PDFs with every OCR engine, score them "
        "against ground truth, and measure the retrieval impact per engine.",
    )
    _add_verbose_option(eval_ocr, default=argparse.SUPPRESS)
    eval_chat = eval_targets.add_parser(
        "chat",
        help="compare the chat engines on multi-turn conversations (per-turn retrieval)",
        description="Play every conversation fixture under each registered chat engine and "
        "score per-turn retrieval against fact-anchored golden refs — the harness that "
        "decides the shipped CHAT_ENGINE default (spec_v3 §4.9).",
    )
    _add_verbose_option(eval_chat, default=argparse.SUPPRESS)
    eval_graph = eval_targets.add_parser(
        "graph",
        help="score the GraphRAG engine over the synthetic message corpus (ADR-017)",
        description="Build the synthetic iMessage fixture corpus, index it with every "
        "registered graph engine, and score their answers by fact and provenance recall "
        "(spec_graphrag §12). Needs the live GPU services — but no Docker stores: graph "
        "engines self-store.",
    )
    _add_verbose_option(eval_graph, default=argparse.SUPPRESS)
    eval_graph.add_argument(
        "--profile",
        choices=("smoke", "full"),
        default="smoke",
        help="corpus size: smoke (~226 scripted messages, minutes per engine) or "
        "full (>10,000 messages across a decade — hours per engine)",
    )
    eval_graph.add_argument(
        "--engine",
        action="append",
        metavar="NAME",
        help="run only this engine (repeatable); defaults to every registered engine. "
        "One engine per session is how the full profile is scheduled",
    )
    eval_graph.add_argument(
        "--mode",
        metavar="NAME",
        help="engine query mode for every question, overriding each adapter's primary "
        "mode (e.g. lightrag's `global` on aggregation questions)",
    )
    eval_graph.add_argument(
        "--skip-build",
        action="store_true",
        help="reuse the existing corpus and each engine's already-built working directory "
        "(and skip the incremental check): re-scores in seconds without re-paying an index",
    )
    eval_graph.add_argument(
        "--message-target",
        type=int,
        metavar="N",
        help="cap the corpus at N messages instead of the profile's size — how an engine "
        "too slow for the uncapped corpus still gets a full-profile seat. The scripted "
        "messages are always present (filler tops up to N), and a capped run uses its own "
        "corpus and working directories (`full-mt1000`), never the uncapped run's",
    )
    return parser

run(argv=None)

Run the varagity command-line interface.

Parameters:

Name	Type	Description	Default
argv	list[str] | None	Argument vector; defaults to sys.argv[1:].	None

Returns:

Type	Description
int	The process exit code (ingest returns 1 if any file failed).

Source code in varagity/cli/app.py

def run(argv: list[str] | None = None) -> int:
    """Run the varagity command-line interface.

    Args:
        argv: Argument vector; defaults to ``sys.argv[1:]``.

    Returns:
        The process exit code (``ingest`` returns 1 if any file failed).
    """
    parser = build_parser()
    args = parser.parse_args(argv)

    settings = get_settings()
    setup_logging(settings.LOG_LEVEL)
    verbose = settings.DEFAULT_VERBOSE if args.verbose is None else args.verbose

    if args.command == "ingest":
        return _run_ingest(verbose, reingest=args.reingest)
    if args.command == "eval":
        if getattr(args, "eval_command", None) == "ocr":
            return _run_eval_ocr(verbose)
        if getattr(args, "eval_command", None) == "chat":
            return _run_eval_chat(verbose)
        if getattr(args, "eval_command", None) == "graph":
            return _run_eval_graph(
                verbose,
                profile=args.profile,
                engines=args.engine,
                mode=args.mode,
                skip_build=args.skip_build,
                message_target=args.message_target,
            )
        return _run_eval(verbose)
    # chat is the default subcommand (spec §13).
    return _run_chat(verbose)

varagity.debug.show

Rich-rendered console helpers backing the verbose= parameter convention.

Varagity separates three output channels (spec §14); this module implements the first — human-facing console output — with three levels:

• 0 — off: render nothing.
• 1 — low: names and counts.
• 2 — high: full metadata, rich panels.

Conventions:

• Every public function in the codebase accepts verbose: int = settings.DEFAULT_VERBOSE and raises :class:ValueError on invalid levels (enforced via :func:check_verbose).
• All rendering lives here as v_<function_name>(...) helpers (e.g. v_discover, v_chunk, v_retrieve), keeping presentation out of business logic. Helpers render nothing at level 0.

Concrete v_<name> helpers land alongside the features they render.

check_verbose(verbose)

Validate a verbose level.

Called at the top of every function that accepts a verbose parameter, so an invalid level fails fast instead of silently rendering nothing.

Parameters:

Name	Type	Description	Default
verbose	int	Requested verbosity; must be 0 (off), 1 (low), or 2 (high).	required

Returns:

Type	Description
int	The validated level, unchanged.

Raises:

Type	Description
ValueError	If verbose is not one of :data:VERBOSE_LEVELS.

Source code in varagity/debug/show.py

def check_verbose(verbose: int) -> int:
    """Validate a ``verbose`` level.

    Called at the top of every function that accepts a ``verbose`` parameter,
    so an invalid level fails fast instead of silently rendering nothing.

    Args:
        verbose: Requested verbosity; must be 0 (off), 1 (low), or 2 (high).

    Returns:
        The validated level, unchanged.

    Raises:
        ValueError: If ``verbose`` is not one of :data:`VERBOSE_LEVELS`.
    """
    if verbose not in VERBOSE_LEVELS:
        raise ValueError(
            f"verbose must be one of {VERBOSE_LEVELS} (0=off, 1=low, 2=high); got {verbose!r}"
        )
    return verbose

trace_badges(trace)

Format a retrieval trace as one-line rank badges (spec_v2 §4.6).

The terminal counterpart of the provenance panel's RankBadges: sem #1 · bm25 #3 · fused 0.94 · rerank +2. An arm that never surfaced the chunk shows —; the rerank badge appears only when the rerank stage ran.

Parameters:

Name	Type	Description	Default
trace	RetrievalTrace	The chunk's rank provenance.	required

Returns:

Type	Description
str	The badge string.

Source code in varagity/debug/show.py

def trace_badges(trace: "RetrievalTrace") -> str:
    """Format a retrieval trace as one-line rank badges (spec_v2 §4.6).

    The terminal counterpart of the provenance panel's ``RankBadges``:
    ``sem #1 · bm25 #3 · fused 0.94 · rerank +2``. An arm that never
    surfaced the chunk shows ``—``; the rerank badge appears only when the
    rerank stage ran.

    Args:
        trace: The chunk's rank provenance.

    Returns:
        The badge string.
    """
    semantic = f"sem #{trace.semantic_rank}" if trace.semantic_rank is not None else "sem —"
    bm25 = f"bm25 #{trace.bm25_rank}" if trace.bm25_rank is not None else "bm25 —"
    parts = [semantic, bm25, f"fused {trace.fused_score:.2f}"]
    if trace.rerank_delta is not None:
        parts.append(f"rerank {trace.rerank_delta:+d}")
    return " · ".join(parts)

v_chunk(chunks, verbose)

Render chunking results (for a chunking strategy's split).

Parameters:

Name	Type	Description	Default
chunks	Sequence[Document]	The chunks produced for one document (langchain Document objects with seeded metadata).	required
verbose	int	0 = nothing; 1 = file → chunk count; 2 = a panel per chunk with its full metadata.	required

Raises:

Type	Description
ValueError	If verbose is invalid.

Source code in varagity/debug/show.py

def v_chunk(chunks: "Sequence[Document]", verbose: int) -> None:
    """Render chunking results (for a chunking strategy's ``split``).

    Args:
        chunks: The chunks produced for one document (langchain
            ``Document`` objects with seeded metadata).
        verbose: 0 = nothing; 1 = file → chunk count; 2 = a panel per chunk
            with its full metadata.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    check_verbose(verbose)
    if verbose == 0 or not chunks:
        return
    file_name = chunks[0].metadata.get("file_name", "<unknown>")
    console.print(f"[bold]{file_name}[/] → {len(chunks)} chunk(s)")
    if verbose == 2:
        for chunk in chunks:
            meta = ", ".join(f"{key}={value}" for key, value in sorted(chunk.metadata.items()))
            console.print(
                Panel(
                    Text(chunk.page_content),
                    title=f"chunk {chunk.metadata.get('chunk_index', '?')}",
                    subtitle=meta,
                    subtitle_align="left",
                )
            )

v_condensed(original, condensed, verbose)

Render the condensed search query (for CondenseContextEngine.prepare).

The terminal counterpart of the evidence panel's "Searched for: …" line (spec_v3 §4.7): whenever retrieval ran on something other than the user's words, the rewrite is shown rather than silently applied.

Parameters:

Name	Type	Description	Default
original	str	The user's question, verbatim (shown at level 2).	required
condensed	str	The standalone search query the condenser produced.	required
verbose	int	0 = nothing; 1 = the condensed query; 2 = a panel pairing it with the original wording it replaced.	required

Raises:

Type	Description
ValueError	If verbose is invalid.

Source code in varagity/debug/show.py

def v_condensed(original: str, condensed: str, verbose: int) -> None:
    """Render the condensed search query (for ``CondenseContextEngine.prepare``).

    The terminal counterpart of the evidence panel's "Searched for: …" line
    (spec_v3 §4.7): whenever retrieval ran on something other than the
    user's words, the rewrite is shown rather than silently applied.

    Args:
        original: The user's question, verbatim (shown at level 2).
        condensed: The standalone search query the condenser produced.
        verbose: 0 = nothing; 1 = the condensed query; 2 = a panel pairing
            it with the original wording it replaced.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    check_verbose(verbose)
    if verbose == 0:
        return
    if verbose == 1:
        console.print(Text.assemble(("Searching with ", "bold"), (condensed, "italic")))
        return
    console.print(
        Panel(
            Text(condensed, style="italic"),
            title="condensed search query",
            subtitle=f"asked: {original}",
            subtitle_align="left",
            style="dim",
        )
    )

v_discover(buckets, verbose)

Render discovery results (for :func:~varagity.ingest.discovery.discover_documents).

Parameters:

Name	Type	Description	Default
buckets	Buckets	The discovered corpus buckets.	required
verbose	int	0 = nothing; 1 = counts per bucket; 2 = also the file list.	required

Raises:

Type	Description
ValueError	If verbose is invalid.

Source code in varagity/debug/show.py

def v_discover(buckets: "Buckets", verbose: int) -> None:
    """Render discovery results (for :func:`~varagity.ingest.discovery.discover_documents`).

    Args:
        buckets: The discovered corpus buckets.
        verbose: 0 = nothing; 1 = counts per bucket; 2 = also the file list.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    check_verbose(verbose)
    if verbose == 0:
        return
    counts = ", ".join(
        f"{len(paths)} {name.replace('_', '-')}" for name, paths in buckets.by_bucket()
    )
    console.print(f"[bold]Discovered[/] {buckets.total} document(s) ({counts})")
    if verbose == 2:
        for bucket_name, paths in buckets.by_bucket():
            for path in paths:
                console.print(f"  [dim]{bucket_name}[/dim] {path}")

v_hyde(query, passage, verbose)

Render the hypothetical passage (for HydeRetriever, ADR-016).

The :func:v_condensed counterpart for the HyDE stage: whenever the dense arm is about to search with generated text rather than the user's words, the substitution is shown rather than silently applied.

Parameters:

Name	Type	Description	Default
query	str	The search query the passage was generated for (shown at level 2).	required
passage	str	The hypothetical answer passage steering the dense arm.	required
verbose	int	0 = nothing; 1 = a one-line notice with the passage's head; 2 = a panel pairing the full passage with the query.	required

Raises:

Type	Description
ValueError	If verbose is invalid.

Source code in varagity/debug/show.py

def v_hyde(query: str, passage: str, verbose: int) -> None:
    """Render the hypothetical passage (for ``HydeRetriever``, ADR-016).

    The :func:`v_condensed` counterpart for the HyDE stage: whenever the
    dense arm is about to search with generated text rather than the
    user's words, the substitution is shown rather than silently applied.

    Args:
        query: The search query the passage was generated for (shown at
            level 2).
        passage: The hypothetical answer passage steering the dense arm.
        verbose: 0 = nothing; 1 = a one-line notice with the passage's
            head; 2 = a panel pairing the full passage with the query.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    check_verbose(verbose)
    if verbose == 0:
        return
    if verbose == 1:
        head = passage if len(passage) <= 80 else passage[:79] + "…"
        notice = ("Dense arm searching with hypothetical passage ", "bold")
        console.print(Text.assemble(notice, (head, "italic")))
        return
    console.print(
        Panel(
            Text(passage, style="italic"),
            title="HyDE hypothetical passage",
            subtitle=f"query: {query}",
            subtitle_align="left",
            style="dim",
        )
    )

v_retrieve(chunks, verbose)

Render retrieval results (for a retriever's retrieve).

Mirrors the reference's v_retrieve_docs: at level 2 each retrieved chunk becomes a panel with its score, source, content, (when the chunk was ingested with CONTEXTUALIZE on) its situating context, and (when the retriever attached one) its :func:trace_badges rank provenance.

Parameters:

Name	Type	Description	Default
chunks	Sequence[RetrievedChunk]	The retrieved chunks, best first.	required
verbose	int	0 = nothing; 1 = retrieved count; 2 = a panel per chunk (score/source/content/context/trace).	required

Raises:

Type	Description
ValueError	If verbose is invalid.

Source code in varagity/debug/show.py

def v_retrieve(chunks: "Sequence[RetrievedChunk]", verbose: int) -> None:
    """Render retrieval results (for a retriever's ``retrieve``).

    Mirrors the reference's ``v_retrieve_docs``: at level 2 each retrieved
    chunk becomes a panel with its score, source, content, (when the chunk
    was ingested with ``CONTEXTUALIZE`` on) its situating context, and
    (when the retriever attached one) its :func:`trace_badges` rank
    provenance.

    Args:
        chunks: The retrieved chunks, best first.
        verbose: 0 = nothing; 1 = retrieved count; 2 = a panel per chunk
            (score/source/content/context/trace).

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    check_verbose(verbose)
    if verbose == 0:
        return
    console.print(f"[bold]Retrieved[/] {len(chunks)} chunk(s)")
    if verbose == 2:
        for rank, chunk in enumerate(chunks, start=1):
            source = str(chunk.metadata.get("source", "<unknown>"))
            page = chunk.metadata.get("page")
            renderables: list[RenderableType] = []
            if chunk.trace is not None:
                renderables.append(Text(trace_badges(chunk.trace), style="bold dim"))
            if chunk.context:
                renderables.append(Panel(Text(chunk.context), title="context", style="dim"))
            renderables.append(Text(chunk.content))
            body: RenderableType = renderables[0] if len(renderables) == 1 else Group(*renderables)
            console.print(
                Panel(
                    body,
                    title=f"match {rank} · score {chunk.score:.4f} · {chunk.chunk_id}",
                    subtitle=source if page is None else f"{source} · page {page}",
                    subtitle_align="left",
                )
            )

v_situate_context(chunk_text, context, verbose)

Render one situating blurb (for :func:~varagity.context.contextual.situate_context).

Level 1 stays quiet — the loader's contextualization sub-progress bar is the per-chunk signal there; level 2 shows each blurb next to a snippet of the chunk it situates.

Parameters:

Name	Type	Description	Default
chunk_text	str	The chunk being situated (snippet shown as the subtitle).	required
context	str	The LLM-generated situating blurb.	required
verbose	int	0/1 = nothing; 2 = a panel per blurb.	required

Raises:

Type	Description
ValueError	If verbose is invalid.

Source code in varagity/debug/show.py

def v_situate_context(chunk_text: str, context: str, verbose: int) -> None:
    """Render one situating blurb (for :func:`~varagity.context.contextual.situate_context`).

    Level 1 stays quiet — the loader's contextualization sub-progress bar is
    the per-chunk signal there; level 2 shows each blurb next to a snippet of
    the chunk it situates.

    Args:
        chunk_text: The chunk being situated (snippet shown as the subtitle).
        context: The LLM-generated situating blurb.
        verbose: 0/1 = nothing; 2 = a panel per blurb.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    check_verbose(verbose)
    if verbose < 2:
        return
    snippet = " ".join(chunk_text.split())
    if len(snippet) > 70:
        snippet = snippet[:69] + "…"
    console.print(
        Panel(
            Text(context) if context else Text("<empty blurb>", style="italic red"),
            title="context blurb",
            subtitle=f"chunk: {snippet}",
            subtitle_align="left",
            style="dim",
        )
    )