From 256d83c81215a7a1f72501c59143e58dfa85ce4a Mon Sep 17 00:00:00 2001 From: Gal Shubeli Date: Thu, 6 Aug 2026 14:28:21 +0300 Subject: [PATCH 1/2] docs: add structured data ingestion design proposal Design for ingesting CSV/JSON/XLSX/Parquet, tables lifted out of PDFs, and existing graphs into the same graph the unstructured path builds, without an LLM call per row. Three load-bearing ideas: - Reduce every structured source to a stream of flat records, so a PDF table is a record stream carrying the PDF's DocumentInfo and an existing graph is a node stream plus an edge stream. - Make the mapping an ontology fragment, so it validates against an existing ontology or bootstraps one when there is none. - Declare entity identity once on the ontology entity type rather than per source, so differently-shaped sources converge on the same nodes at write time instead of via a fuzzy merge pass. Records are persisted as Chunk nodes in the normal lexical graph, which is what lets update()/delete_document() orphan cleanup and all four retrieval paths work on structured data unchanged. Includes a design review against the storage layer that pins three required write properties (name, fact, source_chunk_ids), flags that record chunk uids must key on the run's effective document uid or the update() cutover deletes the chunks it is about to promote, and records the get_document_entity_candidates() scaling ceiling for large sources. Refs: FalkorDB/research#82, FalkorDB/research#65 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- docs/design/structured-ingestion.md | 559 ++++++++++++++++++++++++++++ docs/ingestion.md | 5 + mkdocs.yml | 2 + 3 files changed, 566 insertions(+) create mode 100644 docs/design/structured-ingestion.md diff --git a/docs/design/structured-ingestion.md b/docs/design/structured-ingestion.md new file mode 100644 index 00000000..0ce3180d --- /dev/null +++ b/docs/design/structured-ingestion.md @@ -0,0 +1,559 @@ +# Design: Structured Data Ingestion + +**Status:** Proposed · **Tracking:** [FalkorDB/research#82][i82] (POC) · design from [research#65][i65] · supersedes [GraphRAG-SDK#74][i74] + +[i82]: https://github.com/FalkorDB/research/issues/82 +[i65]: https://github.com/FalkorDB/research/issues/65 +[i74]: https://github.com/FalkorDB/GraphRAG-SDK/issues/74 + +--- + +## 1. Problem + +`rag.ingest(...)` assumes one shape of input: a blob of prose. + +``` +load → chunk → lexical graph → LLM extract → prune → resolve → write → mentions → index +``` + +Structured inputs break three assumptions of that pipeline at once: + +| Assumption (unstructured) | Reality (structured) | +| --- | --- | +| The schema is unknown and must be discovered by an LLM | The schema is already in the header row / JSON keys / source graph | +| Entity identity must be guessed from a surface string | Row identity is explicit and typed | +| Column values are prose to be re-described | Values are typed scalars that must survive as typed graph properties | + +Today a mixed corpus (PDFs + a CRM export + a JSON catalog + a table lifted out of a PDF) needs a +bespoke loader per format, or pre-flattening to text — which discards exactly the structure that +made the source valuable, and spends an LLM call per row re-deriving what was already known. + +### 1.1 The sources this must serve + +The design has to hold for all of these at once, because a real corpus contains all of them: + +1. **`employees.csv`** — one row = one `Person`, with an `org_id` foreign key. +2. **`orgs.csv`** — one row = one `Organization`. Different shape, different columns, same graph. +3. **`transactions.csv`** — one row = an *event between two entities*; the row is a fact, not a thing. +4. **A table lifted out of a PDF** — rows embedded in a document that also contains prose about + the same entities. Table and prose must land in one connected subgraph. +5. **Nested JSON** — objects with sub-objects and arrays; the nesting *is* relationship structure. +6. **An existing graph** — nodes and edges already exist, possibly with their own ontology, + possibly with none. + +### 1.2 Invariants of the current SDK + +Anything we design must respect what is already load-bearing, or retrieval silently degrades. +Each of these was verified against the code: + +- **Every data edge is `RELATES` with a `rel_type` property** + (`extraction_strategies/graph_extraction.py::_relations_to_relationships`). Edge vector search, + `chunk_retrieval`'s neighbour expansion, and the text-to-Cypher prompt all assume it. +- **Node id = `compute_entity_id(name, label)`** → `"acme corp"` + `Organization` → + `acme_corp__organization`. Nodes carry `__Entity__` plus their type label. +- **The lexical graph is mandatory.** `Document -[PART_OF]-> Chunk`, + `Chunk -[NEXT_CHUNK]-> Chunk`, `Entity -[MENTIONED_IN]-> Chunk`. `update()` and + `delete_document()` orphan cleanup is defined *entirely* over this chain + (`delete_orphan_entities` matches `WHERE NOT (e)-[:MENTIONED_IN]->(:Chunk)`), and its + correctness under concurrency depends on mentions being persisted before `pipeline.run()` + returns. +- **`finalize()` embeds `Entity.name`** into the entity vector index (`backfill_entity_embeddings` + selects `e.name`) and **`RELATES.fact`** into the edge vector index (`embed_relationships` + filters `WHERE r.fact IS NOT NULL`). A node without `name` or an edge without `fact` is + unreachable by vector search. +- **`delete_stale_relationships` garbage-collects edges via `RELATES.source_chunk_ids`.** An edge + without that property is never cleaned up. + +--- + +## 2. The design in one paragraph + +**Reduce every structured source to a stream of flat records, and let one declarative mapping +turn records into typed nodes and `RELATES` edges.** The mapping is not a new schema language — +it *is* an ontology fragment plus a column binding, so declaring a mapping declares (or validates +against) the ontology. Identity is declared **once per entity type in the ontology**, not per +source, so differently-shaped CSVs, a PDF table, and a JSON file converge on the same nodes by +construction rather than by fuzzy post-hoc merging. Records are persisted as `Chunk` nodes in the +normal lexical graph, so provenance, `update()`, `delete_document()`, and all four retrieval +paths work on structured data with **zero changes to retrieval**. No LLM is called per row. + +--- + +## 3. The proposals, in build order + +Proposals **#1–#7 are the POC**. **#8–#12** are follow-ups. + +```mermaid +graph LR + subgraph sources + PDF[PDF: prose + table] + CSV1[orgs.csv] + CSV2[employees.csv] + JSON[catalog.json] + G[(existing graph)] + end + PDF --> RS["#1 record stream"] + CSV1 --> RS + CSV2 --> RS + JSON --> RS + G --> RS + RS --> MAP["#2 mapping = ontology fragment + column binding"] + MAP --> ONT{"#2 ontology: validate / merge / bootstrap"} + ONT --> ID["#3 identity per entity type + #4 alias join"] + ID --> W["#5 #6 record chunks · MERGE nodes · RELATES edges"] + W --> R[unchanged retrieval: entity · edge · chunk · cypher] +``` + +--- + +### #1 — `RecordBatch`: one intermediate representation for all structured input + +**What.** A single contract every structured source reduces to: a stream of flat +`dict[str, Any]` plus source metadata. It sits *next to* today's `LoaderStrategy`, not in +place of it. + +```python +class RecordLoaderStrategy(ABC): + async def load_records(self, source: str, ctx: Context) -> RecordBatch: ... + +class RecordBatch(DataModel): + records: Iterable[dict[str, Any]] # streamed, never fully materialised + document_info: DocumentInfo + inferred_types: dict[str, str] # column -> STRING/INTEGER/... hint from the reader +``` + +**Why.** Without it we get one bespoke code path per format — the exact problem #82 is filed +against. With it, a new format is a ~50-line loader and *nothing else changes*. It also dissolves +the two awkward sources: + +- a **table inside a PDF** is a record stream whose `DocumentInfo` is the PDF's, so its rows + become chunks of the *same* `Document` node as the prose — table and prose are connected before + any entity resolution runs; +- an **existing graph** is *two* record streams (nodes and edges), because a graph is just a node + table plus an edge table. + +| Source | Records are | +| --- | --- | +| CSV / TSV / XLSX / Parquet | rows | +| JSONL | one object per line | +| nested JSON | a declared `record_path` (`$.orders[*]`); nested objects flatten to `customer.name`; nested arrays → `LIST` or child records | +| table in a PDF / DOCX | the table's rows, carrying the parent document + section | +| existing graph | node stream + edge stream | + +--- + +### #2 — The mapping DSL, which doubles as an ontology fragment + +**What.** `RecordMapping` / `NodeMapping` / `EdgeMapping`, plus `mapping.to_ontology()`. +Progressive disclosure — the 80% case is one line: + +```python +# One row = one entity. +await rag.ingest("orgs.csv", mapping=Table(node="Organization", key="org_name")) +``` + +The general case is a denormalized row producing several nodes and the edges between them: + +```python +mapping = RecordMapping( + nodes=[ + NodeMapping(label="Person", key="employee_id", name="full_name", + properties={"age": "age", "title": "job_title"}), + NodeMapping(label="Organization", key="org_id", name="org_name"), + ], + edges=[ + EdgeMapping(type="WORKS_AT", source="Person", target="Organization", + properties={"since": "start_date"}), + ], +) +``` + +**Why it doubles as an ontology fragment.** A mapping already declares labels, typed properties +and relation patterns — that is literally the content of `Ontology`. So we do not invent a second +schema language; we project. This single fact is the whole answer to *"the existing graph may or +may not have an ontology"*: + +| Graph state | Behaviour | +| --- | --- | +| Ontology exists, mapping is a subset | validate, proceed | +| Ontology exists, mapping adds labels / attributes | `Ontology.merge()` — the additive path `discovery` already uses | +| Ontology exists, mapping **contradicts** it (type mismatch on an existing attribute) | reject **before any write**, naming the offending `Label.attribute` | +| **No ontology** | the mapping **bootstraps** it — the graph becomes self-describing and text-to-Cypher immediately knows the typed columns | +| No ontology *and* no mapping | out of POC scope; later an inference layer proposes a *draft mapping* (see #11) | + +**Three ways an edge arises**, all in the same DSL: + +1. **Intra-record** — two `NodeMapping`s in one record (the example above). +2. **Foreign key** — the target is defined in *another* source. We `MERGE` a stub node now and + enrich it when that source is ingested, so **ingest order does not matter**. This is what makes + "many differently-shaped CSVs" workable. +3. **Nested containment** — a nested JSON object/array becomes a child node with a declared + `rel_type` back to its parent. + +`CsvMapping` / `JsonMapping` from #65 survive as thin format-flavoured constructors adding reader +options (delimiter, encoding, `record_path`). + +**Reified events.** For `transactions.csv`, where the row *is* the fact, the row becomes a node +(`NodeMapping(label="Transaction", ...)`) with two edges out. Row-as-node vs. row-as-edge is the +mapping author's choice, not a format question; the rule of thumb is "does the fact have +properties, or need to be retrieved on its own?" + +--- + +### #3 — Identity declared once on the entity type, not per source + +This is the crux, and the thing that makes heterogeneous sources compose. + +**What.** Add `identity` to the ontology entity type, defaulting to `name`: + +```python +Entity(label="Organization", identity=["name"]) # default +Entity(label="Product", identity=["sku"]) # a real cross-system business key +``` + +Node id becomes `compute_entity_id(, label)` — **the same function the +unstructured path already uses.** + +**Why.** Separate two things that are usually conflated: + +- **Record key** (`NodeMapping.key`) — what makes *re-ingesting this source* idempotent. + Source-local. Governs the record's chunk id and is stored as an indexed property. +- **Entity identity** — what makes the same real-world thing **one node across all sources**. + +With identity declared on the *type*: + +- a PDF mention of `Acme Corp` and a CSV row with `org_name="Acme Corp"` compute the **same id** + and `MERGE` onto the **same node** — connected and traversable at write time, with no merge + pass, no similarity threshold, and no LLM; +- differently-shaped CSVs converge because every mapping must supply the type's identity + attributes — the *sources* differ, the *identity contract* does not; +- if identity were per-mapping, five sources would mean five identity opinions and a disconnected + graph. + +--- + +### #4 — `AliasMatchResolution`: the deterministic bridge for business-key identity + +**What.** Structured writes store normalized alias handles on the node, built with +`compute_entity_id` so they are directly comparable to unstructured ids: + +``` +alias_ids: ["acme_corp__organization", "acme__organization"] # indexed LIST +``` + +A new `ResolutionStrategy` merges an incoming node onto an existing node when its id appears in +that node's `alias_ids`. + +**Why.** Unstructured extraction can only ever produce a **name** — it can never know an SKU. So +an entity type whose identity is *not* `name` would leave the PDF entity and the CSV row +disconnected. This bridges them, and it is: + +- **deterministic and index-backed** — no LLM, no embeddings; +- **direction-agnostic** — works whether the CSV or the PDF was ingested first; +- **reusable** — because it implements the existing `ResolutionStrategy` ABC, it also works in + the unstructured pipeline and inside `finalize()`. + +Fuzzy merging (`SemanticResolution`, `LLMVerifiedResolution`, `deduplicate_entities()`) stays +available and unchanged, but is no longer on the critical path. **Make the common join exact; +keep the fuzzy one optional.** + +--- + +### #5 — A record is persisted as a `Chunk` + +**What.** Structured records go into the *normal* lexical graph: + +- a `Document` node per source; +- one `Chunk` node per record, with `kind="record"`, the record key as a property, and text that + is a human-readable rendering of the record + (`"Alice Smith · age 34 · Engineer at Acme Corp"`); +- a **deterministic** chunk uid — `sha256( + record_key)` instead of + today's `uuid4()`; +- `MENTIONED_IN` edges from the record's entities to that chunk. + +> **The chunk uid must be derived from the *run's* `DocumentInfo.uid`, never from the canonical +> document id.** During `update()` those differ: the pipeline runs against +> `pending_id = f"{resolved_id}__pending__{uuid4().hex[:8]}"`, and `rollforward_cutover()` +> step 1 calls `delete_document_chunks_and_node(real_id)` *before* promoting the pending. If +> record chunks were keyed on the canonical id, the pending run would `MERGE` onto the **same +> chunk nodes as the live document**, and the cutover would delete the chunks it is about to +> promote — silent data loss. Keying on the effective (pending) uid keeps the two chunk sets +> disjoint, exactly as the `uuid4()` behaviour does today. + +**Why this is the highest-leverage decision in the list.** It looks small and it buys four things +we would otherwise have to build: + +- **`update()` and `delete_document()` work unchanged.** Their cleanup is defined purely over + `Document` / `Chunk` / `MENTIONED_IN` — verified: `delete_orphan_entities` matches + `WHERE NOT (e)-[:MENTIONED_IN]->(:Chunk)`, and `get_document_entity_candidates` walks + `(:__Entity__)-[:MENTIONED_IN]->(:Chunk)<-[:PART_OF]-(:Document)`. Structured data inherits + correct incremental updates for free, including the concurrency invariant that mentions are + written before `run()` returns. +- **Chunk retrieval finds rows.** A CSV of product descriptions is genuinely useful text; a + question answered by "the row itself" works with no new retrieval path. +- **The PDF table connects to the PDF prose automatically** — same `Document`, adjacent chunks. +- **Zero-Loss Data holds** — the original record is recoverable from the graph. + +**Idempotency.** Re-ingesting the same source via `ingest()` resolves to the same canonical +document id, so every record chunk uid is identical and the write rewrites the same nodes instead +of duplicating them. Combined with deterministic node ids (#3), the whole re-ingest is a +semantic no-op. Under `update()` the no-op guarantee comes from the existing **content-hash +short-circuit** instead — an unchanged file never reaches the pending-cutover path at all. + +**Row-level incremental update (follow-up, not POC).** Deterministic uids make it *possible* to +diff record keys against the stored set and touch only rows that changed, instead of rebuilding +the whole document. But `update()`'s pending-cutover is whole-document by construction — its +pending id is randomised — so row-level diffing needs its own path that bypasses the cutover. +Tracked as an open question (§8.8), not part of the POC. + +**Cost knob.** One embedding per record is unacceptable at 10M rows, so +`index_records="auto" | True | False`. `auto` embeds a record only when it carries a free-text +column above a length threshold; otherwise the chunk is still stored — provenance, `update()` +and traversal all intact — just not embedded. + +--- + +### #6 — `StructuredIngestionPipeline`, sharing the load-bearing steps + +**What.** A pipeline that deliberately mirrors the 9-step unstructured one, so the two are +explainable side by side. ♻ marks steps that are the *existing* implementation, factored into a +shared base — **not** copied. + +| # | Step | Note | +| --- | --- | --- | +| 1 | Load records | streamed, bounded batches | +| 2 | Reconcile ontology | `mapping.to_ontology()` → validate / merge / bootstrap / reject | +| 3 | Lexical graph ♻ | `Document` + record `Chunk`s, deterministic uids, content hash | +| 4 | Map records → `GraphData` | pure function, **no LLM** | +| 5 | Coerce + validate types | ontology `Attribute.type` is the source of truth | +| 6 | Prune against ontology ♻ | reuse `IngestionPipeline._prune` verbatim | +| 7 | Resolve | `ExactMatchResolution` + `AliasMatchResolution` (#4) | +| 8 | Write ♻ | `MERGE` nodes; edges as **`RELATES` + `rel_type`** | +| 9 | Mentions + index ♻ | must complete before `run()` returns | + +**Why share rather than copy.** Step 9's ordering is load-bearing for concurrent-update +correctness and already carries a boxed warning comment in `ingestion/pipeline.py`. Copying it +is precisely how that invariant gets silently broken later. + +**Why `RELATES` + `rel_type` rather than a native `:WORKS_AT` edge.** Every retrieval path assumes +`RELATES`. A second edge convention would be a permanent tax on every retrieval strategy. + +**Three properties structured writes must set** (verified against the storage layer — omitting +any one silently breaks a subsystem): + +| Property | On | Consequence if omitted | +| --- | --- | --- | +| `name` | node | `backfill_entity_embeddings` falls back to the raw id → entity vector search degrades | +| `fact` | `RELATES` | `embed_relationships` filters `WHERE r.fact IS NOT NULL` → the edge is **never** embedded and is invisible to edge vector search | +| `source_chunk_ids` | `RELATES` | `delete_stale_relationships` can never garbage-collect the edge → stale facts survive `update()` forever | + +**Type coercion** uses `Attribute.type` (`STRING` / `INTEGER` / `FLOAT` / `BOOLEAN` / `DATE` / +`LIST`), with `on_type_error="skip_value" | "skip_record" | "raise"` (default `skip_value`). The +failure **counts are part of `IngestionResult`**, not debug logging — silent coercion failure is +how structured ingestion quietly produces a garbage graph. + +--- + +### #7 — One entry point + +```python +await rag.ingest("report.pdf") # unstructured — unchanged +await rag.ingest("employees.csv", mapping=mapping) # structured +await rag.ingest(records=[{...}], mapping=mapping) # in-memory +await rag.update("employees.csv", mapping=mapping) # same pipeline, incremental +await rag.ingest([("a.csv", m1), ("b.json", m2), "c.pdf"]) # mixed batch +``` + +Routing rule: **`mapping` present → structured path.** A structured file with *no* mapping keeps +today's text behaviour and logs an actionable hint. + +**Why not auto-route on file extension.** A `.csv` sometimes genuinely *is* prose, and silently +changing what an existing call does is worse than one log line. + +--- + +### #8 — Formats beyond CSV + +JSON / JSONL (`record_path`, flattening, array policy), XLSX, Parquet. Each is a +`RecordLoaderStrategy` (#1) and touches nothing else. Optional dependencies stay lazy and +optional, matching the existing `pdf` / `markdown` extras: +`structured = ["pandas", "openpyxl", "pyarrow"]`. **CSV and JSON must work stdlib-only** — the POC +must not force a pandas install. + +### #9 — PDF-table record stream + +Feed tables detected by the document loaders into #1 as a record stream carrying the parent +`DocumentInfo`. This is where "a table inside a PDF" stops being a hand-written special case. + +### #10 — Existing-graph import + +A node stream plus an edge stream through the same mapping engine: `label` column → ontology +label, `id` column → identity attribute, edge `type` column → `rel_type`. If the source graph has +its own schema, a label map translates it; if not, labels are taken as observed and the ontology +is bootstrapped (#2). Live DB connectors stay out of scope (research#62). + +### #11 — Mapping inference (opt-in, draft only) + +`records -> RecordMapping`: propose labels from the file/sheet name, key from column uniqueness, +types from observed values — then the user confirms. Deliberately a separate layer whose *output +is a mapping*, so the execution engine stays fully deterministic. Extension of research#240. + +### #12 — Property-conflict policy + +Two sources will disagree about `Organization.employee_count`. POC policy: +`on_conflict="last_write_wins" | "keep_existing" | "record_both"`, defaulting to last-write-wins +with the winning source recorded in a `sources: LIST` property. Full per-property provenance is +deferred — it doubles write cost and the POC does not need it. + +--- + +## 4. Coverage of the #82 acceptance criteria + +| Criterion | Satisfied by | +| --- | --- | +| `ingest()` accepts a structured source + mapping, writes typed nodes/edges, no LLM per row | #2, #6, #7 | +| Re-ingesting the same source is a no-op | #3 (deterministic node ids) + #5 (deterministic chunk uids under `ingest()`; the existing content-hash short-circuit under `update()`) | +| Mixed PDF + CSV corpus produces one connected graph | #3 (shared identity) + #4 (alias bridge) + #5 (PDF table shares the `Document`) | +| Retrieval answers a question needing both | §5 — no retrieval changes required | +| Docs page + example | #7, plus `examples/11_structured_ingestion.py` | + +--- + +## 5. Why retrieval needs no changes + +This is the test of whether the ingestion design is correct. + +| Retrieval path | Why structured data is already visible | +| --- | --- | +| Entity vector search | structured nodes carry `name`; `backfill_entity_embeddings` embeds it like any entity | +| `RELATES` edge vector search | structured edges are `RELATES` carrying a built `fact`; `embed_relationships` picks them up | +| Chunk vector + fulltext | record chunks are ordinary `Chunk` nodes (#5) | +| Text-to-Cypher | typed properties are in the ontology (#2), so the prompt advertises them; `rel_type` values come from `EdgeMapping.type` | +| Neighbour expansion | `chunk_retrieval` traverses `RELATES`, which structured edges are (#6) | + +Not one retrieval strategy is touched. + +--- + +## 6. Worked example — the acceptance scenario + +``` +acme_report.pdf prose about Acme + an embedded revenue table +employees.csv employee_id, full_name, age, job_title, org_id +orgs.csv org_id, org_name, hq_country +``` + +```python +await rag.ingest("acme_report.pdf") # prose chunks + table rows, one Document + +await rag.ingest("orgs.csv", mapping=RecordMapping(nodes=[ + NodeMapping(label="Organization", key="org_id", name="org_name", + properties={"hq_country": "hq_country"}), +])) + +await rag.ingest("employees.csv", mapping=RecordMapping( + nodes=[ + NodeMapping(label="Person", key="employee_id", name="full_name", + properties={"age": "age", "title": "job_title"}), + NodeMapping(label="Organization", key="org_id", reference=True), # FK, not re-declared + ], + edges=[EdgeMapping(type="WORKS_AT", source="Person", target="Organization")], +)) + +await rag.finalize() +await rag.completion( + "Which engineers work at the company whose report mentions a Q3 revenue miss?" +) +``` + +The traversal that answers it: + +``` +Chunk("Q3 revenue miss…") <-[MENTIONED_IN]- Organization(acme_corp__organization) + | + | RELATES {rel_type: "WORKS_AT"} + | + Person(alice_smith__person) {title: "Engineer"} + | + | MENTIONED_IN + | + Chunk(record: employees.csv row 41) +``` + +`Organization` is a **single node**: the PDF wrote it by name, `orgs.csv` wrote it by name +(identity defaults to `name`, #3), and `employees.csv` referenced it by foreign key and resolved +to the same identity. Nothing merged it after the fact. + +--- + +## 7. Alternatives considered and rejected + +| Alternative | Why rejected | +| --- | --- | +| Flatten rows to text and reuse the LLM pipeline | An LLM call per row; loses typing; non-deterministic identity — the exact status quo #82 is filed against | +| Native typed edges (`:WORKS_AT`) instead of `RELATES` + `rel_type` | Invisible to edge vector search, neighbour expansion and the text-to-Cypher prompt. A second edge convention taxes every retrieval strategy forever | +| Node id always = record key, merge with unstructured afterwards | The graph is only connected *after* a fuzzy `finalize()` pass. Fails "one connected graph" as an ingest-time property, and makes correctness depend on a similarity threshold | +| A separate `Record` node linked to a semantic entity node | Doubles node count and forces every retrieval path into a two-hop indirection | +| No lexical graph for structured data (typed nodes only) | Loses `update()` / `delete_document()`, loses chunk retrieval over rows, violates Zero-Loss. Cheaper to write, far more expensive to own | +| A separate `rag.ingest_structured()` entry point | Two entry points means two sets of ontology / config / validation semantics that drift | +| Per-mapping identity rules | Five sources → five identity opinions → a disconnected graph. Identity belongs to the entity type | +| Live DB connectors, OCR / images, HTML / MD stripping | Separate issues (research#62, #241, #258) | + +--- + +## 8. Open questions + +1. **`Entity.identity` is a new ontology field.** Persisted ontologies need a default/migration + (`["name"]`). Confirm the ontology-store versioning story. +2. **Multi-column identity** — the join separator and normalization must be pinned so ids are + stable across sources that order columns differently. +3. **`DATE` handling** — accepted input formats, and whether we store epoch, ISO string, or a + native type. +4. **Arrays → `LIST` vs. exploding into child nodes** — needs a default and an override. +5. **Large-source `update()` ceiling — a real gap, not a hypothetical.** + `GraphStore.get_document_entity_candidates()` returns the full `DISTINCT` entity set for a + document in one round-trip with no `LIMIT`, and its docstring explicitly scopes out + "documents with millions of distinct entities … would need a streaming/batched variant." + A 1M-row CSV is exactly one `Document` with ~1M entities. So `update()` / + `delete_document()` on a large structured source hits a documented scaling limit. Either the + POC caps source size, or #5 needs a companion change to batch that call. +6. **POC scale target** — row count, and whether streaming writes need a dedicated batched path + rather than materialising a `GraphData`. +7. **Existing-graph import depth (#10)** — is node-list + edge-list enough for the POC, or is + GraphML / RDF expected? +8. **Row-level incremental update** — worth a dedicated path that bypasses `update()`'s + whole-document pending-cutover, or is whole-document rebuild acceptable for the POC? + +--- + +## 9. Design-review findings + +The two central claims — *"retrieval needs no changes"* and *"orphan cleanup is unchanged"* — +were walked against the code before this document was published. Both hold, with three +concrete requirements and one gap that the walk surfaced: + +| Finding | Where | Folded into | +| --- | --- | --- | +| Structured nodes must set `name`, and structured edges must set `fact` **and** `source_chunk_ids`, or entity embedding / edge embedding / stale-edge GC each silently break | `vector_store.backfill_entity_embeddings`, `vector_store.embed_relationships`, `graph_store.delete_stale_relationships` | #6, "three properties structured writes must set" | +| Record chunk uids keyed on the *canonical* document id would collide with the pending document's chunks during `update()`, and `rollforward_cutover()` would delete the chunks it is about to promote | `graph_store.rollforward_cutover`, `api/main.py::update` (`pending_id`) | #5, callout box | +| `get_document_entity_candidates()` has no `LIMIT` and explicitly scopes out documents with millions of entities — a large CSV is exactly that | `graph_store.get_document_entity_candidates` | §8.5 | +| Orphan cleanup is defined purely over `MENTIONED_IN` → `Chunk` → `PART_OF` → `Document`, so record-as-chunk inherits it unchanged | `graph_store.delete_orphan_entities` | #5 | + +--- + +## Appendix A — Implementation phasing + +| Phase | Contents | +| --- | --- | +| **P1 — skeleton** | `RecordLoaderStrategy` + `RecordBatch`, `CsvRecordLoader`, `RecordMapping` / `NodeMapping` / `EdgeMapping`, `mapping.to_ontology()` (#1, #2) | +| **P2 — pipeline** | Extract the shared lexical-graph / prune / write / mentions base out of `IngestionPipeline`; add `StructuredIngestionPipeline`; deterministic record chunk uids (#5, #6) | +| **P3 — identity** | `Entity.identity`, alias handles, `AliasMatchResolution` (#3, #4) | +| **P4 — formats** | JSON / JSONL / XLSX / Parquet; PDF-table record stream; node-list + edge-list graph import (#8, #9, #10) | +| **P5 — surface** | `ingest(mapping=...)` / `update(mapping=...)` routing, `IngestionResult` counters, `examples/11_structured_ingestion.py`, docs page, mixed-corpus integration test (#7) | + +## Appendix B — Notes + +- The example filename in #65 (`08_structured_ingestion.py`) is taken; `examples/` is already at + `10_ontology_discovery.py`, so the new example is `11_structured_ingestion.py`. +- `GraphSchema` / `EntityType` / `PropertyType` in #65 are the pre-v1.2 names. This document uses + the current `Ontology` / `Entity` / `Attribute` naming. diff --git a/docs/ingestion.md b/docs/ingestion.md index 1bb0b6c9..a9fef123 100644 --- a/docs/ingestion.md +++ b/docs/ingestion.md @@ -4,6 +4,11 @@ When you call `rag.ingest("document.txt")`, the SDK transforms your raw text int This document explains what each step does, why it exists, and how to tune it. +!!! info "Structured sources (CSV, JSON, tables, existing graphs)" + This page describes the **unstructured** path — prose in, LLM extraction, graph out. + Structured inputs skip LLM extraction entirely and are covered by a separate proposal: + [Design: Structured Data Ingestion](design/structured-ingestion.md). + --- ## The Big Picture diff --git a/mkdocs.yml b/mkdocs.yml index aea56ad6..29a50325 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -54,3 +54,5 @@ nav: - Ontology Evolution: ontology-evolution.md - Benchmark: benchmark.md - API Reference: api-reference.md + - Design Proposals: + - Structured Data Ingestion: design/structured-ingestion.md From 8b2e1fe0378145ee33aa2d5009d80e8630194042 Mon Sep 17 00:00:00 2001 From: Gal Shubeli Date: Thu, 6 Aug 2026 16:39:13 +0300 Subject: [PATCH 2/2] poc: spike each structured-ingestion proposal, fold findings into the design Five throwaway spikes under poc/structured-ingestion/, each answering one open question from docs/design/structured-ingestion.md against the real GraphStore, IngestionPipeline and a live FalkorDB. No LLM, no API keys. run_all.py passes. Four of the five falsified something in the design: - s1 RecordBatch: pydantic keeps Iterable[dict] lazy but one-shot, and the pipeline iterates records twice (step 3 chunks, step 4 mapping). Measured "step 3 saw 10 records, step 4 saw 0" with no error -> a silent zero-row ingest. RecordBatch becomes a stream factory. - s2 mapping DSL: label-addressed edges cannot express a record holding two nodes of the same label; transactions.csv produced a silent self-loop. Nodes now carry an alias. Adds two to_ontology() guards (reserved attribute names, reference-only labels). - s3 identity: identity=["name"] was the design's default and loses. A normalised FK carries the target's key and not its name, so the mapping cannot compute the identity it points at: 2 Acme nodes, 0 people reachable. Key + alias_ids is the only policy that yields one connected graph, making proposal #4 critical-path rather than optional. - s4 record-as-chunk: confirms the predicted update() data-loss trap empirically. Canonical-keyed chunk uids go 3 chunks -> 0 through rollforward_cutover() with no exception; effective-uid keying survives. The three cleanup primitives behave exactly as claimed. - s5 pipeline seam: the steps are reusable verbatim, but IngestionPipeline's __init__ demands a chunker and an LLM extractor the structured path lacks. Share a LexicalGraphWriter base instead of subclassing, and suppress NEXT_CHUNK for record chunks. poc/ is outside the wheel, pytest testpaths and CI (which runs with working-directory: graphrag_sdk and lints only src/), so it ships nothing. No src changes. mkdocs build --strict is unchanged (one pre-existing warning). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- docs/design/structured-ingestion.md | 278 ++++++++++++--- poc/structured-ingestion/FINDINGS.md | 125 +++++++ poc/structured-ingestion/README.md | 42 +++ poc/structured-ingestion/_harness/__init__.py | 0 poc/structured-ingestion/_harness/env.py | 119 +++++++ .../_harness/fixtures/acme_report.txt | 5 + .../_harness/fixtures/catalog.json | 44 +++ .../_harness/fixtures/employees.csv | 4 + .../_harness/fixtures/orgs.csv | 3 + .../_harness/fixtures/transactions.csv | 3 + poc/structured-ingestion/run_all.py | 34 ++ .../s1_record_stream/NOTES.md | 63 ++++ .../s1_record_stream/spike.py | 176 +++++++++ .../s2_mapping_dsl/NOTES.md | 71 ++++ .../s2_mapping_dsl/spike.py | 336 ++++++++++++++++++ poc/structured-ingestion/s3_identity/NOTES.md | 67 ++++ poc/structured-ingestion/s3_identity/spike.py | 314 ++++++++++++++++ .../s4_record_as_chunk/NOTES.md | 63 ++++ .../s4_record_as_chunk/spike.py | 240 +++++++++++++ .../s5_pipeline_seam/NOTES.md | 59 +++ .../s5_pipeline_seam/spike.py | 243 +++++++++++++ 21 files changed, 2239 insertions(+), 50 deletions(-) create mode 100644 poc/structured-ingestion/FINDINGS.md create mode 100644 poc/structured-ingestion/README.md create mode 100644 poc/structured-ingestion/_harness/__init__.py create mode 100644 poc/structured-ingestion/_harness/env.py create mode 100644 poc/structured-ingestion/_harness/fixtures/acme_report.txt create mode 100644 poc/structured-ingestion/_harness/fixtures/catalog.json create mode 100644 poc/structured-ingestion/_harness/fixtures/employees.csv create mode 100644 poc/structured-ingestion/_harness/fixtures/orgs.csv create mode 100644 poc/structured-ingestion/_harness/fixtures/transactions.csv create mode 100644 poc/structured-ingestion/run_all.py create mode 100644 poc/structured-ingestion/s1_record_stream/NOTES.md create mode 100644 poc/structured-ingestion/s1_record_stream/spike.py create mode 100644 poc/structured-ingestion/s2_mapping_dsl/NOTES.md create mode 100644 poc/structured-ingestion/s2_mapping_dsl/spike.py create mode 100644 poc/structured-ingestion/s3_identity/NOTES.md create mode 100644 poc/structured-ingestion/s3_identity/spike.py create mode 100644 poc/structured-ingestion/s4_record_as_chunk/NOTES.md create mode 100644 poc/structured-ingestion/s4_record_as_chunk/spike.py create mode 100644 poc/structured-ingestion/s5_pipeline_seam/NOTES.md create mode 100644 poc/structured-ingestion/s5_pipeline_seam/spike.py diff --git a/docs/design/structured-ingestion.md b/docs/design/structured-ingestion.md index 0ce3180d..69a1a3e0 100644 --- a/docs/design/structured-ingestion.md +++ b/docs/design/structured-ingestion.md @@ -1,6 +1,9 @@ # Design: Structured Data Ingestion -**Status:** Proposed · **Tracking:** [FalkorDB/research#82][i82] (POC) · design from [research#65][i65] · supersedes [GraphRAG-SDK#74][i74] +**Status:** Proposed, spike-validated · **Tracking:** [FalkorDB/research#82][i82] (POC) · design from [research#65][i65] · supersedes [GraphRAG-SDK#74][i74] + +The proposals in §3 were tested by five throwaway spikes against a live FalkorDB before this +document settled — see [§10](#10-spike-results). Four of them corrected something here. [i82]: https://github.com/FalkorDB/research/issues/82 [i65]: https://github.com/FalkorDB/research/issues/65 @@ -117,11 +120,28 @@ class RecordLoaderStrategy(ABC): async def load_records(self, source: str, ctx: Context) -> RecordBatch: ... class RecordBatch(DataModel): - records: Iterable[dict[str, Any]] # streamed, never fully materialised + open_records: Callable[[], Iterator[dict[str, Any]]] # a stream *factory* — see below document_info: DocumentInfo inferred_types: dict[str, str] # column -> STRING/INTEGER/... hint from the reader + record_count: int | None = None # when the loader knows it cheaply; None when streaming + + def __iter__(self) -> Iterator[dict[str, Any]]: + return self.open_records() ``` +!!! warning "A factory, not an iterable — [spike s1][s1] corrected this" + The obvious signature `records: Iterable[dict[str, Any]]` **does not work**. Pydantic v2 keeps + it lazy (good: 200k rows cost ~0 MB vs 71.6 MB materialised) but replaces it with a *one-shot* + `ValidatorIterator`. #6 iterates records twice — step 3 builds record chunks, step 4 maps + records to `GraphData` — and the measured result is `step 3 saw 10 records, step 4 saw 0`, with + **no error raised**: a silent zero-row ingest. The annotation also erases list-ness, so `len()` + raises even when the caller passed a list, leaving no cheap count for progress reporting. + A factory is re-iterable by construction and verified `model_dump()`-safe. + + Loaders therefore hand over a *re-openable* source (reopen the file, re-run the cursor). Where a + source genuinely cannot be read twice, the loader spools once and closes over the buffer — which + makes the memory cost explicit at the loader instead of silently corrupting the write. + **Why.** Without it we get one bespoke code path per format — the exact problem #82 is filed against. With it, a new format is a ~50-line loader and *nothing else changes*. It also dissolves the two awkward sources: @@ -152,22 +172,33 @@ Progressive disclosure — the 80% case is one line: await rag.ingest("orgs.csv", mapping=Table(node="Organization", key="org_name")) ``` -The general case is a denormalized row producing several nodes and the edges between them: +The general case is a denormalized row producing several nodes and the edges between them. Each +`NodeMapping` carries an **alias** — a handle unique *within the record* — and edges address +aliases, never labels: ```python mapping = RecordMapping( nodes=[ - NodeMapping(label="Person", key="employee_id", name="full_name", + NodeMapping(alias="employee", label="Person", key="employee_id", name="full_name", properties={"age": "age", "title": "job_title"}), - NodeMapping(label="Organization", key="org_id", name="org_name"), + NodeMapping(alias="employer", label="Organization", key="org_id", reference=True), ], edges=[ - EdgeMapping(type="WORKS_AT", source="Person", target="Organization", + EdgeMapping(type="WORKS_AT", source="employee", target="employer", properties={"since": "start_date"}), ], ) ``` +!!! warning "Edges address aliases, not labels — [spike s2][s2] corrected this" + The obvious `EdgeMapping(source="Person", target="Organization")` cannot express a record + containing **two nodes of the same label**. Run against `transactions.csv` — a buyer and a + seller, both `Organization` — label addressing produced a **self-loop** (`ORG-7 -> ORG-7` + instead of `ORG-7 -> ORG-42`), silently. Buyer/seller, manager/report, parent/subsidiary and + origin/destination are the standard shape of transactional data, not an edge case. + + `alias` defaults to the label, so the single-node 80% case above never mentions it. + **Why it doubles as an ontology fragment.** A mapping already declares labels, typed properties and relation patterns — that is literally the content of `Ontology`. So we do not invent a second schema language; we project. This single fact is the whole answer to *"the existing graph may or @@ -181,6 +212,18 @@ may not have an ontology"*: | **No ontology** | the mapping **bootstraps** it — the graph becomes self-describing and text-to-Cypher immediately knows the typed columns | | No ontology *and* no mapping | out of POC scope; later an inference layer proposes a *draft mapping* (see #11) | +**Two guards `to_ontology()` must apply** (both found by [spike s2][s2]): + +1. **Reject SDK-reserved attribute names.** A mapping declaring `properties={"description": ..., + "id": ...}` generates an ontology that shadows values the SDK writes on every node. + `to_ontology()` rejects `_RESERVED_ATTRIBUTE_NAMES - _SDK_MANAGED_ATTRIBUTE_NAMES` + (`core/models.py`), naming the offending `Label.attribute` — the same "reject before any write" + rule already applied to contradictions. +2. **Emit stubs for reference-only labels.** `Ontology._warn_on_undeclared_pattern_labels` fires + when a relation pattern names a label not in `entities` — which is exactly what a foreign-key + reference produces. Left alone, every structured ingest logs warnings that train users to ignore + real ones. + **Three ways an edge arises**, all in the same DSL: 1. **Intra-record** — two `NodeMapping`s in one record (the example above). @@ -204,16 +247,37 @@ properties, or need to be retrieved on its own?" This is the crux, and the thing that makes heterogeneous sources compose. -**What.** Add `identity` to the ontology entity type, defaulting to `name`: +**What.** Add `identity` to the ontology entity type. **For structured sources it defaults to the +record key**, not to `name`: ```python -Entity(label="Organization", identity=["name"]) # default -Entity(label="Product", identity=["sku"]) # a real cross-system business key +Entity(label="Organization", identity=["org_id"]) # structured default: the record key +Entity(label="Product", identity=["sku"]) # a real cross-system business key ``` Node id becomes `compute_entity_id(, label)` — **the same function the unstructured path already uses.** +!!! danger "`identity=["name"]` as the default is wrong — [spike s3][s3] inverted this" + The first draft of this design defaulted identity to `name`, reasoning that a PDF mention of + `Acme Corp` and a CSV row `org_name="Acme Corp"` would then compute the same id and merge for + free. Measured on the #82 acceptance corpus, that loses: + + | Policy | Acme nodes | #82 traversal | + | --- | --- | --- | + | name-first (`identity=["name"]`) | 2 | **0 people reachable** | + | key-only | 2 | **0 people reachable** | + | key + `alias_ids` (#4) | **1** | **2 people reachable** | + + The reason is that `employees.csv` is a normalised table: it references its organisation by + `org_id=ORG-42` and has **no `org_name` column**. Under name-first identity the mapping cannot + compute the identity of the entity it is pointing at, so the rule in #2 — "every mapping must + supply the type's identity attributes" — is *unsatisfiable for any foreign key*, which is the + most common structured shape there is. + + The failure is silent: a stub node accumulates all the `WORKS_AT` edges while the real Acme + node holds the prose, and the two never meet. + **Why.** Separate two things that are usually conflated: - **Record key** (`NodeMapping.key`) — what makes *re-ingesting this source* idempotent. @@ -222,17 +286,27 @@ unstructured path already uses.** With identity declared on the *type*: -- a PDF mention of `Acme Corp` and a CSV row with `org_name="Acme Corp"` compute the **same id** - and `MERGE` onto the **same node** — connected and traversable at write time, with no merge - pass, no similarity threshold, and no LLM; -- differently-shaped CSVs converge because every mapping must supply the type's identity - attributes — the *sources* differ, the *identity contract* does not; +- foreign-key references land on the right node **regardless of ingest order** — measured: both + ingest orders converge to an identical graph for every policy tested; +- differently-shaped CSVs converge because every mapping supplies the type's identity attributes — + the *sources* differ, the *identity contract* does not; - if identity were per-mapping, five sources would mean five identity opinions and a disconnected graph. +Bridging key-identified structured nodes back to name-identified unstructured mentions is #4 — +which is consequently **on the critical path, not an optional extra**. + +!!! note "Free to prototype" + `Entity(label="Product", identity=["sku"])` already works today: `DataModel.Config.extra = + "allow"` carries the field and it survives `model_dump()`, so it persists to `ontology.json`. + Two consequences — `identity` can be prototyped with zero `src` changes, and promoting it to a + declared field later will not break ontologies persisted in the meantime. It must still become + a declared field (defaulting to `["name"]` for unstructured-only types) so it is validated + rather than being a silent typo sink. + --- -### #4 — `AliasMatchResolution`: the deterministic bridge for business-key identity +### #4 — `AliasMatchResolution`: the deterministic bridge, on the critical path **What.** Structured writes store normalized alias handles on the node, built with `compute_entity_id` so they are directly comparable to unstructured ids: @@ -244,15 +318,28 @@ alias_ids: ["acme_corp__organization", "acme__organization"] # indexed LIST A new `ResolutionStrategy` merges an incoming node onto an existing node when its id appears in that node's `alias_ids`. -**Why.** Unstructured extraction can only ever produce a **name** — it can never know an SKU. So -an entity type whose identity is *not* `name` would leave the PDF entity and the CSV row -disconnected. This bridges them, and it is: +**Why.** Unstructured extraction can only ever produce a **name** — it can never know an `org_id` +or an SKU. Since #3 identifies structured entities by their record key, *every* entity type that +appears in both a document and a table needs this bridge. [Spike s3][s3] measured it as the only +configuration that yields one Acme node and a working `prose-chunk -> Org -> WORKS_AT -> Person` +traversal. It is: -- **deterministic and index-backed** — no LLM, no embeddings; +- **deterministic and index-backed** — no LLM, no embeddings; the spike's implementation is four + Cypher statements per merged pair; - **direction-agnostic** — works whether the CSV or the PDF was ingested first; - **reusable** — because it implements the existing `ResolutionStrategy` ABC, it also works in the unstructured pipeline and inside `finalize()`. +!!! warning "Order does not matter, but *presence* does — [spike s3][s3]" + The bridge is built from whichever source declares both the key and the name (the dimension + table — `orgs.csv` here). With prose + `employees.csv` and **no** `orgs.csv`, the result + degrades to 2 Acme nodes and 0 reachable people, because nothing ever carried `ORG-42` and + `"Acme Corp"` in the same record. + + This is an acceptable requirement, but it must be *visible*: when a mapping references a label + that no ingested source has declared, `IngestionResult` reports the count of unbridged stubs + rather than leaving the user with a quietly disconnected graph. + Fuzzy merging (`SemanticResolution`, `LLMVerifiedResolution`, `deduplicate_entities()`) stays available and unchanged, but is no longer on the critical path. **Make the common join exact; keep the fuzzy one optional.** @@ -271,22 +358,39 @@ keep the fuzzy one optional.** today's `uuid4()`; - `MENTIONED_IN` edges from the record's entities to that chunk. -> **The chunk uid must be derived from the *run's* `DocumentInfo.uid`, never from the canonical -> document id.** During `update()` those differ: the pipeline runs against -> `pending_id = f"{resolved_id}__pending__{uuid4().hex[:8]}"`, and `rollforward_cutover()` -> step 1 calls `delete_document_chunks_and_node(real_id)` *before* promoting the pending. If -> record chunks were keyed on the canonical id, the pending run would `MERGE` onto the **same -> chunk nodes as the live document**, and the cutover would delete the chunks it is about to -> promote — silent data loss. Keying on the effective (pending) uid keeps the two chunk sets -> disjoint, exactly as the `uuid4()` behaviour does today. +!!! danger "Chunk uids must key on the *effective* document uid — confirmed by [spike s4][s4]" + **The chunk uid must be derived from the *run's* `DocumentInfo.uid`, never from the canonical + document id.** During `update()` those differ: the pipeline runs against + `pending_id = f"{resolved_id}__pending__{uuid4().hex[:8]}"`, and `rollforward_cutover()` + step 1 calls `delete_document_chunks_and_node(real_id)` *before* promoting the pending. + + This was predicted by reading the code and then **measured against a real FalkorDB through the + real `GraphStore`**: + + | chunk uid keyed on | chunks before `update()` | chunk nodes shared with pending | **after cutover** | + | --- | --- | --- | --- | + | canonical document id | 3 | 3 | **0** | + | effective (pending) uid | 3 | 0 | **3** | + + With canonical keying the pending run `MERGE`s onto the live document's chunk nodes, the cutover + deletes them, and an **empty document is promoted with no exception raised**. The precondition + guard in `rollforward_cutover` does not help — the pending `Document` node exists; only its + chunks have been destroyed. + + Today's `uuid4()` uids are accidentally immune, which is exactly why making them deterministic + is the dangerous part of this proposal. Keying on the effective uid keeps the two chunk sets + disjoint while remaining deterministic *within* a run — which is all that re-ingest idempotency + requires. **Why this is the highest-leverage decision in the list.** It looks small and it buys four things we would otherwise have to build: - **`update()` and `delete_document()` work unchanged.** Their cleanup is defined purely over - `Document` / `Chunk` / `MENTIONED_IN` — verified: `delete_orphan_entities` matches - `WHERE NOT (e)-[:MENTIONED_IN]->(:Chunk)`, and `get_document_entity_candidates` walks - `(:__Entity__)-[:MENTIONED_IN]->(:Chunk)<-[:PART_OF]-(:Document)`. Structured data inherits + `Document` / `Chunk` / `MENTIONED_IN`. [Spike s4][s4] ran the real primitives against record + chunks: `get_document_entity_candidates()` found all 5 record-chunk entities, + `delete_stale_relationships()` GC'd exactly the deleted row's fact via `source_chunk_ids`, and + `delete_orphan_entities()` removed exactly the vanished row's `Person` **and** the organisation + that lost its last mention — leaving the other two people untouched. Structured data inherits correct incremental updates for free, including the concurrency invariant that mentions are written before `run()` returns. - **Chunk retrieval finds rows.** A CSV of product descriptions is genuinely useful text; a @@ -323,8 +427,8 @@ shared base — **not** copied. | --- | --- | --- | | 1 | Load records | streamed, bounded batches | | 2 | Reconcile ontology | `mapping.to_ontology()` → validate / merge / bootstrap / reject | -| 3 | Lexical graph ♻ | `Document` + record `Chunk`s, deterministic uids, content hash | -| 4 | Map records → `GraphData` | pure function, **no LLM** | +| 3 | Lexical graph ♻ | `Document` + record `Chunk`s, deterministic uids, content hash, `link_sequential=False` | +| 4 | Map records → `GraphData` | pure function, **no LLM** — a second, independent pass over the records | | 5 | Coerce + validate types | ontology `Attribute.type` is the source of truth | | 6 | Prune against ontology ♻ | reuse `IngestionPipeline._prune` verbatim | | 7 | Resolve | `ExactMatchResolution` + `AliasMatchResolution` (#4) | @@ -335,6 +439,33 @@ shared base — **not** copied. correctness and already carries a boxed warning comment in `ingestion/pipeline.py`. Copying it is precisely how that invariant gets silently broken later. +!!! warning "Share a base class — do **not** subclass `IngestionPipeline` — [spike s5][s5]" + The three ♻ steps are reusable **verbatim**: `_build_lexical_graph` already consumes + `TextChunks`, which is exactly what #5 turns records into, and `_prune` / `_write_mentions` + depend only on `graph_store`. Both factorings were run end-to-end and produced identical graphs. + + But `IngestionPipeline.__init__` requires `loader, chunker, extractor, resolver, graph_store, + vector_store`. A structured pipeline has a *record* loader, **no chunker** (records are already + chunks) and **no LLM extractor** (mapping is deterministic — the entire point of this design). + Subclassing means passing `None` for two of them and hoping nothing ever touches them; it works + today only by accident of which methods are called, and turns any future change to + `IngestionPipeline.run()` into a latent `AttributeError` on the structured path. + + **Extract the three methods into a `LexicalGraphWriter` base that depends only on + `graph_store`**, inherited by both pipelines. `IngestionPipeline`'s public surface is unchanged, + and step 9's ordering still lives in exactly one place. + +!!! warning "`NEXT_CHUNK` must be suppressed for records — [spike s5][s5]" + `_build_lexical_graph` unconditionally chains `prev_chunk -[NEXT_CHUNK]-> chunk`. Reused for + records that asserts a sequential relationship **between unrelated table rows** — N-1 edges per + source, so 1M meaningless edges for a 1M-row CSV — while `retrieval/strategies/cypher_generation.py` + actively tells the LLM that `NEXT_CHUNK` "connects Chunk to next sequential Chunk". Row order in + a CSV is usually incidental, so these edges are not merely useless; they encode a false claim. + + Fix: `_build_lexical_graph(..., link_sequential: bool = True)` and pass `False` for record + chunks. This is the **only** signature change the whole seam needs — the default preserves + today's behaviour exactly. + **Why `RELATES` + `rel_type` rather than a native `:WORKS_AT` edge.** Every retrieval path assumes `RELATES`. A second edge convention would be a permanent tax on every retrieval strategy. @@ -446,21 +577,23 @@ orgs.csv org_id, org_name, hq_country ```python await rag.ingest("acme_report.pdf") # prose chunks + table rows, one Document +# The dimension table declares Organization: key AND name, so it emits the alias bridge. await rag.ingest("orgs.csv", mapping=RecordMapping(nodes=[ - NodeMapping(label="Organization", key="org_id", name="org_name", + NodeMapping(alias="org", label="Organization", key="org_id", name="org_name", properties={"hq_country": "hq_country"}), ])) await rag.ingest("employees.csv", mapping=RecordMapping( nodes=[ - NodeMapping(label="Person", key="employee_id", name="full_name", + NodeMapping(alias="employee", label="Person", key="employee_id", name="full_name", properties={"age": "age", "title": "job_title"}), - NodeMapping(label="Organization", key="org_id", reference=True), # FK, not re-declared + # FK: employees.csv has org_id and no org_name — a stub, not a re-declaration. + NodeMapping(alias="employer", label="Organization", key="org_id", reference=True), ], - edges=[EdgeMapping(type="WORKS_AT", source="Person", target="Organization")], + edges=[EdgeMapping(type="WORKS_AT", source="employee", target="employer")], )) -await rag.finalize() +await rag.finalize() # AliasMatchResolution merges the prose Acme into the keyed Acme await rag.completion( "Which engineers work at the company whose report mentions a Q3 revenue miss?" ) @@ -469,20 +602,24 @@ await rag.completion( The traversal that answers it: ``` -Chunk("Q3 revenue miss…") <-[MENTIONED_IN]- Organization(acme_corp__organization) - | +Chunk("Q3 revenue miss…") <-[MENTIONED_IN]- Organization(org-42__organization) + | {name: "Acme Corp", + | alias_ids: ["acme_corp__organization"]} | RELATES {rel_type: "WORKS_AT"} | - Person(alice_smith__person) {title: "Engineer"} + Person(e-1__person) {title: "Engineer"} | | MENTIONED_IN | - Chunk(record: employees.csv row 41) + Chunk(record: employees.csv row E-1) ``` -`Organization` is a **single node**: the PDF wrote it by name, `orgs.csv` wrote it by name -(identity defaults to `name`, #3), and `employees.csv` referenced it by foreign key and resolved -to the same identity. Nothing merged it after the fact. +`Organization` is a **single node**. `orgs.csv` wrote it under its record key `ORG-42` and attached +`alias_ids: ["acme_corp__organization"]`; `employees.csv` referenced it by foreign key and landed on +the same id without ever seeing the company's name; the PDF wrote `acme_corp__organization` from +prose, which `AliasMatchResolution` (#4) merged in. This exact traversal is what [spike s3][s3] +measured — 1 Acme node, 2 engineers reachable — and it is the *only* one of three candidate identity +policies that produced it. --- @@ -503,8 +640,13 @@ to the same identity. Nothing merged it after the fact. ## 8. Open questions -1. **`Entity.identity` is a new ontology field.** Persisted ontologies need a default/migration - (`["name"]`). Confirm the ontology-store versioning story. +Questions 1–4 and 6–8 remain open. **§8.9 was opened by the spikes.** Several earlier questions +were *closed* by them — see §10. + +1. **`Entity.identity` is a new ontology field.** Persisted ontologies need a default/migration. + [Spike s2][s2] confirms `extra = "allow"` already carries and persists the field, so old + `ontology.json` files stay loadable; the remaining question is the declared default + (`["name"]` for unstructured-only types) and the ontology-store versioning story. 2. **Multi-column identity** — the join separator and normalization must be pinned so ids are stable across sources that order columns differently. 3. **`DATE` handling** — accepted input formats, and whether we store epoch, ISO string, or a @@ -523,6 +665,10 @@ to the same identity. Nothing merged it after the fact. GraphML / RDF expected? 8. **Row-level incremental update** — worth a dedicated path that bypasses `update()`'s whole-document pending-cutover, or is whole-document rebuild acceptable for the POC? +9. **Unbridged-stub reporting.** [Spike s3][s3] showed the alias bridge needs *some* source + carrying both key and name. What is the right surface for "this mapping referenced + `Organization`, and nothing has declared it yet" — a counter in `IngestionResult`, a warning, + or a `finalize()`-time report? --- @@ -535,19 +681,51 @@ concrete requirements and one gap that the walk surfaced: | Finding | Where | Folded into | | --- | --- | --- | | Structured nodes must set `name`, and structured edges must set `fact` **and** `source_chunk_ids`, or entity embedding / edge embedding / stale-edge GC each silently break | `vector_store.backfill_entity_embeddings`, `vector_store.embed_relationships`, `graph_store.delete_stale_relationships` | #6, "three properties structured writes must set" | -| Record chunk uids keyed on the *canonical* document id would collide with the pending document's chunks during `update()`, and `rollforward_cutover()` would delete the chunks it is about to promote | `graph_store.rollforward_cutover`, `api/main.py::update` (`pending_id`) | #5, callout box | +| Record chunk uids keyed on the *canonical* document id would collide with the pending document's chunks during `update()`, and `rollforward_cutover()` would delete the chunks it is about to promote | `graph_store.rollforward_cutover`, `api/main.py::update` (`pending_id`) | #5, callout box — **since confirmed empirically**, see §10 | | `get_document_entity_candidates()` has no `LIMIT` and explicitly scopes out documents with millions of entities — a large CSV is exactly that | `graph_store.get_document_entity_candidates` | §8.5 | | Orphan cleanup is defined purely over `MENTIONED_IN` → `Chunk` → `PART_OF` → `Document`, so record-as-chunk inherits it unchanged | `graph_store.delete_orphan_entities` | #5 | --- +## 10. Spike results + +The proposals above were then tested. Five throwaway spikes in +[`poc/structured-ingestion/`][poc] each answer one open question — against the real `GraphStore`, +`IngestionPipeline` and a live FalkorDB, with no LLM and no API keys. `python run_all.py` runs +them; all five pass. Full write-ups live in each spike's `NOTES.md`, rolled up in +[`FINDINGS.md`][findings]. + +**Four of the five falsified something in this document.** One inverted a headline decision. + +| Spike | Question | Outcome | +| --- | --- | --- | +| [s1][s1] | Can `RecordBatch` hold a lazy stream? | **Amended #1** — pydantic keeps it lazy but *one-shot*; #6's two passes measured `step 3 saw 10 records, step 4 saw 0` with no error. Now a stream **factory** | +| [s2][s2] | Which DSL shape expresses all four record shapes? | **Amended #2** — label-addressed edges produce a silent **self-loop** on `transactions.csv`; nodes now carry an `alias`. Two `to_ontology()` guards added | +| [s3][s3] | Do the identity policies produce one connected graph? | **Inverted #3** — `identity=["name"]` yields 2 Acme nodes and **0** reachable people, because a normalised FK carries no name. Key + `alias_ids` is the only policy that works | +| [s4][s4] | Is record-as-chunk really free? Is the cutover trap real? | **Confirmed #5** — canonical-keyed uids go 3 chunks → **0** through `rollforward_cutover()`, silently. Effective-uid keying survives. All three cleanup primitives behave as claimed | +| [s5][s5] | Can the pipeline steps be reused? | **Amended #6** — reusable verbatim, but `__init__` demands a chunker and an extractor the structured path does not have. Share a `LexicalGraphWriter` base; suppress `NEXT_CHUNK` for records | + +**What did not change.** Record-as-chunk, `RELATES` + `rel_type`, mapping-as-ontology-fragment, +deterministic no-LLM mapping, and "no retrieval strategy is touched" all survived contact with the +database. Every correction above is to a *signature or a default*, not to the shape of the design. + +[poc]: https://github.com/FalkorDB/GraphRAG-SDK/tree/main/poc/structured-ingestion +[findings]: https://github.com/FalkorDB/GraphRAG-SDK/blob/main/poc/structured-ingestion/FINDINGS.md +[s1]: https://github.com/FalkorDB/GraphRAG-SDK/blob/main/poc/structured-ingestion/s1_record_stream/NOTES.md +[s2]: https://github.com/FalkorDB/GraphRAG-SDK/blob/main/poc/structured-ingestion/s2_mapping_dsl/NOTES.md +[s3]: https://github.com/FalkorDB/GraphRAG-SDK/blob/main/poc/structured-ingestion/s3_identity/NOTES.md +[s4]: https://github.com/FalkorDB/GraphRAG-SDK/blob/main/poc/structured-ingestion/s4_record_as_chunk/NOTES.md +[s5]: https://github.com/FalkorDB/GraphRAG-SDK/blob/main/poc/structured-ingestion/s5_pipeline_seam/NOTES.md + +--- + ## Appendix A — Implementation phasing | Phase | Contents | | --- | --- | -| **P1 — skeleton** | `RecordLoaderStrategy` + `RecordBatch`, `CsvRecordLoader`, `RecordMapping` / `NodeMapping` / `EdgeMapping`, `mapping.to_ontology()` (#1, #2) | -| **P2 — pipeline** | Extract the shared lexical-graph / prune / write / mentions base out of `IngestionPipeline`; add `StructuredIngestionPipeline`; deterministic record chunk uids (#5, #6) | -| **P3 — identity** | `Entity.identity`, alias handles, `AliasMatchResolution` (#3, #4) | +| **P1 — skeleton** | `RecordLoaderStrategy` + `RecordBatch` (stream **factory**, s1), `CsvRecordLoader`, `RecordMapping` / `NodeMapping` (with `alias`, s2) / `EdgeMapping`, `mapping.to_ontology()` with both guards (#1, #2) | +| **P2 — pipeline** | Extract a `LexicalGraphWriter` base out of `IngestionPipeline` (s5 — a base, *not* a subclass); add `link_sequential` kwarg; `StructuredIngestionPipeline`; record chunk uids keyed on the **effective** document uid (#5, #6) | +| **P3 — identity** | `Entity.identity` defaulting to the record key for structured sources, alias handles, `AliasMatchResolution` — on the critical path, not optional (#3, #4) | | **P4 — formats** | JSON / JSONL / XLSX / Parquet; PDF-table record stream; node-list + edge-list graph import (#8, #9, #10) | | **P5 — surface** | `ingest(mapping=...)` / `update(mapping=...)` routing, `IngestionResult` counters, `examples/11_structured_ingestion.py`, docs page, mixed-corpus integration test (#7) | diff --git a/poc/structured-ingestion/FINDINGS.md b/poc/structured-ingestion/FINDINGS.md new file mode 100644 index 00000000..22474cfc --- /dev/null +++ b/poc/structured-ingestion/FINDINGS.md @@ -0,0 +1,125 @@ +# FINDINGS — what the spikes changed about the design + +Five spikes, each answering one open question against real code and a real FalkorDB. +`python run_all.py` → **all 5 pass**. + +Four of the five falsified something in `docs/design/structured-ingestion.md`. One of those +(s3) inverts a headline decision. + +| # | Spike | Verdict on the design | +| --- | --- | --- | +| s1 | record stream | **amend** — `RecordBatch` must expose a stream *factory*; as written it silently writes zero rows | +| s2 | mapping DSL | **amend** — edges must address node *aliases*, not labels; `to_ontology()` needs two guards | +| s3 | identity | **invert** — `identity=["name"]` as the default is wrong; key + `alias_ids` is the only policy that works | +| s4 | record-as-chunk | **confirmed, incl. the predicted data-loss trap** | +| s5 | pipeline seam | **amend** — share a base class, don't subclass; one new kwarg for `NEXT_CHUNK` | + +--- + +## 1. `RecordBatch` must be a stream factory → proposal #1 + +Pydantic v2 keeps an `Iterable[dict]` field lazy (200k rows: ~0 MB vs 71.6 MB materialised) but +wraps it in a one-shot `ValidatorIterator`. Proposal #6 iterates records **twice** — step 3 builds +record chunks, step 4 maps records to `GraphData`. Measured result: + +``` +step 3 saw 10 records, step 4 saw 0 — no error raised +``` + +Silent zero-row ingest. Also, the annotation erases list-ness: `len()` raises even when the caller +passed a list, so no cheap record count is available for progress or `IngestionResult`. + +**Change:** `open_records: Callable[[], Iterator[dict]]` + `record_count: int | None`. Verified +re-iterable and `model_dump()`-safe. *(details: `s1_record_stream/NOTES.md`)* + +## 2. Edges must address aliases, not labels → proposal #2 + +`EdgeMapping(source="Organization", target="Organization")` cannot express a transaction with a +buyer and a seller. Executed against `transactions.csv` it produced a **self-loop** +(`ORG-7 -> ORG-7` instead of `ORG-7 -> ORG-42`) with no error. Buyer/seller, manager/report, +parent/subsidiary is the standard shape of transactional data, not an edge case. + +**Change:** `NodeMapping(alias=..., ...)`, `EdgeMapping(source=, target=)`, alias +defaulting to the label so the 80% case is untouched. Two further guards on `to_ontology()`: reject +`_RESERVED_ATTRIBUTE_NAMES` (a mapping can otherwise shadow SDK-written keys like `id` and +`description`), and emit stubs for reference-only labels (otherwise `Ontology`'s own validator +warns on every ingest — 2 warnings from one mapping in the spike). + +Bonus: `Entity(label=..., identity=[...])` **already works** via `Config.extra = "allow"` and +survives `model_dump()`, so #3 is prototypable with zero `src` changes and old `ontology.json` +files stay loadable. *(details: `s2_mapping_dsl/NOTES.md`)* + +## 3. Identity: the design's default is backwards → proposals #3, #4 + +Measured on the #82 acceptance corpus, three policies × two ingest orders: + +| Policy | Acme nodes | #82 traversal | +| --- | --- | --- | +| name-first (`identity=["name"]`, **the design's default**) | 2 | **0 people reachable** | +| key-only | 2 | **0 people reachable** | +| key + `alias_ids` + resolve pass | **1** | **2 people reachable** | + +Cause: `employees.csv` references its org by `org_id=ORG-42` and has no `org_name` column. Under +name-first identity the mapping **cannot compute the identity of the entity it points at**, so +proposal #2's rule "each mapping must supply the type's identity attributes" is unsatisfiable for +any normalised foreign key. The result is a stub node holding all the `WORKS_AT` edges, sitting +next to the real Acme node that holds the prose — the exact failure #82 exists to prevent, silent. + +**Change:** structured writes are **key-identified**, and `alias_ids` moves from optional bridge to +**critical path**. Caveat, also measured: with no source carrying key *and* name together, the +bridge degrades to 2 nodes / 0 reachable. The honest contract is **ingest order does not matter, +but presence does** — and unbridged stubs must be reported in `IngestionResult` rather than left +silent. *(details: `s3_identity/NOTES.md`)* + +## 4. The predicted cutover trap is real → proposal #5 + +The design review predicted, from reading code, that deterministic chunk uids keyed on the +canonical document id would destroy data during `update()`. Run against real FalkorDB through the +real `GraphStore`: + +| chunk uid keyed on | chunks before | shared with pending | **after cutover** | +| --- | --- | --- | --- | +| canonical doc id | 3 | 3 | **0** | +| effective (pending) doc id | 3 | 0 | **3** | + +The pending run `MERGE`s onto the live document's chunk nodes; `rollforward_cutover()` step 1 +deletes them; step 3 promotes an empty document. No exception. Today's `uuid4()` uids are +accidentally immune — which is why making them deterministic is the dangerous part of #5. + +Everything else in #5 **holds**: `get_document_entity_candidates()` found 5 record-chunk entities, +`delete_stale_relationships()` GC'd exactly the removed row's fact, `delete_orphan_entities()` +removed exactly Carol and the org that lost its last mention. + +**Change:** promote the §5 callout from "review finding" to a hard rule with this measurement +attached. §8.8 (row-level incremental update is *not* free) stays refuted. +*(details: `s4_record_as_chunk/NOTES.md`)* + +## 5. Share a base class, don't subclass → proposal #6 + +`_build_lexical_graph` / `_prune` / `_write_mentions` are reusable **verbatim** — the first already +consumes `TextChunks`, which is what proposal #5 turns records into. Both factorings produced +identical graphs (9 nodes, 6 `MENTIONED_IN`). + +But `IngestionPipeline.__init__` requires `loader, chunker, extractor, resolver, graph_store, +vector_store`. A structured pipeline has no chunker and no LLM extractor, so subclassing means +passing `None` and hoping — it works today only by accident of which methods are called. + +**Change:** extract the three methods into a `LexicalGraphWriter` base depending only on +`graph_store`. Plus one new kwarg: `_build_lexical_graph(..., link_sequential: bool = True)`, since +reusing it as-is chains `NEXT_CHUNK` between unrelated CSV rows (N-1 edges asserting a sequence +that doesn't exist, while `cypher_generation.py` tells the LLM those edges mean "next sequential +Chunk"). *(details: `s5_pipeline_seam/NOTES.md`)* + +--- + +## What did **not** change + +The core architecture survived contact with the database. Record-as-chunk, `RELATES` + `rel_type`, +mapping-as-ontology-fragment, deterministic no-LLM mapping, and "no retrieval strategy is touched" +all held up. Every correction above is a change to a *signature or a default*, not to the shape of +the design — which is the outcome a spike round is supposed to produce. + +## Disposal + +Once these are folded into `docs/design/structured-ingestion.md` and the implementation lands, +delete `poc/`. Nothing imports it; nothing ships it. diff --git a/poc/structured-ingestion/README.md b/poc/structured-ingestion/README.md new file mode 100644 index 00000000..5f882cb5 --- /dev/null +++ b/poc/structured-ingestion/README.md @@ -0,0 +1,42 @@ +# Structured-ingestion spikes + +Throwaway experiments backing [`docs/design/structured-ingestion.md`](../../docs/design/structured-ingestion.md) +(tracking: FalkorDB/research#82). + +**This folder is disposable.** Nothing here ships. It is outside the wheel +(`[tool.hatch.build.targets.wheel] packages = ["src/graphrag_sdk"]`), outside pytest +(`testpaths = ["tests"]`), and outside CI (every job sets `working-directory: graphrag_sdk` +and lints only `src/`). Delete the whole directory once the findings are folded into the design +and the real implementation lands. + +## Why it exists + +The design doc makes claims. These spikes exist to **find out where the claims are wrong** before +we build on them — each folder answers *one specific open question* and records a decision, rather +than re-implementing a proposal. Five spikes, not twelve; a proposal without a genuine open +question does not get a folder. + +| Spike | Proposal | The question it answers | +| --- | --- | --- | +| `s1_record_stream` | #1 | Can `RecordBatch` actually hold a lazy record stream, or does the model layer consume it? What does streaming buy in memory? | +| `s2_mapping_dsl` | #2 | Which DSL shape expresses all four real-world record shapes without special cases — and does `to_ontology()` round-trip into a valid `Ontology`? | +| `s3_identity` | #3, #4 | Do the three candidate identity policies actually produce one connected graph? Measured, not argued. | +| `s4_record_as_chunk` | #5 | Does record-as-chunk really inherit orphan cleanup unchanged — and is the predicted `update()` cutover data-loss trap real? | +| `s5_pipeline_seam` | #6 | Can `StructuredIngestionPipeline` reuse the existing pipeline's steps as-is, or do the signatures need to change? | + +## Running + +```bash +cd poc/structured-ingestion +python run_all.py # everything; DB spikes skip if FalkorDB is unreachable +python s3_identity/spike.py # one spike +``` + +Spikes needing a graph use `FALKOR_HOST` / `FALKOR_PORT` (default `localhost:6379`) and write to +throwaway graph names prefixed `poc_`. `docker compose up -d falkordb` from the repo root starts one. + +No API keys are needed anywhere — the harness supplies a deterministic fake embedder, and the +spikes never call an LLM (which is the point of the whole design). + +Each spike prints its findings and writes them to its own `NOTES.md`. [`FINDINGS.md`](FINDINGS.md) +rolls up the decisions that feed back into the design doc. diff --git a/poc/structured-ingestion/_harness/__init__.py b/poc/structured-ingestion/_harness/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/poc/structured-ingestion/_harness/env.py b/poc/structured-ingestion/_harness/env.py new file mode 100644 index 00000000..150878ac --- /dev/null +++ b/poc/structured-ingestion/_harness/env.py @@ -0,0 +1,119 @@ +"""Shared harness for the structured-ingestion spikes. + +Importing this module pins *this worktree's* ``src`` ahead of any installed +graphrag_sdk, so spikes exercise the branch under review rather than whatever +``pip install -e`` happens to point at. +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path + +# poc/structured-ingestion/_harness/env.py -> repo root +REPO_ROOT = Path(__file__).resolve().parents[3] +SDK_SRC = REPO_ROOT / "graphrag_sdk" / "src" +FIXTURES = Path(__file__).resolve().parent / "fixtures" + +if str(SDK_SRC) not in sys.path: + sys.path.insert(0, str(SDK_SRC)) + +FALKOR_HOST = os.getenv("FALKOR_HOST", "localhost") +FALKOR_PORT = int(os.getenv("FALKOR_PORT", "6379")) + + +def sdk_is_local() -> bool: + """True when ``graphrag_sdk`` resolves inside this worktree.""" + import graphrag_sdk + + return str(SDK_SRC) in str(Path(graphrag_sdk.__file__).resolve()) + + +def falkor_available() -> bool: + """Cheap reachability probe so DB spikes can skip instead of exploding.""" + import socket + + try: + with socket.create_connection((FALKOR_HOST, FALKOR_PORT), timeout=2): + return True + except OSError: + return False + + +def connection(graph_name: str): + """A FalkorDBConnection against a throwaway ``poc_``-prefixed graph.""" + from graphrag_sdk.core.connection import ConnectionConfig, FalkorDBConnection + + if not graph_name.startswith("poc_"): + raise ValueError(f"refusing non-throwaway graph name {graph_name!r}; use a poc_ prefix") + return FalkorDBConnection( + ConnectionConfig(host=FALKOR_HOST, port=FALKOR_PORT, graph_name=graph_name) + ) + + +async def reset_graph(conn) -> None: + """Drop everything in the connected graph. Only ever called on poc_ graphs.""" + await conn.query("MATCH (n) DETACH DELETE n") + + +class FakeEmbedder: + """Deterministic, dependency-free embedder. + + Spikes care about *what gets written and traversed*, never about vector + quality, so a hash-derived vector is sufficient and keeps the whole folder + runnable with no API keys. + """ + + def __init__(self, dimension: int = 8) -> None: + self.dimension = dimension + self.calls = 0 + + @property + def model_name(self) -> str: + return "fake-embedder" + + def _vec(self, text: str) -> list[float]: + import hashlib + + digest = hashlib.sha256(text.encode("utf-8")).digest() + return [digest[i] / 255.0 for i in range(self.dimension)] + + async def aembed_query(self, text: str) -> list[float]: + self.calls += 1 + return self._vec(text) + + async def aembed_documents(self, texts: list[str]) -> list[list[float]]: + self.calls += len(texts) + return [self._vec(t) for t in texts] + + +# ── tiny reporting helpers ─────────────────────────────────────── + + +class Report: + """Collects PASS/FAIL checks so a spike ends with a verdict, not a wall of prints.""" + + def __init__(self, title: str) -> None: + self.title = title + self.lines: list[str] = [] + self.failures = 0 + print(f"\n=== {title} ===") + + def check(self, ok: bool, label: str, detail: str = "") -> bool: + mark = "PASS" if ok else "FAIL" + if not ok: + self.failures += 1 + line = f"[{mark}] {label}" + (f" — {detail}" if detail else "") + self.lines.append(line) + print(line) + return ok + + def note(self, text: str) -> None: + self.lines.append(f" {text}") + print(f" {text}") + + def verdict(self) -> int: + status = "OK" if self.failures == 0 else f"{self.failures} FAILED" + print(f"--- {self.title}: {status}") + return 1 if self.failures else 0 diff --git a/poc/structured-ingestion/_harness/fixtures/acme_report.txt b/poc/structured-ingestion/_harness/fixtures/acme_report.txt new file mode 100644 index 00000000..b677c8b3 --- /dev/null +++ b/poc/structured-ingestion/_harness/fixtures/acme_report.txt @@ -0,0 +1,5 @@ +Acme Corp reported a Q3 revenue miss, attributing the shortfall to delayed +enterprise renewals. Globex, its largest counterparty, expanded its own +services division over the same period. + +Alice Smith, an engineer at Acme Corp, presented the remediation plan. diff --git a/poc/structured-ingestion/_harness/fixtures/catalog.json b/poc/structured-ingestion/_harness/fixtures/catalog.json new file mode 100644 index 00000000..aa04cbba --- /dev/null +++ b/poc/structured-ingestion/_harness/fixtures/catalog.json @@ -0,0 +1,44 @@ +{ + "catalog": { + "version": 3 + }, + "products": [ + { + "sku": "SKU-1", + "name": "Anvil", + "price": 99.5, + "sold_by": { + "org_id": "ORG-42", + "org_name": "Acme Corp" + }, + "tags": [ + "hardware", + "heavy" + ] + }, + { + "sku": "SKU-2", + "name": "Rocket Skates", + "price": 249.0, + "sold_by": { + "org_id": "ORG-42", + "org_name": "Acme Corp" + }, + "tags": [ + "hardware" + ] + }, + { + "sku": "SKU-9", + "name": "Consulting Hour", + "price": 180.0, + "sold_by": { + "org_id": "ORG-7", + "org_name": "Globex" + }, + "tags": [ + "services" + ] + } + ] +} diff --git a/poc/structured-ingestion/_harness/fixtures/employees.csv b/poc/structured-ingestion/_harness/fixtures/employees.csv new file mode 100644 index 00000000..12a78872 --- /dev/null +++ b/poc/structured-ingestion/_harness/fixtures/employees.csv @@ -0,0 +1,4 @@ +employee_id,full_name,age,job_title,org_id,start_date +E-1,Alice Smith,34,Engineer,ORG-42,2019-04-01 +E-2,Bob Jones,45,CFO,ORG-42,2015-11-15 +E-3,Carol White,29,Engineer,ORG-7,2021-06-30 diff --git a/poc/structured-ingestion/_harness/fixtures/orgs.csv b/poc/structured-ingestion/_harness/fixtures/orgs.csv new file mode 100644 index 00000000..55932899 --- /dev/null +++ b/poc/structured-ingestion/_harness/fixtures/orgs.csv @@ -0,0 +1,3 @@ +org_id,org_name,hq_country,employee_count +ORG-42,Acme Corp,US,1200 +ORG-7,Globex,GB,340 diff --git a/poc/structured-ingestion/_harness/fixtures/transactions.csv b/poc/structured-ingestion/_harness/fixtures/transactions.csv new file mode 100644 index 00000000..750ab161 --- /dev/null +++ b/poc/structured-ingestion/_harness/fixtures/transactions.csv @@ -0,0 +1,3 @@ +txn_id,buyer_org_id,seller_org_id,amount,currency,closed_on +TXN-100,ORG-7,ORG-42,250000,USD,2024-02-11 +TXN-101,ORG-42,ORG-7,90000,GBP,2024-05-02 diff --git a/poc/structured-ingestion/run_all.py b/poc/structured-ingestion/run_all.py new file mode 100644 index 00000000..3f25b8a5 --- /dev/null +++ b/poc/structured-ingestion/run_all.py @@ -0,0 +1,34 @@ +"""Run every spike. DB-backed spikes skip cleanly when FalkorDB is unreachable.""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +HERE = Path(__file__).resolve().parent +SPIKES = [ + "s1_record_stream", + "s2_mapping_dsl", + "s3_identity", + "s4_record_as_chunk", + "s5_pipeline_seam", +] + + +def main() -> int: + failures = [] + for spike in SPIKES: + rc = subprocess.call([sys.executable, str(HERE / spike / "spike.py")]) + if rc != 0: + failures.append(spike) + print("\n" + "=" * 60) + if failures: + print(f"FAILED: {', '.join(failures)}") + return 1 + print(f"All {len(SPIKES)} spikes passed.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/poc/structured-ingestion/s1_record_stream/NOTES.md b/poc/structured-ingestion/s1_record_stream/NOTES.md new file mode 100644 index 00000000..8ba8433b --- /dev/null +++ b/poc/structured-ingestion/s1_record_stream/NOTES.md @@ -0,0 +1,63 @@ +# s1 — record stream shape · DECIDED + +**Question.** Proposal #1 declares `records: Iterable[dict[str, Any]]` on a pydantic `DataModel` +and asserts it is "streamed, never fully materialised". Is that true, and is it safe? + +**Run:** `python s1_record_stream/spike.py` (no DB, no keys). All checks pass. + +## What actually happens + +Pydantic v2.11 does **not** materialise an `Iterable[...]` field — it replaces whatever you pass +with a `pydantic_core.ValidatorIterator`. Streaming works, and works well: + +| 200,000 rows | peak memory | +| --- | --- | +| streamed | ~0.0 MB | +| materialised via `list()` | 71.6 MB | + +`repr()`, `model_dump()` and `model_copy()` all leave the stream intact, so incidental logging is +not a hazard. So far the design's claim holds. + +## The two things the design got wrong + +**1 — the field is one-shot, and proposal #6 iterates it twice.** +The 9-step pipeline consumes records in step 3 (build record `Chunk`s) and again in step 4 +(map records → `GraphData`). Over a `ValidatorIterator` the second pass yields nothing: + +``` +step 3 saw 10 records, step 4 saw 0 — no error raised +``` + +No exception, no warning — ingestion would report success and write **zero nodes**. This is the +worst possible failure shape and it is latent in the design as written. + +**2 — the annotation erases list-ness.** +Even when the caller passes a fully materialised `list`, the field comes back as a +`ValidatorIterator` and `len()` raises `TypeError`. Any downstream code wanting a cheap record +count (progress reporting, `IngestionResult.records_processed`, batch sizing) cannot have one. + +## Decision + +Use a **stream factory**, not a stream: + +```python +class RecordBatch(DataModel): + open_records: Callable[[], Iterator[dict[str, Any]]] # re-iterable by construction + document_info: DocumentInfo + inferred_types: dict[str, str] + record_count: int | None = None # set when the loader knows it cheaply; None when streaming + + def __iter__(self) -> Iterator[dict[str, Any]]: + return self.open_records() +``` + +Verified in the spike: this shape survives `model_dump()` **and** the two-pass pipeline, returning +10/10 records on both passes. + +A loader then hands over a re-openable source (reopen the file handle / re-run the cursor) rather +than a live generator, which is also the honest contract — a CSV *can* be read twice, and where it +genuinely cannot (a network stream), the loader spools once and closes over the buffer, making the +cost explicit at the loader instead of silently corrupting the write. + +**Feeds back into the design:** proposal #1's `RecordBatch` signature, and a note on #6 that +step 3 and step 4 are two independent passes. diff --git a/poc/structured-ingestion/s1_record_stream/spike.py b/poc/structured-ingestion/s1_record_stream/spike.py new file mode 100644 index 00000000..9da82422 --- /dev/null +++ b/poc/structured-ingestion/s1_record_stream/spike.py @@ -0,0 +1,176 @@ +"""s1 — can ``RecordBatch`` actually carry a *lazy* record stream? + +Proposal #1 says: + + class RecordBatch(DataModel): + records: Iterable[dict[str, Any]] # streamed, never fully materialised + +That is a claim about pydantic behaviour, and the whole streaming story rests on +it. The failure mode that would matter in production is not "it raises" — it is +"something innocuous silently consumes the stream and ingestion writes zero +rows". So this spike attacks exactly that. + +Questions: + Q1 Does constructing the model consume the generator? + Q2 Is the field re-iterable, or one-shot? + Q3 Does an incidental ``model_dump()`` / ``repr()`` / ``model_copy()`` — the + kind of thing logging and result objects do — eat the records? + Q4 What does streaming actually buy in peak memory at POC scale? +""" + +from __future__ import annotations + +import sys +import tracemalloc +from collections.abc import Callable, Iterable, Iterator +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from _harness.env import Report # noqa: E402 + +from graphrag_sdk.core.models import DataModel # noqa: E402 + +N = 200_000 + + +def rows(n: int = N) -> Iterator[dict[str, Any]]: + for i in range(n): + yield {"employee_id": f"E-{i}", "full_name": f"Person {i}", "age": 20 + i % 50} + + +# ── Candidate A: the design's literal shape ────────────────────── + + +class BatchIterable(DataModel): + records: Iterable[dict[str, Any]] + source: str + + +# ── Candidate B: a factory, so the stream is re-iterable ───────── + + +class BatchFactory(DataModel): + open_records: Callable[[], Iterator[dict[str, Any]]] + source: str + + def __iter__(self): # type: ignore[override] + return self.open_records() + + +def main() -> int: + r = Report("s1 — record stream shape") + + # Q1 — construction must not drain the generator. + gen = rows(10) + batch = BatchIterable(records=gen, source="employees.csv") + consumed_at_construction = sum(1 for _ in gen) + r.check( + consumed_at_construction in (0, 10), + "construction does not silently drop records", + f"generator still yields {consumed_at_construction} after model init", + ) + r.note(f"field type after validation: {type(batch.records).__name__}") + lazy = consumed_at_construction == 10 or type(batch.records).__name__ != "list" + r.check(lazy, "records field stays lazy (not materialised to a list)") + + # Q2 — one-shot? + b2 = BatchIterable(records=rows(10), source="x") + first = list(b2.records) + second = list(b2.records) + r.check( + len(first) == 10, + "first iteration yields every record", + f"{len(first)} records", + ) + one_shot = len(second) == 0 + r.check( + True, + "second iteration behaviour recorded", + f"re-iteration yields {len(second)} records -> {'ONE-SHOT' if one_shot else 're-iterable'}", + ) + + # Q3 — the dangerous one. Does incidental inspection eat the stream? + for label, poke in ( + ("repr()", lambda b: repr(b)), + ("model_dump()", lambda b: b.model_dump()), + ("model_copy()", lambda b: b.model_copy()), + ): + b = BatchIterable(records=rows(10), source="x") + try: + poke(b) + survived = len(list(b.records)) + r.check( + survived == 10, + f"{label} leaves the stream intact", + f"{survived}/10 records survive", + ) + except Exception as exc: # noqa: BLE001 + r.check(False, f"{label} raised", f"{type(exc).__name__}: {exc}") + + # Candidate B under the same abuse. + fb = BatchFactory(open_records=lambda: rows(10), source="x") + fb.model_dump() + r.check( + len(list(fb)) == 10 and len(list(fb)) == 10, + "factory shape survives model_dump() AND is re-iterable", + ) + + # Q2b — does the annotation erase list-ness even for an eager caller? + eager = BatchIterable(records=[{"a": 1}, {"a": 2}], source="x") + try: + length: int | None = len(eager.records) # type: ignore[arg-type] + except TypeError: + length = None + r.check( + length is None, + "Iterable[dict] erases list-ness: len() fails even when a list was passed", + f"type is {type(eager.records).__name__}; downstream code can never cheaply count records", + ) + + # Q5 — the consequence that actually bites. Proposal #6 iterates records + # twice: step 3 builds record chunks, step 4 maps records -> GraphData. + def two_pass(batch) -> tuple[int, int]: + recs = batch.records if isinstance(batch, BatchIterable) else batch + chunks = sum(1 for _ in recs) # step 3 + nodes = sum(1 for _ in recs) # step 4 + return chunks, nodes + + chunks, nodes = two_pass(BatchIterable(records=rows(10), source="employees.csv")) + r.check( + nodes == 0, + "two-pass pipeline over a one-shot stream silently writes ZERO nodes", + f"step 3 saw {chunks} records, step 4 saw {nodes} — no error raised", + ) + fb2 = BatchFactory(open_records=lambda: rows(10), source="employees.csv") + r.check( + (sum(1 for _ in fb2), sum(1 for _ in fb2)) == (10, 10), + "factory shape survives the same two-pass pipeline", + ) + + # Q4 — peak memory, streamed vs materialised. + def peak(fn) -> float: + tracemalloc.start() + fn() + peak_bytes = tracemalloc.get_traced_memory()[1] + tracemalloc.stop() + return peak_bytes / 1e6 + + streamed = peak(lambda: sum(1 for _ in BatchIterable(records=rows(), source="s").records)) + materialised = peak( + lambda: sum(1 for _ in BatchIterable(records=list(rows()), source="s").records) + ) + r.note( + f"{N:,} rows — streamed peak {streamed:.1f} MB · materialised peak {materialised:.1f} MB" + ) + r.check( + streamed < materialised / 10, + "streaming keeps peak memory an order of magnitude below materialising", + f"{materialised / max(streamed, 1e-6):.0f}x reduction", + ) + + return r.verdict() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/poc/structured-ingestion/s2_mapping_dsl/NOTES.md b/poc/structured-ingestion/s2_mapping_dsl/NOTES.md new file mode 100644 index 00000000..a6c9f37e --- /dev/null +++ b/poc/structured-ingestion/s2_mapping_dsl/NOTES.md @@ -0,0 +1,71 @@ +# s2 — mapping DSL shape · DECIDED + +**Question.** Proposal #2's `EdgeMapping(type=..., source="Person", target="Organization")` +addresses the record's nodes **by label**. Does that survive real record shapes? + +**Run:** `python s2_mapping_dsl/spike.py` (no DB, no keys). All checks pass. + +Four shapes from the fixture corpus, executed rather than argued about: + +| | Record shape | Label-addressed (design as written) | +| --- | --- | --- | +| R1 | `orgs.csv` — one row, one entity | works | +| R2 | `employees.csv` — two nodes + FK edge | works | +| R3 | `transactions.csv` — reified event, **two `Organization`s in one record** | **broken** | +| R4 | `catalog.json` — nested object | works after dotted flattening | + +## The defect: label is not a unique handle inside a record + +A transaction has a buyer and a seller, both `Organization`. With only a label to resolve by, the +edge resolver has no way to pick — and produces a **self-loop**: + +``` +want: BOUGHT_FROM ORG-7 -> ORG-42 +produced: BOUGHT_FROM ORG-7 -> ORG-7 +``` + +Silently wrong, not an error. And this is not an exotic case — buyer/seller, manager/report, +parent/subsidiary, origin/destination are the standard shape of any transactional or hierarchical +table, which is most of what "structured data" means in practice. + +## Decision: nodes get an `alias`; edges address aliases + +```python +RecordMapping( + nodes=[ + NodeMapping(alias="txn", label="Transaction", key="txn_id"), + NodeMapping(alias="buyer", label="Organization", key="buyer_org_id", reference=True), + NodeMapping(alias="seller", label="Organization", key="seller_org_id", reference=True), + ], + edges=[EdgeMapping(type="BOUGHT_FROM", source="buyer", target="seller")], +) +``` + +Verified: produces `BOUGHT_FROM ORG-7 -> ORG-42` and `INVOLVES_BUYER TXN-100 -> ORG-7` correctly. +`alias` defaults to the label, so the 80% single-node case in proposal #2 is unchanged and nobody +writing `orgs.csv` ever types an alias. + +R4 also confirms nested JSON needs **no new DSL concept** — flatten to `sold_by.org_id` and the +same alias machinery applies. Nested containment does not need to be its own third edge kind. + +## Two more traps found in `to_ontology()` + +**Reserved attribute names.** A mapping declaring `properties={"description": ..., "id": ...}` +generates an ontology that shadows SDK-written values on every node. `_RESERVED_ATTRIBUTE_NAMES` +in `core/models.py` lists ten such keys. `to_ontology()` must **reject** +`_RESERVED_ATTRIBUTE_NAMES - _SDK_MANAGED_ATTRIBUTE_NAMES`, naming the offending `Label.attribute` — +this is the same "reject before any write" rule the design already applies to contradictions. + +**Reference-only labels warn.** `Ontology`'s own `_warn_on_undeclared_pattern_labels` validator +fires when a relation pattern names a label not in `entities` — which is exactly what an FK +reference produces. The spike captured 2 warnings from one mapping. `to_ontology()` must emit bare +`Entity` stubs for reference labels, or be merged into the live ontology before validation, or +every structured ingest logs noise that trains users to ignore real warnings. + +## Free win for proposal #3 + +`Entity(label="Product", identity=["sku"])` **already works** — `DataModel.Config.extra = "allow"` +carries it, and it survives `model_dump()`, so it persists to `ontology.json`. Two consequences: +`identity` is prototypable with zero `src` changes, and adding it as a real field later will not +break ontologies persisted in the meantime. It must still become a *declared* field defaulting to +`["name"]` so it is validated rather than a silent typo sink. diff --git a/poc/structured-ingestion/s2_mapping_dsl/spike.py b/poc/structured-ingestion/s2_mapping_dsl/spike.py new file mode 100644 index 00000000..47a37d3c --- /dev/null +++ b/poc/structured-ingestion/s2_mapping_dsl/spike.py @@ -0,0 +1,336 @@ +"""s2 — which mapping DSL shape survives all four real record shapes? + +Proposal #2 proposes: + + RecordMapping(nodes=[NodeMapping(...)], edges=[EdgeMapping(type=..., source="Person", + target="Organization")]) + +`source`/`target` address nodes **by label**. This spike executes candidate DSLs +against four record shapes taken from the fixture corpus and checks the edges +they actually produce, rather than arguing about readability: + + R1 orgs.csv one row -> one entity (the 80% case) + R2 employees.csv one row -> two nodes + FK edge + R3 transactions.csv one row -> a reified event with TWO edges to the SAME label + R4 catalog.json nested object -> child node + containment edge + +R3 is the case that decides it: a transaction has a buyer *and* a seller, both +`Organization`. + +It also checks the two things a generated ontology can silently get wrong: +reserved attribute names, and whether `Entity.identity` (proposal #3) can even +be carried by today's model. +""" + +from __future__ import annotations + +import csv +import json +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from _harness.env import FIXTURES, Report # noqa: E402 + +from graphrag_sdk.core.models import ( # noqa: E402 + _RESERVED_ATTRIBUTE_NAMES, + Attribute, + Entity, + Ontology, + Relation, +) + +# ── Candidate A: proposal #2 verbatim — edges address nodes by LABEL ── + + +@dataclass +class NodeA: + label: str + key: str + name: str | None = None + properties: dict[str, str] = field(default_factory=dict) + reference: bool = False + + +@dataclass +class EdgeA: + type: str + source: str # a LABEL + target: str # a LABEL + properties: dict[str, str] = field(default_factory=dict) + + +@dataclass +class MappingA: + nodes: list[NodeA] + edges: list[EdgeA] = field(default_factory=list) + + def apply(self, record: dict[str, Any]) -> tuple[list[tuple[str, str]], list[tuple]]: + """-> ([(label, key_value)], [(rel_type, src_key, tgt_key)])""" + built: list[tuple[str, str]] = [] + by_label: dict[str, list[str]] = {} + for n in self.nodes: + if n.key not in record: + continue + kv = str(record[n.key]) + built.append((n.label, kv)) + by_label.setdefault(n.label, []).append(kv) + edges = [] + for e in self.edges: + # The design gives us only a label to resolve with. + src = by_label.get(e.source, []) + tgt = by_label.get(e.target, []) + if not src or not tgt: + continue + # Ambiguity is unresolvable here — take the first, which is the + # bug this spike is looking for. + edges.append((e.type, src[0], tgt[0])) + return built, edges + + +# ── Candidate A': same, but nodes carry an ALIAS ───────────────── + + +@dataclass +class NodeB: + alias: str + label: str + key: str + name: str | None = None + properties: dict[str, str] = field(default_factory=dict) + reference: bool = False + + +@dataclass +class EdgeB: + type: str + source: str # an ALIAS + target: str # an ALIAS + properties: dict[str, str] = field(default_factory=dict) + + +@dataclass +class MappingB: + nodes: list[NodeB] + edges: list[EdgeB] = field(default_factory=list) + + def apply(self, record: dict[str, Any]) -> tuple[list[tuple[str, str]], list[tuple]]: + built: list[tuple[str, str]] = [] + by_alias: dict[str, str] = {} + for n in self.nodes: + if n.key not in record: + continue + kv = str(record[n.key]) + built.append((n.label, kv)) + by_alias[n.alias] = kv + edges = [ + (e.type, by_alias[e.source], by_alias[e.target]) + for e in self.edges + if e.source in by_alias and e.target in by_alias + ] + return built, edges + + def to_ontology(self) -> Ontology: + label_props: dict[str, dict[str, str]] = {} + identity: dict[str, list[str]] = {} + for n in self.nodes: + if n.reference: + continue + props = label_props.setdefault(n.label, {}) + props.update(dict.fromkeys(n.properties, "STRING")) + if n.name: + props["name"] = "STRING" + identity[n.label] = ["name"] if n.name else [n.key] + entities = [ + Entity( + label=lbl, + properties=[Attribute(name=p, type=t) for p, t in props.items()], + identity=identity[lbl], # proposal #3 — does extra="allow" carry it? + ) + for lbl, props in label_props.items() + ] + alias_label = {n.alias: n.label for n in self.nodes} + relations = [ + Relation( + label=e.type, + patterns=[(alias_label[e.source], alias_label[e.target])], + properties=[Attribute(name=p) for p in e.properties], + ) + for e in self.edges + ] + return Ontology(entities=entities, relations=relations) + + +def read_csv(name: str) -> list[dict[str, Any]]: + with open(FIXTURES / name, newline="", encoding="utf-8") as fh: + return list(csv.DictReader(fh)) + + +def main() -> int: + r = Report("s2 — mapping DSL shape") + + orgs = read_csv("orgs.csv") + employees = read_csv("employees.csv") + transactions = read_csv("transactions.csv") + catalog = json.loads((FIXTURES / "catalog.json").read_text())["products"] + + # R1 — one row, one entity. Both shapes handle it. + a1 = MappingA(nodes=[NodeA(label="Organization", key="org_id", name="org_name")]) + nodes, edges = a1.apply(orgs[0]) + r.check(nodes == [("Organization", "ORG-42")] and edges == [], "R1 orgs.csv — trivial for A") + + # R2 — two nodes + FK edge, all distinct labels. + a2 = MappingA( + nodes=[ + NodeA(label="Person", key="employee_id", name="full_name"), + NodeA(label="Organization", key="org_id", reference=True), + ], + edges=[EdgeA(type="WORKS_AT", source="Person", target="Organization")], + ) + _, edges = a2.apply(employees[0]) + r.check(edges == [("WORKS_AT", "E-1", "ORG-42")], "R2 employees.csv — A resolves the FK edge") + + # R3 — the decider. Two Organizations in one record. + a3 = MappingA( + nodes=[ + NodeA(label="Transaction", key="txn_id"), + NodeA(label="Organization", key="buyer_org_id", reference=True), + NodeA(label="Organization", key="seller_org_id", reference=True), + ], + edges=[ + EdgeA(type="BOUGHT_FROM", source="Organization", target="Organization"), + EdgeA(type="INVOLVES_BUYER", source="Transaction", target="Organization"), + ], + ) + _, edges = a3.apply(transactions[0]) + want_buyer, want_seller = "ORG-7", "ORG-42" + self_loop = any(e[1] == e[2] for e in edges) + r.check( + self_loop, + "R3 transactions.csv — label-addressed edges COLLAPSE to a self-loop", + f"produced {edges} · buyer={want_buyer} seller={want_seller}", + ) + r.note("label is not a unique handle inside a record; the design has no way to say which one") + + b3 = MappingB( + nodes=[ + NodeB(alias="txn", label="Transaction", key="txn_id"), + NodeB(alias="buyer", label="Organization", key="buyer_org_id", reference=True), + NodeB(alias="seller", label="Organization", key="seller_org_id", reference=True), + ], + edges=[ + EdgeB(type="BOUGHT_FROM", source="buyer", target="seller"), + EdgeB(type="INVOLVES_BUYER", source="txn", target="buyer"), + ], + ) + _, edges = b3.apply(transactions[0]) + r.check( + edges == [("BOUGHT_FROM", "ORG-7", "ORG-42"), ("INVOLVES_BUYER", "TXN-100", "ORG-7")], + "R3 transactions.csv — alias-addressed edges resolve correctly", + str(edges), + ) + + # R4 — nested JSON, after flattening. Same alias machinery, no new concept. + flat = [ + { + "sku": p["sku"], + "name": p["name"], + "sold_by.org_id": p["sold_by"]["org_id"], + "tags": p["tags"], + } + for p in catalog + ] + b4 = MappingB( + nodes=[ + NodeB(alias="prod", label="Product", key="sku", name="name"), + NodeB(alias="vendor", label="Organization", key="sold_by.org_id", reference=True), + ], + edges=[EdgeB(type="SOLD_BY", source="prod", target="vendor")], + ) + _, edges = b4.apply(flat[0]) + r.check( + edges == [("SOLD_BY", "SKU-1", "ORG-42")], + "R4 catalog.json — dotted flattening needs no new DSL concept", + ) + + # ── the generated ontology ─────────────────────────────────── + import logging + + class _Capture(logging.Handler): + def __init__(self) -> None: + super().__init__() + self.messages: list[str] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.messages.append(record.getMessage()) + + cap = _Capture() + logging.getLogger("graphrag_sdk.core.models").addHandler(cap) + onto = b3.to_ontology() + logging.getLogger("graphrag_sdk.core.models").removeHandler(cap) + r.check( + {e.label for e in onto.entities} == {"Transaction"}, + "to_ontology() emits only non-reference nodes", + f"entities={[e.label for e in onto.entities]} " + "(Organization is a reference, declared by orgs.csv)", + ) + r.check( + [rel.label for rel in onto.relations] == ["BOUGHT_FROM", "INVOLVES_BUYER"], + "to_ontology() round-trips into a real Ontology with directional patterns", + ) + r.check( + any("not declared in ontology.entities" in m for m in cap.messages), + "a reference-only label makes Ontology's own validator warn on every ingest", + f"{len(cap.messages)} warning(s), e.g. " + "'Organization ... not declared in ontology.entities'", + ) + r.note( + "=> to_ontology() must emit bare Entity stubs for reference labels, or be merged into " + "the live ontology before validation runs — otherwise every structured ingest logs noise" + ) + + # proposal #3's new field, on today's model. + ent = Entity(label="Product", identity=["sku"]) + carried = getattr(ent, "identity", None) + r.check( + carried == ["sku"], + "Entity accepts an `identity` field today via Config.extra='allow'", + "so #3 is prototypable with zero src changes, and old ontology.json still loads", + ) + r.check( + "identity" in Entity(label="P", identity=["sku"]).model_dump(), + "`identity` survives model_dump(), so it persists to ontology.json", + ) + r.note( + "but it is untyped/unvalidated as an extra — the real change must add it as a " + "declared field defaulting to ['name']" + ) + + # reserved-name collision: the trap a generated ontology walks straight into. + bad = MappingB( + nodes=[ + NodeB( + alias="o", + label="Organization", + key="org_id", + name="org_name", + properties={"description": "hq_country", "id": "org_id"}, + ) + ] + ) + declared = {a.name for e in bad.to_ontology().entities for a in e.properties} + collisions = (declared & _RESERVED_ATTRIBUTE_NAMES) - {"name"} + r.check( + bool(collisions), + "a mapping can silently declare SDK-reserved attributes", + f"collides on {sorted(collisions)} — these shadow SDK-written values on every node", + ) + r.note("=> to_ontology() must reject _RESERVED_ATTRIBUTE_NAMES minus _SDK_MANAGED ones") + + return r.verdict() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/poc/structured-ingestion/s3_identity/NOTES.md b/poc/structured-ingestion/s3_identity/NOTES.md new file mode 100644 index 00000000..676e0f9a --- /dev/null +++ b/poc/structured-ingestion/s3_identity/NOTES.md @@ -0,0 +1,67 @@ +# s3 — entity identity · DECIDED (design correction) + +**Question.** Proposal #3 defaults `Entity.identity` to `["name"]` and frames proposal #4's +`alias_ids` as an *optional* bridge for the minority of types whose identity is a business key. +Measured against a real FalkorDB, that is backwards. + +**Run:** `python s3_identity/spike.py` (needs FalkorDB; no keys). All checks pass. + +Three policies x two ingest orders over the #82 acceptance corpus: + +| Policy | Acme nodes | #82 traversal | order-independent | +| --- | --- | --- | --- | +| P1 name-first (`identity=["name"]`, the design's default) | **2** | **0 people reachable** | yes | +| P2 key-only | **2** | **0 people reachable** | yes | +| P3 key + `alias_ids` + resolve pass | **1** | **2 people reachable** | yes | + +## Why name-first loses: normalised FKs carry keys, not names + +`employees.csv` is a perfectly ordinary normalised table: + +``` +employee_id,full_name,age,job_title,org_id,start_date +E-1,Alice Smith,34,Engineer,ORG-42,2019-04-01 +``` + +It references an organisation by `org_id=ORG-42`. It does **not** contain `org_name`. So under +`identity=["name"]` the employees mapping *cannot compute the identity of the entity it points at* — +there is no name in the record to compute it from. Proposal #2's rule "each mapping must supply the +type's identity attributes" is unsatisfiable for any normalised foreign key, which is the single +most common structured-data shape there is. + +The result is not an error. It is a stub node `org-42__organization` sitting next to the real +`acme_corp__organization`, with `WORKS_AT` attached to the stub — so the prose about Acme's Q3 +revenue miss and the engineers who work there are in the same graph and **not connected**. That is +precisely the acceptance criterion #82 exists to test, silently failing. + +P2 fails the mirror image: structured sources converge on `org-42__organization`, but LLM +extraction can only ever produce a name, so the prose entity is stranded. + +## Decision + +1. **Structured writes are key-identified.** `NodeMapping.key` produces the node id via + `compute_entity_id(key_value, label)`. This is what makes FK stubs land on the right node + regardless of ingest order — confirmed: both orders converge for all three policies. +2. **`alias_ids` is on the critical path, not optional.** Any mapping that has both the key and a + name emits `alias_ids=[compute_entity_id(name, label)]`, and the resolve pass merges the + name-identified node from unstructured extraction into it. This is the *only* configuration + tested that yields one Acme node and a working traversal. +3. Resolution stays deterministic and index-backed — the spike's implementation is four Cypher + statements per merged pair, no LLM and no embeddings, exactly as proposal #4 claims. + +So proposal #3's headline should be inverted: identity is declared on the type, but its **default +for structured sources is the record key**, and #4 is what makes the graph connected rather than a +nicety for SKU-shaped types. + +## The caveat, measured + +P3 is not magic. With prose + `employees.csv` and **no** `orgs.csv`, the result degrades to +2 Acme nodes and 0 reachable people — the alias bridge has nothing to be built from, because no +source carried both `ORG-42` and `"Acme Corp"` in the same record. + +The precise contract is therefore: **ingest order does not matter, but presence does.** An entity +type needs at least one source that declares it (key *and* name) for its FK stubs and its +unstructured mentions to converge. That is a reasonable requirement — it is just the dimension +table — but it must be stated in the design and surfaced at ingest time: if a mapping references a +label that no ingested source has ever declared, the run should report the count of unbridged +stubs rather than leave the user with a quietly disconnected graph. diff --git a/poc/structured-ingestion/s3_identity/spike.py b/poc/structured-ingestion/s3_identity/spike.py new file mode 100644 index 00000000..e1432d61 --- /dev/null +++ b/poc/structured-ingestion/s3_identity/spike.py @@ -0,0 +1,314 @@ +"""s3 — do the candidate identity policies actually produce ONE connected graph? + +Proposal #3 puts identity on the entity type, defaulting to ``name``. +Proposal #4 adds ``alias_ids`` as a deterministic bridge. Both are arguments. +This spike measures them against a real FalkorDB. + +The corpus is the #82 acceptance scenario: + acme_report.txt prose -> Organization("Acme Corp"), Person("Alice Smith") [name-only ids] + orgs.csv ORG-42 -> "Acme Corp" [key AND name] + employees.csv E-1 Alice Smith -> org_id=ORG-42 [key ONLY — no org_name] + +That last line is the whole experiment. A normalised FK table carries the +*key* of its target and not its *name*, so under name-first identity the +employees mapping physically cannot compute the identity of the organisation +it is pointing at. + +Three policies x two ingest orders, scored on: + * how many nodes end up representing Acme Corp (want: 1) + * whether the #82 query traverses prose-chunk -> Org -> WORKS_AT -> Person + * whether the two ingest orders converge to the same graph +""" + +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import csv # noqa: E402 + +from _harness.env import FIXTURES, Report, connection, falkor_available, reset_graph # noqa: E402 + +from graphrag_sdk.core.models import GraphNode, GraphRelationship # noqa: E402 +from graphrag_sdk.ingestion.extraction_strategies.entity_extractors import ( # noqa: E402 + compute_entity_id, +) +from graphrag_sdk.storage.graph_store import GraphStore # noqa: E402 + +CHUNK_ID = "chunk-prose-1" +DOC_ID = "doc-acme-report" + + +def rows(name: str) -> list[dict[str, str]]: + with open(FIXTURES / name, newline="", encoding="utf-8") as fh: + return list(csv.DictReader(fh)) + + +async def write_unstructured(store: GraphStore) -> None: + """What the LLM pipeline produces from acme_report.txt: name-derived ids only.""" + await store.upsert_nodes( + [ + GraphNode(id=DOC_ID, label="Document", properties={"path": "acme_report.txt"}), + GraphNode( + id=CHUNK_ID, + label="Chunk", + properties={"text": "Acme Corp reported a Q3 revenue miss...", "index": 0}, + ), + GraphNode( + id=compute_entity_id("Acme Corp", "Organization"), + label="Organization", + properties={"name": "Acme Corp"}, + ), + ] + ) + await store.upsert_relationships( + [ + GraphRelationship(start_node_id=DOC_ID, end_node_id=CHUNK_ID, type="PART_OF"), + GraphRelationship( + start_node_id=compute_entity_id("Acme Corp", "Organization"), + end_node_id=CHUNK_ID, + type="MENTIONED_IN", + ), + ] + ) + + +# ── the three identity policies ────────────────────────────────── +# +# Each returns the node id a mapping would compute for a target entity, +# given whatever columns that particular record actually has. + + +def id_name_first(label: str, key_value: str, name_value: str | None) -> str: + # identity=["name"]; falls back to the key when the record has no name column. + return compute_entity_id(name_value or key_value, label) + + +def id_key_only(label: str, key_value: str, name_value: str | None) -> str: + return compute_entity_id(key_value, label) + + +POLICIES = { + "P1_name_first": id_name_first, + "P2_key_only": id_key_only, + "P3_key_plus_alias": id_key_only, # same ids; differs by writing alias_ids + a resolve pass +} + + +async def ingest_orgs(store: GraphStore, policy: str) -> None: + nodes = [] + for row in rows("orgs.csv"): + nid = POLICIES[policy]("Organization", row["org_id"], row["org_name"]) + props = { + "name": row["org_name"], + "org_id": row["org_id"], + "hq_country": row["hq_country"], + } + if policy == "P3_key_plus_alias": + props["alias_ids"] = [compute_entity_id(row["org_name"], "Organization")] + nodes.append(GraphNode(id=nid, label="Organization", properties=props)) + await store.upsert_nodes(nodes) + + +async def ingest_employees(store: GraphStore, policy: str) -> None: + nodes, rels = [], [] + for row in rows("employees.csv"): + pid = POLICIES[policy]("Person", row["employee_id"], row["full_name"]) + nodes.append( + GraphNode( + id=pid, + label="Person", + properties={ + "name": row["full_name"], + "employee_id": row["employee_id"], + "title": row["job_title"], + "age": int(row["age"]), + }, + ) + ) + # The FK stub. employees.csv has org_id and NOT org_name. + oid = POLICIES[policy]("Organization", row["org_id"], None) + nodes.append(GraphNode(id=oid, label="Organization", properties={"name": row["org_id"]})) + rels.append( + GraphRelationship( + start_node_id=pid, + end_node_id=oid, + type="RELATES", + properties={ + "rel_type": "WORKS_AT", + "fact": f"({row['full_name']}, WORKS_AT, {row['org_id']})", + "source_chunk_ids": [f"rec-employees-{row['employee_id']}"], + "src_name": row["full_name"], + "tgt_name": row["org_id"], + }, + ) + ) + await store.upsert_nodes(nodes) + await store.upsert_relationships(rels) + + +# ── proposal #4: deterministic alias resolution ────────────────── + + +async def resolve_aliases(store: GraphStore) -> int: + """Merge any node whose id appears in another node's alias_ids. + + Deterministic, index-friendly, no LLM and no embeddings — the property + proposal #4 claims. This is what would run inside finalize(). + """ + res = await store.query_raw( + "MATCH (keep:__Entity__) WHERE keep.alias_ids IS NOT NULL " + "UNWIND keep.alias_ids AS alias " + "MATCH (dup:__Entity__ {id: alias}) WHERE dup.id <> keep.id " + "RETURN keep.id AS keep, dup.id AS dup" + ) + pairs = [(r[0], r[1]) for r in (res.result_set or [])] + for keep, dup in pairs: + # rewire outgoing RELATES, incoming RELATES, and MENTIONED_IN + await store.query_raw( + "MATCH (d:__Entity__ {id:$dup})-[r:RELATES]->(o) MATCH (k:__Entity__ {id:$keep}) " + "MERGE (k)-[n:RELATES {rel_type: r.rel_type}]->(o) " + "SET n.fact = r.fact, n.source_chunk_ids = r.source_chunk_ids DELETE r", + {"dup": dup, "keep": keep}, + ) + await store.query_raw( + "MATCH (o)-[r:RELATES]->(d:__Entity__ {id:$dup}) MATCH (k:__Entity__ {id:$keep}) " + "MERGE (o)-[n:RELATES {rel_type: r.rel_type}]->(k) " + "SET n.fact = r.fact, n.source_chunk_ids = r.source_chunk_ids DELETE r", + {"dup": dup, "keep": keep}, + ) + await store.query_raw( + "MATCH (d:__Entity__ {id:$dup})-[r:MENTIONED_IN]->(c:Chunk) " + "MATCH (k:__Entity__ {id:$keep}) MERGE (k)-[:MENTIONED_IN]->(c) DELETE r", + {"dup": dup, "keep": keep}, + ) + # keep the human-readable name from whichever side actually has one + await store.query_raw( + "MATCH (d:__Entity__ {id:$dup}), (k:__Entity__ {id:$keep}) " + "SET k.name = coalesce(k.name, d.name) DETACH DELETE d", + {"dup": dup, "keep": keep}, + ) + return len(pairs) + + +# ── measurement ────────────────────────────────────────────────── + + +async def measure(store: GraphStore) -> dict[str, int]: + acme = await store.query_raw( + "MATCH (o:Organization) WHERE o.name IN ['Acme Corp','ORG-42'] OR o.org_id = 'ORG-42' " + "RETURN count(o) AS c" + ) + # the #82 acceptance traversal: prose chunk -> Org -> WORKS_AT -> engineer + hops = await store.query_raw( + "MATCH (c:Chunk {id:$cid})<-[:MENTIONED_IN]-(o:Organization)" + "<-[r:RELATES]-(p:Person) WHERE r.rel_type = 'WORKS_AT' " + "RETURN count(DISTINCT p) AS c", + {"cid": CHUNK_ID}, + ) + orgs = await store.query_raw("MATCH (o:Organization) RETURN count(o) AS c") + return { + "acme_nodes": acme.result_set[0][0], + "reachable_people": hops.result_set[0][0], + "total_orgs": orgs.result_set[0][0], + } + + +async def run(policy: str, order: str) -> dict[str, int]: + conn = connection(f"poc_s3_{policy.lower()}_{order}") + store = GraphStore(conn) + await reset_graph(conn) + await write_unstructured(store) + if order == "orgs_first": + await ingest_orgs(store, policy) + await ingest_employees(store, policy) + else: + await ingest_employees(store, policy) + await ingest_orgs(store, policy) + if policy == "P3_key_plus_alias": + await resolve_aliases(store) + result = await measure(store) + await conn.close() + return result + + +async def main() -> int: + r = Report("s3 — entity identity") + if not falkor_available(): + r.note("SKIPPED — no FalkorDB on FALKOR_HOST:FALKOR_PORT") + return 0 + + results: dict[str, dict[str, dict[str, int]]] = {} + for policy in POLICIES: + results[policy] = {} + for order in ("orgs_first", "employees_first"): + results[policy][order] = await run(policy, order) + + for policy, per_order in results.items(): + a, b = per_order["orgs_first"], per_order["employees_first"] + r.note( + f"{policy:<20} orgs_first={a} employees_first={b}", + ) + + # 1. one node for Acme + for policy, per_order in results.items(): + ok = per_order["orgs_first"]["acme_nodes"] == 1 + r.check( + ok if policy == "P3_key_plus_alias" else True, + f"{policy}: Acme Corp is a single node", + f"{per_order['orgs_first']['acme_nodes']} node(s)" + ("" if ok else " <-- duplicated"), + ) + + # 2. the acceptance traversal + for policy, per_order in results.items(): + reach = per_order["orgs_first"]["reachable_people"] + ok = reach > 0 + r.check( + ok if policy == "P3_key_plus_alias" else True, + f"{policy}: #82 traversal prose-chunk -> Org -> WORKS_AT -> Person", + f"{reach} people reachable" + ("" if ok else " <-- prose is disconnected"), + ) + + # 3. order independence + for policy, per_order in results.items(): + same = per_order["orgs_first"] == per_order["employees_first"] + r.check( + same if policy != "P1_name_first" else True, + f"{policy}: ingest order does not change the final graph", + "converged" if same else "ORDER-DEPENDENT", + ) + + winner = "P3_key_plus_alias" + w = results[winner]["orgs_first"] + r.check( + w["acme_nodes"] == 1 and w["reachable_people"] > 0, + f"{winner} is the only policy that satisfies both #82 criteria", + str(w), + ) + + # 4. the honest caveat — P3 needs SOME source carrying both key and name. + conn = connection("poc_s3_p3_nobridge") + store = GraphStore(conn) + await reset_graph(conn) + await write_unstructured(store) + await ingest_employees(store, "P3_key_plus_alias") # FK-only source, no orgs.csv + await resolve_aliases(store) + nobridge = await measure(store) + await conn.close() + r.check( + nobridge["acme_nodes"] == 2 and nobridge["reachable_people"] == 0, + "P3 degrades when NO source carries both the key and the name", + f"{nobridge} — the alias bridge has nothing to be built from", + ) + r.note( + "=> alias_ids must be emitted by whichever mapping declares the entity (orgs.csv), " + "and FK-only references stay stubs until that source arrives. Order still does not " + "matter; presence does." + ) + return r.verdict() + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/poc/structured-ingestion/s4_record_as_chunk/NOTES.md b/poc/structured-ingestion/s4_record_as_chunk/NOTES.md new file mode 100644 index 00000000..73102b27 --- /dev/null +++ b/poc/structured-ingestion/s4_record_as_chunk/NOTES.md @@ -0,0 +1,63 @@ +# s4 — record-as-chunk & the `update()` cutover · CONFIRMED + +**Question.** Proposal #5 claims record-as-chunk makes `update()` / `delete_document()` work +unchanged. The design review *predicted*, from reading `api/main.py:2104` and +`rollforward_cutover()`, that deterministic chunk uids keyed on the canonical document id would +silently destroy data. A prediction from reading code is a hypothesis; this spike runs it against +a real FalkorDB through the real `GraphStore`. + +**Run:** `python s4_record_as_chunk/spike.py` (needs FalkorDB; no keys). All checks pass. + +## The trap is real + +| chunk uid keyed on | chunks before `update()` | chunk nodes shared with pending | chunks after cutover | +| --- | --- | --- | --- | +| **canonical** document id | 3 | **3** | **0** | +| **effective** (pending) document id | 3 | 0 | **3** | + +Mechanism, now observed rather than inferred: + +1. `update()` writes the new version under `pending_id = f"{resolved_id}__pending__{uuid4().hex[:8]}"`. +2. With canonical-keyed uids, the pending run's `MERGE` lands on the **same `Chunk` nodes** as the + live document — the spike measured all 3 chunks carrying `PART_OF` from *both* Documents. +3. `rollforward_cutover()` step 1 calls `delete_document_chunks_and_node(real_id)`, whose Cypher is + `MATCH (:Document {id})-[:PART_OF]->(c:Chunk) DETACH DELETE c` — it deletes the shared nodes. +4. Step 3 renames the pending to the canonical id. It is promoted **with zero chunks**. + +No exception. `rollforward_cutover`'s precondition guard doesn't help — the pending Document node +exists, it is only its chunks that have been destroyed. The user sees a successful `update()` and a +document whose every record has vanished, along with the provenance that orphan cleanup depends on. + +Today's `uuid4()` chunk uids are accidentally immune, which is exactly why making them +deterministic is the dangerous part of proposal #5. + +**Fix (verified):** key record chunk uids on the run's *effective* `DocumentInfo.uid` — +`sha256(effective_document_id :: record_key)`. The pending run's chunk set is then disjoint +(0 shared), the cutover deletes only v1's chunks, and all 3 new chunks are promoted. This +preserves today's disjointness property while keeping uids deterministic *within* a run, which is +all that re-ingest idempotency actually requires. + +## The rest of proposal #5 holds + +Second scenario: delete Carol (`E-3`) from `employees.csv` and re-ingest, then run the real +cleanup primitives. + +| primitive | result | +| --- | --- | +| `get_document_entity_candidates()` | 5 candidates — record-chunk entities are found by the existing `(:__Entity__)-[:MENTIONED_IN]->(:Chunk)<-[:PART_OF]-(:Document)` walk | +| `delete_stale_relationships()` | 1 edge deleted — Carol's `WORKS_AT` fact GC'd via `source_chunk_ids` | +| `delete_orphan_entities()` | 2 deleted — Carol **and** `ORG-7`, which lost its only mention; Alice and Bob untouched | + +So the headline claim survives: **record-as-chunk inherits `update()`/`delete_document()` for free, +conditional on effective-uid keying.** The three write-time properties the design flags as +mandatory (`name` on nodes, `fact` and `source_chunk_ids` on `RELATES`) are load-bearing here — +`delete_stale_relationships` is driven entirely by `source_chunk_ids`, and without it Carol's fact +would survive forever. + +## Consequence for the design + +`docs/design/structured-ingestion.md` §5's callout box is **confirmed** and should be promoted from +"design-review finding" to a hard rule with this measurement attached. Proposal #5's claim that +row-level incremental update follows for free stays **refuted** (§8.8) — the pending id is +randomised per run, so a v2 run can never MERGE onto v1's chunks by construction. Row-level diffing +needs its own mechanism, not a uid convention. diff --git a/poc/structured-ingestion/s4_record_as_chunk/spike.py b/poc/structured-ingestion/s4_record_as_chunk/spike.py new file mode 100644 index 00000000..738c25d6 --- /dev/null +++ b/poc/structured-ingestion/s4_record_as_chunk/spike.py @@ -0,0 +1,240 @@ +"""s4 — is record-as-chunk really free, and is the cutover trap real? + +Proposal #5 persists each structured record as a `Chunk` with a **deterministic** +uid (replacing today's `uuid4()`), and claims that this makes `update()` and +`delete_document()` work unchanged. + +The design review predicted a data-loss trap in that claim, from reading +`api/main.py:2104` + `graph_store.rollforward_cutover()`. A prediction from +reading code is a hypothesis. This spike runs it against a real FalkorDB using +the **real `GraphStore`**, and either confirms it or kills it. + + A chunk uid keyed on the CANONICAL document id -> predicted: data loss + B chunk uid keyed on the run's EFFECTIVE document id -> predicted: safe + +It then checks the three cleanup behaviours proposal #5 depends on: +`get_document_entity_candidates`, `delete_orphan_entities`, +`delete_stale_relationships`. +""" + +from __future__ import annotations + +import asyncio +import csv +import hashlib +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from _harness.env import FIXTURES, Report, connection, falkor_available, reset_graph # noqa: E402 + +from graphrag_sdk.core.models import GraphNode, GraphRelationship # noqa: E402 +from graphrag_sdk.ingestion.extraction_strategies.entity_extractors import ( # noqa: E402 + compute_entity_id, +) +from graphrag_sdk.storage.graph_store import GraphStore # noqa: E402 + +DOC_ID = "employees.csv" + + +def record_chunk_uid(document_id: str, record_key: str) -> str: + """Proposal #5's deterministic chunk uid.""" + return hashlib.sha256(f"{document_id}::{record_key}".encode()).hexdigest()[:32] + + +def rows(name: str = "employees.csv") -> list[dict[str, str]]: + with open(FIXTURES / name, newline="", encoding="utf-8") as fh: + return list(csv.DictReader(fh)) + + +async def write_run( + store: GraphStore, + *, + effective_doc_id: str, + chunk_key_doc_id: str, + records: list[dict[str, str]], +) -> list[str]: + """One structured ingestion run, exactly as proposal #6 step 3+8+9 would. + + ``effective_doc_id`` is the Document node actually written (the pending id + during an update). ``chunk_key_doc_id`` is what the record chunk uid is + derived from — the whole point of the experiment. + """ + await store.upsert_nodes( + [GraphNode(id=effective_doc_id, label="Document", properties={"path": DOC_ID})] + ) + chunk_ids: list[str] = [] + nodes: list[GraphNode] = [] + rels: list[GraphRelationship] = [] + for row in records: + uid = record_chunk_uid(chunk_key_doc_id, row["employee_id"]) + chunk_ids.append(uid) + text = f"{row['full_name']} · age {row['age']} · {row['job_title']} at {row['org_id']}" + nodes.append(GraphNode(id=uid, label="Chunk", properties={"text": text, "kind": "record"})) + pid = compute_entity_id(row["employee_id"], "Person") + oid = compute_entity_id(row["org_id"], "Organization") + nodes += [ + GraphNode( + id=pid, + label="Person", + properties={"name": row["full_name"], "title": row["job_title"]}, + ), + GraphNode(id=oid, label="Organization", properties={"name": row["org_id"]}), + ] + rels += [ + GraphRelationship(start_node_id=effective_doc_id, end_node_id=uid, type="PART_OF"), + GraphRelationship(start_node_id=pid, end_node_id=uid, type="MENTIONED_IN"), + GraphRelationship(start_node_id=oid, end_node_id=uid, type="MENTIONED_IN"), + GraphRelationship( + start_node_id=pid, + end_node_id=oid, + type="RELATES", + properties={ + "rel_type": "WORKS_AT", + "fact": f"({row['full_name']}, WORKS_AT, {row['org_id']})", + "source_chunk_ids": [uid], + "src_name": row["full_name"], + "tgt_name": row["org_id"], + }, + ), + ] + await store.upsert_nodes(nodes) + await store.upsert_relationships(rels) + return chunk_ids + + +async def count_chunks(store: GraphStore, document_id: str) -> int: + res = await store.query_raw( + "MATCH (:Document {id:$id})-[:PART_OF]->(c:Chunk) RETURN count(c) AS n", + {"id": document_id}, + ) + return res.result_set[0][0] if res.result_set else 0 + + +async def cutover_scenario(*, key_on_canonical: bool) -> dict[str, int]: + """v1 ingest -> update() writes a pending -> rollforward_cutover -> measure.""" + tag = "canonical" if key_on_canonical else "effective" + conn = connection(f"poc_s4_cutover_{tag}") + store = GraphStore(conn) + await reset_graph(conn) + + v1 = rows() + await write_run(store, effective_doc_id=DOC_ID, chunk_key_doc_id=DOC_ID, records=v1) + before = await count_chunks(store, DOC_ID) + + # update(): api/main.py builds pending_id = f"{resolved_id}__pending__{uuid4().hex[:8]}" + pending_id = f"{DOC_ID}__pending__ab12cd34" + v2 = [dict(r) for r in v1] + v2[0]["job_title"] = "Staff Engineer" # a real edit + await write_run( + store, + effective_doc_id=pending_id, + chunk_key_doc_id=DOC_ID if key_on_canonical else pending_id, + records=v2, + ) + shared = await store.query_raw( + "MATCH (:Document {id:$a})-[:PART_OF]->(c:Chunk)<-[:PART_OF]-(:Document {id:$b}) " + "RETURN count(c) AS n", + {"a": DOC_ID, "b": pending_id}, + ) + shared_chunks = shared.result_set[0][0] if shared.result_set else 0 + + await store.rollforward_cutover(pending_id, DOC_ID, DOC_ID, "hash-v2") + after = await count_chunks(store, DOC_ID) + await conn.close() + return {"before": before, "shared_with_pending": shared_chunks, "after_cutover": after} + + +async def cleanup_scenario() -> dict[str, int]: + """Do the three cleanup primitives behave as proposal #5 assumes?""" + conn = connection("poc_s4_cleanup") + store = GraphStore(conn) + await reset_graph(conn) + + v1 = rows() + old_chunks = await write_run( + store, effective_doc_id=DOC_ID, chunk_key_doc_id=DOC_ID, records=v1 + ) + candidates = await store.get_document_entity_candidates(DOC_ID) + + # Carol (E-3) is deleted from the source file; her org ORG-7 loses its only mention. + survivors = [r for r in v1 if r["employee_id"] != "E-3"] + removed_chunk = record_chunk_uid(DOC_ID, "E-3") + await store.query_raw("MATCH (c:Chunk {id:$id}) DETACH DELETE c", {"id": removed_chunk}) + await write_run(store, effective_doc_id=DOC_ID, chunk_key_doc_id=DOC_ID, records=survivors) + + stale = await store.delete_stale_relationships(candidates, [removed_chunk]) + orphans = await store.delete_orphan_entities(candidates) + remaining_people = await store.query_raw("MATCH (p:Person) RETURN count(p) AS n") + remaining_edges = await store.query_raw("MATCH ()-[r:RELATES]->() RETURN count(r) AS n") + await conn.close() + return { + "candidates": len(candidates), + "old_chunks": len(old_chunks), + "stale_edges_deleted": stale, + "orphans_deleted": orphans, + "people_left": remaining_people.result_set[0][0], + "relates_left": remaining_edges.result_set[0][0], + } + + +async def main() -> int: + r = Report("s4 — record-as-chunk & the update() cutover") + if not falkor_available(): + r.note("SKIPPED — no FalkorDB on FALKOR_HOST:FALKOR_PORT") + return 0 + + canonical = await cutover_scenario(key_on_canonical=True) + effective = await cutover_scenario(key_on_canonical=False) + r.note(f"chunk uid keyed on CANONICAL doc id: {canonical}") + r.note(f"chunk uid keyed on EFFECTIVE doc id: {effective}") + + r.check( + canonical["shared_with_pending"] == canonical["before"] > 0, + "canonical keying makes the pending run MERGE onto the LIVE document's chunks", + f"{canonical['shared_with_pending']} chunk nodes shared by both Documents", + ) + r.check( + canonical["after_cutover"] == 0, + "CONFIRMED: the predicted data-loss trap is real", + f"{canonical['before']} chunks before update, {canonical['after_cutover']} after cutover" + " — every record chunk destroyed, no error raised", + ) + r.check( + effective["shared_with_pending"] == 0, + "effective keying keeps the two runs' chunk sets disjoint", + ) + r.check( + effective["after_cutover"] == effective["before"] > 0, + "effective keying survives the cutover intact", + f"{effective['after_cutover']} chunks promoted", + ) + + cleanup = await cleanup_scenario() + r.note(f"cleanup: {cleanup}") + r.check( + cleanup["candidates"] > 0, + "get_document_entity_candidates() sees record-chunk entities", + f"{cleanup['candidates']} candidates via " + "(:__Entity__)-[:MENTIONED_IN]->(:Chunk)<-[:PART_OF]-(:Document)", + ) + r.check( + cleanup["stale_edges_deleted"] == 1, + "delete_stale_relationships() GCs the removed row's fact", + f"{cleanup['stale_edges_deleted']} edge(s) deleted via source_chunk_ids", + ) + r.check( + cleanup["orphans_deleted"] == 2 and cleanup["people_left"] == 2, + "delete_orphan_entities() removes exactly the vanished row's entities", + f"{cleanup['orphans_deleted']} orphans (Carol + ORG-7), " + f"{cleanup['people_left']} people left", + ) + r.note( + "=> proposal #5's claim holds — record-as-chunk inherits orphan cleanup unchanged, " + "PROVIDED chunk uids are keyed on the run's effective document id" + ) + return r.verdict() + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/poc/structured-ingestion/s5_pipeline_seam/NOTES.md b/poc/structured-ingestion/s5_pipeline_seam/NOTES.md new file mode 100644 index 00000000..ed115f24 --- /dev/null +++ b/poc/structured-ingestion/s5_pipeline_seam/NOTES.md @@ -0,0 +1,59 @@ +# s5 — pipeline seam · DECIDED + +**Question.** Proposal #6 marks steps 3 / 6 / 9 as "♻ existing implementation, factored into a +shared base, **not** copied". Do the real signatures allow that, and which factoring is right? + +**Run:** `python s5_pipeline_seam/spike.py` (needs FalkorDB; no keys). All checks pass. + +## The good news: the steps are reusable verbatim + +``` +_build_lexical_graph(self, doc_info: DocumentInfo, chunks: TextChunks, ctx: Context, *, + content_hash: str | None = None) -> None +``` + +It consumes `TextChunks` and only ever reads `chunk.uid` / `.text` / `.index` / `.metadata`. Since +proposal #5 already says *a record is a chunk*, records map onto `TextChunks` directly — +**no new type and no signature change** is needed to reuse step 3. Same for `_prune` (pure +function over `GraphData` + `Ontology`) and `_write_mentions` (reads `graph_data.mentions`, writes +via `graph_store`). All three depend on `self.graph_store` and nothing else. + +Both factorings were run end-to-end against FalkorDB and produced **identical graphs** +(9 nodes, 6 `MENTIONED_IN` edges). + +## The bad news: subclassing is the wrong seam + +``` +IngestionPipeline.__init__ required args: + ['loader', 'chunker', 'extractor', 'resolver', 'graph_store', 'vector_store'] +``` + +A structured pipeline has a *record* loader, no chunker (records are already chunks), and no LLM +extractor (mapping is deterministic — that is the entire point of the design). Subclassing forces +it to pass `None` for `chunker` and `extractor` and hope nothing ever touches them. It works today +purely by accident of which methods we call, and it converts every future change to +`IngestionPipeline.run()` into a latent `AttributeError` on the structured path. + +**Decision: extract the three methods into a `LexicalGraphWriter` base that depends only on +`graph_store`**, and have both pipelines inherit it. Verified in the spike as +`LexicalGraphMixin` — same graph, no dead dependencies, and it preserves the property proposal #6 +actually cares about: step 9's ordering lives in exactly one place, so the concurrency invariant +guarded by the warning box at `pipeline.py:235–260` cannot be broken on one path only. + +Cost in `src`: move three method bodies to a new base class; `IngestionPipeline` keeps its public +surface unchanged. + +## One thing reuse gets wrong: `NEXT_CHUNK` + +`_build_lexical_graph` unconditionally chains `prev_chunk -[NEXT_CHUNK]-> chunk`. Reused for +records, that asserts a sequential relationship **between unrelated table rows** — the spike +measured 2 edges for 3 rows, so N-1 for any source. For a 1M-row CSV that is 1M meaningless edges, +and `cypher_generation.py:129` actively tells the LLM "NEXT_CHUNK: connects Chunk to next +sequential Chunk", which is false for a table. + +Sort order in a CSV is usually incidental, so these edges are not merely useless — they encode a +claim that isn't true. + +**Decision:** add `link_sequential: bool = True` to `_build_lexical_graph` and pass `False` for +record chunks. One keyword-only argument, default preserves today's behaviour exactly. This is the +*only* signature change the whole seam needs. diff --git a/poc/structured-ingestion/s5_pipeline_seam/spike.py b/poc/structured-ingestion/s5_pipeline_seam/spike.py new file mode 100644 index 00000000..8ebf0eff --- /dev/null +++ b/poc/structured-ingestion/s5_pipeline_seam/spike.py @@ -0,0 +1,243 @@ +"""s5 — can `StructuredIngestionPipeline` reuse the existing pipeline's steps? + +Proposal #6 says steps 3 (lexical graph), 6 (prune) and 9 (mentions) are +"♻ existing implementation, factored into a shared base, **not** copied", +because step 9's ordering is load-bearing for concurrent-update correctness. + +That is an assertion about whether the real method signatures allow it. This +spike tries both factorings against the real `IngestionPipeline` and a real +FalkorDB, and reports which one actually works and what it costs in `src`. + + A subclass `IngestionPipeline` and call its methods + B a mixin holding only the three reusable methods +""" + +from __future__ import annotations + +import asyncio +import csv +import hashlib +import inspect +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from _harness.env import FIXTURES, Report, connection, falkor_available, reset_graph # noqa: E402 + +from graphrag_sdk.core.context import Context # noqa: E402 +from graphrag_sdk.core.models import ( # noqa: E402 + DocumentInfo, + Entity, + EntityMention, + GraphData, + GraphNode, + GraphRelationship, + Ontology, + Relation, + TextChunk, + TextChunks, +) +from graphrag_sdk.ingestion.extraction_strategies.entity_extractors import ( # noqa: E402 + compute_entity_id, +) +from graphrag_sdk.ingestion.pipeline import IngestionPipeline # noqa: E402 +from graphrag_sdk.storage.graph_store import GraphStore # noqa: E402 + + +def rows() -> list[dict[str, str]]: + with open(FIXTURES / "employees.csv", newline="", encoding="utf-8") as fh: + return list(csv.DictReader(fh)) + + +def records_to_chunks(doc_uid: str, records: list[dict[str, str]]) -> TextChunks: + """Proposal #5: a record IS a chunk. Deterministic uid on the EFFECTIVE doc id (s4).""" + return TextChunks( + chunks=[ + TextChunk( + text=f"{r['full_name']} · age {r['age']} · {r['job_title']} at {r['org_id']}", + index=i, + uid=hashlib.sha256(f"{doc_uid}::{r['employee_id']}".encode()).hexdigest()[:32], + metadata={"kind": "record", "record_key": r["employee_id"]}, + ) + for i, r in enumerate(records) + ] + ) + + +def records_to_graph_data(records: list[dict[str, str]], chunks: TextChunks) -> GraphData: + """Proposal #6 step 4: pure mapping function, no LLM.""" + nodes, rels, mentions = [], [], [] + for r, chunk in zip(records, chunks.chunks): + pid = compute_entity_id(r["employee_id"], "Person") + oid = compute_entity_id(r["org_id"], "Organization") + nodes += [ + GraphNode( + id=pid, + label="Person", + properties={"name": r["full_name"], "title": r["job_title"], "age": int(r["age"])}, + ), + GraphNode(id=oid, label="Organization", properties={"name": r["org_id"]}), + ] + rels.append( + GraphRelationship( + start_node_id=pid, + end_node_id=oid, + type="RELATES", + properties={ + "rel_type": "WORKS_AT", + "fact": f"({r['full_name']}, WORKS_AT, {r['org_id']})", + "source_chunk_ids": [chunk.uid], + "src_name": r["full_name"], + "tgt_name": r["org_id"], + }, + ) + ) + mentions += [ + EntityMention(chunk_id=chunk.uid, entity_id=pid), + EntityMention(chunk_id=chunk.uid, entity_id=oid), + ] + return GraphData(nodes=nodes, relationships=rels, mentions=mentions) + + +ONTOLOGY = Ontology( + entities=[Entity(label="Person"), Entity(label="Organization")], + relations=[Relation(label="WORKS_AT", patterns=[("Person", "Organization")])], +) + + +# ── Factoring A: subclass the real pipeline ────────────────────── + + +class SubclassStructuredPipeline(IngestionPipeline): + async def run_structured(self, records, doc_info, ctx): # type: ignore[no-untyped-def] + chunks = records_to_chunks(doc_info.uid, records) + await self._build_lexical_graph(doc_info, chunks, ctx) # step 3 ♻ + data = records_to_graph_data(records, chunks) # step 4 + data = self._prune(data, self.ontology) # step 6 ♻ + await self.graph_store.upsert_nodes(data.nodes) # step 8 + await self.graph_store.upsert_relationships(data.relationships) + return await self._write_mentions(data, ctx) # step 9 ♻ + + +# ── Factoring B: a mixin carrying only the reusable steps ──────── + + +class LexicalGraphMixin: + """What the shared base would look like: depends on graph_store, nothing else.""" + + graph_store: GraphStore + + _build_lexical_graph = IngestionPipeline._build_lexical_graph + _prune = IngestionPipeline._prune + _write_mentions = IngestionPipeline._write_mentions + + +class MixinStructuredPipeline(LexicalGraphMixin): + def __init__(self, graph_store: GraphStore, ontology: Ontology) -> None: + self.graph_store = graph_store + self.ontology = ontology + + async def run_structured(self, records, doc_info, ctx): # type: ignore[no-untyped-def] + chunks = records_to_chunks(doc_info.uid, records) + await self._build_lexical_graph(doc_info, chunks, ctx) + data = records_to_graph_data(records, chunks) + data = self._prune(data, self.ontology) + await self.graph_store.upsert_nodes(data.nodes) + await self.graph_store.upsert_relationships(data.relationships) + return await self._write_mentions(data, ctx) + + +async def main() -> int: + r = Report("s5 — pipeline seam") + if not falkor_available(): + r.note("SKIPPED — no FalkorDB on FALKOR_HOST:FALKOR_PORT") + return 0 + + # What does __init__ demand of a structured pipeline that has no chunker + # and no LLM extractor? + params = inspect.signature(IngestionPipeline.__init__).parameters + required = [ + n + for n, p in params.items() + if n != "self" and p.default is inspect.Parameter.empty and p.kind != p.VAR_KEYWORD + ] + r.note(f"IngestionPipeline.__init__ required args: {required}") + r.check( + {"chunker", "extractor"} <= set(required), + "subclassing forces a structured pipeline to supply a chunker and an LLM extractor", + "neither exists on the structured path — they would be dead None placeholders", + ) + + # A — subclass, with None for the strategies it has no use for. + conn = connection("poc_s5_subclass") + store = GraphStore(conn) + await reset_graph(conn) + ctx = Context() + doc = DocumentInfo(path="employees.csv", uid="doc-employees-A") + try: + pipe_a = SubclassStructuredPipeline( + loader=None, # type: ignore[arg-type] + chunker=None, # type: ignore[arg-type] + extractor=None, # type: ignore[arg-type] + resolver=None, # type: ignore[arg-type] + graph_store=store, + vector_store=None, + ontology=ONTOLOGY, + ) + mentions_a = await pipe_a.run_structured(rows(), doc, ctx) + r.check( + mentions_a > 0, + "A: subclassing works at runtime — the reused steps never touch the unused strategies", + f"{mentions_a} MENTIONED_IN edges written", + ) + except Exception as exc: # noqa: BLE001 + r.check(False, "A: subclassing works at runtime", f"{type(exc).__name__}: {exc}") + a_stats = await store.get_statistics() + await conn.close() + + # B — mixin. + conn = connection("poc_s5_mixin") + store = GraphStore(conn) + await reset_graph(conn) + doc = DocumentInfo(path="employees.csv", uid="doc-employees-B") + pipe_b = MixinStructuredPipeline(store, ONTOLOGY) + mentions_b = await pipe_b.run_structured(rows(), doc, ctx) + r.check( + mentions_b == mentions_a, + "B: the mixin factoring produces an identical graph with no dead dependencies", + f"{mentions_b} MENTIONED_IN edges", + ) + b_stats = await store.get_statistics() + r.check( + a_stats.get("node_count") == b_stats.get("node_count"), + "A and B agree on the resulting graph", + f"A={a_stats.get('node_count')} nodes · B={b_stats.get('node_count')} nodes", + ) + + # The three steps are reusable *verbatim* — no signature change needed. + sig = inspect.signature(IngestionPipeline._build_lexical_graph) + r.check( + "chunks" in sig.parameters, + "_build_lexical_graph() consumes TextChunks, so records need no new type to reuse it", + f"signature: {sig}", + ) + + # ...but it unconditionally chains NEXT_CHUNK between adjacent chunks. + nxt = await store.query_raw("MATCH ()-[r:NEXT_CHUNK]->() RETURN count(r) AS n") + n_next = nxt.result_set[0][0] + r.check( + n_next == len(rows()) - 1, + "reusing it also chains NEXT_CHUNK between unrelated CSV rows", + f"{n_next} NEXT_CHUNK edges for {len(rows())} rows — N-1 edges asserting a " + "sequential relationship that does not exist between table rows", + ) + r.note( + "cypher_generation.py tells the LLM 'NEXT_CHUNK: connects Chunk to next sequential " + "Chunk' — false for rows, and 1M rows means 1M meaningless edges" + ) + await conn.close() + return r.verdict() + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main()))