Façades
The four entry points: ingest, query, evaluate, and the assembler.
Four entry points. The first three are the primitives; RagSage assembles them.
Ingestion
The ingestion façade — one document in, retrievable chunks out.
IngestionPipeline is the whole write-path of the engine expressed as a
single public method, ingest(). It orchestrates the
ports and owns the policy between them — dedup, per-page routing,
contextualisation, and the fan-out to the three stores — while every step that
touches a model, a file, or a database lives behind a port it was handed.
This is the primary write seam: a Seam-1 test wires fakes for each port and
asserts on the returned IngestionResult and the resulting store state,
never on anything private here.
class ragsage.ingestion.IngestionResult(document: Document, chunk_count: int, route_counts: dict[~ragsage.models.PageRoute, int]=, deduplicated: bool = False)
Bases: object
What ingest reports back: identity, size, and how pages were read.
deduplicated is True when the content hash already existed and
nothing was reprocessed. route_counts records how many pages took each
parser route, which is what makes per-page routing observable to a test
without reaching inside the pipeline.
document : Document
chunk_count : int
route_counts : dict[PageRoute, int]
deduplicated : bool = False
class ragsage.ingestion.IngestionPipeline(*, parser: DocumentParser, classifier: PageClassifier, chunker: Chunker, contextualizer: Contextualizer, embedder: Embedder, vector_store: VectorStore, lexical_store: LexicalStore, document_store: DocumentStore, llm: LLMClient, cache: Cache, tracer: Tracer | None = None)
Bases: object
Ingests documents into a scope’s stores through injected adapters.
async ingest(source: RawSource, scope: Scope, config: IngestionConfig | None = None) → IngestionResult
Parse, route, chunk, contextualise, embed, and store one document.
Returns immediately with a dedup result if the same content already
lives in scope — a failed or repeated upload never reprocesses.
Query
The query façade — a question in, a grounded, cited answer out.
QueryEngine is the whole read-path: embed the question, retrieve
densely and lexically, fuse the two rankings, rerank, assemble a bounded
context, and have the model answer only from it. It owns the retrieval
policy; the models and stores it uses are all injected ports.
It answers two ways over one implementation: stream() yields the answer
as it is produced — tokens, then resolved citations, then usage, then a
terminal complete event — for a backend to relay over SSE; query()
drains that same stream into a single Answer. Building one on the
other is deliberate, so the streamed and buffered answers can never diverge in
prompt, citations, or not-found behaviour.
Four behaviours here are load-bearing and tested through query() and
stream():
- Hybrid fusion — dense and lexical results combine by Reciprocal Rank Fusion, so a chunk strong in either channel surfaces.
- Citation binding — each context chunk gets a stable marker, and the answer’s citations are exactly the markers the model used.
- Honest not-found — when retrieval is empty (or below
min_score), the engine returns an ungrounded not-found answer without inventing one. - Small talk short-circuit — a message that was never a question about the
corpus (a greeting, a thanks, “what can you do?”) is answered conversationally
without searching, because reporting a miss on a greeting is a lie about what
the engine did. See
ragsage.smalltalk; the guarantee above is untouched for every message that gate doesn’t recognise.
ragsage.query.NOT_FOUND_MESSAGE = "I couldn't find an answer to that in your documents."
Returned verbatim when the corpus can’t support an answer.
ragsage.query.reciprocal_rank_fusion(*ranked_lists: list[ScoredChunk]) → list[ScoredChunk]
Fuse several ranked lists into one by Reciprocal Rank Fusion.
A chunk’s fused score is sum(1 / (K + rank)) over the lists it appears
in (rank 0-based). Chunks are keyed by id, so the same chunk retrieved by
both channels reinforces rather than duplicates. Pure and deterministic —
the fusion logic is unit-testable without any adapter.
class ragsage.query.QueryEngine(*, embedder: Embedder, vector_store: VectorStore, lexical_store: LexicalStore, reranker: Reranker, llm: LLMClient, rewriter: QueryRewriter | None = None, tracer: Tracer | None = None)
Bases: object
Answers questions against a scope’s stores through injected adapters.
async query(question: str, scope: Scope, options: QueryOptions | None = None, *, history: Sequence[Turn] = ()) → Answer
Retrieve, rerank, and generate a grounded Answer.
A thin buffer over stream(): it drains the same event stream and
returns its terminal AnswerComplete as an Answer, so
the buffered result is byte-for-byte the streamed one. history carries
prior conversation turns so a follow-up resolves against them.
async stream(question: str, scope: Scope, options: QueryOptions | None = None, *, history: Sequence[Turn] = ()) → AsyncIterator[AnswerToken | Citation | Usage | AnswerComplete]
Stream a grounded answer for question as it is produced.
Yields, in order: an AnswerToken per model fragment, then one
Citation per resolved marker, then a Usage, then a
terminal AnswerComplete. When retrieval can’t support an answer
the not-found message is still streamed as tokens and the run completes
ungrounded with no citations — so a consumer needs no special case.
A message that isn’t a question about the corpus at all short-circuits
before any of that: is_small_talk() recognises
greetings, acknowledgements and “what can you do?”, and
_converse() answers them conversationally without embedding,
retrieving, or rewriting. That check runs on the raw message, ahead of
the rewriter — condensing “thanks” against a document thread would
manufacture a question the user never asked.
“Insufficient” is judged twice: retrieval may return nothing (below
min_score), and the model may decline even on retrieved context by
replying with the not-found message. Both resolve to grounded=False
with no citations — a nearest-neighbour store almost always returns
something, so the model’s own refusal is the second, load-bearing gate.
When history is non-empty the question is first condensed into a
standalone one (so “what about its pricing?” becomes something retrieval
can match); that resolved question drives both retrieval and generation.
The whole run is traced under one tenant-tagged root span, with a child span per stage — rewrite, retrieve, rerank, prompt, generate — each carrying what it produced. That is what lets a wrong answer be tracked to the stage that failed: whether the right chunk was never retrieved, was dropped in rerank, never reached the prompt, or was in context yet the model still answered wrong.
Evaluation
The evaluation façade — measure answer quality on a fixed golden set.
Evaluator runs a labelled dataset through a QueryEngine
and scores the answers it gets back on the four standard RAG metrics —
faithfulness, answer relevancy, context precision, and context
recall — alongside the answer/citation correctness backbone the Seam-1 tests
rely on. It is deliberately a thin, offline harness over the public query seam:
the same deterministic-fakes setup the tests use, so quality can be tracked
against a golden set with no network and gated in CI against agreed thresholds.
The metrics here are reference-based and deterministic — computed from the
answer, its citations, and the golden labels by content-token overlap, not by
an LLM judge. That is a deliberate fit for this stack: the whole engine runs
offline against fakes in CI (no API keys), so a judge-model RAGAS/deepeval run
would not run there. Swapping in an LLM-judged scorer for a richer, subjective
read is a follow-up that plugs in behind this same Evaluator seam; what
lives here is the gate that can run on every push.
class ragsage.evaluation.EvalExample(question: str, expected_answer: str | None = None, ground_truth_answer: str | None = None, relevant_document_ids: frozenset[str] = )
Bases: object
One labelled question: what to ask and what a right answer looks like.
expected_answer is a substring the answer should contain (case-folded).
ground_truth_answer is the reference answer relevancy and faithfulness
score against — for an out-of-corpus question set it to the engine’s
not-found message, so an honest refusal scores as the relevant response.
relevant_document_ids are the documents a good answer should draw on
(context precision/recall). Any may be omitted to score only the others.
question : str
expected_answer : str | None = None
ground_truth_answer : str | None = None
relevant_document_ids : frozenset[str]
class ragsage.evaluation.EvalThresholds(faithfulness: float = 0.7, answer_relevancy: float = 0.6, context_precision: float = 0.7, context_recall: float = 0.7)
Bases: object
The agreed metric floors an eval run is gated against in CI.
Defaults are conservative pass marks for the deterministic golden set; a
deploy can tighten them. EvalReport.check() compares aggregate means
against these.
faithfulness : float = 0.7
answer_relevancy : float = 0.6
context_precision : float = 0.7
context_recall : float = 0.7
class ragsage.evaluation.EvalCaseResult(example: EvalExample, answer: Answer, answer_matched: bool, faithfulness: float, answer_relevancy: float, context_precision: float, context_recall: float)
Bases: object
The scored outcome for a single example.
example : EvalExample
answer : Answer
answer_matched : bool
faithfulness : float
answer_relevancy : float
context_precision : float
context_recall : float
property citation_precision : float
property citation_recall : float
class ragsage.evaluation.ThresholdCheck(passed: bool, failures: tuple[str, ...])
Bases: object
Whether an EvalReport cleared its thresholds, and what didn’t.
passed : bool
failures : tuple[str, ...]
class ragsage.evaluation.EvalReport(results: Sequence[EvalCaseResult], answer_accuracy: float, mean_faithfulness: float, mean_answer_relevancy: float, mean_context_precision: float, mean_context_recall: float, grounded_rate: float)
Bases: object
Aggregate scores across the dataset, plus the per-case breakdown.
results : Sequence[EvalCaseResult]
answer_accuracy : float
mean_faithfulness : float
mean_answer_relevancy : float
mean_context_precision : float
mean_context_recall : float
grounded_rate : float
property mean_citation_precision : float
property mean_citation_recall : float
check(thresholds: EvalThresholds) → ThresholdCheck
Compare the aggregate means against thresholds.
Returns which metrics fell short (empty when all clear), so CI can print the offenders and exit non-zero on any miss.
class ragsage.evaluation.Evaluator(engine: QueryEngine, scope: Scope, *, options: QueryOptions | None = None)
Bases: object
Scores a QueryEngine against a labelled dataset for a scope.
async evaluate(dataset: Sequence[EvalExample]) → EvalReport
Run every example through the engine and aggregate the scores.
A built-in golden set and a runnable offline eval, shipped in the library.
The eval harness needs something to score. This module carries a small,
hand-labelled corpus plus the questions a correct pipeline must answer (and one
it must decline), so ragsage eval — and the CI test — run the full
ingest-then-score loop against the in-memory ragsage.fakes with no
network, exactly as the CLI proves the rest of the engine.
Examples reference their relevant documents by filename; run_golden_eval()
ingests the corpus, maps each filename to the document id the parser assigned,
and hands real EvalExamples to the evaluator. A
deploy points the same machinery at its own corpus and labels by supplying a
different GoldenSet.
class ragsage.goldens.GoldenExample(question: str, expected_answer: str | None = None, ground_truth_answer: str | None = None, relevant_sources: frozenset[str] = )
Bases: object
A labelled question keyed to its relevant documents by filename.
question : str
expected_answer : str | None = None
ground_truth_answer : str | None = None
relevant_sources : frozenset[str]
class ragsage.goldens.GoldenSet(corpus: ~collections.abc.Mapping[str, bytes], examples: ~collections.abc.Sequence[~ragsage.goldens.GoldenExample], thresholds: ~ragsage.evaluation.EvalThresholds = )
Bases: object
A corpus, its labelled questions, and the thresholds to gate against.
corpus : Mapping[str, bytes]
examples : Sequence[GoldenExample]
thresholds : EvalThresholds
ragsage.goldens.default_golden_set() → GoldenSet
The library’s built-in golden set — a small multi-topic corpus.
async ragsage.goldens.run_golden_eval(golden: GoldenSet | None = None, *, contextualizer: Contextualizer | None = None, config: IngestionConfig | None = None) → EvalReport
Ingest the golden corpus into fresh fakes and score its questions.
Wires a FakeEngineKit exactly as the CLI does, ingests every corpus
document, resolves each example’s filenames to the ids the parser assigned,
and returns the evaluator’s EvalReport. Fully offline and
deterministic — the same run every time, so it can gate CI.
contextualizer and config exist so one corpus can be scored under
different contextualisation arms — the comparison
HeadingWindowContextualizer had to justify
itself against. They default to the fake and to
IngestionConfig’s own defaults, so an existing caller
sees no change.
The assembler
Not re-exported from the top-level package, deliberately: importing it pulls SQLAlchemy,
asyncpg and the provider SDKs, and import ragsage is guaranteed to stay free of that stack.
RagSage — one entry point that assembles the engine from a config object.
Before this, a consumer had to hand-wire a parser, a classifier, a chunker, a contextualizer, an embedder, a reranker, an LLM client, a rewriter, three stores and a cache before it could ingest a single document. That wiring is the same every time, so it belongs in the library.
This is an assembler, not a replacement. The ports stay public,
IngestionPipeline and QueryEngine
stay directly constructible, and every adapter RagSage.from_config() builds
can be swapped. RagSage.pipeline_for() and RagSage.engine_for() are
public for exactly that reason — a consumer that outgrows the façade reaches past
it rather than forking it. The fakes, the CLI and the offline test suite never touch
this module, which is the standing evidence that it stayed optional.
Not re-exported from the ``ragsage`` namespace, deliberately: importing this
pulls SQLAlchemy, asyncpg and the provider SDKs (and through voyageai, numpy
and HuggingFace tokenizers). import ragsage and import ragsage.parsing
must stay free of that stack — see tests/test_dependency_guard.py. Reach for it
explicitly:
from ragsage.sage import RagSage, RagSageConfigWhy the stores are built per operation rather than once: they hold a session, and a session carries the namespace binding that makes Row-Level Security work. A long-lived store would either pin one namespace forever or need its session swapped underneath it. So each call opens a scoped session, builds the two façades around it, and closes it — the compute adapters, which are stateless with respect to the caller, are built once and reused.
class ragsage.sage.RagSageConfig(postgres: PostgresConfig, providers: ProviderConfig, ingestion: IngestionConfig = , query: QueryOptions = )
Bases: object
Everything RagSage.from_config() needs, composed from the two halves.
Composed rather than flattened on purpose. PostgresConfig (ticket 01) and
ProviderConfig (ticket 04) already validate their own fields and are
independently useful — a consumer injecting a custom embedder still wants the
storage config. Flattening would mean two declarations of every setting and a
second place for them to drift.
Neither half reads the environment, here or anywhere: wiring env vars to fields is the consumer’s job at its own composition root (ADR-0002).
ingestion and query carry the policy defaults — chunk size, whether to
contextualize, how many candidates to retrieve — so a caller can set them once
instead of passing them to every call. Both remain overridable per call.
postgres : PostgresConfig
providers : ProviderConfig
ingestion : IngestionConfig
query : QueryOptions
class ragsage.sage.ComputeKit(parser: DocumentParser, classifier: PageClassifier, chunker: Chunker, contextualizer: Contextualizer, embedder: Embedder, reranker: Reranker, llm: LLMClient, rewriter: QueryRewriter | None = None, cache: Cache = , tracer: Tracer | None = None)
Bases: object
The adapters that do not depend on a database session.
Stateless with respect to the caller, so one instance serves every operation.
Frozen, so overriding one is dataclasses.replace() rather than a
thirteen-argument constructor call:
kit = replace(ComputeKit.from_config(config), embedder=MyEmbedder())
sage = RagSage.from_config(config, compute=kit)parser : DocumentParser
classifier : PageClassifier
chunker : Chunker
contextualizer : Contextualizer
embedder : Embedder
reranker : Reranker
llm : LLMClient
rewriter : QueryRewriter | None = None
cache : Cache
tracer : Tracer | None = None
classmethod from_config(config: RagSageConfig) → ComputeKit
Build the default adapters: ragsage’s own parser, Voyage, and OpenAI.
classmethod with_model_contextualizer(config: RagSageConfig) → ComputeKit
The same kit, but contextualizing with OpenAI instead of headings.
Offered because it is the one default a quality-focused consumer is most likely to want changed, and discovering the right client to pass otherwise means reading this module.
class ragsage.sage.StoreKit(vector_store: Callable[[AsyncSession], VectorStore], lexical_store: Callable[[AsyncSession], LexicalStore], document_store: Callable[[AsyncSession], DocumentStore])
Bases: object
How to build the three session-bound stores.
Factories rather than instances, because a store holds a session and a session holds the namespace binding RLS enforces. Overriding one — a consumer with its own vector database — means supplying a factory that ignores the session.
vector_store : Callable[[AsyncSession], VectorStore]
lexical_store : Callable[[AsyncSession], LexicalStore]
document_store : Callable[[AsyncSession], DocumentStore]
classmethod from_config(config: RagSageConfig) → StoreKit
class ragsage.sage.RagSage(*, database: Database, compute: ComputeKit, stores: StoreKit, ingestion: IngestionConfig | None = None, query: QueryOptions | None = None)
Bases: object
The assembled engine: migrate, ingest, query, stream, purge.
classmethod from_config(config: RagSageConfig, *, database: Database | None = None, compute: ComputeKit | None = None, stores: StoreKit | None = None) → RagSage
Assemble the default engine, with any layer replaceable.
Each argument defaults to the standard build. Pass compute to swap model
adapters, stores to swap persistence, database to share a pool with
the consumer’s own.
pipeline_for(session: AsyncSession) → IngestionPipeline
The ingestion pipeline bound to one scoped session.
Public so a consumer can drive the pipeline directly — inside its own transaction, say — without reimplementing this wiring.
engine_for(session: AsyncSession) → QueryEngine
The query engine bound to one scoped session.
async migrate() → None
Create ragsage’s table, indexes, role and RLS policy. Idempotent.
async dispose() → None
Close the connection pool. Call on shutdown.
async ingest(source: RawSource, scope: Scope, config: IngestionConfig | None = None) → IngestionResult
Ingest one document into scope, in a single scoped transaction.
async query(question: str, scope: Scope, options: QueryOptions | None = None, *, history: Sequence[Turn] = ()) → Answer
Answer question from scope’s corpus.
async stream(question: str, scope: Scope, options: QueryOptions | None = None, *, history: Sequence[Turn] = ()) → AsyncIterator[AnswerToken | Citation | Usage | AnswerComplete]
Stream the answer, holding the scoped session open for the whole stream.
The session must outlive the generator — retrieval happens as the stream is
consumed, not before it starts — so the transaction closes when the consumer
stops iterating. A caller that abandons the stream early still releases the
connection, because the async with unwinds on generator close.
async delete_document(scope: Scope, document_id: str) → None
Remove one document’s chunks from scope. Idempotent.
The consumer’s own delete path needs this: its lifecycle row and the chunks now live in different tables on different connections, so deleting the row no longer takes the chunks with it. All three stores address the same rows, so this drives them together rather than leaving the caller to guess which one owns the purge.
async purge(scope: Scope) → None
Delete every chunk in scope’s namespace. Idempotent.
Load-bearing rather than convenience. ADR-0002 dropped the
chunks.user_id -> users.id ON DELETE CASCADE foreign key, so deleting a
consumer’s user no longer purges their chunks for free. Nothing deletes
users today; this exists so the path is available the day account deletion
ships, instead of being discovered missing then.