Adapters
What ships in the box: the parser, Voyage/OpenAI, Postgres, the fakes.
What ships in the box. Each of these is one implementation of a port, not a privileged part of the engine — swapping any of them changes no ragsage code.
Parsing and chunking
Heuristic and model-free, deliberately: the import graph carries no torch, no numpy ≥ 2, no transformers, no onnxruntime and no magika, which is a portability promise enforced by a test that probes the graph in a clean subprocess. See Failure modes for what that costs.
Parsing: the page-route classifier and the heuristic parse/chunk backend.
ragsage owns the parse/chunk ports and their fakes; this package holds the
library’s real, model-free implementations of them:
LayoutPageClassifier— the parser-agnosticTEXT/VISIONrouting policy (reads only aPageLayout).HeuristicBackend— a pure-Python backend that satisfies both theDocumentParserandChunkerports, with a format router and the shared two-pass structure-aware Markdown chunker.
Both are re-exported here so callers import from ragsage.parsing regardless of
the internal module split.
class ragsage.parsing.DocumentFormat(*values)
Bases: StrEnum
The parse path a source is routed to.
TEXT covers both plain text and Markdown — they share one read-as-is
path. The remaining members name the structured formats later issues fill
in; until then the backend rejects them explicitly rather than mis-parsing.
TEXT = 'text'
PDF = 'pdf'
DOCX = 'docx'
PPTX = 'pptx'
HTML = 'html'
UNKNOWN = 'unknown'
class ragsage.parsing.HeuristicBackend(*, encoding_name: str = 'cl100k_base')
Bases: object
Both the DocumentParser and the Chunker port, no ML dependency.
encoding_name names the tiktoken encoding the chunker bounds token
counts against; it is loaded lazily on first chunk so importing this module
(which the composition root does) costs nothing.
parse(source: RawSource) → ParsedDocument
Read a source into per-page Markdown and stash it for the chunker.
chunk(document: Document, pages: Sequence[Page], *, size: int, overlap: int) → Sequence[Chunk]
Structure-aware chunks over the resolved pages, page by page.
size is the per-chunk token budget and overlap the token overlap
for oversized prose; the two-pass chunker keeps tables whole and carries
each section’s heading path onto its chunks. pages is the resolved
page list the pipeline built — a page that routed to vision arrives here
with its transcription filled into text — so chunking reads from it
(not the stash) and a scanned PDF page ends up as retrievable as a typed
one. The stash is merely discarded here so parsing does not retain pages
for the backend’s lifetime — a document whose pages came from the parse
cache was never stashed, and that is fine.
Ordinals run densely across every page so chunk ids stay unique.
class ragsage.parsing.LayoutPageClassifier(*, min_text_chars: int = 24, max_image_area_ratio: float = 0.5, min_bitmap_area: float = 50_000.0)
Bases: object
PageClassifier — routes each page on whether it has a usable text layer.
A born-digital page carries enough extractable text to trust and route as
TEXT. A scanned, photographed, or handwritten page carries little or no
text and is dominated by a raster bitmap; it has no usable text layer and,
when the parser rendered an image for it, routes to the premium vision model.
The decision reads the PageLayout the parser
measured, against three tunable thresholds so the policy is inspectable and
unit-testable without any parser:
min_text_chars— below this many characters the text layer is treated as absent (a scan’s stray or hidden-OCR characters don’t make it “typed”).max_image_area_ratio— at or above this fraction of the page covered by images the page is image-dominated, so any thin text is not the real content and the page is read by vision instead.min_bitmap_area— a single bitmap at least this large (in the page’s own units) is a page-filling scan; paired with thin text it forces the vision route even if the image-area ratio reads low.
With no layout measured (a flow format with no page grid, or a hand-built
Page) it falls back to whether a text layer is
present at all — the same signal the earlier two-path scaffold used.
classify(page: Page) → PageRoute
ragsage.parsing.route(source: RawSource, content: bytes) → DocumentFormat
Resolve source to a DocumentFormat.
Tries, in order of trustworthiness: the declared media_type, the filename
extension, then pure-Python magic-byte sniffing. If none identifies a known
container but the bytes decode as text, it is treated as TEXT (a bare
.txt/.md upload with no media type and an unknown extension still
ingests); otherwise the format is UNKNOWN.
Contextualizing
Contextual retrieval without a model call — the structure is already in hand.
Anthropic-style contextual retrieval improves a chunk’s embedding by prefixing a
sentence that says where the chunk sits in its document. The obvious way to
write that sentence is to ask an LLM, which is what the Contextualizer
port was shaped for — and what the backend’s adapter does. But that costs one
model call per chunk, adds latency to every ingest, and cannot run in an offline
test: today the only Contextualizer shipped in the library is
FakeContextualizer, which is a stand-in, so
IngestionConfig.contextualize=True buys a standalone CLI user nothing real.
HeadingWindowContextualizer fills that hole with string assembly rather
than inference. The two-pass chunker (ragsage.parsing.chunking) already
carries each section’s heading path onto every chunk it emits, and the ingestion
pipeline already hands the contextualizer the document’s full_text. Between
them that is enough to say, deterministically, which document, which section,
and what was written immediately either side — most of what an LLM-written
context sentence conveys, at zero model cost, zero latency, and with the same
output for the same input every time. That determinism is the point: it is
testable in the offline CI an LLM contextualizer can never run in.
The window is bounded by a token budget counted with
make_token_counter() — deliberately the same
tiktoken encoding the chunker bounds chunks against, so “512-token chunks
plus 128 tokens of context” means one thing across the pipeline rather than two.
When the budget bites, text is dropped at whole paragraph boundaries and, only if
a single paragraph still overflows, at whole sentence boundaries. Nothing is ever
cut mid-word: a half-word contributes a garbage token to the embedding, which is
the opposite of what contextualisation is for.
It is weaker than an LLM-written context sentence — it repeats source prose
instead of summarising it, and it adds tokens to embed_text. It is offered as
the honest default for a caller with no model budget, not as a replacement for
the LLM adapter where one is available.
class ragsage.contextualizing.HeadingWindowContextualizer(*, window: int = 2, max_context_tokens: int = 128, encoding_name: str = _DEFAULT_ENCODING)
Bases: object
A Contextualizer built from structure, not inference.
window is how many neighbouring paragraphs may be drawn from each side of
the chunk (0 disables the window entirely, leaving heading context only).
max_context_tokens is the hard ceiling on everything prepended to the
chunk — the heading header and both windows — so a caller can reason about
what contextualisation costs their embedder. encoding_name names the
tiktoken encoding those tokens are counted with; it defaults to the
chunker’s, and there is rarely a reason to change one without the other.
The encoding is loaded lazily on first use, so constructing this at a composition root costs nothing.
async contextualize(document: Document, chunk: Chunk, *, full_text: str) → str
Return the text to embed: a bounded context header, then the chunk verbatim.
async only because the port is — nothing here awaits, touches the
network, or reads a clock, so the same arguments always produce the same
string.
Models: Voyage and OpenAI
Core dependencies rather than an extra, so pip install ragsage plus two API keys is a
working engine. Their configuration is on the Configuration page.
The shipped production adapters: Voyage for retrieval, OpenAI for generation.
Five adapters cover the five model-shaped ports — Embedder, Reranker,
LLMClient, Contextualizer, QueryRewriter — so pip install ragsage
plus two API keys is a working engine, not a pile of interfaces waiting for the
consumer to write 750 lines of adapter code (ADR-0002). They are core, not an
extra: an engine you have to assemble before it does anything is reusable in
principle and unusable in practice.
They remain adapters, not the engine. Each conforms to its port by shape, the
ports stay public, and the in-memory ragsage.fakes are still a complete
alternative — swap one of these for a local embedder or a different provider and
nothing upstream changes.
Importing this package pulls the provider SDKs, and with voyageai comes
numpy and HuggingFace tokenizers. That is why the adapters live behind
this namespace and are not re-exported from the top-level ragsage package:
import ragsage and import ragsage.parsing stay free of that stack, which
is the portability promise ADR-0001 kept and tests/test_dependency_guard.py
enforces. Import ragsage.providers only where you actually intend to talk to
a provider.
type ragsage.providers.CallConfig = RunnableConfig
class ragsage.providers.OpenAIContextualizer(client: ChatOpenAI, *, config: CallConfig | None = None)
Bases: object
Contextualizer — situates a chunk with a cheap OpenAI model.
The whole document is sent once as a system message and the per-chunk ask as
the human message. That keeps the long, shared document body as the literal
prompt prefix across every chunk of the same document, so OpenAI’s automatic
prefix caching reuses it — no Anthropic-style cache_control annotation is
needed (or sent). The embed text is the model’s context prepended to the
verbatim chunk; if the model returns nothing usable, the raw chunk is embedded
unchanged rather than polluting the vector with empty context.
async contextualize(document: Document, chunk: Chunk, *, full_text: str) → str
class ragsage.providers.OpenAILLMClient(*, generation: ChatOpenAI, vision: ChatOpenAI, config: CallConfig | None = None)
Bases: object
LLMClient — streamed generation and multimodal vision transcription.
Generation streams token-by-token from the generation client so a caller can relay each delta straight to its own transport; transcription sends the page image to the vision client as a multimodal message and returns the whole reply at once. Two clients rather than one because they differ in streaming and in token budget.
async generate(prompt: str) → AsyncIterator[str]
async transcribe(image: PageImage) → str
class ragsage.providers.OpenAIQueryRewriter(client: ChatOpenAI, *, config: CallConfig | None = None)
Bases: object
QueryRewriter — condenses a follow-up against history with a cheap model.
The conversation is rendered as a short transcript and handed to the same
cheap OpenAI tier contextualization uses; the model returns a standalone
question that QueryEngine retrieves and generates on.
With no history there is nothing to resolve, so the question is returned
unchanged and no call is made at all. If the model returns nothing usable, the
original question is used rather than retrieving on empty text.
async rewrite(question: str, history: Sequence[Turn]) → str
class ragsage.providers.ProviderClients(embeddings: VoyageAIEmbeddings, rerank: VoyageAIRerank, generation: ChatOpenAI, contextualize: ChatOpenAI, rewrite: ChatOpenAI, vision: ChatOpenAI)
Bases: object
The LangChain clients, one per model role, built once as singletons.
Held together so a composition root can construct them in one call and hand
the whole bundle to the adapters that bind roles around them. Generation and
vision are distinct ChatOpenAI instances because they differ in
whether they stream and how many tokens they may emit; contextualize and
rewrite share the cheap tier but cap output differently.
embeddings : VoyageAIEmbeddings
rerank : VoyageAIRerank
generation : ChatOpenAI
contextualize : ChatOpenAI
rewrite : ChatOpenAI
vision : ChatOpenAI
class ragsage.providers.ProviderConfig(openai_api_key: str, voyage_api_key: str, embedding_model: str = 'voyage-3-large', rerank_model: str = 'rerank-2', contextualize_model: str = 'gpt-4o-mini', generation_model: str = 'gpt-4o', vision_model: str = 'gpt-4o', timeout_seconds: float = 60.0, context_max_tokens: int = 128, rewrite_max_tokens: int = 256, vision_max_tokens: int = 4096)
Bases: object
Keys, per-role model ids, and call limits for the Voyage/OpenAI adapters.
The roles are split because they are priced and shaped differently, not for
symmetry: embedding and reranking go to Voyage; generation, contextualization,
query rewriting, and vision go to OpenAI, with contextualization and rewriting
sharing a cheap tier. embedding_model is pinned per corpus — changing it
invalidates every stored vector — so it is named here rather than defaulted at
each call site.
An empty API key is permitted and means “no calls to this provider will
succeed”. The provider SDKs reject an empty key at construction, so
build_provider_clients() substitutes a
placeholder: the clients build and the process boots (dev, CI, and any harness
that swaps fakes in at the ports never touches a model), while a real call
fails with an auth error at request time rather than at import time.
openai_api_key : str
voyage_api_key : str
embedding_model : str = 'voyage-3-large'
rerank_model : str = 'rerank-2'
contextualize_model : str = 'gpt-4o-mini'
generation_model : str = 'gpt-4o'
vision_model : str = 'gpt-4o'
timeout_seconds : float = 60.0
context_max_tokens : int = 128
rewrite_max_tokens : int = 256
vision_max_tokens : int = 4096
class ragsage.providers.VoyageEmbedder(client: VoyageAIEmbeddings)
Bases: object
Embedder over Voyage’s batch embeddings endpoint.
The vectors it returns are as wide as the configured embedding model makes them, which is why the model is pinned per corpus: a store’s vector column is sized to one model’s output and re-indexing is the only way to change it.
async embed(texts: Sequence[str]) → Sequence[tuple[float, ...]]
class ragsage.providers.VoyageReranker(client: VoyageAIRerank)
Bases: object
Reranker over Voyage’s rerank endpoint (a cross-encoder).
Each candidate is handed to Voyage as a Document
tagged with its position in the input list; the reranked replies carry that tag
and a relevance_score back, so the adapter rebuilds the
ScoredChunk list from its own candidates by index — the model only
ever sees text, never the engine’s chunk identities. A reply whose tag is
missing or out of range is dropped rather than trusted, so a malformed
response can never mislabel a citation. The client scores every candidate (it was built with top_k=None);
the port’s per-call top_k is applied here.
async rerank(query: str, candidates: Sequence[ScoredChunk], *, top_k: int) → Sequence[ScoredChunk]
ragsage.providers.build_provider_clients(config: ProviderConfig) → ProviderClients
Construct the per-role provider clients from config (opens no socket).
Every OpenAI role shares one key and one timeout; likewise every Voyage role.
The rerank client is built with top_k=None so it scores and returns all
candidates — VoyageReranker honours the
per-call top_k itself, since the Reranker port passes it per request
rather than fixing it per deployment.
Storage: Postgres
pgvector dense retrieval over an HNSW cosine index, alongside a generated tsvector lexical
column with GIN, both Row-Level-Security-scoped by namespace. PostgresConfig is on the
Configuration page.
Postgres storage: the pool, the scoped session, and the config that names them.
ragsage owns its storage end to end (ADR-0002). This package holds the part that is not about rows at all — how a connection is obtained and what has to be true of a transaction before a store may run inside it.
This package is not re-exported from ``ragsage``, on purpose. Importing the
top-level package, or ragsage.parsing, must not drag in SQLAlchemy or the
Postgres driver: the parser and the pipelines run offline over the in-memory
fakes, and a consumer that only parses should not need a database installed to
import the library. Reach for storage explicitly:
from ragsage.storage import Database, PostgresConfig
db = Database(PostgresConfig(dsn="postgresql://user:pw@host/db", app_role="rag_app"))
async with db.session(scope) as session:
... # runs as the non-privileged role, pinned to scope.namespaceclass ragsage.storage.Database(config: PostgresConfig, engine: AsyncEngine | None = None)
Bases: object
ragsage’s pool: one engine, one session factory, one way in.
Construct it from a PostgresConfig and hand it to the stores. An
engine may be injected instead of created — for tests, or for a consumer
that genuinely wants both libraries sharing one pool against
max_connections.
property config : PostgresConfig
property engine : AsyncEngine
session(scope: Scope) → AbstractAsyncContextManager[AsyncSession]
Open a transaction pinned to scope, with RLS actively enforced.
The one door scoped work goes through. See
open_scoped_session() for what that costs
and why both settings are transaction-local.
owner_connection() → AsyncIterator[AsyncConnection]
Open an owner-privileged transaction — DDL only, RLS not enforced.
No SET LOCAL ROLE, no isolation variable: this connection sees every
namespace. It exists so ragsage can create its own table, indexes, role
and policy (ticket 02’s migrate). Reading or writing rows through it
would bypass the guarantee the rest of this package exists to provide.
async migrate() → None
Create ragsage’s table, indexes, role and RLS policy. Idempotent.
Safe to call on every start. Runs on an owner connection because only the
owner may issue this DDL — see ragsage.storage.schema.
async ping() → bool
Return True if a trivial round-trip to the database succeeds.
SELECT 1 on a real connection proves the socket, the credentials and
the database are all live — not merely that a DSN was configured. Any
failure becomes False so a caller’s health check can report degraded
rather than raise.
async dispose() → None
Close every pooled connection. Call on shutdown.
class ragsage.storage.PgDocumentStore(session: AsyncSession)
Bases: object
The document-level port, deliberately minimal.
Document identity is the consumer’s, not the library’s, and that boundary is
kept rather than quietly moved inward: find_by_hash returns None and
save persists nothing. The consumer owns the lifecycle row — status, storage
key, dedup at upload — so re-implementing dedup here would mean two answers to
“have I seen this content”, which is worse than none.
That is why ragsage owns exactly one table. get/list reconstruct what
was ingested from the chunks that exist, which is all the port needs, and
delete purges them.
One consequence worth naming: Document.metadata therefore has nowhere to
go. Chunk.metadata round-trips through the JSONB column, but a
document-level payload would need a documents table ragsage has decided not to
own. A consumer that needs it keeps it on its own row.
async save(scope: Scope, document: Document, chunks_saved: Sequence[Chunk]) → None
async find_by_hash(scope: Scope, content_hash: str) → Document | None
async get(scope: Scope, document_id: str) → Document | None
async list(scope: Scope) → Sequence[Document]
async delete(scope: Scope, document_id: str) → None
class ragsage.storage.PgLexicalStore(session: AsyncSession, *, text_search_config: str)
Bases: object
Keyword search over the database-generated lexical column.
index is a no-op, and that is not a stub: lexical is a generated column
on rows PgVectorStore already wrote in the same transaction, so there
is nothing separate to persist. Writing here would mean a second copy of the
same text.
The text-search configuration must match the one the generated column was built
with — a query parsed as simple against a column stemmed as english
silently under-matches. It is therefore required, with no default: a default
here would be a second declaration of
text_search_config, and the two
could drift without anything failing loudly.
async index(scope: Scope, chunks_to_index: Sequence[Chunk]) → None
async search(scope: Scope, query: str, *, k: int) → Sequence[ScoredChunk]
Rank matches by ts_rank; an unindexable query matches nothing.
The blank short-circuit avoids a pointless round trip.
websearch_to_tsquery handles everything else — quotes, or, a bare
- — without raising, so an arbitrary user question is safe to pass
straight through.
async delete(scope: Scope, document_id: str) → None
class ragsage.storage.PgVectorStore(session: AsyncSession, *, embedding_model: str = '')
Bases: object
Dense vectors: the write path for every column, plus ANN search.
All three stores share one table, and this is the one that writes it. The
corpus’s pinned embedding_model is recorded on every row so the pin is
inspectable and a corpus cannot silently mix models or widths.
async upsert(scope: Scope, records: Sequence[EmbeddedChunk]) → None
Replace every chunk of the incoming documents, then insert the new set.
Delete-then-insert rather than ON CONFLICT, and the delete is scoped to
the document, not to the incoming chunk refs. That distinction is the
whole point: refs are <document>:<ordinal>, so a re-ingest at a larger
chunk_size yields fewer refs, and the previous run’s high-ordinal rows
are not among them. Deleting only the incoming refs would leave that tail
behind — still searchable, so a re-index would serve chunks from two
different chunkings at once. Deleting by document makes the write a true
replacement, and idempotent on retry rather than doubling.
A caller upserting one document’s chunks in several batches therefore has to pass them as one call; that is the honest trade for making re-index correct, and the pipeline already does it.
async search(scope: Scope, query_vector: tuple[float, ...], *, k: int) → Sequence[ScoredChunk]
Nearest neighbours by cosine distance, returned as cosine similarity.
cosine_distance must match the index’s vector_cosine_ops, or the
HNSW index is built and never used. The score is inverted to a similarity
so higher is better, which is what RRF fusion and the lexical half assume.
async delete(scope: Scope, document_id: str) → None
class ragsage.storage.PostgresConfig(dsn: str, app_role: str = 'ragsage_app', isolation_variable: str = 'ragsage.namespace', embedding_dim: int = 1024, text_search_config: str = 'english', pool_size: int = 5, max_overflow: int = 5, pool_pre_ping: bool = True)
Bases: object
Everything ragsage needs to own a connection pool against one database.
- Parameters:
- dsn (str) – The connection URL ragsage connects with. It connects as the owner of
its own table — owners bypass Row-Level Security, which is why every
scoped operation drops into
app_rolefirst (seeopen_scoped_session()). - app_role (str) – The non-privileged Postgres role scoped work runs under. Not a login
identity: it is a privilege context the owner switches into with
SET LOCAL ROLEso RLS is actually enforced. Reaches interpolated SQL, so it is guarded as a plain identifier. - isolation_variable (str) – The run-time setting carrying
Scope.namespacefor the duration of a transaction, and the value ragsage’s RLS policy compares rows against. Configurable because two isolation variables coexist in the deployment this was extracted from — the consumer’sapp.user_idover its own tables, and this one over ragsage’s — and they must not be confused. The default is ragsage-owned for exactly that reason. - embedding_dim (int) – The width of the
embeddingcolumn. Fixed at table-creation time and pinned per corpus — changing it invalidates every stored vector, so it belongs in config rather than as a literal in the DDL. Must match whatever the configured embedder actually returns. - text_search_config (str) – The Postgres text-search configuration the generated
lexicalcolumn is built with ('english'was hardcoded in the consuming backend). It has to be named explicitly because the two-argument form ofto_tsvectoris the IMMUTABLE one, and aSTOREDgenerated column accepts nothing else — the one-argument form depends on a session setting and is only STABLE. Changing it requires rebuilding the column. - pool_size – Pool sizing. Deliberately explicit and deliberately modest: ragsage’s
pool is a second pool reaching the same Postgres as the consumer’s, and
the pair have to fit inside
max_connectionstogether. - max_overflow – Pool sizing. Deliberately explicit and deliberately modest: ragsage’s
pool is a second pool reaching the same Postgres as the consumer’s, and
the pair have to fit inside
max_connectionstogether. - pool_pre_ping (bool) – Check a pooled connection is alive before handing it out. On by default — connections idle across a database restart or a proxy timeout otherwise fail the first query of an unlucky request.
- dsn (str) – The connection URL ragsage connects with. It connects as the owner of
its own table — owners bypass Row-Level Security, which is why every
scoped operation drops into
dsn : str
app_role : str = 'ragsage_app'
isolation_variable : str = 'ragsage.namespace'
embedding_dim : int = 1024
text_search_config : str = 'english'
pool_size : int = 5
max_overflow : int = 5
pool_pre_ping : bool = True
exception ragsage.storage.SchemaMismatch
Bases: RuntimeError
An existing table disagrees with the config it is being migrated against.
Raised instead of letting migrate() succeed misleadingly. CREATE TABLE IF NOT EXISTS is a no-op on a table that already exists, so it silently accepts a
config whose embedding_dim differs from the column that is really there —
and the disagreement then surfaces at the first insert, thousands of parsed
tokens later, as expected 1024 dimensions, not 384 from deep inside the
driver. Failing at migrate() puts the error where the mistake is.
class ragsage.storage.Statement(sql: str, parameters: Mapping[str, str]=)
Bases: object
One SQL statement plus its bound parameters.
A plain value so the isolation preamble can be inspected — asserted on in a unit test, logged, or replayed by ticket-02’s migration — without a live database and without reaching into a session mock.
sql : str
parameters : Mapping[str, str]
ragsage.storage.create_engine(config: PostgresConfig) → AsyncEngine
Create the pooled async engine described by config.
ragsage.storage.create_sessionmaker(engine: AsyncEngine) → async_sessionmaker[AsyncSession]
Build the session factory bound to engine.
expire_on_commit=False so rows read inside a scoped session stay usable
after it commits — the scoped session commits on the way out of its context
manager, and its caller is still holding what it returned.
async ragsage.storage.existing_embedding_dim(connection: AsyncConnection) → int | None
The width of the live embedding column, or None if there is no table.
Read from pg_attribute.atttypmod, which for a pgvector column is the
declared dimension (unlike varchar, it carries no length offset). -1
means the column was declared unsized as bare vector, which no ragsage
migration produces — treated as unknown rather than as a mismatch, so a table
created by something else is not condemned on this evidence alone.
ragsage.storage.isolation_preamble(config: PostgresConfig, scope: Scope) → tuple[Statement, ...]
Return the statements that pin a transaction to scope under RLS.
The role is interpolated because Postgres has no bind form for an identifier;
it goes through safe_identifier() here, at the interpolation site, even
though PostgresConfig already validated it. The setting name, by
contrast, is an ordinary text argument to set_config and is bound like
any other parameter — nothing of the namespace or the variable name is ever
spliced into SQL text.
async ragsage.storage.migrate(connection: AsyncConnection, config: PostgresConfig) → None
Create the table, indexes, role and policy. Idempotent.
Takes an owner connection rather than a Database
so a consumer already holding one — running this inside a wider startup
transaction, say — is not forced to open a second.
Raises SchemaMismatch if the table already exists at a different
embedding_dim. That check runs first, before any DDL: a caller who has
misconfigured the width should not have indexes rebuilt or a policy recreated
on the way to being told so.
ragsage.storage.migration_statements(config: PostgresConfig) → tuple[str, ...]
Every statement migrate() runs, in order.
Separated from execution so the whole schema is inspectable — and assertable — without a live database.
ragsage.storage.open_scoped_session(sessionmaker: async_sessionmaker[AsyncSession], config: PostgresConfig, scope: Scope) → AsyncIterator[AsyncSession]
Open a session pinned to one Scope, RLS enforced.
Commits when the block exits cleanly and rolls back when it raises; either way the role and the isolation variable are gone before the connection returns to the pool. Every ragsage store goes through here, so the isolation posture is defined exactly once.
async ragsage.storage.purge_namespace(session: AsyncSession) → None
Delete every chunk row visible to this session — i.e. one whole namespace.
Carries no WHERE at all, which looks alarming and is exactly right: the RLS
policy scopes the statement to the session’s namespace. A hand-written predicate
here would be a second, weaker copy of a guarantee the database already makes.
Idempotent — deleting nothing is a successful delete.
ragsage.storage.safe_identifier(name: str) → str
Return name if it is a plain SQL identifier, else raise ValueError.
Used for the app role, which reaches SET LOCAL ROLE "<role>" as
interpolated text. Returning the name (rather than just asserting) keeps the
guard at the interpolation site, where a reader can see that nothing
unchecked reaches the SQL string.
ragsage.storage.safe_setting_name(name: str) → str
Return name if it is a valid custom Postgres setting, else raise.
A run-time-defined setting must be qualified — prefix.name, exactly one
dot — because Postgres only accepts assignments to unknown parameters when
they carry an extension-style prefix. Requiring the prefix here means a
typo’d isolation variable fails when the config is built, not on the first
query against a policy that then silently matches nothing.
Caching
Keys and wire format for caching parse output, over the existing Cache port.
Parsing is the most expensive deterministic step in ingestion — the whole PDF stack runs, per page — and its result is a pure function of the document’s bytes and the parser that read them. That makes it exactly the thing worth caching, and re-indexing is what makes it matter: the embedding model is pinned per corpus, so changing it (or changing chunk size) currently means re-parsing every document to get chunks to re-embed. A parse cache is what makes that affordable.
Two design decisions are load-bearing.
Keyed by content hash, never by path or filename. The same bytes moved,
renamed, or re-uploaded under a different name must hit. RAG-Anything normalises
image paths to basenames for precisely this reason; we already compute a
content-derived id in ragsage.parsing.identity, so the hash is free.
The parser is part of the key, the chunking parameters are not. The key
carries a fingerprint of the parser that produced the entry plus a format version
for this module, so swapping parsers or changing the wire format misses rather
than deserialising something stale into a shape that no longer matches. Chunk
size and overlap are deliberately absent: what is cached is the parse output —
pages — and chunking runs on every ingest regardless. A caller who changes
chunk_size gets new chunks immediately, so there is no staleness for the key
to guard against. Including them would only lower the hit rate for nothing.
The Cache port is unchanged — still str -> str | None.
That is why this module exists: pages have to survive a round trip through a
single string, and every decode failure has to degrade to a miss rather than an
exception, because the port’s stated contract is that the engine is correct with
the cache absent, wrong, or empty.
ragsage.caching.PARSE_CACHE_VERSION = 1
Bumped when the encoding below changes shape. An old entry then misses rather than decoding into something subtly wrong — cheaper than a migration, and the only safe option for a cache we do not enumerate or clear.
ragsage.caching.parser_fingerprint(parser: object) → str
A stable-enough identity for the parser that produced a cache entry.
The DocumentParser port has no version field — adding
one would burden every adapter for this feature’s benefit — so the import path
of the parser’s class stands in. It distinguishes the cases that matter in
practice: a different parser, a fake versus the real backend, a consumer’s own
implementation.
What it does not catch is the same class changing its output, e.g. a
heuristic tuned between releases. PARSE_CACHE_VERSION covers this
module’s own format; a parser whose behaviour changes needs its consumer to
accept one stale generation or key on something more specific. Documented
rather than papered over.
ragsage.caching.parse_cache_key(content_hash: str, parser: object) → str
The cache key for one document’s parse output.
ragsage.caching.encode_parsed(parsed: ParsedDocument) → str
Serialise parse output to a single JSON string for the cache.
Image bytes are base64-encoded rather than dropped: a scanned page’s
PageImage is what the vision route needs, and a cache
entry that silently lost it would turn a hit into an empty document. It does
make entries for image-heavy documents large — sizing the cache is the
deployment’s call, and a value too big for the backing store simply fails to
set, which degrades to a miss.
ragsage.caching.decode_parsed(payload: str) → ParsedDocument | None
Deserialise a cache entry, or return None if it cannot be trusted.
Every malformed, truncated, or foreign payload becomes a miss. A cache is shared, long-lived, and outside our control; treating a bad entry as absent is the only behaviour that keeps the port’s “correct without the cache” contract true. Deliberately broad: there is no failure here worth propagating over simply parsing the document again.
Fakes
A genuine in-memory implementation of every port. Not only a testing convenience: they are what makes the engine runnable the moment it is installed, and the standing proof that the library is free of web, tenant and provider concepts.
In-memory, offline adapters for every port — the standalone proof.
These are fakes, not mocks: each one is a genuine, working implementation that happens to run in memory with no network. Together they let the pipelines run the full ingest-and-query loop deterministically — which is exactly what the Seam-1 tests and the CLI need. They are shipped in the package (not hidden in tests) so pip install ragsage gives you a runnable engine out of the box.
The retrieval is real enough to be meaningful: the embedder hashes tokens into a
sparse vector so cosine similarity tracks term overlap, and unrelated text
scores zero — which is what lets the honest “not found” path actually fire. The
stores serialise to plain dicts so the CLI can persist a corpus between the
ingest and query commands.
Nothing here is tuned for quality; swap in real adapters (voyage embeddings, Cohere rerank, pgvector) behind the same ports for that.
class ragsage.fakes.FakeDocumentParser
Bases: object
Reads a source’s bytes into pages, splitting on form-feeds.
A text source becomes one page per form-feed-delimited block. A source whose
media type or name says “image” becomes a single vision page carrying the
bytes for FakeLLMClient.transcribe() to read — so both parser routes
are exercisable from real inputs.
parse(source: RawSource) → ParsedDocument
class ragsage.fakes.FakePageClassifier
Bases: object
Routes a page by whether it has a usable text layer.
When the parser measured a PageLayout, this reads the
same signals the real classifier does — a page with little text that a raster
image dominates goes to vision; otherwise the presence of a text layer
decides. With no layout it falls back to whether text is non-blank, which
is all the offline fakes and the CLI need.
classify(page: Page) → PageRoute
class ragsage.fakes.FakeChunker
Bases: object
Word-window chunker: size words per chunk with overlap carried over.
Tokens stand in for the real tokeniser’s tokens; the windowing and stable per-document chunk ids are what the pipeline actually depends on.
chunk(document: Document, pages: Sequence[Page], *, size: int, overlap: int) → Sequence[Chunk]
class ragsage.fakes.FakeContextualizer
Bases: object
Prepends a document-situating tag to the chunk’s embed text.
async contextualize(document: Document, chunk: Chunk, *, full_text: str) → str
class ragsage.fakes.FakeEmbedder(*, dim: int = _EMBED_DIM)
Bases: object
Deterministic hashing embedder — cosine tracks term overlap.
dim sizes the output vector; it defaults to the library’s own width but is
configurable so a caller whose store fixes a different dimension (e.g. a real
pgvector column) can produce matching vectors without forking this algorithm.
async embed(texts: Sequence[str]) → Sequence[tuple[float, ...]]
class ragsage.fakes.FakeReranker
Bases: object
Cross-encoder stand-in: rescore by query-term overlap, drop the misses.
Uses a different signal (raw token overlap) than the dense embedder, so it genuinely reorders fused candidates and prunes ones with no lexical tie to the query — the pruning is what backstops the not-found path.
async rerank(query: str, candidates: Sequence[ScoredChunk], *, top_k: int) → Sequence[ScoredChunk]
class ragsage.fakes.FakeLLMClient
Bases: object
An extractive reader: answers verbatim from the most relevant source.
generate parses the numbered sources out of the grounded prompt, picks
the ones sharing terms with the question, and streams back the best source’s
text followed by [n] markers for every relevant source — giving a
grounded, citable answer with zero model calls. transcribe decodes the
image bytes as the “vision” reading of a scanned page.
It also answers the small-talk prompt, which carries a Message: line
instead of sources: a fixed friendly sentence, so the conversational path is
exercisable offline exactly like the grounded one.
SMALL_TALK_REPLY = "Hello! Ask me anything about your uploaded documents and I'll answer with citations."
What the fake replies to a conversational message. Fixed, so tests can assert on it without pinning the wording of a real model.
async generate(prompt: str) → AsyncIterator[str]
async transcribe(image: PageImage) → str
class ragsage.fakes.FakeQueryRewriter
Bases: object
Condenses a follow-up by folding in the content words of prior turns.
A real rewriter would ask a model to resolve “its” to “the Acme plan”; this deterministic stand-in appends the content words said earlier (that the follow-up doesn’t already carry) to the question. That is enough for the overlap-based retrieval and reranker fakes to match the passage the pronoun pointed at — so the multi-turn path is exercisable offline, exactly like the rest of the engine.
async rewrite(question: str, history: Sequence[Turn]) → str
class ragsage.fakes.FakeVectorStore
Bases: object
In-memory dense store; cosine search, scoped and filterable.
async upsert(scope: Scope, records: Sequence[EmbeddedChunk]) → None
async search(scope: Scope, query_vector: tuple[float, ...], *, k: int) → Sequence[ScoredChunk]
async delete(scope: Scope, document_id: str) → None
snapshot() → dict[str, Any]
restore(data: dict[str, Any]) → None
class ragsage.fakes.FakeLexicalStore
Bases: object
In-memory keyword store; overlap search, scoped and filterable.
async index(scope: Scope, chunks: Sequence[Chunk]) → None
async search(scope: Scope, query: str, *, k: int) → Sequence[ScoredChunk]
async delete(scope: Scope, document_id: str) → None
snapshot() → dict[str, Any]
restore(data: dict[str, Any]) → None
class ragsage.fakes.FakeDocumentStore
Bases: object
In-memory document registry: dedup, listing, and purge.
async save(scope: Scope, document: Document, chunks: Sequence[Chunk]) → None
async find_by_hash(scope: Scope, content_hash: str) → Document | None
async get(scope: Scope, document_id: str) → Document | None
async list(scope: Scope) → Sequence[Document]
async delete(scope: Scope, document_id: str) → None
snapshot() → dict[str, Any]
restore(data: dict[str, Any]) → None
class ragsage.fakes.FakeCache
Bases: object
A plain in-memory dict cache.
async get(key: str) → str | None
async set(key: str, value: str) → None
class ragsage.fakes.RecordedSpan(name: str, tenant: str | None, attributes: dict[str, object]=, children: list[RecordedSpan] = , error: str | None = None)
Bases: object
One captured trace node — a stage’s name, tenant tag, and attributes.
Mutable on purpose: FakeSpan.set() updates attributes and
FakeSpan.child() appends to children as a stage runs, so after a
query the tree mirrors exactly what the engine emitted.
name : str
tenant : str | None
attributes : dict[str, object]
children : list[RecordedSpan]
error : str | None = None
class ragsage.fakes.FakeSpan(record: RecordedSpan)
Bases: object
A Span that writes into a RecordedSpan.
Children inherit the parent’s tenant tag, so the captured tree is proof that every node of a trace is tenant-scoped — the guarantee the observability tests assert against.
set(**attributes: object) → None
child(name: str, **attributes: object) → FakeSpan
class ragsage.fakes.FakeTracer
Bases: object
Records flat events and the span tree so tests can assert what ran.
events/names are the point-event breadcrumbs (unchanged); spans
holds each request’s root RecordedSpan with its nested stages, and
all_spans()/find_span() walk that tree.
events : list[tuple[str, dict[str, object]]]
spans : list[RecordedSpan]
event(name: str, **fields: object) → None
span(name: str, **attributes: object) → FakeSpan
names() → list[str]
all_spans() → list[RecordedSpan]
Every recorded span, roots and descendants, in pre-order.
find_span(name: str) → RecordedSpan | None
The first recorded span with name, searched depth-first, or None.
class ragsage.fakes.FakeEngineKit
Bases: object
Every fake, pre-instantiated and shared, ready to wire a pipeline.
A single object holding one instance of each adapter so a caller (a test or
the CLI) can construct an IngestionPipeline and a
QueryEngine that read and write the same in-memory
stores. snapshot()/restore() persist the whole corpus as one
JSON-able dict.
snapshot() → dict[str, Any]
restore(data: dict[str, Any]) → None
ragsage.fakes.all_fakes() → Iterable[object]
Every fake adapter type — used to assert one exists per port.