Orchestration (Prefect)

varagity.pipeline

Prefect orchestration layer: thin @flow/@task adapters (spec §9–10).

Business logic lives in the plain modules (ingest.loader, retrieval, generation); this package wraps their stage functions as tracked, retryable Prefect task runs and exposes the two flows the CLI invokes directly — no worker or deployment needed (spec §21 #8).

Importing this package exports settings.PREFECT_API_URL into the process environment before prefect is imported: Prefect captures its environment at import time, so a later export would be silently ignored and runs would never reach the compose prefect service. In-container the variable is already set (env_file) and the export is a no-op; on the host it carries the .env value into Prefect. With no server reachable, Prefect 3 falls back to an ephemeral in-process API, so flows still run without the stack (untracked once the process exits).

GraphBatches

Opaque carrier keeping a parsed archive out of the flow-run record.

Prefect serializes every flow-call parameter into the flow run it creates (Flow.serialize_parameters), and the server rejects records over 512 KB — a decade-scale archive of parsed messages serializes to megabytes, so passing the batches bare fails the build at flow-run creation, before the engine sees a single document. This box is deliberately not JSON-encodable (__slots__, no mapping or iteration protocol), which lands it on the same "<GraphBatches>" placeholder path the live GraphService handle already rides; the real objects still reach the flow in-process, untouched.

Attributes:

Name	Type	Description
batches	Sequence[MessageBatch]	The parsed source files, exactly as given.

Source code in varagity/pipeline/graph_flow.py

class GraphBatches:
    """Opaque carrier keeping a parsed archive out of the flow-run record.

    Prefect serializes every flow-call parameter into the flow run it
    creates (``Flow.serialize_parameters``), and the server rejects records
    over 512 KB — a decade-scale archive of parsed messages serializes to
    megabytes, so passing the batches bare fails the build at flow-run
    creation, before the engine sees a single document. This box is
    deliberately *not* JSON-encodable (``__slots__``, no mapping or
    iteration protocol), which lands it on the same ``"<GraphBatches>"``
    placeholder path the live ``GraphService`` handle already rides; the
    real objects still reach the flow in-process, untouched.

    Attributes:
        batches: The parsed source files, exactly as given.
    """

    __slots__ = ("batches",)

    batches: Sequence[MessageBatch]

    def __init__(self, batches: Sequence[MessageBatch]) -> None:
        """Box the batches.

        Args:
            batches: Parsed source files bound for the build.
        """
        self.batches = batches

__init__(batches)

Box the batches.

Parameters:

Name	Type	Description	Default
batches	Sequence[MessageBatch]	Parsed source files bound for the build.	required

Source code in varagity/pipeline/graph_flow.py

def __init__(self, batches: Sequence[MessageBatch]) -> None:
    """Box the batches.

    Args:
        batches: Parsed source files bound for the build.
    """
    self.batches = batches

chat_eval_flow(verbose=None)

Run the multi-turn chat-engine eval as a tracked flow run.

Parameters:

Name	Type	Description	Default
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (see
dict[str, Any]	func:varagity.eval.evaluate.run_chat_eval).

Source code in varagity/pipeline/eval_flow.py

@flow(name="eval-chat", validate_parameters=False)
def chat_eval_flow(verbose: int | None = None) -> dict[str, Any]:
    """Run the multi-turn chat-engine eval as a tracked flow run.

    Args:
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (see
        :func:`varagity.eval.evaluate.run_chat_eval`).
    """
    return run_chat_eval(ingest=ingest_flow, verbose=verbose)

graph_build_flow(service, batches, *, prune_removed=True, verbose=0)

Upsert a parsed message corpus into the graph as a tracked flow run.

Parameters:

Name	Type	Description	Default
service	GraphService	The process's graph service (holds the session and the single-flight write lock).	required
batches	GraphBatches	The boxed parsed source files (bare sequences would serialize into the flow-run record, which the server caps at 512 KB), guid-merged by the session before rendering.	required
prune_removed	bool	Whether batches render the whole corpus. A bounded build passes False — its render is partial on purpose, and pruning on its say-so would delete the rest of the archive.	True
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
BuildReport	What the build did (messages seen, wall clock, caught failures).

Source code in varagity/pipeline/graph_flow.py

@flow(name="graph-build", validate_parameters=False)
def graph_build_flow(
    service: GraphService,
    batches: GraphBatches,
    *,
    prune_removed: bool = True,
    verbose: int = 0,
) -> BuildReport:
    """Upsert a parsed message corpus into the graph as a tracked flow run.

    Args:
        service: The process's graph service (holds the session and the
            single-flight write lock).
        batches: The boxed parsed source files (bare sequences would
            serialize into the flow-run record, which the server caps at
            512 KB), guid-merged by the session before rendering.
        prune_removed: Whether ``batches`` render the whole corpus. A bounded
            build passes ``False`` — its render is partial on purpose, and
            pruning on its say-so would delete the rest of the archive.
        verbose: Validated console verbosity (0–2).

    Returns:
        What the build did (messages seen, wall clock, caught failures).
    """
    logger = get_run_logger()
    source_files = batches.batches
    logger.info(
        "graph build starting: %d source file(s), %s render",
        len(source_files),
        "full-corpus" if prune_removed else "bounded",
    )
    report = service.build(source_files, verbose=verbose, prune_removed=prune_removed)
    logger.info(
        "graph build finished: %d message(s) in %.1fs, %d failure(s)",
        report.messages_seen,
        report.wall_clock_s,
        len(report.failures),
    )
    return report

graph_eval_flow(profile='smoke', engines=None, mode=None, skip_build=False, message_target=None, verbose=None)

Run the graph bake-off as a tracked flow run (spec_graphrag §12).

Parameters:

Name	Type	Description	Default
profile	str	Fixture corpus profile (smoke or full).	'smoke'
engines	Sequence[str] | None	Engine names to run; None runs every registered engine.	None
mode	str | None	Engine query mode override for every question.	None
skip_build	bool	Reuse the existing corpus and working directories instead of re-indexing.	False
message_target	int | None	Cap the corpus at this many messages instead of the profile's own size (a capped run gets its own corpus and working directories); None uses the profile's size.	None
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (see
dict[str, Any]	func:varagity.eval.graph_eval.run_graph_eval).

Source code in varagity/pipeline/eval_flow.py

@flow(name="eval-graph", validate_parameters=False)
def graph_eval_flow(
    profile: str = "smoke",
    engines: Sequence[str] | None = None,
    mode: str | None = None,
    skip_build: bool = False,
    message_target: int | None = None,
    verbose: int | None = None,
) -> dict[str, Any]:
    """Run the graph bake-off as a tracked flow run (spec_graphrag §12).

    Args:
        profile: Fixture corpus profile (``smoke`` or ``full``).
        engines: Engine names to run; ``None`` runs every registered engine.
        mode: Engine query mode override for every question.
        skip_build: Reuse the existing corpus and working directories
            instead of re-indexing.
        message_target: Cap the corpus at this many messages instead of the
            profile's own size (a capped run gets its own corpus and working
            directories); ``None`` uses the profile's size.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (see
        :func:`varagity.eval.graph_eval.run_graph_eval`).
    """
    return run_graph_eval(
        profile=profile,
        engines=engines,
        mode=mode,
        skip_build=skip_build,
        message_target=message_target,
        verbose=verbose,
    )

graph_export_flow(dest, *, on_progress=None, service=None, stale=None)

Export the message graph's working directory into a new bundle as a tracked flow run.

The copy runs under the graph service's write lock, so no build can change a file mid-copy while graph chat keeps answering. Needs no model service and opens no session — only postgres, for the stale flag.

Parameters:

Name	Type	Description	Default
dest	Path	The exports folder (<GRAPH_DOCS_PATH>/varagity-exports).	required
on_progress	GraphExportProgressCallback | None	Called with (stage, current, total) per stage (snapshot → pack, counting bytes → finalize).	None
service	GraphService | None	The graph service whose workdir to copy; the process's own (the graph's single writer) when omitted.	None
stale	bool | None	The graph-stale flag to record; when omitted, it is read from app_settings once the files are copied, so a source deleted mid-export is recorded.	None

Returns:

Type	Description
GraphExportReport	The published bundle and how long the export took.

Source code in varagity/pipeline/backup_flow.py

@flow(name="graph-export", validate_parameters=False)
def graph_export_flow(
    dest: Path,
    *,
    on_progress: GraphExportProgressCallback | None = None,
    service: GraphService | None = None,
    stale: bool | None = None,
) -> GraphExportReport:
    """Export the message graph's working directory into a new bundle as a tracked flow run.

    The copy runs under the graph service's write lock, so no build can
    change a file mid-copy while graph chat keeps answering. Needs no model
    service and opens no session — only postgres, for the stale flag.

    Args:
        dest: The exports folder (``<GRAPH_DOCS_PATH>/varagity-exports``).
        on_progress: Called with ``(stage, current, total)`` per stage
            (``snapshot`` → ``pack``, counting bytes → ``finalize``).
        service: The graph service whose workdir to copy; the process's own
            (the graph's single writer) when omitted.
        stale: The graph-stale flag to record; when omitted, it is read from
            ``app_settings`` once the files are copied, so a source deleted
            mid-export is recorded.

    Returns:
        The published bundle and how long the export took.
    """
    logger = get_run_logger()

    def read_stale() -> bool:
        if stale is not None:
            return stale
        with AppSettingsStore() as settings_store:
            return settings_store.is_graph_stale()

    active = service if service is not None else get_graph_service()
    result = export_graph(active, dest, read_stale=read_stale, on_progress=on_progress)
    counts = result.counts
    logger.info(
        "graph export wrote %s: %d file(s), %d thread-day(s), %d bytes in %.1fs",
        result.bundle.name,
        result.files,
        counts.get("thread_days", 0),
        result.bundle.size_bytes,
        result.wall_clock_s,
    )
    return result

graph_query_stream_flow(query, service, *, history=(), engine=None, llm=None, mode=None, verbose=None, on_retrieved=None, on_delta, should_abort=None, on_stats=None)

Answer one question from the graph, streaming generation deltas.

:func:varagity.pipeline.query_flow.query_stream_flow's graph twin, backing POST /api/chat with corpus="graph": condense (the same task, so a follow-up is rewritten into a standalone search query exactly as it is for chunk RAG) → graph retrieve → on_retrieved (the SSE retrieval event: evidence before prose) → the same streaming generate task over a graph-shaped context.

Generation is deliberately the repo's own, not the engine's (ADR-017's retrieval-only decision): the answer is grounded in the transcript days the graph cited, cites them by [SOURCE], and streams.

Parameters:

Name	Type	Description	Default
query	str	The user's question.	required
service	GraphService	The process's graph service.	required
history	Sequence[Turn]	Prior conversation turns, oldest first; the chat engine reads them when preparing the search query.	()
engine	ChatEngine | None	Chat engine preparing the search query; resolved from settings.CHAT_ENGINE when omitted.	None
llm	LLMClient | None	Chat client; resolved via the model registry when omitted.	None
mode	str | None	Engine query mode; defaults to settings.GRAPH_QUERY_MODE.	None
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None
on_retrieved	Callable[[GraphRetrieval], None] | None	Optional hook called with the graph retrieval before generation starts.	None
on_delta	Callable[[Kind, str], None]	Called with each classified (kind, text) fragment.	required
should_abort	Callable[[], bool] | None	Polled between deltas; True stops generation early and marks the state aborted.	None
on_stats	Callable[[GenerationTimings], None] | None	Called with llama.cpp's cumulative decode counters as they arrive.	None

Returns:

Type	Description
StreamedGraphState	The completed :class:StreamedGraphState.

Raises:

Type	Description
ValueError	If verbose is invalid.
KeyError	If settings.CHAT_ENGINE names an unregistered engine.
GraphUnavailable	If the engine's session cannot be opened (the route resolves this before streaming, so reaching it here means the session died mid-turn).

Source code in varagity/pipeline/graph_flow.py

@flow(name="graph-query-stream", validate_parameters=False)
def graph_query_stream_flow(
    query: str,
    service: GraphService,
    *,
    history: Sequence[Turn] = (),
    engine: ChatEngine | None = None,
    llm: LLMClient | None = None,
    mode: str | None = None,
    verbose: int | None = None,
    on_retrieved: Callable[[GraphRetrieval], None] | None = None,
    on_delta: Callable[[Kind, str], None],
    should_abort: Callable[[], bool] | None = None,
    on_stats: Callable[[GenerationTimings], None] | None = None,
) -> StreamedGraphState:
    """Answer one question from the graph, streaming generation deltas.

    :func:`varagity.pipeline.query_flow.query_stream_flow`'s graph twin,
    backing ``POST /api/chat`` with ``corpus="graph"``: condense (the same
    task, so a follow-up is rewritten into a standalone search query exactly
    as it is for chunk RAG) → graph retrieve → ``on_retrieved`` (the SSE
    ``retrieval`` event: evidence before prose) → the same streaming
    generate task over a graph-shaped context.

    Generation is deliberately the repo's own, not the engine's (ADR-017's
    retrieval-only decision): the answer is grounded in the transcript days
    the graph cited, cites them by ``[SOURCE]``, and streams.

    Args:
        query: The user's question.
        service: The process's graph service.
        history: Prior conversation turns, oldest first; the chat engine
            reads them when preparing the search query.
        engine: Chat engine preparing the search query; resolved from
            ``settings.CHAT_ENGINE`` when omitted.
        llm: Chat client; resolved via the model registry when omitted.
        mode: Engine query mode; defaults to ``settings.GRAPH_QUERY_MODE``.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.
        on_retrieved: Optional hook called with the graph retrieval before
            generation starts.
        on_delta: Called with each classified ``(kind, text)`` fragment.
        should_abort: Polled between deltas; ``True`` stops generation early
            and marks the state ``aborted``.
        on_stats: Called with llama.cpp's cumulative decode counters as they
            arrive.

    Returns:
        The completed :class:`StreamedGraphState`.

    Raises:
        ValueError: If ``verbose`` is invalid.
        KeyError: If ``settings.CHAT_ENGINE`` names an unregistered engine.
        varagity.graph.service.GraphUnavailable: If the engine's session
            cannot be opened (the route resolves this *before* streaming, so
            reaching it here means the session died mid-turn).
    """
    from varagity.chat import get_chat_engine  # deferred: keeps the import graph acyclic

    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    active_engine = engine if engine is not None else get_chat_engine(settings.CHAT_ENGINE)
    prepared = condense_query_task(active_engine, query, history=history, llm=llm, verbose=verbose)
    retrieval = graph_retrieve_task(
        service,
        prepared.search_query,
        mode=mode if mode is not None else settings.GRAPH_QUERY_MODE,
        verbose=verbose,
    )
    if on_retrieved is not None:
        on_retrieved(retrieval)
    formatted_context = graph_answer_context(
        retrieval,
        max_chars=answer_context_max_chars(settings.LLM_CONTEXT_TOKENS, settings.MAX_TOKENS),
    )
    result = generate_answer_stream_task(
        prepared.original_query,  # always the user's words (spec_v3 §4.2)
        [],  # the grounding is the graph's, pre-rendered above
        llm=llm,
        formatted_context=formatted_context,
        on_delta=on_delta,
        should_abort=should_abort,
        on_stats=on_stats,
        verbose=verbose,
    )
    return StreamedGraphState(
        query=query,
        prepared=prepared,
        retrieval=retrieval,
        formatted_context=formatted_context,
        answer=result["answer"],
        reasoning=result["reasoning"],
        aborted=result["aborted"],
        usage=result["usage"],
        tokens_per_second=result["tokens_per_second"],
    )

graph_restore_flow(bundle, *, run_id, on_progress=None, service=None, settings_store=None)

Restore the message graph from a bundle as a tracked flow run.

Extracts the bundle's workdir beside the live one and swaps it in under the graph service's write and open locks — the previous graph comes back if the restored one won't open — then, and only once all of that succeeded, writes the graph-stale flag the bundle recorded. Needs no model service: only postgres, for the flag.

Parameters:

Name	Type	Description	Default
bundle	Path	The bundle file (inside <GRAPH_DOCS_PATH>/varagity-exports).	required
run_id	str	The restore run's id (names the staging and previous folders beside the workdir).	required
on_progress	GraphRestoreProgressCallback | None	Called with (stage, current, total) per stage (validate → extract, counting bytes → swap → open).	None
service	GraphService | None	The graph service whose workdir to replace; the process's own (the graph's single writer) when omitted.	None
settings_store	AppSettingsStore | None	Where the stale flag is written; a connection of its own is opened (and closed) when omitted.	None

Returns:

Type	Description
GraphRestore	What was restored, with the stale flag it left.

Source code in varagity/pipeline/backup_flow.py

@flow(name="graph-restore", validate_parameters=False)
def graph_restore_flow(
    bundle: Path,
    *,
    run_id: str,
    on_progress: GraphRestoreProgressCallback | None = None,
    service: GraphService | None = None,
    settings_store: AppSettingsStore | None = None,
) -> GraphRestore:
    """Restore the message graph from a bundle as a tracked flow run.

    Extracts the bundle's workdir beside the live one and swaps it in under
    the graph service's write and open locks — the previous graph comes back
    if the restored one won't open — then, and only once all of that
    succeeded, writes the graph-stale flag the bundle recorded. Needs no
    model service: only postgres, for the flag.

    Args:
        bundle: The bundle file (inside ``<GRAPH_DOCS_PATH>/varagity-exports``).
        run_id: The restore run's id (names the staging and previous folders
            beside the workdir).
        on_progress: Called with ``(stage, current, total)`` per stage
            (``validate`` → ``extract``, counting bytes → ``swap`` →
            ``open``).
        service: The graph service whose workdir to replace; the process's
            own (the graph's single writer) when omitted.
        settings_store: Where the stale flag is written; a connection of its
            own is opened (and closed) when omitted.

    Returns:
        What was restored, with the stale flag it left.
    """
    logger = get_run_logger()
    active = service if service is not None else get_graph_service()
    result = restore_graph(active, bundle, run_id=run_id, on_progress=on_progress)
    # The bundle's own flag: the graph is exactly what was exported, so it is
    # stale exactly when it was then (a source deleted before the export).
    stale = result.manifest.stale
    if settings_store is not None:
        settings_store.set_graph_stale(stale)
    else:
        with AppSettingsStore() as owned_settings:
            owned_settings.set_graph_stale(stale)
    logger.info(
        "graph restore loaded %s: %d file(s), %d thread-day(s) in %.1fs%s",
        result.bundle_name,
        result.files,
        result.counts.get("thread_days", 0),
        result.wall_clock_s,
        " — the graph is flagged stale" if stale else "",
    )
    return replace(result, stale=stale)

ocr_benchmark_flow(verbose=None)

Run the OCR engine benchmark as a tracked flow run.

Parameters:

Name	Type	Description	Default
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (see
dict[str, Any]	func:varagity.eval.ocr_benchmark.run_ocr_benchmark).

Source code in varagity/pipeline/eval_flow.py

@flow(name="eval-ocr", validate_parameters=False)
def ocr_benchmark_flow(verbose: int | None = None) -> dict[str, Any]:
    """Run the OCR engine benchmark as a tracked flow run.

    Args:
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (see
        :func:`varagity.eval.ocr_benchmark.run_ocr_benchmark`).
    """
    return run_ocr_benchmark(ingest=ingest_flow, verbose=verbose)

query_stream_flow(query, *, history=(), engine=None, retriever=None, llm=None, k=None, verbose=None, on_retrieved=None, on_delta, should_abort=None, on_stats=None)

Answer one question with tracked stages, streaming generation deltas.

:func:query_flow's streaming twin, backing POST /api/chat (spec_v2 §4.3): identical condense/embed/retrieve staging (on_retrieved still fires before generation — the SSE retrieval event), then the streaming generate task hands each classified delta to on_delta while it runs.

Parameters:

Name	Type	Description	Default
query	str	The user's question.	required
history	Sequence[Turn]	Prior conversation turns, oldest first; the chat engine reads them when preparing the search query (empty by default — a first turn, or a history-less caller).	()
engine	ChatEngine | None	Chat engine preparing the search query; resolved from settings.CHAT_ENGINE when omitted.	None
retriever	Retriever | None	Retrieval method; resolved from settings.RETRIEVAL_METHOD when omitted.	None
llm	LLMClient | None	Chat client; resolved via the model registry when omitted.	None
k	int | None	Chunks to retrieve; defaults to settings.TOP_K.	None
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None
on_retrieved	Callable[[list[RetrievedChunk]], None] | None	Optional hook called with the retrieved chunks before generation.	None
on_delta	Callable[[Kind, str], None]	Called with each classified (kind, text) fragment, in stream order (kind is "reasoning" or "answer").	required
should_abort	Callable[[], bool] | None	Polled between deltas; returning True stops generation early and marks the state aborted.	None
on_stats	Callable[[GenerationTimings], None] | None	Called with llama.cpp's cumulative decode counters as they arrive (once per chunk — throttle before rendering); never fires on a server that reports no timings.	None

Returns:

Type	Description
StreamedQueryState	The completed :class:StreamedQueryState.

Raises:

Type	Description
ValueError	If verbose is invalid.
KeyError	If settings.RETRIEVAL_METHOD names an unregistered retrieval method, or settings.CHAT_ENGINE an unregistered chat engine.

Source code in varagity/pipeline/query_flow.py

@flow(name="query-stream", validate_parameters=False)
def query_stream_flow(
    query: str,
    *,
    history: Sequence[Turn] = (),
    engine: ChatEngine | None = None,
    retriever: Retriever | None = None,
    llm: LLMClient | None = None,
    k: int | None = None,
    verbose: int | None = None,
    on_retrieved: Callable[[list[RetrievedChunk]], None] | None = None,
    on_delta: Callable[[Kind, str], None],
    should_abort: Callable[[], bool] | None = None,
    on_stats: Callable[[GenerationTimings], None] | None = None,
) -> StreamedQueryState:
    """Answer one question with tracked stages, streaming generation deltas.

    :func:`query_flow`'s streaming twin, backing ``POST /api/chat``
    (spec_v2 §4.3): identical condense/embed/retrieve staging
    (``on_retrieved`` still fires before generation — the SSE ``retrieval``
    event), then the streaming generate task hands each classified delta to
    ``on_delta`` while it runs.

    Args:
        query: The user's question.
        history: Prior conversation turns, oldest first; the chat engine
            reads them when preparing the search query (empty by default —
            a first turn, or a history-less caller).
        engine: Chat engine preparing the search query; resolved from
            ``settings.CHAT_ENGINE`` when omitted.
        retriever: Retrieval method; resolved from
            ``settings.RETRIEVAL_METHOD`` when omitted.
        llm: Chat client; resolved via the model registry when omitted.
        k: Chunks to retrieve; defaults to ``settings.TOP_K``.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.
        on_retrieved: Optional hook called with the retrieved chunks before
            generation.
        on_delta: Called with each classified ``(kind, text)`` fragment, in
            stream order (``kind`` is ``"reasoning"`` or ``"answer"``).
        should_abort: Polled between deltas; returning ``True`` stops
            generation early and marks the state ``aborted``.
        on_stats: Called with llama.cpp's cumulative decode counters as
            they arrive (once per chunk — throttle before rendering);
            never fires on a server that reports no ``timings``.

    Returns:
        The completed :class:`StreamedQueryState`.

    Raises:
        ValueError: If ``verbose`` is invalid.
        KeyError: If ``settings.RETRIEVAL_METHOD`` names an unregistered
            retrieval method, or ``settings.CHAT_ENGINE`` an unregistered
            chat engine.
    """
    prep = _prepare_retrieval(
        query,
        history=history,
        engine=engine,
        retriever=retriever,
        llm=llm,
        k=k,
        verbose=verbose,
        on_retrieved=on_retrieved,
    )
    try:
        stage_started = time.perf_counter()
        result = generate_answer_stream_task(
            prep.prepared.original_query,  # always the user's words (spec_v3 §4.2)
            prep.chunks,
            llm=llm,
            formatted_context=prep.formatted_context,
            on_delta=on_delta,
            should_abort=should_abort,
            on_stats=on_stats,
            verbose=prep.verbose,
        )
        metrics.observe_query_stage("generate", prep.method, time.perf_counter() - stage_started)
    except Exception:
        metrics.count_query(prep.method, "error")
        raise
    metrics.count_query(prep.method, "aborted" if result["aborted"] else "ok")
    usage = result["usage"] or {}
    metrics.count_llm_tokens(usage.get("prompt_tokens"), usage.get("completion_tokens"))
    return StreamedQueryState(
        query=query,
        prepared=prep.prepared,
        query_vector=prep.query_vector,
        retrieved=prep.chunks,
        formatted_context=prep.formatted_context,
        answer=result["answer"],
        reasoning=result["reasoning"],
        aborted=result["aborted"],
        usage=result["usage"],
        tokens_per_second=result["tokens_per_second"],
    )

rag_export_flow(dest, *, ingest_settings, on_progress=None, store=None, stale=None)

Export the chunk corpus into a new bundle as a tracked flow run.

Parameters:

Name	Type	Description	Default
dest	Path	The exports folder (<DOCS_PATH>/varagity-exports).	required
ingest_settings	dict[str, str]	The reingest-affecting settings at export, in env-string form (the runner reads them before the stale flag, so a settings change racing the export can only make the bundle more conservative about staleness, never less).	required
on_progress	ProgressCallback | None	Called with (stage, current, total) per stage.	None
store	ContextualVectorDB | None	The vector store to snapshot; a connection of its own is opened (and closed) when omitted.	None
stale	bool | None	The corpus-stale flag at export; read from app_settings when omitted.	None

Returns:

Type	Description
RagExport	The published bundle and how long the export took.

Source code in varagity/pipeline/backup_flow.py

@flow(name="rag-export", validate_parameters=False)
def rag_export_flow(
    dest: Path,
    *,
    ingest_settings: dict[str, str],
    on_progress: ProgressCallback | None = None,
    store: ContextualVectorDB | None = None,
    stale: bool | None = None,
) -> RagExport:
    """Export the chunk corpus into a new bundle as a tracked flow run.

    Args:
        dest: The exports folder (``<DOCS_PATH>/varagity-exports``).
        ingest_settings: The reingest-affecting settings at export, in
            env-string form (the runner reads them before the stale flag,
            so a settings change racing the export can only make the bundle
            *more* conservative about staleness, never less).
        on_progress: Called with ``(stage, current, total)`` per stage.
        store: The vector store to snapshot; a connection of its own is
            opened (and closed) when omitted.
        stale: The corpus-stale flag at export; read from ``app_settings``
            when omitted.

    Returns:
        The published bundle and how long the export took.
    """
    logger = get_run_logger()
    if stale is None:
        with AppSettingsStore() as settings_store:
            stale = settings_store.is_corpus_stale()
    if store is not None:
        result = export_rag(dest, store, ingest_settings, stale=stale, on_progress=on_progress)
    else:
        with ContextualVectorDB() as owned:
            result = export_rag(dest, owned, ingest_settings, stale=stale, on_progress=on_progress)
    counts = result.counts
    logger.info(
        "rag export wrote %s: %d document(s), %d chunk(s), %d bytes in %.1fs",
        result.bundle.name,
        counts.get("documents", 0),
        counts.get("chunks", 0),
        result.bundle.size_bytes,
        result.wall_clock_s,
    )
    return result

rag_restore_flow(bundle, *, ingest_settings, on_progress=None, store=None, bm25=None, settings_store=None)

Restore the chunk corpus from a bundle as a tracked flow run.

Replaces both stores' corpus with the bundle's rows (re-rooted onto this stack's DOCS_PATH) and rebuilds the BM25 index from them, then writes the corpus-stale flag the bundle implies: stale if it was stale at export, or if its rows were made under other reingest-affecting settings than this stack's. The flag follows the rows: it is written once they committed — after a complete restore, and also when only the BM25 rebuild failed (the rows are the bundle's by then, so the previous corpus's flag would misdescribe them) — and never after a restore that rolled back.

Parameters:

Name	Type	Description	Default
bundle	Path	The bundle file (inside <DOCS_PATH>/varagity-exports).	required
ingest_settings	Callable[[], Mapping[str, str]]	Provider of this stack's reingest-affecting settings, in env-string form — called when the stale flag is written, so a settings change made during the restore is judged against the restored rows too.	required
on_progress	RestoreProgressCallback | None	Called with (stage, current, total) per stage (validate → load → index → bm25).	None
store	ContextualVectorDB | None	The vector store to replace; a connection of its own is opened (and closed) when omitted.	None
bm25	ElasticsearchBM25 | None	The BM25 store to rebuild; one is constructed (and closed) when omitted.	None
settings_store	AppSettingsStore | None	Where the stale flag is written; a connection of its own is opened (and closed) when omitted.	None

Returns:

Type	Description
RagRestore	What was restored, with the stale flag it left.

Source code in varagity/pipeline/backup_flow.py

@flow(name="rag-restore", validate_parameters=False)
def rag_restore_flow(
    bundle: Path,
    *,
    ingest_settings: Callable[[], Mapping[str, str]],
    on_progress: RestoreProgressCallback | None = None,
    store: ContextualVectorDB | None = None,
    bm25: ElasticsearchBM25 | None = None,
    settings_store: AppSettingsStore | None = None,
) -> RagRestore:
    """Restore the chunk corpus from a bundle as a tracked flow run.

    Replaces both stores' corpus with the bundle's rows (re-rooted onto this
    stack's ``DOCS_PATH``) and rebuilds the BM25 index from them, then
    writes the corpus-stale flag the bundle implies: stale if it was stale
    at export, or if its rows were made under other reingest-affecting
    settings than this stack's. The flag follows the rows: it is written
    once they committed — after a complete restore, and also when only the
    BM25 rebuild failed (the rows are the bundle's by then, so the previous
    corpus's flag would misdescribe them) — and never after a restore that
    rolled back.

    Args:
        bundle: The bundle file (inside ``<DOCS_PATH>/varagity-exports``).
        ingest_settings: Provider of this stack's reingest-affecting
            settings, in env-string form — called when the stale flag is
            written, so a settings change made *during* the restore is
            judged against the restored rows too.
        on_progress: Called with ``(stage, current, total)`` per stage
            (``validate`` → ``load`` → ``index`` → ``bm25``).
        store: The vector store to replace; a connection of its own is
            opened (and closed) when omitted.
        bm25: The BM25 store to rebuild; one is constructed (and closed)
            when omitted.
        settings_store: Where the stale flag is written; a connection of its
            own is opened (and closed) when omitted.

    Returns:
        What was restored, with the stale flag it left.
    """
    logger = get_run_logger()
    docs_root = Path(get_settings().DOCS_PATH)

    def write_stale(manifest: BundleManifest) -> bool:
        stale = derive_corpus_stale(manifest, ingest_settings())
        if settings_store is not None:
            settings_store.set_corpus_stale(stale)
        else:
            with AppSettingsStore() as owned_settings:
                owned_settings.set_corpus_stale(stale)
        return stale

    try:
        with ExitStack() as stack:
            active_store = store if store is not None else stack.enter_context(ContextualVectorDB())
            active_bm25 = bm25 if bm25 is not None else stack.enter_context(ElasticsearchBM25())
            result = restore_rag(
                bundle, active_store, active_bm25, docs_root, on_progress=on_progress
            )
    except BM25RebuildFailed as error:
        # The rows committed before the rebuild failed: they are the bundle's
        # now, so the flag must describe them too — then the run still fails.
        if error.manifest is not None:
            try:
                stale = write_stale(error.manifest)
            except Exception:  # the rebuild's failure stays the run's answer
                logger.exception("rag restore could not write the corpus-stale flag")
            else:
                logger.warning(
                    "rag restore committed the rows of %s but not the BM25 index%s",
                    bundle.name,
                    " — the corpus is flagged stale" if stale else "",
                )
        raise
    stale = write_stale(result.manifest)
    logger.info(
        "rag restore loaded %s: %d document(s), %d chunk(s), %d in BM25 in %.1fs%s",
        result.bundle_name,
        result.documents,
        result.chunks,
        result.bm25_indexed,
        result.wall_clock_s,
        " — the corpus is flagged stale" if stale else "",
    )
    return replace(result, stale=stale)

varagity.pipeline.ingest_flow

Prefect ingestion flow: the spec §9 pipeline as tracked task runs.

The flow delegates to :func:varagity.ingest.loader.ingest_corpus — the single orchestration loop — passing an :class:~varagity.ingest.loader.IngestStages bundle whose entries are @task-wrapped equivalents of the loader's stage functions, so every stage of every file becomes a task run visible at the Prefect UI (:4200) with state, duration, and logs.

Model/store tasks (contextualize, embed, store) carry retries=2 with exponential backoff. This is a second retry layer on purpose: the tenacity retries inside the clients cover transient HTTP failures within one call, while task retries re-run the whole stage after the client has given up — e.g. an Elasticsearch restart mid-ingest (both store writes are idempotent, so a re-run is safe). Discovery/parse/chunk are local and deterministic; retrying them cannot help, so they carry none.

Result caching is disabled on every task (NO_CACHE): the stages are side-effecting calls against live services, and their inputs include unhashable handles (store/client objects, the rich progress display) that make Prefect's default input-hash cache policy log an error per run — a cache hit could never be correct here anyway.

The contextualize and store tasks are Prometheus probe points (spec_v2 §6.2): blurb latency per document, plus document/chunk counters at the moment a document actually lands in both stores.

chunk_document_task(raw, *, verbose)

Task wrapper over chunking (spec §9.3).

Parameters:

Name	Type	Description	Default
raw	RawDocument	The parsed document.	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
list[Document]	The chunks, provenance seeded into each chunk's metadata.

Source code in varagity/pipeline/ingest_flow.py

@task(name="chunk_document", cache_policy=NO_CACHE)
def chunk_document_task(raw: RawDocument, *, verbose: int) -> list[Document]:
    """Task wrapper over chunking (spec §9.3).

    Args:
        raw: The parsed document.
        verbose: Validated console verbosity.

    Returns:
        The chunks, provenance seeded into each chunk's metadata.
    """
    chunks = loader.chunk_document(raw, verbose=verbose)
    get_run_logger().info(
        "chunked %s into %d chunk(s)", raw.source_meta.get("file_name"), len(chunks)
    )
    return chunks

contextualize_chunks_task(*, document_text, chunk_texts, llm, file_name, progress, verbose)

Task wrapper over situating-blurb generation (spec §9.4).

Parameters:

Name	Type	Description	Default
document_text	str	The parent document's full extracted text.	required
chunk_texts	list[str]	The document's chunk texts, in order.	required
llm	LLMClient | None	Chat client, or None for the identity path (settings.CONTEXTUALIZE off).	required
file_name	str	Source file name (labels the sub-progress bar).	required
progress	Progress	The run's progress display.	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
list[str | None]	One context blurb per chunk (all None when llm is None).

Source code in varagity/pipeline/ingest_flow.py

@task(
    name="contextualize_chunks",
    retries=2,
    retry_delay_seconds=exponential_backoff(backoff_factor=2),
    cache_policy=NO_CACHE,
)
def contextualize_chunks_task(
    *,
    document_text: str,
    chunk_texts: list[str],
    llm: LLMClient | None,
    file_name: str,
    progress: Progress,
    verbose: int,
) -> list[str | None]:
    """Task wrapper over situating-blurb generation (spec §9.4).

    Args:
        document_text: The parent document's full extracted text.
        chunk_texts: The document's chunk texts, in order.
        llm: Chat client, or ``None`` for the identity path
            (``settings.CONTEXTUALIZE`` off).
        file_name: Source file name (labels the sub-progress bar).
        progress: The run's progress display.
        verbose: Validated console verbosity.

    Returns:
        One context blurb per chunk (all ``None`` when ``llm`` is ``None``).
    """
    started = time.perf_counter()
    contexts = loader.contextualize_chunks(
        document_text=document_text,
        chunk_texts=chunk_texts,
        llm=llm,
        file_name=file_name,
        progress=progress,
        verbose=verbose,
    )
    logger = get_run_logger()
    if llm is None:
        logger.info("CONTEXTUALIZE off — identity path for %s", file_name)
    else:
        # The identity path is not contextualization — only real blurb
        # generation feeds the spec §9.4 throughput-cost histogram.
        metrics.observe_contextualize(time.perf_counter() - started)
        logger.info("contextualized %d chunk(s) of %s", len(contexts), file_name)
    return contexts

discover_documents_task(docs_path, verbose=None)

Task wrapper over corpus discovery (spec §9.1).

Parameters:

Name	Type	Description	Default
docs_path	str	Directory to scan.	required
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
Buckets	The bucketed file paths.

Source code in varagity/pipeline/ingest_flow.py

@task(name="discover_documents", cache_policy=NO_CACHE)
def discover_documents_task(docs_path: str, verbose: int | None = None) -> Buckets:
    """Task wrapper over corpus discovery (spec §9.1).

    Args:
        docs_path: Directory to scan.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The bucketed file paths.
    """
    buckets = loader.discover_documents(docs_path, verbose=verbose)
    get_run_logger().info(
        "discovered %d file(s) under %s (%s)",
        buckets.total,
        docs_path,
        ", ".join(f"{len(paths)} {name}" for name, paths in buckets.by_bucket()),
    )
    return buckets

embed_chunks_task(texts, *, embeddings, verbose)

Task wrapper over passage embedding (spec §9.5).

Parameters:

Name	Type	Description	Default
texts	list[str]	The contextualized_content of each chunk, in order.	required
embeddings	EmbeddingsClient	The embeddings client.	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
list[list[float]]	One embedding vector per text, in order.

Source code in varagity/pipeline/ingest_flow.py

@task(
    name="embed_chunks",
    retries=2,
    retry_delay_seconds=exponential_backoff(backoff_factor=2),
    cache_policy=NO_CACHE,
)
def embed_chunks_task(
    texts: list[str], *, embeddings: EmbeddingsClient, verbose: int
) -> list[list[float]]:
    """Task wrapper over passage embedding (spec §9.5).

    Args:
        texts: The ``contextualized_content`` of each chunk, in order.
        embeddings: The embeddings client.
        verbose: Validated console verbosity.

    Returns:
        One embedding vector per text, in order.
    """
    vectors = loader.embed_chunks(texts, embeddings=embeddings, verbose=verbose)
    get_run_logger().info("embedded %d passage(s)", len(vectors))
    return vectors

ingest_flow(docs_path=None, *, store=None, bm25=None, embeddings=None, llm=None, reingest=False, verbose=None, stages=None, on_file=None)

Ingest the corpus with every stage tracked as a Prefect task run.

A thin @flow shell over :func:varagity.ingest.loader.ingest_corpus (same parameters, same behavior — one file failing is counted, logged, and visible as a failed task run, never aborting the corpus).

Parameter validation is off because callers (tests, the eval harness) inject duck-typed store/client fakes that pydantic would reject; the flow's inputs are already-validated internals.

Parameters:

Name	Type	Description	Default
docs_path	str | None	Corpus directory; defaults to settings.DOCS_PATH.	None
store	ContextualVectorDB | None	Vector store to write to; constructed from settings (and closed on return) when omitted.	None
bm25	ElasticsearchBM25 | None	BM25 store to write to; constructed from settings (and closed on return) when omitted.	None
embeddings	EmbeddingsClient | None	Embeddings client; resolved via the model registry when omitted.	None
llm	LLMClient | None	Chat client for contextualization; resolved via the model registry when omitted. Unused when settings.CONTEXTUALIZE is off.	None
reingest	bool	Delete each discovered document's previous ingest (from both stores) and re-process it.	False
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None
stages	IngestStages | None	Per-stage call seam; this flow's Prefect @task-wrapped stages when omitted. The API's ingest runner passes event-emitting wrappers around those tasks (spec_v2 §4.2 live progress), so tracking and streaming compose.	None
on_file	Callable[[Path, str, int], None] | None	Per-file outcome observer, passed through to the loader.	None

Returns:

Type	Description
IngestSummary	The run's counters.

Source code in varagity/pipeline/ingest_flow.py

@flow(name="ingest", validate_parameters=False)
def ingest_flow(
    docs_path: str | None = None,
    *,
    store: ContextualVectorDB | None = None,
    bm25: ElasticsearchBM25 | None = None,
    embeddings: EmbeddingsClient | None = None,
    llm: LLMClient | None = None,
    reingest: bool = False,
    verbose: int | None = None,
    stages: IngestStages | None = None,
    on_file: Callable[[Path, str, int], None] | None = None,
) -> IngestSummary:
    """Ingest the corpus with every stage tracked as a Prefect task run.

    A thin ``@flow`` shell over :func:`varagity.ingest.loader.ingest_corpus`
    (same parameters, same behavior — one file failing is counted, logged,
    and visible as a failed task run, never aborting the corpus).

    Parameter validation is off because callers (tests, the eval
    harness) inject duck-typed store/client fakes that pydantic would
    reject; the flow's inputs are already-validated internals.

    Args:
        docs_path: Corpus directory; defaults to ``settings.DOCS_PATH``.
        store: Vector store to write to; constructed from settings (and
            closed on return) when omitted.
        bm25: BM25 store to write to; constructed from settings (and closed
            on return) when omitted.
        embeddings: Embeddings client; resolved via the model registry when
            omitted.
        llm: Chat client for contextualization; resolved via the model
            registry when omitted. Unused when ``settings.CONTEXTUALIZE``
            is off.
        reingest: Delete each discovered document's previous ingest (from
            both stores) and re-process it.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.
        stages: Per-stage call seam; this flow's Prefect ``@task``-wrapped
            stages when omitted. The API's ingest runner passes
            event-emitting wrappers *around* those tasks (spec_v2 §4.2
            live progress), so tracking and streaming compose.
        on_file: Per-file outcome observer, passed through to the loader.

    Returns:
        The run's counters.
    """
    return ingest_corpus(
        docs_path,
        store=store,
        bm25=bm25,
        embeddings=embeddings,
        llm=llm,
        reingest=reingest,
        verbose=verbose,
        stages=stages if stages is not None else TASK_STAGES,
        on_file=on_file,
    )

parse_document_task(parser, path, *, verbose)

Task wrapper over text extraction (spec §9.2).

Parameters:

Name	Type	Description	Default
parser	Parser	Parser for the file's discovery bucket.	required
path	Path	The file to parse.	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
RawDocument	The extracted document.

Source code in varagity/pipeline/ingest_flow.py

@task(name="parse_document", cache_policy=NO_CACHE)
def parse_document_task(parser: Parser, path: Path, *, verbose: int) -> RawDocument:
    """Task wrapper over text extraction (spec §9.2).

    Args:
        parser: Parser for the file's discovery bucket.
        path: The file to parse.
        verbose: Validated console verbosity.

    Returns:
        The extracted document.
    """
    raw = loader.parse_document(parser, path, verbose=verbose)
    get_run_logger().info(
        "parsed %s → %d chars (extraction=%s)",
        path.name,
        len(raw.text),
        raw.source_meta.get("extraction", "text"),
    )
    return raw

store_chunks_task(records, vectors, *, store, bm25)

Task wrapper over the dual-store write (spec §9.6).

Both writes share this one task boundary: a partial failure fails the task loudly and the retry re-runs both writes (idempotent — see :func:varagity.ingest.loader.store_chunks for the ordering rationale).

Parameters:

Name	Type	Description	Default
records	list[ChunkRecord]	The document's chunk records (non-empty).	required
vectors	list[list[float]]	One embedding per record, in order.	required
store	ContextualVectorDB	The vector store.	required
bm25	ElasticsearchBM25	The BM25 store.	required

Source code in varagity/pipeline/ingest_flow.py

@task(
    name="store_chunks",
    retries=2,
    retry_delay_seconds=exponential_backoff(backoff_factor=2),
    cache_policy=NO_CACHE,
)
def store_chunks_task(
    records: list[ChunkRecord],
    vectors: list[list[float]],
    *,
    store: ContextualVectorDB,
    bm25: ElasticsearchBM25,
) -> None:
    """Task wrapper over the dual-store write (spec §9.6).

    Both writes share this one task boundary: a partial failure fails the
    task loudly and the retry re-runs both writes (idempotent — see
    :func:`varagity.ingest.loader.store_chunks` for the ordering rationale).

    Args:
        records: The document's chunk records (non-empty).
        vectors: One embedding per record, in order.
        store: The vector store.
        bm25: The BM25 store.
    """
    loader.store_chunks(records, vectors, store=store, bm25=bm25)
    # Counted here — after both writes — so the corpus-growth counters
    # reflect documents that actually landed, not attempts.
    head = records[0]
    metrics.count_ingested_document(head.file_type, head.extraction)
    metrics.count_ingested_chunks(head.chunking_strategy, len(records))
    get_run_logger().info(
        "stored %d chunk(s) of %s in both stores", len(records), records[0].file_name
    )

varagity.pipeline.query_flow

Prefect query flow: the spec §10.1 pipeline as tracked task runs.

Condense → embed → retrieve → generate, each a task run, returning the spec §10.1 state dict. The condense stage (spec_v3 §4.2) runs the configured chat engine over the turn and its history to decide the search query; retrieval sees prepared.search_query while the answer prompt always gets prepared.original_query — the user's words are never rewritten for generation. The stage is always in the graph (plan decision

14): under the simple engine it is a no-LLM pass-through, so a

tracked run is its only cost. Query embedding — encapsulated inside the retrievers on the plain path (:func:varagity.generation.answer.answer_query) — is hoisted into its own tracked stage here via the retrievers' :meth:~varagity.retrieval.base.Retriever.encode_query seam, which also fills the state's query_vector (None for bm25: nothing to encode).

:func:query_stream_flow is the flow's streaming twin (spec_v2 §4.3): the same embed/retrieve stages, with generation swapped for a tracked task that hands deltas to a callback while it runs — the task boundary (and its run log) is preserved while tokens flow out to the HTTP API's SSE stream.

Unlike the ingestion flow's model/store tasks, these tasks carry no Prefect-level retries: the query path is interactive, the clients already retry transient HTTP failures internally (tenacity), and stacking task retries on top would multiply the wait before a hard failure surfaces at the prompt. Result caching is disabled (NO_CACHE) for the same reasons as the ingestion tasks: live-service calls with unhashable client/retriever inputs.

Both flows are Prometheus probe points (spec_v2 §6.2): they time each stage around its task call, record the retrieved chunks' scores (+ rerank movement), and count the flow's outcome — the streaming flow additionally counts the server-reported token usage it already returns.

StreamedQueryState

Bases: QueryState

The spec §10.1 state dict, extended with streaming outcomes.

Attributes:

Name	Type	Description
reasoning	str	The captured <think> stream ("" when none).
aborted	bool	True when the client aborted generation mid-stream.
usage	dict[str, int] | None	Server-reported token counts, or None when unreported.
tokens_per_second	float | None	Final decode throughput reported by llama.cpp, or None when the model server reports no timings.

Source code in varagity/pipeline/query_flow.py

class StreamedQueryState(QueryState):
    """The spec §10.1 state dict, extended with streaming outcomes.

    Attributes:
        reasoning: The captured ``<think>`` stream (``""`` when none).
        aborted: ``True`` when the client aborted generation mid-stream.
        usage: Server-reported token counts, or ``None`` when unreported.
        tokens_per_second: Final decode throughput reported by llama.cpp,
            or ``None`` when the model server reports no ``timings``.
    """

    reasoning: str
    aborted: bool
    usage: dict[str, int] | None
    tokens_per_second: float | None

condense_query_task(engine, query, *, history, llm, verbose)

Task wrapper over the chat engine's query preparation (spec_v3 §4.2).

Always in the graph, whatever the engine (plan decision #14): simple returns the identity split with no LLM call, so the tracked stage is ~free — and the registry stays a seam rather than a special case.

Parameters:

Name	Type	Description	Default
engine	ChatEngine	The chat engine (owns the condense decision).	required
query	str	The user's question, verbatim.	required
history	Sequence[Turn]	Prior conversation turns, oldest first (empty on a first turn, and from history-less callers like the CLI).	required
llm	LLMClient | None	Chat client for engines that condense; None lets them resolve one via the model registry.	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
PreparedQuery	The two-string split downstream stages retrieve and answer with.

Source code in varagity/pipeline/query_flow.py

@task(name="condense_query", cache_policy=NO_CACHE)
def condense_query_task(
    engine: ChatEngine,
    query: str,
    *,
    history: Sequence[Turn],
    llm: LLMClient | None,
    verbose: int,
) -> PreparedQuery:
    """Task wrapper over the chat engine's query preparation (spec_v3 §4.2).

    Always in the graph, whatever the engine (plan decision #14): ``simple``
    returns the identity split with no LLM call, so the tracked stage is
    ~free — and the registry stays a seam rather than a special case.

    Args:
        engine: The chat engine (owns the condense decision).
        query: The user's question, verbatim.
        history: Prior conversation turns, oldest first (empty on a first
            turn, and from history-less callers like the CLI).
        llm: Chat client for engines that condense; ``None`` lets them
            resolve one via the model registry.
        verbose: Validated console verbosity.

    Returns:
        The two-string split downstream stages retrieve and answer with.
    """
    prepared = engine.prepare(query, history=history, llm=llm, verbose=verbose)
    logger = get_run_logger()
    if prepared.condensed:
        logger.info("condensed the search query → %r", prepared.search_query)
    else:
        logger.info("search query is the user's words, verbatim")
    return prepared

embed_query_task(retriever, query, *, verbose)

Task wrapper over query encoding (spec §10.1 step 2).

Parameters:

Name	Type	Description	Default
retriever	Retriever	The retrieval method (owns its query encoding).	required
query	str	The user's question, unformatted.	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
list[float] | None	The query embedding, or None for methods that never encode
list[float] | None	queries (bm25).

Source code in varagity/pipeline/query_flow.py

@task(name="embed_query", cache_policy=NO_CACHE)
def embed_query_task(retriever: Retriever, query: str, *, verbose: int) -> list[float] | None:
    """Task wrapper over query encoding (spec §10.1 step 2).

    Args:
        retriever: The retrieval method (owns its query encoding).
        query: The user's question, unformatted.
        verbose: Validated console verbosity.

    Returns:
        The query embedding, or ``None`` for methods that never encode
        queries (``bm25``).
    """
    vector = retriever.encode_query(query, verbose=verbose)
    logger = get_run_logger()
    if vector is None:
        logger.info("retrieval method encodes no query vector")
    else:
        logger.info("embedded query → %d-dim vector", len(vector))
    return vector

generate_answer_stream_task(query, chunks, *, llm, formatted_context, on_delta, should_abort, on_stats, verbose)

Task wrapper over streamed answer generation (spec_v2 §4.3).

The task boundary is preserved while deltas flow out through on_delta — the run log records the outcome exactly like the non-streaming twin. A client-side abort is a deliberate stop, not a failure: the task completes normally with aborted=True.

Parameters:

Name	Type	Description	Default
query	str	The user's question.	required
chunks	list[RetrievedChunk]	The retrieved chunks grounding the answer.	required
llm	LLMClient | None	Chat client; resolved via the model registry when None.	required
formatted_context	str	The pre-built context block (spec §10.2).	required
on_delta	Callable[[Kind, str], None]	Called with each classified (kind, text) fragment.	required
should_abort	Callable[[], bool] | None	Polled between deltas; True stops generation.	required
on_stats	Callable[[GenerationTimings], None] | None	Called with llama.cpp's decode counters as they arrive.	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
StreamedAnswer	The completed :class:~varagity.generation.answer.StreamedAnswer.

Source code in varagity/pipeline/query_flow.py

@task(name="generate_answer_stream", cache_policy=NO_CACHE)
def generate_answer_stream_task(
    query: str,
    chunks: list[RetrievedChunk],
    *,
    llm: LLMClient | None,
    formatted_context: str,
    on_delta: Callable[[Kind, str], None],
    should_abort: Callable[[], bool] | None,
    on_stats: Callable[[GenerationTimings], None] | None,
    verbose: int,
) -> StreamedAnswer:
    """Task wrapper over streamed answer generation (spec_v2 §4.3).

    The task boundary is preserved while deltas flow out through
    ``on_delta`` — the run log records the outcome exactly like the
    non-streaming twin. A client-side abort is a deliberate stop, not a
    failure: the task completes normally with ``aborted=True``.

    Args:
        query: The user's question.
        chunks: The retrieved chunks grounding the answer.
        llm: Chat client; resolved via the model registry when ``None``.
        formatted_context: The pre-built context block (spec §10.2).
        on_delta: Called with each classified ``(kind, text)`` fragment.
        should_abort: Polled between deltas; ``True`` stops generation.
        on_stats: Called with llama.cpp's decode counters as they arrive.
        verbose: Validated console verbosity.

    Returns:
        The completed :class:`~varagity.generation.answer.StreamedAnswer`.
    """
    result = generate_answer_stream(
        query,
        chunks,
        on_delta=on_delta,
        llm=llm,
        formatted_context=formatted_context,
        should_abort=should_abort,
        on_stats=on_stats,
        verbose=verbose,
    )
    logger = get_run_logger()
    if result["aborted"]:
        logger.info("generation aborted by the client after %d chars", len(result["answer"]))
    else:
        logger.info(
            "streamed %d-char answer (%d-char reasoning) from %d chunk(s)",
            len(result["answer"]),
            len(result["reasoning"]),
            len(chunks),
        )
    return result

generate_answer_task(query, chunks, *, llm, formatted_context, verbose)

Task wrapper over grounded answer generation (spec §10.1 steps 5–6).

Parameters:

Name	Type	Description	Default
query	str	The user's question.	required
chunks	list[RetrievedChunk]	The retrieved chunks grounding the answer.	required
llm	LLMClient | None	Chat client; resolved via the model registry when None.	required
formatted_context	str	The pre-built context block (spec §10.2).	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
str	The generated, <think>-stripped answer.

Source code in varagity/pipeline/query_flow.py

@task(name="generate_answer", cache_policy=NO_CACHE)
def generate_answer_task(
    query: str,
    chunks: list[RetrievedChunk],
    *,
    llm: LLMClient | None,
    formatted_context: str,
    verbose: int,
) -> str:
    """Task wrapper over grounded answer generation (spec §10.1 steps 5–6).

    Args:
        query: The user's question.
        chunks: The retrieved chunks grounding the answer.
        llm: Chat client; resolved via the model registry when ``None``.
        formatted_context: The pre-built context block (spec §10.2).
        verbose: Validated console verbosity.

    Returns:
        The generated, ``<think>``-stripped answer.
    """
    answer = generate_answer(
        query, chunks, llm=llm, formatted_context=formatted_context, verbose=verbose
    )
    get_run_logger().info("generated %d-char answer from %d chunk(s)", len(answer), len(chunks))
    return answer

query_flow(query, *, history=(), engine=None, retriever=None, llm=None, k=None, verbose=None, on_retrieved=None)

Answer one question with every stage tracked as a Prefect task run.

The tracked twin of :func:varagity.generation.answer.answer_query (same state dict); the composition differs in query embedding and chat engine preparation each running as their own stage. Parameter validation is off because callers inject duck-typed retriever/client fakes that pydantic would reject; the flow's inputs are already-validated internals.

Parameters:

Name	Type	Description	Default
query	str	The user's question.	required
history	Sequence[Turn]	Prior conversation turns, oldest first; the chat engine reads them when preparing the search query (empty by default — a first turn, or a history-less caller).	()
engine	ChatEngine | None	Chat engine preparing the search query; resolved from settings.CHAT_ENGINE when omitted.	None
retriever	Retriever | None	Retrieval method; resolved from settings.RETRIEVAL_METHOD when omitted.	None
llm	LLMClient | None	Chat client; resolved via the model registry when omitted.	None
k	int | None	Chunks to retrieve; defaults to settings.TOP_K.	None
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None
on_retrieved	Callable[[list[RetrievedChunk]], None] | None	Optional hook called with the retrieved chunks before generation (the CLI's matches table — spec §10.1 step 4).	None

Returns:

Type	Description
QueryState	The completed :class:~varagity.generation.answer.QueryState.

Raises:

Type	Description
ValueError	If verbose is invalid.
KeyError	If settings.RETRIEVAL_METHOD names an unregistered retrieval method, or settings.CHAT_ENGINE an unregistered chat engine.

Source code in varagity/pipeline/query_flow.py

@flow(name="query", validate_parameters=False)
def query_flow(
    query: str,
    *,
    history: Sequence[Turn] = (),
    engine: ChatEngine | None = None,
    retriever: Retriever | None = None,
    llm: LLMClient | None = None,
    k: int | None = None,
    verbose: int | None = None,
    on_retrieved: Callable[[list[RetrievedChunk]], None] | None = None,
) -> QueryState:
    """Answer one question with every stage tracked as a Prefect task run.

    The tracked twin of :func:`varagity.generation.answer.answer_query`
    (same state dict); the composition differs in query embedding and chat
    engine preparation each running as their own stage. Parameter
    validation is off because callers inject duck-typed retriever/client
    fakes that pydantic would reject; the flow's inputs are
    already-validated internals.

    Args:
        query: The user's question.
        history: Prior conversation turns, oldest first; the chat engine
            reads them when preparing the search query (empty by default —
            a first turn, or a history-less caller).
        engine: Chat engine preparing the search query; resolved from
            ``settings.CHAT_ENGINE`` when omitted.
        retriever: Retrieval method; resolved from
            ``settings.RETRIEVAL_METHOD`` when omitted.
        llm: Chat client; resolved via the model registry when omitted.
        k: Chunks to retrieve; defaults to ``settings.TOP_K``.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.
        on_retrieved: Optional hook called with the retrieved chunks before
            generation (the CLI's matches table — spec §10.1 step 4).

    Returns:
        The completed :class:`~varagity.generation.answer.QueryState`.

    Raises:
        ValueError: If ``verbose`` is invalid.
        KeyError: If ``settings.RETRIEVAL_METHOD`` names an unregistered
            retrieval method, or ``settings.CHAT_ENGINE`` an unregistered
            chat engine.
    """
    prep = _prepare_retrieval(
        query,
        history=history,
        engine=engine,
        retriever=retriever,
        llm=llm,
        k=k,
        verbose=verbose,
        on_retrieved=on_retrieved,
    )
    try:
        stage_started = time.perf_counter()
        answer = generate_answer_task(
            prep.prepared.original_query,  # always the user's words (spec_v3 §4.2)
            prep.chunks,
            llm=llm,
            formatted_context=prep.formatted_context,
            verbose=prep.verbose,
        )
        metrics.observe_query_stage("generate", prep.method, time.perf_counter() - stage_started)
    except Exception:
        metrics.count_query(prep.method, "error")
        raise
    metrics.count_query(prep.method, "ok")
    return QueryState(
        query=query,
        prepared=prep.prepared,
        query_vector=prep.query_vector,
        retrieved=prep.chunks,
        formatted_context=prep.formatted_context,
        answer=answer,
    )

query_stream_flow(query, *, history=(), engine=None, retriever=None, llm=None, k=None, verbose=None, on_retrieved=None, on_delta, should_abort=None, on_stats=None)

Answer one question with tracked stages, streaming generation deltas.

:func:query_flow's streaming twin, backing POST /api/chat (spec_v2 §4.3): identical condense/embed/retrieve staging (on_retrieved still fires before generation — the SSE retrieval event), then the streaming generate task hands each classified delta to on_delta while it runs.

Parameters:

Name	Type	Description	Default
query	str	The user's question.	required
history	Sequence[Turn]	Prior conversation turns, oldest first; the chat engine reads them when preparing the search query (empty by default — a first turn, or a history-less caller).	()
engine	ChatEngine | None	Chat engine preparing the search query; resolved from settings.CHAT_ENGINE when omitted.	None
retriever	Retriever | None	Retrieval method; resolved from settings.RETRIEVAL_METHOD when omitted.	None
llm	LLMClient | None	Chat client; resolved via the model registry when omitted.	None
k	int | None	Chunks to retrieve; defaults to settings.TOP_K.	None
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None
on_retrieved	Callable[[list[RetrievedChunk]], None] | None	Optional hook called with the retrieved chunks before generation.	None
on_delta	Callable[[Kind, str], None]	Called with each classified (kind, text) fragment, in stream order (kind is "reasoning" or "answer").	required
should_abort	Callable[[], bool] | None	Polled between deltas; returning True stops generation early and marks the state aborted.	None
on_stats	Callable[[GenerationTimings], None] | None	Called with llama.cpp's cumulative decode counters as they arrive (once per chunk — throttle before rendering); never fires on a server that reports no timings.	None

Returns:

Type	Description
StreamedQueryState	The completed :class:StreamedQueryState.

Raises:

Type	Description
ValueError	If verbose is invalid.
KeyError	If settings.RETRIEVAL_METHOD names an unregistered retrieval method, or settings.CHAT_ENGINE an unregistered chat engine.

Source code in varagity/pipeline/query_flow.py

@flow(name="query-stream", validate_parameters=False)
def query_stream_flow(
    query: str,
    *,
    history: Sequence[Turn] = (),
    engine: ChatEngine | None = None,
    retriever: Retriever | None = None,
    llm: LLMClient | None = None,
    k: int | None = None,
    verbose: int | None = None,
    on_retrieved: Callable[[list[RetrievedChunk]], None] | None = None,
    on_delta: Callable[[Kind, str], None],
    should_abort: Callable[[], bool] | None = None,
    on_stats: Callable[[GenerationTimings], None] | None = None,
) -> StreamedQueryState:
    """Answer one question with tracked stages, streaming generation deltas.

    :func:`query_flow`'s streaming twin, backing ``POST /api/chat``
    (spec_v2 §4.3): identical condense/embed/retrieve staging
    (``on_retrieved`` still fires before generation — the SSE ``retrieval``
    event), then the streaming generate task hands each classified delta to
    ``on_delta`` while it runs.

    Args:
        query: The user's question.
        history: Prior conversation turns, oldest first; the chat engine
            reads them when preparing the search query (empty by default —
            a first turn, or a history-less caller).
        engine: Chat engine preparing the search query; resolved from
            ``settings.CHAT_ENGINE`` when omitted.
        retriever: Retrieval method; resolved from
            ``settings.RETRIEVAL_METHOD`` when omitted.
        llm: Chat client; resolved via the model registry when omitted.
        k: Chunks to retrieve; defaults to ``settings.TOP_K``.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.
        on_retrieved: Optional hook called with the retrieved chunks before
            generation.
        on_delta: Called with each classified ``(kind, text)`` fragment, in
            stream order (``kind`` is ``"reasoning"`` or ``"answer"``).
        should_abort: Polled between deltas; returning ``True`` stops
            generation early and marks the state ``aborted``.
        on_stats: Called with llama.cpp's cumulative decode counters as
            they arrive (once per chunk — throttle before rendering);
            never fires on a server that reports no ``timings``.

    Returns:
        The completed :class:`StreamedQueryState`.

    Raises:
        ValueError: If ``verbose`` is invalid.
        KeyError: If ``settings.RETRIEVAL_METHOD`` names an unregistered
            retrieval method, or ``settings.CHAT_ENGINE`` an unregistered
            chat engine.
    """
    prep = _prepare_retrieval(
        query,
        history=history,
        engine=engine,
        retriever=retriever,
        llm=llm,
        k=k,
        verbose=verbose,
        on_retrieved=on_retrieved,
    )
    try:
        stage_started = time.perf_counter()
        result = generate_answer_stream_task(
            prep.prepared.original_query,  # always the user's words (spec_v3 §4.2)
            prep.chunks,
            llm=llm,
            formatted_context=prep.formatted_context,
            on_delta=on_delta,
            should_abort=should_abort,
            on_stats=on_stats,
            verbose=prep.verbose,
        )
        metrics.observe_query_stage("generate", prep.method, time.perf_counter() - stage_started)
    except Exception:
        metrics.count_query(prep.method, "error")
        raise
    metrics.count_query(prep.method, "aborted" if result["aborted"] else "ok")
    usage = result["usage"] or {}
    metrics.count_llm_tokens(usage.get("prompt_tokens"), usage.get("completion_tokens"))
    return StreamedQueryState(
        query=query,
        prepared=prep.prepared,
        query_vector=prep.query_vector,
        retrieved=prep.chunks,
        formatted_context=prep.formatted_context,
        answer=result["answer"],
        reasoning=result["reasoning"],
        aborted=result["aborted"],
        usage=result["usage"],
        tokens_per_second=result["tokens_per_second"],
    )

retrieve_task(retriever, query, *, k, query_vector, verbose)

Task wrapper over retrieval (spec §10.1 steps 3–4).

Parameters:

Name	Type	Description	Default
retriever	Retriever	The retrieval method.	required
query	str	The user's question.	required
k	int	Number of chunks to return.	required
query_vector	list[float] | None	The embed stage's output (reused instead of re-encoding; ignored by bm25).	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
list[RetrievedChunk]	The top-k chunks, best first.

Source code in varagity/pipeline/query_flow.py

@task(name="retrieve", cache_policy=NO_CACHE)
def retrieve_task(
    retriever: Retriever,
    query: str,
    *,
    k: int,
    query_vector: list[float] | None,
    verbose: int,
) -> list[RetrievedChunk]:
    """Task wrapper over retrieval (spec §10.1 steps 3–4).

    Args:
        retriever: The retrieval method.
        query: The user's question.
        k: Number of chunks to return.
        query_vector: The embed stage's output (reused instead of
            re-encoding; ignored by ``bm25``).
        verbose: Validated console verbosity.

    Returns:
        The top-k chunks, best first.
    """
    chunks = retriever.retrieve(query, k=k, verbose=verbose, query_vector=query_vector)
    logger = get_run_logger()
    if chunks:
        logger.info(
            "retrieved %d chunk(s); best: %s (score %.4f)",
            len(chunks),
            chunks[0].metadata.get("file_name", "<unknown>"),
            chunks[0].score,
        )
    else:
        logger.warning("retrieved no chunks — is the corpus ingested?")
    return chunks

varagity.pipeline.graph_flow

Prefect graph flows: the build and the query path (spec_graphrag §5.2, §4.3).

The build is a thin @flow shell over :meth:varagity.graph.service.GraphService.build, for the same reason :mod:varagity.pipeline.eval_flow is one: the work itself belongs to the service (which owns the session, the write lock, and the manifest diff), while Prefect owns the record that a build ran, how long it took, and whether it failed. A multi-day backfill with no run in the UI would be an invisible job. Deliberately one flow run per build attempt, not per document: the engine enqueues documents into its own durable status store and processes them in batches inside a single call, so per-document task runs would have to reach inside the engine's pipeline. Per-document progress is reported instead by the runner's status sampler (:mod:varagity.api.graph_runner), which reads the engine's own document counts — tracking and streaming compose, as they do for ingest.

The query path is :func:varagity.pipeline.query_flow.query_stream_flow with one stage swapped: condense → graph retrieve → stream the answer. The condense and generation stages are the same tasks the chunk path runs, so a graph turn inherits the search/answer query split (the original words always drive the prompt — spec_v3 §4.2), the [SOURCE] citation contract, <think> splitting, client-abort handling, and token accounting without a second implementation of any of them. Only the middle stage differs, because only the middle stage is actually different.

Parameter validation is off (the service is a live handle, not a serializable payload — and the batches ride in :class:GraphBatches, so the flow-run record stores a placeholder rather than megabytes of messages) and there are no Prefect-level retries: an engine that fails mid-backfill must surface, not silently re-run hours of extraction — and a re-called build resumes from the durable statuses anyway. The query path carries none for the query-path reason instead: it is interactive, and the clients already retry transient HTTP failures themselves.

GraphBatches

Opaque carrier keeping a parsed archive out of the flow-run record.

Prefect serializes every flow-call parameter into the flow run it creates (Flow.serialize_parameters), and the server rejects records over 512 KB — a decade-scale archive of parsed messages serializes to megabytes, so passing the batches bare fails the build at flow-run creation, before the engine sees a single document. This box is deliberately not JSON-encodable (__slots__, no mapping or iteration protocol), which lands it on the same "<GraphBatches>" placeholder path the live GraphService handle already rides; the real objects still reach the flow in-process, untouched.

Attributes:

Name	Type	Description
batches	Sequence[MessageBatch]	The parsed source files, exactly as given.

Source code in varagity/pipeline/graph_flow.py

class GraphBatches:
    """Opaque carrier keeping a parsed archive out of the flow-run record.

    Prefect serializes every flow-call parameter into the flow run it
    creates (``Flow.serialize_parameters``), and the server rejects records
    over 512 KB — a decade-scale archive of parsed messages serializes to
    megabytes, so passing the batches bare fails the build at flow-run
    creation, before the engine sees a single document. This box is
    deliberately *not* JSON-encodable (``__slots__``, no mapping or
    iteration protocol), which lands it on the same ``"<GraphBatches>"``
    placeholder path the live ``GraphService`` handle already rides; the
    real objects still reach the flow in-process, untouched.

    Attributes:
        batches: The parsed source files, exactly as given.
    """

    __slots__ = ("batches",)

    batches: Sequence[MessageBatch]

    def __init__(self, batches: Sequence[MessageBatch]) -> None:
        """Box the batches.

        Args:
            batches: Parsed source files bound for the build.
        """
        self.batches = batches

__init__(batches)

Box the batches.

Parameters:

Name	Type	Description	Default
batches	Sequence[MessageBatch]	Parsed source files bound for the build.	required

Source code in varagity/pipeline/graph_flow.py

def __init__(self, batches: Sequence[MessageBatch]) -> None:
    """Box the batches.

    Args:
        batches: Parsed source files bound for the build.
    """
    self.batches = batches

StreamedGraphState

Bases: TypedDict

Outcome of one streamed graph turn (the chunk path's state, graph-shaped).

Attributes:

Name	Type	Description
query	str	The user's question.
prepared	PreparedQuery	The chat engine's search/answer query split (spec_v3 §4.2) — the same split the chunk path makes.
retrieval	GraphRetrieval	What the graph returned (the SSE event's payload and the turn's persisted evidence both come from here).
formatted_context	str	The grounding context the answer was written from.
answer	str	The generated, <think>-stripped answer.
reasoning	str	The captured <think> stream ("" when none).
aborted	bool	True when the client aborted generation mid-stream.
usage	dict[str, int] | None	Server-reported token counts, or None when unreported.
tokens_per_second	float | None	Final decode throughput, or None.

Source code in varagity/pipeline/graph_flow.py

class StreamedGraphState(TypedDict):
    """Outcome of one streamed graph turn (the chunk path's state, graph-shaped).

    Attributes:
        query: The user's question.
        prepared: The chat engine's search/answer query split (spec_v3
            §4.2) — the same split the chunk path makes.
        retrieval: What the graph returned (the SSE event's payload and the
            turn's persisted evidence both come from here).
        formatted_context: The grounding context the answer was written
            from.
        answer: The generated, ``<think>``-stripped answer.
        reasoning: The captured ``<think>`` stream (``""`` when none).
        aborted: ``True`` when the client aborted generation mid-stream.
        usage: Server-reported token counts, or ``None`` when unreported.
        tokens_per_second: Final decode throughput, or ``None``.
    """

    query: str
    prepared: PreparedQuery
    retrieval: GraphRetrieval
    formatted_context: str
    answer: str
    reasoning: str
    aborted: bool
    usage: dict[str, int] | None
    tokens_per_second: float | None

graph_build_flow(service, batches, *, prune_removed=True, verbose=0)

Upsert a parsed message corpus into the graph as a tracked flow run.

Parameters:

Name	Type	Description	Default
service	GraphService	The process's graph service (holds the session and the single-flight write lock).	required
batches	GraphBatches	The boxed parsed source files (bare sequences would serialize into the flow-run record, which the server caps at 512 KB), guid-merged by the session before rendering.	required
prune_removed	bool	Whether batches render the whole corpus. A bounded build passes False — its render is partial on purpose, and pruning on its say-so would delete the rest of the archive.	True
verbose	int	Validated console verbosity (0–2).	0

Returns:

Type	Description
BuildReport	What the build did (messages seen, wall clock, caught failures).

Source code in varagity/pipeline/graph_flow.py

@flow(name="graph-build", validate_parameters=False)
def graph_build_flow(
    service: GraphService,
    batches: GraphBatches,
    *,
    prune_removed: bool = True,
    verbose: int = 0,
) -> BuildReport:
    """Upsert a parsed message corpus into the graph as a tracked flow run.

    Args:
        service: The process's graph service (holds the session and the
            single-flight write lock).
        batches: The boxed parsed source files (bare sequences would
            serialize into the flow-run record, which the server caps at
            512 KB), guid-merged by the session before rendering.
        prune_removed: Whether ``batches`` render the whole corpus. A bounded
            build passes ``False`` — its render is partial on purpose, and
            pruning on its say-so would delete the rest of the archive.
        verbose: Validated console verbosity (0–2).

    Returns:
        What the build did (messages seen, wall clock, caught failures).
    """
    logger = get_run_logger()
    source_files = batches.batches
    logger.info(
        "graph build starting: %d source file(s), %s render",
        len(source_files),
        "full-corpus" if prune_removed else "bounded",
    )
    report = service.build(source_files, verbose=verbose, prune_removed=prune_removed)
    logger.info(
        "graph build finished: %d message(s) in %.1fs, %d failure(s)",
        report.messages_seen,
        report.wall_clock_s,
        len(report.failures),
    )
    return report

graph_query_stream_flow(query, service, *, history=(), engine=None, llm=None, mode=None, verbose=None, on_retrieved=None, on_delta, should_abort=None, on_stats=None)

Answer one question from the graph, streaming generation deltas.

:func:varagity.pipeline.query_flow.query_stream_flow's graph twin, backing POST /api/chat with corpus="graph": condense (the same task, so a follow-up is rewritten into a standalone search query exactly as it is for chunk RAG) → graph retrieve → on_retrieved (the SSE retrieval event: evidence before prose) → the same streaming generate task over a graph-shaped context.

Generation is deliberately the repo's own, not the engine's (ADR-017's retrieval-only decision): the answer is grounded in the transcript days the graph cited, cites them by [SOURCE], and streams.

Parameters:

Name	Type	Description	Default
query	str	The user's question.	required
service	GraphService	The process's graph service.	required
history	Sequence[Turn]	Prior conversation turns, oldest first; the chat engine reads them when preparing the search query.	()
engine	ChatEngine | None	Chat engine preparing the search query; resolved from settings.CHAT_ENGINE when omitted.	None
llm	LLMClient | None	Chat client; resolved via the model registry when omitted.	None
mode	str | None	Engine query mode; defaults to settings.GRAPH_QUERY_MODE.	None
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None
on_retrieved	Callable[[GraphRetrieval], None] | None	Optional hook called with the graph retrieval before generation starts.	None
on_delta	Callable[[Kind, str], None]	Called with each classified (kind, text) fragment.	required
should_abort	Callable[[], bool] | None	Polled between deltas; True stops generation early and marks the state aborted.	None
on_stats	Callable[[GenerationTimings], None] | None	Called with llama.cpp's cumulative decode counters as they arrive.	None

Returns:

Type	Description
StreamedGraphState	The completed :class:StreamedGraphState.

Raises:

Type	Description
ValueError	If verbose is invalid.
KeyError	If settings.CHAT_ENGINE names an unregistered engine.
GraphUnavailable	If the engine's session cannot be opened (the route resolves this before streaming, so reaching it here means the session died mid-turn).

Source code in varagity/pipeline/graph_flow.py

@flow(name="graph-query-stream", validate_parameters=False)
def graph_query_stream_flow(
    query: str,
    service: GraphService,
    *,
    history: Sequence[Turn] = (),
    engine: ChatEngine | None = None,
    llm: LLMClient | None = None,
    mode: str | None = None,
    verbose: int | None = None,
    on_retrieved: Callable[[GraphRetrieval], None] | None = None,
    on_delta: Callable[[Kind, str], None],
    should_abort: Callable[[], bool] | None = None,
    on_stats: Callable[[GenerationTimings], None] | None = None,
) -> StreamedGraphState:
    """Answer one question from the graph, streaming generation deltas.

    :func:`varagity.pipeline.query_flow.query_stream_flow`'s graph twin,
    backing ``POST /api/chat`` with ``corpus="graph"``: condense (the same
    task, so a follow-up is rewritten into a standalone search query exactly
    as it is for chunk RAG) → graph retrieve → ``on_retrieved`` (the SSE
    ``retrieval`` event: evidence before prose) → the same streaming
    generate task over a graph-shaped context.

    Generation is deliberately the repo's own, not the engine's (ADR-017's
    retrieval-only decision): the answer is grounded in the transcript days
    the graph cited, cites them by ``[SOURCE]``, and streams.

    Args:
        query: The user's question.
        service: The process's graph service.
        history: Prior conversation turns, oldest first; the chat engine
            reads them when preparing the search query.
        engine: Chat engine preparing the search query; resolved from
            ``settings.CHAT_ENGINE`` when omitted.
        llm: Chat client; resolved via the model registry when omitted.
        mode: Engine query mode; defaults to ``settings.GRAPH_QUERY_MODE``.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.
        on_retrieved: Optional hook called with the graph retrieval before
            generation starts.
        on_delta: Called with each classified ``(kind, text)`` fragment.
        should_abort: Polled between deltas; ``True`` stops generation early
            and marks the state ``aborted``.
        on_stats: Called with llama.cpp's cumulative decode counters as they
            arrive.

    Returns:
        The completed :class:`StreamedGraphState`.

    Raises:
        ValueError: If ``verbose`` is invalid.
        KeyError: If ``settings.CHAT_ENGINE`` names an unregistered engine.
        varagity.graph.service.GraphUnavailable: If the engine's session
            cannot be opened (the route resolves this *before* streaming, so
            reaching it here means the session died mid-turn).
    """
    from varagity.chat import get_chat_engine  # deferred: keeps the import graph acyclic

    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    active_engine = engine if engine is not None else get_chat_engine(settings.CHAT_ENGINE)
    prepared = condense_query_task(active_engine, query, history=history, llm=llm, verbose=verbose)
    retrieval = graph_retrieve_task(
        service,
        prepared.search_query,
        mode=mode if mode is not None else settings.GRAPH_QUERY_MODE,
        verbose=verbose,
    )
    if on_retrieved is not None:
        on_retrieved(retrieval)
    formatted_context = graph_answer_context(
        retrieval,
        max_chars=answer_context_max_chars(settings.LLM_CONTEXT_TOKENS, settings.MAX_TOKENS),
    )
    result = generate_answer_stream_task(
        prepared.original_query,  # always the user's words (spec_v3 §4.2)
        [],  # the grounding is the graph's, pre-rendered above
        llm=llm,
        formatted_context=formatted_context,
        on_delta=on_delta,
        should_abort=should_abort,
        on_stats=on_stats,
        verbose=verbose,
    )
    return StreamedGraphState(
        query=query,
        prepared=prepared,
        retrieval=retrieval,
        formatted_context=formatted_context,
        answer=result["answer"],
        reasoning=result["reasoning"],
        aborted=result["aborted"],
        usage=result["usage"],
        tokens_per_second=result["tokens_per_second"],
    )

graph_retrieve_task(service, query, *, mode, verbose)

Task wrapper over one graph retrieval (spec_graphrag §4.3).

The chunk path's embed+retrieve stages collapse into this one: the engine owns its own keyword extraction, embedding, and fusion inside a single call, and splitting the wrapper would only invent task boundaries the engine does not actually have.

Parameters:

Name	Type	Description	Default
service	GraphService	The process's graph service (reads take no build lock, so this is answered during a backfill).	required
query	str	The search query — the chat engine's search_query.	required
mode	str	The engine query mode (GRAPH_QUERY_MODE).	required
verbose	int	Validated console verbosity.	required

Returns:

Type	Description
GraphRetrieval	The evidence and transcript passages the answer will be grounded in.

Source code in varagity/pipeline/graph_flow.py

@task(name="graph_retrieve", cache_policy=NO_CACHE)
def graph_retrieve_task(
    service: GraphService, query: str, *, mode: str, verbose: int
) -> GraphRetrieval:
    """Task wrapper over one graph retrieval (spec_graphrag §4.3).

    The chunk path's embed+retrieve stages collapse into this one: the
    engine owns its own keyword extraction, embedding, and fusion inside a
    single call, and splitting the wrapper would only invent task boundaries
    the engine does not actually have.

    Args:
        service: The process's graph service (reads take no build lock, so
            this is answered during a backfill).
        query: The search query — the chat engine's ``search_query``.
        mode: The engine query mode (``GRAPH_QUERY_MODE``).
        verbose: Validated console verbosity.

    Returns:
        The evidence and transcript passages the answer will be grounded in.
    """
    retrieval = service.retrieve(query, mode=mode, verbose=verbose)
    logger = get_run_logger()
    if retrieval.excerpts or retrieval.evidence.relations:
        logger.info(
            "graph retrieved %d entities, %d relations, %d transcript(s) (mode %s)",
            len(retrieval.evidence.entities),
            len(retrieval.evidence.relations),
            len(retrieval.excerpts),
            retrieval.mode,
        )
    else:
        logger.warning("the graph returned no evidence — is the corpus built?")
    return retrieval

varagity.pipeline.backup_flow

Prefect backup flows: export and restore, for both corpora (ADR-019).

Thin @flow shells over :func:varagity.backup.rag.export_rag, :func:varagity.backup.rag.restore_rag, :func:varagity.backup.graph.export_graph and :func:varagity.backup.graph.restore_graph, for the reason the graph build is one: the work belongs to the backup package (one corpus snapshot and one all-or-nothing bundle write; one replacing transaction and one index rebuild; one copy of the graph's workdir under its write lock; one extract and one swap of it), while Prefect owns the record that a backup ran, how long it took, and whether it failed. One flow run per operation — not a task per member or per stage: an export is a single consistent read and a restore a single transaction (or a single swap), and slicing either into task runs would invent boundaries they do not have. Per-stage progress travels as the API runner's progress events instead (tracking and streaming compose, as they do for ingest and the graph build).

No Prefect-level retries: a failed export leaves no bundle behind (the partial is removed) and a failed restore leaves the previous corpus (its transaction rolled back, or its previous workdir put back) — or, failing in the BM25 rebuild after the commit, says so — so the honest response is to surface the failure and let the user press the button again, not to repeat the work silently. Parameter validation is off because the stores (and the graph service) are live handles, not serializable payloads.

graph_export_flow(dest, *, on_progress=None, service=None, stale=None)

Export the message graph's working directory into a new bundle as a tracked flow run.

The copy runs under the graph service's write lock, so no build can change a file mid-copy while graph chat keeps answering. Needs no model service and opens no session — only postgres, for the stale flag.

Parameters:

Name	Type	Description	Default
dest	Path	The exports folder (<GRAPH_DOCS_PATH>/varagity-exports).	required
on_progress	GraphExportProgressCallback | None	Called with (stage, current, total) per stage (snapshot → pack, counting bytes → finalize).	None
service	GraphService | None	The graph service whose workdir to copy; the process's own (the graph's single writer) when omitted.	None
stale	bool | None	The graph-stale flag to record; when omitted, it is read from app_settings once the files are copied, so a source deleted mid-export is recorded.	None

Returns:

Type	Description
GraphExportReport	The published bundle and how long the export took.

Source code in varagity/pipeline/backup_flow.py

@flow(name="graph-export", validate_parameters=False)
def graph_export_flow(
    dest: Path,
    *,
    on_progress: GraphExportProgressCallback | None = None,
    service: GraphService | None = None,
    stale: bool | None = None,
) -> GraphExportReport:
    """Export the message graph's working directory into a new bundle as a tracked flow run.

    The copy runs under the graph service's write lock, so no build can
    change a file mid-copy while graph chat keeps answering. Needs no model
    service and opens no session — only postgres, for the stale flag.

    Args:
        dest: The exports folder (``<GRAPH_DOCS_PATH>/varagity-exports``).
        on_progress: Called with ``(stage, current, total)`` per stage
            (``snapshot`` → ``pack``, counting bytes → ``finalize``).
        service: The graph service whose workdir to copy; the process's own
            (the graph's single writer) when omitted.
        stale: The graph-stale flag to record; when omitted, it is read from
            ``app_settings`` once the files are copied, so a source deleted
            mid-export is recorded.

    Returns:
        The published bundle and how long the export took.
    """
    logger = get_run_logger()

    def read_stale() -> bool:
        if stale is not None:
            return stale
        with AppSettingsStore() as settings_store:
            return settings_store.is_graph_stale()

    active = service if service is not None else get_graph_service()
    result = export_graph(active, dest, read_stale=read_stale, on_progress=on_progress)
    counts = result.counts
    logger.info(
        "graph export wrote %s: %d file(s), %d thread-day(s), %d bytes in %.1fs",
        result.bundle.name,
        result.files,
        counts.get("thread_days", 0),
        result.bundle.size_bytes,
        result.wall_clock_s,
    )
    return result

graph_restore_flow(bundle, *, run_id, on_progress=None, service=None, settings_store=None)

Restore the message graph from a bundle as a tracked flow run.

Extracts the bundle's workdir beside the live one and swaps it in under the graph service's write and open locks — the previous graph comes back if the restored one won't open — then, and only once all of that succeeded, writes the graph-stale flag the bundle recorded. Needs no model service: only postgres, for the flag.

Parameters:

Name	Type	Description	Default
bundle	Path	The bundle file (inside <GRAPH_DOCS_PATH>/varagity-exports).	required
run_id	str	The restore run's id (names the staging and previous folders beside the workdir).	required
on_progress	GraphRestoreProgressCallback | None	Called with (stage, current, total) per stage (validate → extract, counting bytes → swap → open).	None
service	GraphService | None	The graph service whose workdir to replace; the process's own (the graph's single writer) when omitted.	None
settings_store	AppSettingsStore | None	Where the stale flag is written; a connection of its own is opened (and closed) when omitted.	None

Returns:

Type	Description
GraphRestore	What was restored, with the stale flag it left.

Source code in varagity/pipeline/backup_flow.py

@flow(name="graph-restore", validate_parameters=False)
def graph_restore_flow(
    bundle: Path,
    *,
    run_id: str,
    on_progress: GraphRestoreProgressCallback | None = None,
    service: GraphService | None = None,
    settings_store: AppSettingsStore | None = None,
) -> GraphRestore:
    """Restore the message graph from a bundle as a tracked flow run.

    Extracts the bundle's workdir beside the live one and swaps it in under
    the graph service's write and open locks — the previous graph comes back
    if the restored one won't open — then, and only once all of that
    succeeded, writes the graph-stale flag the bundle recorded. Needs no
    model service: only postgres, for the flag.

    Args:
        bundle: The bundle file (inside ``<GRAPH_DOCS_PATH>/varagity-exports``).
        run_id: The restore run's id (names the staging and previous folders
            beside the workdir).
        on_progress: Called with ``(stage, current, total)`` per stage
            (``validate`` → ``extract``, counting bytes → ``swap`` →
            ``open``).
        service: The graph service whose workdir to replace; the process's
            own (the graph's single writer) when omitted.
        settings_store: Where the stale flag is written; a connection of its
            own is opened (and closed) when omitted.

    Returns:
        What was restored, with the stale flag it left.
    """
    logger = get_run_logger()
    active = service if service is not None else get_graph_service()
    result = restore_graph(active, bundle, run_id=run_id, on_progress=on_progress)
    # The bundle's own flag: the graph is exactly what was exported, so it is
    # stale exactly when it was then (a source deleted before the export).
    stale = result.manifest.stale
    if settings_store is not None:
        settings_store.set_graph_stale(stale)
    else:
        with AppSettingsStore() as owned_settings:
            owned_settings.set_graph_stale(stale)
    logger.info(
        "graph restore loaded %s: %d file(s), %d thread-day(s) in %.1fs%s",
        result.bundle_name,
        result.files,
        result.counts.get("thread_days", 0),
        result.wall_clock_s,
        " — the graph is flagged stale" if stale else "",
    )
    return replace(result, stale=stale)

rag_export_flow(dest, *, ingest_settings, on_progress=None, store=None, stale=None)

Export the chunk corpus into a new bundle as a tracked flow run.

Parameters:

Name	Type	Description	Default
dest	Path	The exports folder (<DOCS_PATH>/varagity-exports).	required
ingest_settings	dict[str, str]	The reingest-affecting settings at export, in env-string form (the runner reads them before the stale flag, so a settings change racing the export can only make the bundle more conservative about staleness, never less).	required
on_progress	ProgressCallback | None	Called with (stage, current, total) per stage.	None
store	ContextualVectorDB | None	The vector store to snapshot; a connection of its own is opened (and closed) when omitted.	None
stale	bool | None	The corpus-stale flag at export; read from app_settings when omitted.	None

Returns:

Type	Description
RagExport	The published bundle and how long the export took.

Source code in varagity/pipeline/backup_flow.py

@flow(name="rag-export", validate_parameters=False)
def rag_export_flow(
    dest: Path,
    *,
    ingest_settings: dict[str, str],
    on_progress: ProgressCallback | None = None,
    store: ContextualVectorDB | None = None,
    stale: bool | None = None,
) -> RagExport:
    """Export the chunk corpus into a new bundle as a tracked flow run.

    Args:
        dest: The exports folder (``<DOCS_PATH>/varagity-exports``).
        ingest_settings: The reingest-affecting settings at export, in
            env-string form (the runner reads them before the stale flag,
            so a settings change racing the export can only make the bundle
            *more* conservative about staleness, never less).
        on_progress: Called with ``(stage, current, total)`` per stage.
        store: The vector store to snapshot; a connection of its own is
            opened (and closed) when omitted.
        stale: The corpus-stale flag at export; read from ``app_settings``
            when omitted.

    Returns:
        The published bundle and how long the export took.
    """
    logger = get_run_logger()
    if stale is None:
        with AppSettingsStore() as settings_store:
            stale = settings_store.is_corpus_stale()
    if store is not None:
        result = export_rag(dest, store, ingest_settings, stale=stale, on_progress=on_progress)
    else:
        with ContextualVectorDB() as owned:
            result = export_rag(dest, owned, ingest_settings, stale=stale, on_progress=on_progress)
    counts = result.counts
    logger.info(
        "rag export wrote %s: %d document(s), %d chunk(s), %d bytes in %.1fs",
        result.bundle.name,
        counts.get("documents", 0),
        counts.get("chunks", 0),
        result.bundle.size_bytes,
        result.wall_clock_s,
    )
    return result

rag_restore_flow(bundle, *, ingest_settings, on_progress=None, store=None, bm25=None, settings_store=None)

Restore the chunk corpus from a bundle as a tracked flow run.

Replaces both stores' corpus with the bundle's rows (re-rooted onto this stack's DOCS_PATH) and rebuilds the BM25 index from them, then writes the corpus-stale flag the bundle implies: stale if it was stale at export, or if its rows were made under other reingest-affecting settings than this stack's. The flag follows the rows: it is written once they committed — after a complete restore, and also when only the BM25 rebuild failed (the rows are the bundle's by then, so the previous corpus's flag would misdescribe them) — and never after a restore that rolled back.

Parameters:

Name	Type	Description	Default
bundle	Path	The bundle file (inside <DOCS_PATH>/varagity-exports).	required
ingest_settings	Callable[[], Mapping[str, str]]	Provider of this stack's reingest-affecting settings, in env-string form — called when the stale flag is written, so a settings change made during the restore is judged against the restored rows too.	required
on_progress	RestoreProgressCallback | None	Called with (stage, current, total) per stage (validate → load → index → bm25).	None
store	ContextualVectorDB | None	The vector store to replace; a connection of its own is opened (and closed) when omitted.	None
bm25	ElasticsearchBM25 | None	The BM25 store to rebuild; one is constructed (and closed) when omitted.	None
settings_store	AppSettingsStore | None	Where the stale flag is written; a connection of its own is opened (and closed) when omitted.	None

Returns:

Type	Description
RagRestore	What was restored, with the stale flag it left.

Source code in varagity/pipeline/backup_flow.py

@flow(name="rag-restore", validate_parameters=False)
def rag_restore_flow(
    bundle: Path,
    *,
    ingest_settings: Callable[[], Mapping[str, str]],
    on_progress: RestoreProgressCallback | None = None,
    store: ContextualVectorDB | None = None,
    bm25: ElasticsearchBM25 | None = None,
    settings_store: AppSettingsStore | None = None,
) -> RagRestore:
    """Restore the chunk corpus from a bundle as a tracked flow run.

    Replaces both stores' corpus with the bundle's rows (re-rooted onto this
    stack's ``DOCS_PATH``) and rebuilds the BM25 index from them, then
    writes the corpus-stale flag the bundle implies: stale if it was stale
    at export, or if its rows were made under other reingest-affecting
    settings than this stack's. The flag follows the rows: it is written
    once they committed — after a complete restore, and also when only the
    BM25 rebuild failed (the rows are the bundle's by then, so the previous
    corpus's flag would misdescribe them) — and never after a restore that
    rolled back.

    Args:
        bundle: The bundle file (inside ``<DOCS_PATH>/varagity-exports``).
        ingest_settings: Provider of this stack's reingest-affecting
            settings, in env-string form — called when the stale flag is
            written, so a settings change made *during* the restore is
            judged against the restored rows too.
        on_progress: Called with ``(stage, current, total)`` per stage
            (``validate`` → ``load`` → ``index`` → ``bm25``).
        store: The vector store to replace; a connection of its own is
            opened (and closed) when omitted.
        bm25: The BM25 store to rebuild; one is constructed (and closed)
            when omitted.
        settings_store: Where the stale flag is written; a connection of its
            own is opened (and closed) when omitted.

    Returns:
        What was restored, with the stale flag it left.
    """
    logger = get_run_logger()
    docs_root = Path(get_settings().DOCS_PATH)

    def write_stale(manifest: BundleManifest) -> bool:
        stale = derive_corpus_stale(manifest, ingest_settings())
        if settings_store is not None:
            settings_store.set_corpus_stale(stale)
        else:
            with AppSettingsStore() as owned_settings:
                owned_settings.set_corpus_stale(stale)
        return stale

    try:
        with ExitStack() as stack:
            active_store = store if store is not None else stack.enter_context(ContextualVectorDB())
            active_bm25 = bm25 if bm25 is not None else stack.enter_context(ElasticsearchBM25())
            result = restore_rag(
                bundle, active_store, active_bm25, docs_root, on_progress=on_progress
            )
    except BM25RebuildFailed as error:
        # The rows committed before the rebuild failed: they are the bundle's
        # now, so the flag must describe them too — then the run still fails.
        if error.manifest is not None:
            try:
                stale = write_stale(error.manifest)
            except Exception:  # the rebuild's failure stays the run's answer
                logger.exception("rag restore could not write the corpus-stale flag")
            else:
                logger.warning(
                    "rag restore committed the rows of %s but not the BM25 index%s",
                    bundle.name,
                    " — the corpus is flagged stale" if stale else "",
                )
        raise
    stale = write_stale(result.manifest)
    logger.info(
        "rag restore loaded %s: %d document(s), %d chunk(s), %d in BM25 in %.1fs%s",
        result.bundle_name,
        result.documents,
        result.chunks,
        result.bm25_indexed,
        result.wall_clock_s,
        " — the corpus is flagged stale" if stale else "",
    )
    return replace(result, stale=stale)

varagity.pipeline.eval_flow

Prefect evaluation flows: tracked runs of the spec §16 harness.

Thin @flow shells over :func:varagity.eval.evaluate.run_matrix, :func:varagity.eval.evaluate.run_chat_eval, :func:varagity.eval.ocr_benchmark.run_ocr_benchmark and :func:varagity.eval.graph_eval.run_graph_eval, passing the tracked :func:~varagity.pipeline.ingest_flow.ingest_flow through the harness's ingest seam — every eval ingest (two for the matrix, one each for the chat eval and per OCR engine) is a subflow with per-stage task runs at the Prefect UI, and the eval run itself is one flow run. The graph bake-off has no ingest seam to thread: graph engines self-store, so its flow is a pure tracking shell.

Like the other flows, parameter validation is off (duck-typed internals) and the measurement work carries no Prefect-level retries: an unreachable GPU service or Docker daemon should fail the run loudly, not back off — the clients already retry transient HTTP failures internally.

chat_eval_flow(verbose=None)

Run the multi-turn chat-engine eval as a tracked flow run.

Parameters:

Name	Type	Description	Default
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (see
dict[str, Any]	func:varagity.eval.evaluate.run_chat_eval).

Source code in varagity/pipeline/eval_flow.py

@flow(name="eval-chat", validate_parameters=False)
def chat_eval_flow(verbose: int | None = None) -> dict[str, Any]:
    """Run the multi-turn chat-engine eval as a tracked flow run.

    Args:
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (see
        :func:`varagity.eval.evaluate.run_chat_eval`).
    """
    return run_chat_eval(ingest=ingest_flow, verbose=verbose)

eval_flow(verbose=None)

Run the 7-configuration retrieval matrix as a tracked flow run.

Parameters:

Name	Type	Description	Default
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (see
dict[str, Any]	func:varagity.eval.evaluate.run_matrix).

Source code in varagity/pipeline/eval_flow.py

@flow(name="eval-matrix", validate_parameters=False)
def eval_flow(verbose: int | None = None) -> dict[str, Any]:
    """Run the 7-configuration retrieval matrix as a tracked flow run.

    Args:
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (see
        :func:`varagity.eval.evaluate.run_matrix`).
    """
    return run_matrix(ingest=ingest_flow, verbose=verbose)

graph_eval_flow(profile='smoke', engines=None, mode=None, skip_build=False, message_target=None, verbose=None)

Run the graph bake-off as a tracked flow run (spec_graphrag §12).

Parameters:

Name	Type	Description	Default
profile	str	Fixture corpus profile (smoke or full).	'smoke'
engines	Sequence[str] | None	Engine names to run; None runs every registered engine.	None
mode	str | None	Engine query mode override for every question.	None
skip_build	bool	Reuse the existing corpus and working directories instead of re-indexing.	False
message_target	int | None	Cap the corpus at this many messages instead of the profile's own size (a capped run gets its own corpus and working directories); None uses the profile's size.	None
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (see
dict[str, Any]	func:varagity.eval.graph_eval.run_graph_eval).

Source code in varagity/pipeline/eval_flow.py

@flow(name="eval-graph", validate_parameters=False)
def graph_eval_flow(
    profile: str = "smoke",
    engines: Sequence[str] | None = None,
    mode: str | None = None,
    skip_build: bool = False,
    message_target: int | None = None,
    verbose: int | None = None,
) -> dict[str, Any]:
    """Run the graph bake-off as a tracked flow run (spec_graphrag §12).

    Args:
        profile: Fixture corpus profile (``smoke`` or ``full``).
        engines: Engine names to run; ``None`` runs every registered engine.
        mode: Engine query mode override for every question.
        skip_build: Reuse the existing corpus and working directories
            instead of re-indexing.
        message_target: Cap the corpus at this many messages instead of the
            profile's own size (a capped run gets its own corpus and working
            directories); ``None`` uses the profile's size.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (see
        :func:`varagity.eval.graph_eval.run_graph_eval`).
    """
    return run_graph_eval(
        profile=profile,
        engines=engines,
        mode=mode,
        skip_build=skip_build,
        message_target=message_target,
        verbose=verbose,
    )

ocr_benchmark_flow(verbose=None)

Run the OCR engine benchmark as a tracked flow run.

Parameters:

Name	Type	Description	Default
verbose	int | None	Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.	None

Returns:

Type	Description
dict[str, Any]	The results document (see
dict[str, Any]	func:varagity.eval.ocr_benchmark.run_ocr_benchmark).

Source code in varagity/pipeline/eval_flow.py

@flow(name="eval-ocr", validate_parameters=False)
def ocr_benchmark_flow(verbose: int | None = None) -> dict[str, Any]:
    """Run the OCR engine benchmark as a tracked flow run.

    Args:
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The results document (see
        :func:`varagity.eval.ocr_benchmark.run_ocr_benchmark`).
    """
    return run_ocr_benchmark(ingest=ingest_flow, verbose=verbose)