Ingestion
IR/text → stores. Structured fact ingestion with a batch manifest ledger, narrative chunk ingestion, and an SME human review-queue state machine — all idempotent.
The ingestion domain takes extraction output (StyledGrid IR, narrative text) and writes it into the stores. It has three lanes: structured (numeric facts), narrative (document chunks), and review (the human-in-the-loop queue that catches everything the deterministic path is unsure about). Every lane is idempotent — re-running an ingest must not double-write.
Package: src/ragspine/ingestion/. Contract: src/ragspine/ingestion/CLAUDE.md.
Layout
Structured ingestion
structured/ingestion.py orchestrates one document end to end:
extract → normalize (glossary) → color tags → upsert. There are two entry points:
ingest_excel(path, store, registry, queue, *, dry_run=False, extractor_version="xlsx_styled@1", manifest=None, batch_id=None)— xlsx-only.ingest_file(path, store, registry, queue, *, dry_run=False, manifest=None, batch_id=None, valid_as_of=None, grid_extractor=None)— the unified multi-format dispatcher: it routes by suffix to the right extractor (xlsx / xlsm / pptx, or PDF via the router) and reuses the shared ingest logic.
Both return an IngestReport (a counts object, not raw numbers):
Prop
Type
Inside, each grid is color-tagged via the active mapping (apply_mapping), turned into
Fact objects, and written with
store.upsert_facts(...). Facts are stamped with their lineage —
source_doc_id, source_locator, source_file_hash, extractor_version,
mapping_version, and review_status=REVIEW_AUTO_APPROVED.
from ragspine.ingestion.structured.ingestion import ingest_file
report = ingest_file("report.xlsx", store, registry, queue)
print(report.n_facts_ingested, report.n_enqueued_review)When ingestion enqueues review instead
The structured path is conservative. A file routes to the review queue rather than auto-ingesting when:
- no grid resolves an entity yet the file contains extractable data — reason "实体无法解析，需人工指认" (entity unresolvable, needs human identification);
- the file has colored cells but the scope has no active color mapping — reason "颜色映射未确认，需 SME 确认图例" (color mapping unconfirmed, SME must confirm the legend);
- a PDF looks like a PowerPoint export (ask for the pptx source) or is a scan needing OCR.
With dry_run=True, extraction and reporting run fully but n_facts_ingested and
n_enqueued_review stay 0 — the store and queue are untouched.
The batch manifest ledger
structured/ingestion_manifest.py records what ran. ManifestStore (sqlite,
manifest_batch + manifest_input tables) opens a batch, logs each input file, and closes
the batch with a final status and duration. Each batch is a ManifestRecord:
| field | meaning |
|---|---|
batch_id | caller-supplied, or auto batch-{uuid4 hex[:12]} |
status | running → done / failed |
inputs | per-file {path, hash, format, …} rows |
n_facts · n_warnings · n_failed | aggregate counts |
duration_s · failures | timing + per-file errors |
API: open_batch(batch_id=None), record_input(...), close_batch(batch_id, status="done"),
get_batch(id), list_batches(). Two observability helpers ride alongside:
compute_metrics(manifest_store, queue, store) (fact totals, review backlog, confidence
buckets, warning rate) and list_versions(store, registry) (active extractor versions +
color mappings).
Where idempotency actually lives. The contract calls the manifest "the guard," and it is the
audit ledger of every run (path / hash / counts / failures). But the literal no-double-write
guarantee comes from the fact store's unique-key upsert (store.upsert_facts, keyed on
dim_key): re-running a batch re-extracts and
re-upserts, and the unique key keeps the store from growing. batch_id is not content-derived
— it is caller-supplied or a random uuid.
Narrative ingestion
The narrative lane is two modules with a clean split:
Pure, deterministic text extraction — zero OCR, zero LLM, no store. extract_narrative(path)
dispatches by suffix (SUPPORTED_SUFFIXES = {.pptx, .pdf, .docx, .docm, .txt}) to the matching
extractor and returns a NarrativeDoc: doc_id, file_hash, a list of NarrativeSegment
(text + source_locator), skipped_pages, and warnings. Locators look like
'slide={N},frame={M}', 'slide={N},notes', 'page={N}', or (for plain text) 'para={N}'.
NarrativeDoc.to_text() joins segments with blank lines — that string is the chunking input
contract.
Plain .txt files take the same narrative path — treated as continuous prose, never coerced
into structured facts. extract_txt_narrative(path) reads UTF-8 (with errors="replace"),
splits on blank lines into paragraph blocks, normalizes each block, and emits one
NarrativeSegment per non-empty block with a 1-based source_locator="para={N}". It is
zero-dependency and deterministic.
Batch orchestration: extract → chunk → write chunk store, idempotent and dry-runnable.
ingest_narrative(inputs, store, *, meta_by_doc=None, dry_run=False, chunker=None) accepts a
folder, a file, or a list, and returns a NarrativeIngestReport (a list of per-file
FileReport, plus counts()). With chunker=None, each file uses
chunk_document(doc.to_text(), doc_meta); an injected Chunker can select parent-child,
sentence-window, layout, domain, or semantic behavior. The resulting hierarchy/window fields are
persisted in the chunk store rather than reconstructed only in memory. Chunks are
written with store.replace_doc_chunks(...) into the
ChunkStore.
Per-file status is one of ingested / skipped / failed / no_text. Idempotency uses a
narrative_doc table (doc_id → file_hash) in the same sqlite DB: if the recorded hash
matches the file, the file is skipped without re-extracting. meta_by_doc keys are
validated against ALLOWED_META_KEYS (title, topic, entity, geography, period,
language, sensitivity, valid_as_of) — unknown fields raise ValueError. The period is
taken from metadata or inferred from the filename via period_from_filename(name).
Sensitivity is applied here: an explicit meta["sensitivity"] wins, otherwise
classify_sensitivity(...) from common runs. This is what
later lets retrieval enforce RESTRICTED isolation.
The review queue
review/review_queue.py is the SME human-review state machine that catches everything
the deterministic path is unsure about — low-confidence OCR, cross-channel conflicts,
unconfirmed color mappings, unresolvable entities. It is sqlite-backed (same DB as the fact
store, different tables: review_item + an append-only review_audit).
The state machine has three string states and two transitions:
pending ──approve──▶ approved (terminal)
pending ──reject───▶ rejected (terminal)STATUS_PENDING = "pending", STATUS_APPROVED = "approved", STATUS_REJECTED = "rejected".
Approved and rejected are terminal — re-processing a terminal item (or acting on a
non-existent one) raises IllegalTransitionError.
A ReviewItem carries reason, payload (JSON), locator, priority (default 100,
lower = reviewed sooner), id, status, actor, note, and corrected_value.
API:
| method | effect |
|---|---|
enqueue(reason, payload, locator, priority=100) -> int | insert a pending item + write an enqueue audit row |
list_pending() -> list[ReviewItem] | pending items, ordered priority ASC, id ASC |
approve(item_id, actor, note=None) | → approved |
reject(item_id, actor, note=None, corrected_value=None) | → rejected (optionally record a correction) |
get(item_id) · audit_trail(item_id) | fetch item / append-only AuditRecord history |
Every transition appends an AuditRecord (enqueue / approve / reject) — the trail is
append-only and never mutated, so review history is fully reconstructable. See
Review queue in the glossary.
Source connectors
Both lanes ingest from local paths by default, but the source is itself a pluggable seam.
ingestion/source/connector.py defines the SourceConnector Protocol — a single
iter_documents() -> Iterable[RawDoc] that yields each document as a frozen RawDoc
(source_doc_id, locator, content: bytes, content_type, metadata).
The default FilesystemConnector is zero-dependency. InMemoryConnector supplies documents
in-process (tests / fixtures). HttpConnector and NotionConnector reach remote knowledge
bases and lazy-import httpx behind the [connectors] extra. Pick one with
make_source_connector(spec, **kwargs) or RAGSPINE_SOURCE_CONNECTOR; third-party connectors
register through the ragspine.source_connectors entry-point group (built-in names win on a
clash). See Extension points → SourceConnector.
from ragspine.ingestion.source.connector import make_source_connector
# None / "none" → None; "filesystem"/"fs" → local walk; "http"/"notion" → remote ([connectors]).
connector = make_source_connector("filesystem")
for raw in connector.iter_documents():
... # raw is a RawDoc: source_doc_id, locator, content bytes, content_type, metadataThe [connectors] extra pulls in only httpx (permissive-licensed, lazy-imported). With no
connector selected, ingestion reads local paths exactly as before — byte-identical.
Invariants this domain upholds
- Idempotent ingestion — structured re-runs upsert on
dim_key; narrative re-runs skip on matchingfile_hash. Re-ingesting never doubles the store. - Provenance preserved — every fact and chunk keeps its
source_doc_id+ locator. - Conservative auto-ingest — anything ambiguous (entity / mapping / confidence / conflict) goes to a human, not silently into the store.
- Append-only audit — review transitions are recorded, never overwritten.
Related
Extraction
Documents → a frozen StyledGrid IR. Style- and color-aware xlsx/pptx/pdf extractors, per-page PDF routing, a versioned color-semantics registry, and dual-channel cross-checking into a review queue.
Storage
The sqlite persistence layer — a numeric fact store and a narrative chunk store, both with full source lineage. The Fact dataclass, the dim_key upsert key, and deterministic found/not-found reads.