From dfbca40730b8cf9cc015f569d1ddd9c7226ed2ce Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Fri, 17 Apr 2026 15:18:22 +0200 Subject: [PATCH 1/8] =?UTF-8?q?feat(neo4j):=20Day=201=20=E2=80=94=20compos?= =?UTF-8?q?e=20service,=20driver,=20schema=20bootstrap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add Neo4j as an optional graph-native storage layer (ingestion profile). Introduces infra/neo4j with a singleton async driver wrapper and an idempotent bootstrap of constraints + indexes, wired into the FastAPI lifespan. Integration tests skip when no live Neo4j is reachable. Refs #186 --- .env.example | 5 + docker-compose.dev.yml | 27 ++ docker-compose.yml | 23 + docs/design/neo4j-integration.md | 435 ++++++++++++++++++ document-parser/infra/neo4j/__init__.py | 10 + document-parser/infra/neo4j/driver.py | 48 ++ document-parser/infra/neo4j/schema.py | 51 ++ document-parser/infra/settings.py | 6 + document-parser/main.py | 31 +- document-parser/requirements.txt | 1 + document-parser/tests/neo4j/__init__.py | 0 document-parser/tests/neo4j/conftest.py | 36 ++ .../tests/neo4j/test_document_roundtrip.py | 32 ++ document-parser/tests/neo4j/test_driver.py | 10 + document-parser/tests/neo4j/test_schema.py | 38 ++ 15 files changed, 752 insertions(+), 1 deletion(-) create mode 100644 docs/design/neo4j-integration.md create mode 100644 document-parser/infra/neo4j/__init__.py create mode 100644 document-parser/infra/neo4j/driver.py create mode 100644 document-parser/infra/neo4j/schema.py create mode 100644 document-parser/tests/neo4j/__init__.py create mode 100644 document-parser/tests/neo4j/conftest.py create mode 100644 document-parser/tests/neo4j/test_document_roundtrip.py create mode 100644 document-parser/tests/neo4j/test_driver.py create mode 100644 document-parser/tests/neo4j/test_schema.py diff --git a/.env.example b/.env.example index b423dce..d7f515c 100644 --- a/.env.example +++ b/.env.example @@ -48,3 +48,8 @@ # Embedding vector dimension (default: 384 for Granite Embedding 30M / all-MiniLM-L6-v2) # EMBEDDING_DIMENSION=384 + +# Neo4j — graph-native document structure (used by docker-compose.dev.yml) +# NEO4J_URI=bolt://neo4j:7687 +# NEO4J_USER=neo4j +# NEO4J_PASSWORD=changeme diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 9255c3b..9a23fff 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -9,6 +9,26 @@ # ============================================================================= services: + # --- Neo4j (graph-native document structure) --- + neo4j: + image: neo4j:5.15-community + environment: + NEO4J_AUTH: ${NEO4J_USER:-neo4j}/${NEO4J_PASSWORD:-changeme} + NEO4J_server_memory_heap_initial__size: 512m + NEO4J_server_memory_heap_max__size: 1g + ports: + - "7474:7474" + - "7687:7687" + volumes: + - neo4j_data:/data + - neo4j_logs:/logs + healthcheck: + test: ["CMD-SHELL", "cypher-shell -u $${NEO4J_USER:-neo4j} -p $${NEO4J_PASSWORD:-changeme} 'RETURN 1' || exit 1"] + interval: 10s + timeout: 5s + retries: 10 + start_period: 30s + # --- OpenSearch (single-node, security disabled for local dev) --- opensearch: image: opensearchproject/opensearch:2 @@ -77,12 +97,17 @@ services: BATCH_PAGE_SIZE: ${BATCH_PAGE_SIZE:-10} OPENSEARCH_URL: http://opensearch:9200 EMBEDDING_URL: http://embedding:8001 + NEO4J_URI: bolt://neo4j:7687 + NEO4J_USER: ${NEO4J_USER:-neo4j} + NEO4J_PASSWORD: ${NEO4J_PASSWORD:-changeme} command: ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"] depends_on: opensearch: condition: service_healthy embedding: condition: service_healthy + neo4j: + condition: service_healthy deploy: resources: limits: @@ -105,6 +130,8 @@ services: volumes: opensearch_data: + neo4j_data: + neo4j_logs: uploads_data: db_data: frontend_node_modules: diff --git a/docker-compose.yml b/docker-compose.yml index 918bfb9..49c9880 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,4 +1,22 @@ services: + # --- Neo4j (graph-native document structure) --- + neo4j: + profiles: ["ingestion"] + image: neo4j:5.15-community + environment: + NEO4J_AUTH: ${NEO4J_USER:-neo4j}/${NEO4J_PASSWORD:-changeme} + NEO4J_server_memory_heap_initial__size: 512m + NEO4J_server_memory_heap_max__size: 1g + volumes: + - neo4j_data:/data + - neo4j_logs:/logs + healthcheck: + test: ["CMD-SHELL", "cypher-shell -u $${NEO4J_USER:-neo4j} -p $${NEO4J_PASSWORD:-changeme} 'RETURN 1' || exit 1"] + interval: 10s + timeout: 5s + retries: 10 + start_period: 30s + # --- OpenSearch (single-node, security disabled) --- opensearch: profiles: ["ingestion"] @@ -53,6 +71,9 @@ services: BATCH_PAGE_SIZE: ${BATCH_PAGE_SIZE:-10} OPENSEARCH_URL: ${OPENSEARCH_URL:-} EMBEDDING_URL: ${EMBEDDING_URL:-} + NEO4J_URI: ${NEO4J_URI:-} + NEO4J_USER: ${NEO4J_USER:-neo4j} + NEO4J_PASSWORD: ${NEO4J_PASSWORD:-changeme} deploy: resources: limits: @@ -69,5 +90,7 @@ services: volumes: opensearch_data: + neo4j_data: + neo4j_logs: uploads_data: db_data: diff --git a/docs/design/neo4j-integration.md b/docs/design/neo4j-integration.md new file mode 100644 index 0000000..1965408 --- /dev/null +++ b/docs/design/neo4j-integration.md @@ -0,0 +1,435 @@ +# Neo4j integration — Docling-Studio v0.5.0 + +Design doc for Neo4j integration targeting release 0.5.0. +Target: Hackernoon hackathon demo (Neo4j partner). + +--- + +## 1. Context and goals + +### Already in Docling-Studio +- Ingestion pipeline: Docling parser → chunking (HybridChunker) → embedding → OpenSearch (vector index) +- Vue 3 + FastAPI UI +- Debug view to inspect/edit chunks before retrieval +- Docker compose with existing services + +### What we add in v0.5.0 +- Neo4j as **graph-native storage** of the document structure +- A new ingestion layer that stores the DoclingDocument tree faithfully as a graph +- Minimal UI to visualize the graph (demo value to the judges) +- Compose pipeline with Neo4j + +### Why graph-native (hackathon positioning) +> Most document AI tools store parsed content as flat chunks in a vector DB. +> Docling-Studio v0.5 introduces a graph-native storage layer on top of Neo4j, +> preserving the full hierarchical structure of documents as first-class citizens. +> This unlocks hybrid retrieval, agentic navigation, and structural debugging — +> impossible with chunk-only stores. + +### Out of scope for v0.5.0 (roadmap mention only) +- EnrichmentWriter (entities / summaries / keywords via docling-agent) — v0.6.0 +- Agent reasoning trace viewer — v0.6.0 +- RAG hybrid (graph traversal + vector) — v0.7.0 +- Document versioning — v0.7.0+ + +--- + +## 2. Architectural principles + +### Port & adapter, with nuance + +**Write side**: one `Writer` port, **composable stages** (not alternative adapters). +Pipelines A and B are additive, not exclusive. + +``` + CORE (always) Pipeline A (RAG) Pipeline B (agent-ready, v0.6+) + ┌─────────────┐ ┌────────────────┐ ┌───────────────────┐ + │ TreeWriter │ ─────▶ │ ChunkWriter │ │ EnrichmentWriter │ + │ │ │ (existing │ │ (via docling- │ + │ │ │ OpenSearch + │ │ agent, v0.6+) │ + │ │ │ adds chunks │ │ │ + │ │ │ to Neo4j) │ │ │ + └─────────────┘ └────────────────┘ └───────────────────┘ +``` + +```python +# docling_studio/ingestion/pipeline.py +class Writer(Protocol): + def write(self, doc: DoclingDocument, ctx: IngestionContext) -> None: ... + +# Explicit composition per use case +def build_pipeline(config: PipelineConfig) -> list[Writer]: + writers = [TreeWriter(neo4j_driver)] + if config.rag_enabled: + writers.append(ChunkWriter(neo4j_driver, chunker, embedder, opensearch)) + if config.enrichment_enabled: # v0.6.0+ + writers.append(EnrichmentWriter(neo4j_driver, docling_agent)) + return writers +``` + +**Read side**: two distinct ports (same Neo4j backend, different queries). + +```python +class RAGRetrievalPort(Protocol): + def search(self, query: str, k: int) -> list[Chunk]: ... + def similar(self, chunk_id: str, k: int) -> list[Chunk]: ... + +class TreeNavigationPort(Protocol): # v0.6.0+ + def get_outline(self, doc_id: str) -> Tree: ... + def read_node(self, ref: str) -> Element: ... + def list_children(self, ref: str) -> list[Element]: ... + def walk(self, ref: str, depth: int) -> SubTree: ... +``` + +--- + +## 3. Neo4j schema + +### Constraints & indexes (created at boot) + +```cypher +// Uniqueness +CREATE CONSTRAINT document_id IF NOT EXISTS + FOR (d:Document) REQUIRE d.id IS UNIQUE; + +CREATE CONSTRAINT element_composite IF NOT EXISTS + FOR (e:Element) REQUIRE (e.doc_id, e.self_ref) IS UNIQUE; + +CREATE CONSTRAINT page_composite IF NOT EXISTS + FOR (p:Page) REQUIRE (p.doc_id, p.page_no) IS UNIQUE; + +CREATE CONSTRAINT chunk_id IF NOT EXISTS + FOR (c:Chunk) REQUIRE c.id IS UNIQUE; + +// Full-text index (element text search) +CREATE FULLTEXT INDEX element_text IF NOT EXISTS + FOR (e:Element) ON EACH [e.text]; + +// Simple indexes for per-doc queries +CREATE INDEX element_doc IF NOT EXISTS FOR (e:Element) ON (e.doc_id); +CREATE INDEX chunk_doc IF NOT EXISTS FOR (c:Chunk) ON (c.doc_id); +``` + +### Data model + +```cypher +// Root document +(:Document { + id: string, // UUID or PDF hash + title: string, + source_uri: string, // path or S3 + ingested_at: datetime, + docling_version: string, + stages_applied: list, // ["tree", "chunks"] etc. + last_tree_write: datetime, + last_chunk_write: datetime, + tenant_id: string // simple multi-tenancy +}) + +// All tree elements (shared :Element label + specific label) +(:Element:SectionHeader {doc_id, self_ref, text, level, prov_page, prov_bbox}) +(:Element:Paragraph {doc_id, self_ref, text, prov_page, prov_bbox}) +(:Element:Table {doc_id, self_ref, caption, cells_json, prov_page, prov_bbox}) +(:Element:Figure {doc_id, self_ref, caption, image_uri, prov_page, prov_bbox}) +(:Element:ListItem {doc_id, self_ref, text, marker, prov_page, prov_bbox}) +(:Element:Formula {doc_id, self_ref, latex, text, prov_page, prov_bbox}) + +// Page for layout provenance +(:Page {doc_id, page_no, width, height}) + +// Chunks (Pipeline A) +(:Chunk { + id, doc_id, + text, + chunk_index, + embedding_ref, // id in OpenSearch (no inline duplication) + token_count +}) +``` + +### Relations + +```cypher +// Hierarchical structure +(:Document)-[:HAS_ROOT]->(:Element) +(:Element)-[:PARENT_OF {order: int}]->(:Element) // order preserves sequence +(:Element)-[:NEXT]->(:Element) // DFS pre-order reading + +// Layout +(:Element)-[:ON_PAGE]->(:Page) + +// Pipeline A (chunking) +(:Document)-[:HAS_CHUNK]->(:Chunk) +(:Chunk)-[:DERIVED_FROM]->(:Element) // back-reference; a chunk can span multiple elements +``` + +### Decisions + +| Decision | Choice | Rationale | +|----------|-------|---------------| +| Element composite key | `(doc_id, self_ref)` | self_ref not unique across docs | +| Multi-tenancy | `tenant_id` property on Document | Simple, filterable, migrable to multi-db later | +| Table cells | `cells_json` property | v0.5 KISS. May model `(Table)-[:HAS_CELL]->(Cell)` in v0.6+ | +| Reading order | `[:NEXT]` chain + `{order}` on `PARENT_OF` | Both views useful | +| Versioning | None (replace strategy on re-upload) | v0.5 KISS | +| APOC | Not required | Pure Cypher is sufficient for v0.5 | + +### Re-ingestion strategy + +```cypher +// Before ingesting, wipe existing +MATCH (d:Document {id: $doc_id}) +OPTIONAL MATCH (d)-[:HAS_ROOT|HAS_CHUNK]->() +DETACH DELETE d +// Then re-walk cleanly +``` + +--- + +## 4. Implementation plan (3 days) + +### Day 1 — Infra + schema + +- [ ] Add `neo4j` service to `docker-compose.yml` (`neo4j:5.15-community`, persistent volume, healthcheck) +- [ ] Add env vars (`NEO4J_URI`, `NEO4J_USER`, `NEO4J_PASSWORD`) to `.env.example` +- [ ] Create module `docling_studio/storage/neo4j/`: + - `driver.py` — neo4j-python driver wrapper (connection pool, context manager) + - `schema.py` — idempotent `bootstrap_schema()` (CREATE CONSTRAINT / INDEX at startup) + - `__init__.py` with exports +- [ ] Hook `bootstrap_schema()` in FastAPI startup +- [ ] Basic integration tests: + - Driver connection + - Schema bootstrap (idempotence verified) + - Simple round-trip: write Document, read Document, delete Document + +**Deliverable:** docker compose boots with healthy Neo4j, schema in place at init. + +### Day 2 — TreeWriter (write + read) + +- [ ] `storage/neo4j/tree_writer.py` — `DoclingDocument → Neo4j` walker + - `write_document(doc, tenant_id, driver)` in a transaction + - DFS pre-order, batched `MERGE` for perf + - Pages first, then Elements, then `PARENT_OF` / `NEXT` / `ON_PAGE` relations + - Dynamic labels based on `node.label` (`SectionHeader`, `Paragraph`, …) +- [ ] `storage/neo4j/tree_reader.py` — inverse walker `Neo4j → DoclingDocument` + - `read_document(doc_id, driver) -> DoclingDocument` + - Loads all Elements + Pages, rebuilds the Pydantic structure + - Prerequisite for v0.6 (feeding docling-agent from Neo4j) +- [ ] Integrate into existing ingestion pipeline: + - Add TreeWriter as first stage of `IngestionPipeline` + - `neo4j_enabled: bool` config toggle +- [ ] Round-trip tests: + - 3–4 varied PDFs (academic, invoice, report) + - Assertion: `doc_original == read_document(write_document(doc_original))` + - Beware dates, bbox floats (tolerance) + +**Deliverable:** A PDF uploaded to Docling-Studio is fully present in Neo4j and rebuildable. + +### Day 3 — UI + ChunkWriter + packaging + +- [ ] `storage/neo4j/chunk_writer.py`: + - After existing chunking, push each Chunk to Neo4j + - Create `(:Chunk)-[:DERIVED_FROM]->(:Element)` via source element `self_ref` + - Do NOT duplicate embeddings (stay in OpenSearch, keep `embedding_ref`) +- [ ] Frontend: new "Graph view" tab in debug panel + - Vue component with `cytoscape` (lighter, better layout API) + - FastAPI endpoint `/api/documents/{doc_id}/graph` returns nodes + edges around a scope (whole doc or subtree) + - View: vertical tree, colors per node type, click-to-zoom, hover details +- [ ] Per-document "Graph-ready" / "RAG-ready" badge in list +- [ ] README update: + - "Graph storage with Neo4j" section + - Schema diagram (Mermaid or image) + - 2–3 Cypher examples like "find all paragraphs under section X that mention Y" + - Neo4j badge in features list +- [ ] (bonus if time) "Query explorer" dev tab for live demo: Cypher editor + results + +**Deliverable:** release 0.5.0 with Neo4j visible, functional, documented. + +--- + +## 5. Proposed code structure + +``` +docling_studio/ +├── storage/ +│ ├── neo4j/ +│ │ ├── __init__.py +│ │ ├── driver.py # connection management +│ │ ├── schema.py # bootstrap_schema() +│ │ ├── tree_writer.py # DoclingDocument -> Neo4j +│ │ ├── tree_reader.py # Neo4j -> DoclingDocument +│ │ ├── chunk_writer.py # Chunks -> Neo4j +│ │ └── queries.py # shared Cypher queries +│ ├── opensearch/ # (existing) +│ └── ports.py # Writer, RAGRetrievalPort protocols +├── ingestion/ +│ └── pipeline.py # IngestionPipeline composing Writers +├── api/ +│ └── graph.py # /api/documents/{id}/graph +└── frontend/ + └── components/ + └── GraphView.vue # cytoscape + graph API fetch +``` + +--- + +## 6. Docker compose (added excerpt) + +```yaml +services: + neo4j: + image: neo4j:5.15-community + environment: + NEO4J_AUTH: ${NEO4J_USER:-neo4j}/${NEO4J_PASSWORD:-changeme} + NEO4J_PLUGINS: '["apoc"]' + NEO4J_server_memory_heap_initial__size: 512m + NEO4J_server_memory_heap_max__size: 1g + ports: + - "7474:7474" # Browser UI (demo) + - "7687:7687" # Bolt protocol + volumes: + - neo4j_data:/data + - neo4j_logs:/logs + healthcheck: + test: ["CMD-SHELL", "cypher-shell -u neo4j -p $${NEO4J_PASSWORD:-changeme} 'RETURN 1' || exit 1"] + interval: 10s + timeout: 5s + retries: 10 + + docling-studio-backend: + depends_on: + neo4j: + condition: service_healthy + environment: + NEO4J_URI: bolt://neo4j:7687 + NEO4J_USER: neo4j + NEO4J_PASSWORD: ${NEO4J_PASSWORD:-changeme} + +volumes: + neo4j_data: + neo4j_logs: +``` + +--- + +## 7. Tests + +### Unit tests +- `tests/storage/neo4j/test_schema.py` — bootstrap is idempotent +- `tests/storage/neo4j/test_tree_writer.py` — round-trip on synthetic DoclingDocument +- `tests/storage/neo4j/test_chunk_writer.py` — chunks written with correct `DERIVED_FROM` + +### Integration tests +- `tests/integration/test_ingestion_pipeline.py` — full pipeline on a real PDF +- PDF fixtures: 1 academic (complex heading hierarchy), 1 invoice (tables), 1 report (lists) + +### E2E (bonus) +- Upload PDF via UI → check structure in Neo4j Browser + +--- + +## 8. Open decisions to settle before coding + +1. **Neo4j edition**: Community (free) or AuraDB (managed) ? + - Rec: Community in Docker for v0.5.0 dev/demo. AuraDB mentioned as prod option. + +2. **Chunks: duplicate embeddings in Neo4j or OpenSearch ref ?** + - Rec: OpenSearch ref (avoid duplication; OpenSearch remains source of truth for vectors). In v0.6+, consider native Neo4j vector index. + +3. **Graph view UI: cytoscape or vis-network ?** + - Rec: **cytoscape** — lighter, better layout API, used by Neo4j itself. + +4. **Graph endpoint: return full doc or paginate ?** + - Rec: full doc for v0.5 (reasonable cap at 200 pages). Pagination in v0.6. + +5. **Error strategy**: if Neo4j is down at ingestion, fail or degrade gracefully ? + - Rec: **fail fast** for v0.5 (avoid silent inconsistencies). `neo4j_required: bool` config option. + +--- + +## 9. Hooks for later (v0.6.0+ — don't implement but prepare) + +**EnrichmentWriter (v0.6)** — will need: +- The reader (Neo4j → DoclingDocument) to re-materialize the doc, feed docling-agent, re-patch enrichments +- A stage addable to `IngestionPipeline` without touching other stages +- An `:Entity` label (not created in v0.5 but schema-compatible) + +**Agent reasoning trace viewer (v0.6)** — will need: +- An event stream (WebSocket) that v0.5 already prepares via the reactive UI +- A node_ref ↔ Element correlation in Neo4j (our composite `self_ref` key is enough) + +**TreeNavigationPort (v0.7)** — will need: +- Optimized Cypher queries for descendant/ancestor walk (indexes already provisioned) + +--- + +## 10. v0.5.0 success criteria + +**Must have:** +- [ ] A PDF uploaded to Docling-Studio is in Neo4j with structure preserved +- [ ] Neo4j Browser shows the graph and is manually explorable +- [ ] A graph visual in the Docling-Studio UI works +- [ ] `docker compose up` works zero-config +- [ ] README mentions Neo4j and describes the schema + +**Nice to have (decreasing priority):** +- [ ] Graph-ready / RAG-ready badge per doc +- [ ] Live query explorer in the UI +- [ ] 2–3 example queries in README that do something impossible with vector-only + +**For the hackathon (post-release):** +- [ ] 60s video: upload PDF → structure in Neo4j → cross-doc query impossible in vector-only +- [ ] HackerNoon post explaining "graph-native documents" positioning +- [ ] Explicit Neo4j partnership mention + +--- + +## 11. Fundamental architectural decisions recap + +| Question | Answer | +|----------|---------| +| Is Neo4j source of truth or cache ? | **Source of truth** for structure. OpenSearch remains source of truth for embeddings. | +| Does chunking go away ? | No, v0.5.0 keeps existing chunking. "Chunkless" is Pipeline B, v0.6+. | +| Can it be toggled per doc ? | Yes — `stages_applied` on Document + pipeline config | +| What about OpenSearch ? | Stays, stores vectors. Neo4j tracks `(:Chunk)-[:DERIVED_FROM]->(:Element)` links. | +| Multi-tenancy ? | `tenant_id` property on Document, Cypher filter | +| Versioning ? | None for v0.5.0 — replace strategy on re-upload | + +--- + +## Appendix — Demo queries + +### Query 1 — All "Methods" sections across documents +```cypher +MATCH (d:Document)-[:HAS_ROOT]->(:Element)-[:PARENT_OF*]->(s:SectionHeader) +WHERE toLower(s.text) CONTAINS 'method' +RETURN d.title, s.text, s.level +``` + +### Query 2 — Context of a chunk (parent + siblings) +```cypher +MATCH (c:Chunk {id: $chunk_id})-[:DERIVED_FROM]->(e:Element) +MATCH (e)<-[:PARENT_OF]-(parent:Element) +MATCH (parent)-[:PARENT_OF]->(sibling:Element) +RETURN parent, collect(sibling) AS siblings +``` + +### Query 3 — All tables from a document type +```cypher +MATCH (d:Document)-[:HAS_ROOT]->(:Element)-[:PARENT_OF*]->(t:Table) +WHERE d.source_uri CONTAINS 'invoices/' +RETURN d.title, t.caption, t.cells_json +``` + +### Query 4 — Direct children of a section (ordered) +```cypher +MATCH (s:Element {doc_id: $doc_id, self_ref: $section_ref}) +MATCH (s)-[pof:PARENT_OF]->(child) +RETURN child +ORDER BY pof.order +``` + +--- + +*Single reference doc for Neo4j v0.5.0 implementation. +Read this first in the implementation thread.* diff --git a/document-parser/infra/neo4j/__init__.py b/document-parser/infra/neo4j/__init__.py new file mode 100644 index 0000000..23be98d --- /dev/null +++ b/document-parser/infra/neo4j/__init__.py @@ -0,0 +1,10 @@ +"""Neo4j storage adapter — graph-native document structure. + +Provides a thin driver wrapper, idempotent schema bootstrap, and +walkers between DoclingDocument and the graph model. +""" + +from infra.neo4j.driver import Neo4jDriver, close_driver, get_driver +from infra.neo4j.schema import bootstrap_schema + +__all__ = ["Neo4jDriver", "bootstrap_schema", "close_driver", "get_driver"] diff --git a/document-parser/infra/neo4j/driver.py b/document-parser/infra/neo4j/driver.py new file mode 100644 index 0000000..0745d59 --- /dev/null +++ b/document-parser/infra/neo4j/driver.py @@ -0,0 +1,48 @@ +"""Async Neo4j driver wrapper. + +Owns a single `AsyncDriver` per process. Callers acquire it via +`get_driver()` and must call `close_driver()` at shutdown. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +from neo4j import AsyncDriver, AsyncGraphDatabase + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class Neo4jDriver: + driver: AsyncDriver + database: str = "neo4j" + + +_instance: Neo4jDriver | None = None + + +async def get_driver(uri: str, user: str, password: str, database: str = "neo4j") -> Neo4jDriver: + """Return the process-wide driver, creating it on first call. + + Verifies connectivity once at creation — raises if the server is unreachable. + """ + global _instance + if _instance is not None: + return _instance + + driver = AsyncGraphDatabase.driver(uri, auth=(user, password)) + await driver.verify_connectivity() + logger.info("Neo4j driver connected to %s (db=%s)", uri, database) + _instance = Neo4jDriver(driver=driver, database=database) + return _instance + + +async def close_driver() -> None: + global _instance + if _instance is None: + return + await _instance.driver.close() + _instance = None + logger.info("Neo4j driver closed") diff --git a/document-parser/infra/neo4j/schema.py b/document-parser/infra/neo4j/schema.py new file mode 100644 index 0000000..9307d3b --- /dev/null +++ b/document-parser/infra/neo4j/schema.py @@ -0,0 +1,51 @@ +"""Idempotent Neo4j schema bootstrap. + +Runs at backend startup. All statements use `IF NOT EXISTS`, so calling +this multiple times is safe — it's the contract integration tests rely on. +""" + +from __future__ import annotations + +import logging + +from infra.neo4j.driver import Neo4jDriver + +logger = logging.getLogger(__name__) + + +CONSTRAINTS: tuple[str, ...] = ( + "CREATE CONSTRAINT document_id IF NOT EXISTS " + "FOR (d:Document) REQUIRE d.id IS UNIQUE", + "CREATE CONSTRAINT element_composite IF NOT EXISTS " + "FOR (e:Element) REQUIRE (e.doc_id, e.self_ref) IS UNIQUE", + "CREATE CONSTRAINT page_composite IF NOT EXISTS " + "FOR (p:Page) REQUIRE (p.doc_id, p.page_no) IS UNIQUE", + "CREATE CONSTRAINT chunk_id IF NOT EXISTS " + "FOR (c:Chunk) REQUIRE c.id IS UNIQUE", +) + +INDEXES: tuple[str, ...] = ( + "CREATE INDEX element_doc IF NOT EXISTS FOR (e:Element) ON (e.doc_id)", + "CREATE INDEX chunk_doc IF NOT EXISTS FOR (c:Chunk) ON (c.doc_id)", +) + +FULLTEXT_INDEXES: tuple[str, ...] = ( + "CREATE FULLTEXT INDEX element_text IF NOT EXISTS " + "FOR (e:Element) ON EACH [e.text]", +) + + +async def bootstrap_schema(neo: Neo4jDriver) -> None: + """Create constraints and indexes required by the graph model. + + Idempotent: safe to call on every startup. + """ + async with neo.driver.session(database=neo.database) as session: + for stmt in (*CONSTRAINTS, *INDEXES, *FULLTEXT_INDEXES): + await session.run(stmt) + logger.info( + "Neo4j schema bootstrapped (%d constraints, %d indexes, %d fulltext)", + len(CONSTRAINTS), + len(INDEXES), + len(FULLTEXT_INDEXES), + ) diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index cc322fe..b114045 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -25,6 +25,9 @@ class Settings: batch_page_size: int = 0 # 0 = disabled, > 0 = pages per batch opensearch_url: str = "" # empty = disabled embedding_url: str = "" # empty = disabled (e.g. http://localhost:8001) + neo4j_uri: str = "" # empty = disabled (e.g. bolt://neo4j:7687) + neo4j_user: str = "neo4j" + neo4j_password: str = "changeme" opensearch_default_limit: int = 1000 # max chunks returned by get_chunks embedding_dimension: int = 384 # Granite Embedding 30M / all-MiniLM-L6-v2 upload_dir: str = "./uploads" @@ -102,6 +105,9 @@ class Settings: batch_page_size=int(os.environ.get("BATCH_PAGE_SIZE", "10")), opensearch_url=os.environ.get("OPENSEARCH_URL", ""), embedding_url=os.environ.get("EMBEDDING_URL", ""), + neo4j_uri=os.environ.get("NEO4J_URI", ""), + neo4j_user=os.environ.get("NEO4J_USER", "neo4j"), + neo4j_password=os.environ.get("NEO4J_PASSWORD", "changeme"), opensearch_default_limit=int(os.environ.get("OPENSEARCH_DEFAULT_LIMIT", "1000")), embedding_dimension=int(os.environ.get("EMBEDDING_DIMENSION", "384")), upload_dir=os.environ.get("UPLOAD_DIR", "./uploads"), diff --git a/document-parser/main.py b/document-parser/main.py index a163e85..1c41921 100644 --- a/document-parser/main.py +++ b/document-parser/main.py @@ -93,6 +93,28 @@ def _build_analysis_service( ) +async def _init_neo4j(): + """Initialize the Neo4j driver and bootstrap schema — skip if not configured.""" + if not settings.neo4j_uri: + logger.info("Neo4j disabled (NEO4J_URI not set)") + return None + + from infra.neo4j import Neo4jDriver, bootstrap_schema, get_driver + + try: + neo = await get_driver( + settings.neo4j_uri, + settings.neo4j_user, + settings.neo4j_password, + ) + await bootstrap_schema(neo) + logger.info("Neo4j ready (uri=%s)", settings.neo4j_uri) + return neo + except Exception: + logger.exception("Neo4j init failed — continuing without graph storage") + return None + + def _build_ingestion_service() -> IngestionService | None: """Build the ingestion service — only if embedding + opensearch are configured.""" if not settings.embedding_url or not settings.opensearch_url: @@ -150,8 +172,15 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: if ingestion_service is not None: app.include_router(ingestion_router) logger.info("Ingestion router mounted") + app.state.neo4j = await _init_neo4j() logger.info("Docling Studio backend ready (engine=%s)", settings.conversion_engine) - yield + try: + yield + finally: + if app.state.neo4j is not None: + from infra.neo4j import close_driver + + await close_driver() app = FastAPI( diff --git a/document-parser/requirements.txt b/document-parser/requirements.txt index 2319b46..1137357 100644 --- a/document-parser/requirements.txt +++ b/document-parser/requirements.txt @@ -8,3 +8,4 @@ aiosqlite>=0.20.0,<1.0.0 httpx>=0.27.0,<1.0.0 pypdfium2>=4.0.0,<5.0.0 opensearch-py[async]>=2.6.0,<3.0.0 +neo4j>=5.15.0,<6.0.0 diff --git a/document-parser/tests/neo4j/__init__.py b/document-parser/tests/neo4j/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/document-parser/tests/neo4j/conftest.py b/document-parser/tests/neo4j/conftest.py new file mode 100644 index 0000000..9c77d44 --- /dev/null +++ b/document-parser/tests/neo4j/conftest.py @@ -0,0 +1,36 @@ +"""Shared fixtures for Neo4j integration tests. + +These tests are skipped unless a live Neo4j is reachable via NEO4J_TEST_URI +(defaulting to bolt://localhost:7687). CI spins up `neo4j:5.15-community` +alongside the job; locally run `docker compose -f docker-compose.dev.yml up neo4j`. +""" + +from __future__ import annotations + +import os + +import pytest + +from infra.neo4j import close_driver, get_driver + + +def _cfg() -> tuple[str, str, str]: + return ( + os.environ.get("NEO4J_TEST_URI", "bolt://localhost:7687"), + os.environ.get("NEO4J_TEST_USER", "neo4j"), + os.environ.get("NEO4J_TEST_PASSWORD", "changeme"), + ) + + +@pytest.fixture +async def neo4j_driver(): + uri, user, password = _cfg() + try: + neo = await get_driver(uri, user, password) + except Exception as exc: + pytest.skip(f"Neo4j not reachable at {uri}: {exc}") + # Wipe DB before each test — integration tests assume an empty graph. + async with neo.driver.session(database=neo.database) as session: + await session.run("MATCH (n) DETACH DELETE n") + yield neo + await close_driver() diff --git a/document-parser/tests/neo4j/test_document_roundtrip.py b/document-parser/tests/neo4j/test_document_roundtrip.py new file mode 100644 index 0000000..1303d85 --- /dev/null +++ b/document-parser/tests/neo4j/test_document_roundtrip.py @@ -0,0 +1,32 @@ +"""Minimal Document node round-trip — validates the driver + schema end-to-end.""" + +from __future__ import annotations + +from infra.neo4j.schema import bootstrap_schema + + +async def test_document_write_read_delete(neo4j_driver): + await bootstrap_schema(neo4j_driver) + + async with neo4j_driver.driver.session(database=neo4j_driver.database) as session: + await session.run( + "CREATE (d:Document {id: $id, title: $title, tenant_id: $tenant})", + id="doc-42", + title="Round-trip fixture", + tenant="default", + ) + + result = await session.run( + "MATCH (d:Document {id: $id}) RETURN d.title AS title, d.tenant_id AS tenant", + id="doc-42", + ) + record = await result.single() + assert record is not None + assert record["title"] == "Round-trip fixture" + assert record["tenant"] == "default" + + await session.run("MATCH (d:Document {id: $id}) DETACH DELETE d", id="doc-42") + gone = await ( + await session.run("MATCH (d:Document {id: $id}) RETURN d", id="doc-42") + ).single() + assert gone is None diff --git a/document-parser/tests/neo4j/test_driver.py b/document-parser/tests/neo4j/test_driver.py new file mode 100644 index 0000000..abb42ea --- /dev/null +++ b/document-parser/tests/neo4j/test_driver.py @@ -0,0 +1,10 @@ +"""Neo4j driver connectivity smoke test.""" + +from __future__ import annotations + + +async def test_driver_connects_and_runs_cypher(neo4j_driver): + async with neo4j_driver.driver.session(database=neo4j_driver.database) as session: + result = await session.run("RETURN 1 AS x") + record = await result.single() + assert record["x"] == 1 diff --git a/document-parser/tests/neo4j/test_schema.py b/document-parser/tests/neo4j/test_schema.py new file mode 100644 index 0000000..4e6e2c2 --- /dev/null +++ b/document-parser/tests/neo4j/test_schema.py @@ -0,0 +1,38 @@ +"""Schema bootstrap is idempotent and produces the expected constraints/indexes.""" + +from __future__ import annotations + +from infra.neo4j.schema import CONSTRAINTS, FULLTEXT_INDEXES, INDEXES, bootstrap_schema + + +async def _count_schema(neo4j_driver) -> tuple[int, int]: + async with neo4j_driver.driver.session(database=neo4j_driver.database) as session: + constraints = await (await session.run("SHOW CONSTRAINTS")).data() + indexes = await (await session.run("SHOW INDEXES")).data() + return len(constraints), len(indexes) + + +async def test_bootstrap_is_idempotent(neo4j_driver): + await bootstrap_schema(neo4j_driver) + first = await _count_schema(neo4j_driver) + + # Running a second time must not duplicate anything. + await bootstrap_schema(neo4j_driver) + second = await _count_schema(neo4j_driver) + + assert first == second + # Sanity: we created at least what we declared. + assert first[0] >= len(CONSTRAINTS) + assert first[1] >= len(INDEXES) + len(FULLTEXT_INDEXES) + + +async def test_document_id_is_unique(neo4j_driver): + await bootstrap_schema(neo4j_driver) + async with neo4j_driver.driver.session(database=neo4j_driver.database) as session: + await session.run("CREATE (d:Document {id: 'doc-1', title: 'first'})") + with_err: Exception | None = None + try: + await session.run("CREATE (d:Document {id: 'doc-1', title: 'dup'})") + except Exception as exc: + with_err = exc + assert with_err is not None, "unique constraint on Document.id must reject duplicates" From 34743906883e90ad0b35336c156e7654b095557f Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Fri, 17 Apr 2026 15:21:23 +0200 Subject: [PATCH 2/8] docs(neo4j): ADR-001 graph viz lib + 200-page endpoint cap Document the Cytoscape.js vs vis-network / Neovis / D3 / Reagraph analysis for the graph view, and make the 200-page cap on /api/documents/{id}/graph explicit (HTTP 413 + truncated flag beyond the cap). Refs #186 --- .../ADR-001-graph-visualization-library.md | 142 ++++++++++++++++++ docs/design/neo4j-integration.md | 8 +- 2 files changed, 146 insertions(+), 4 deletions(-) create mode 100644 docs/architecture/adrs/ADR-001-graph-visualization-library.md diff --git a/docs/architecture/adrs/ADR-001-graph-visualization-library.md b/docs/architecture/adrs/ADR-001-graph-visualization-library.md new file mode 100644 index 0000000..7931300 --- /dev/null +++ b/docs/architecture/adrs/ADR-001-graph-visualization-library.md @@ -0,0 +1,142 @@ +# ADR-001: Graph visualization library for the Neo4j graph view + +**Date**: 2026-04-17 +**Status**: Proposed +**Deciders**: Pier-Jean Malandrino + +## Context + +v0.5.0 introduces Neo4j as a graph-native storage layer for parsed documents +(see [docs/design/neo4j-integration.md](../../design/neo4j-integration.md) +and [#186](https://github.com/scub-france/Docling-Studio/issues/186)). We need +an in-app visualization of that graph: the `DoclingDocument` tree as rendered +in Neo4j, with nodes colored by element type (`SectionHeader`, `Paragraph`, +`Table`, `Figure`, `ListItem`, `Formula`) and edges (`PARENT_OF`, `NEXT`, +`ON_PAGE`, `HAS_CHUNK`, `DERIVED_FROM`). + +The view lives in the existing Vue 3 debug panel. It is the **primary demo +artifact** for the Hackernoon hackathon (Neo4j partner), so polish matters as +much as correctness. + +### Constraints + +- Vue 3 + Vite frontend, no framework change +- Must render the full tree of a 200-page document (worst case ≈ a few + thousand nodes; see graph endpoint cap in the design doc §8.4) +- Needs a **clean hierarchical layout** — documents are trees, not arbitrary + graphs; a good tree layout is the single biggest UX lever +- Needs per-node styling (shape + color by label), click, hover, zoom, pan +- Must be installable without Java/Python-side changes +- License compatible with the repo (MIT-ish preferred) + +### Non-goals for v0.5.0 + +- 3D rendering +- Force-directed simulation as the primary layout (we have a tree) +- Editing nodes in place (read-only view) +- Rendering millions of nodes + +## Decision + +Use **Cytoscape.js** via a thin Vue wrapper (`vue-cytoscape` or a bespoke +`GraphView.vue` that imports `cytoscape` directly and uses the +`dagre`/`breadthfirst` layouts). + +## Consequences + +### Positive + +- Battle-tested library (13k+ GitHub stars, maintained since 2013, used by + Neo4j's own "Bloom"-style visualizations in the community) +- First-class support for hierarchical layouts via `cytoscape-dagre` (hub-and- + spoke / tree) and built-in `breadthfirst` — both map naturally to our + `PARENT_OF` structure +- CSS-like selector syntax for styling (`node[label = "Table"] { ... }`), + which is pleasant to evolve as we add node types +- Permissive licensing (MIT) +- Headless mode available, so it can be tested outside a DOM (Jest + jsdom + works cleanly) +- Active ecosystem: `cytoscape-cola`, `cytoscape-klay`, `cytoscape-popper` for + tooltips, all maintained +- Bundle size is reasonable for a demo: ~300 KB min+gz for core + dagre, well + below our current frontend budget + +### Negative + +- Styling DSL is powerful but has its own syntax to learn; not plain CSS +- Large graphs (>10k nodes) benefit from canvas+WebGL libraries + (sigma.js, reagraph) — we are explicitly not in that regime for v0.5, but + we would need to swap if we later visualize the cross-document graph +- No Vue 3 component library that is both maintained and popular — we wrap it + ourselves in `GraphView.vue` (the wrapper is ~50 LOC, so this is minor) + +### Neutral + +- Not "Neo4j-branded": we do not use Neovis.js, which is a thin Cytoscape + wrapper around the Bolt protocol. Our graph API already returns shaped + JSON, so the Neovis convenience is not worth the lock-in +- We take on one runtime dependency (`cytoscape` + `cytoscape-dagre`) + +## Alternatives Considered + +### Alternative 1: vis-network (vis.js) + +- **Pros**: Very easy to get started, built-in physics, shipped by Neo4j + Browser historically +- **Cons**: Maintenance has been rocky (original vis.js split into several + forks; `vis-network` is the maintained branch but releases are sparse); + hierarchical layout is OK but less configurable than dagre; styling API is + less expressive; TypeScript types lag behind the JS API +- **Why rejected**: Hierarchical layout quality is the single most important + criterion for a document tree, and vis-network is clearly a notch below + Cytoscape + dagre here. Maintenance trajectory is also a concern for a + release we want to keep shipping on + +### Alternative 2: Neovis.js + +- **Pros**: Built by Neo4j Labs, connects directly to a Bolt endpoint, nice + out-of-the-box "Neo4j look" +- **Cons**: Wraps Cytoscape anyway, so everything it can do we can do with + Cytoscape directly; expects the browser to talk Bolt, which forces us to + expose Neo4j creds in the frontend OR to proxy Bolt through the backend + (both worse than our current "backend returns JSON" design); limited + customization compared to raw Cytoscape +- **Why rejected**: The auth story is a non-starter for a hackathon demo we + want to show publicly, and we lose nothing vs. Cytoscape by going one + layer lower + +### Alternative 3: D3 (d3-hierarchy + d3-force) + +- **Pros**: Maximum flexibility; beautiful, publication-grade output; full + SVG control +- **Cons**: Much more code for the same result — layout, zoom, pan, hover, + selection all hand-rolled; steeper learning curve for future contributors + to the project; no built-in graph data model +- **Why rejected**: We're building a product feature, not a data-viz + artefact. The time budget (1 day of Day 3) doesn't fit a D3 build-your-own + +### Alternative 4: Reagraph / react-force-graph / sigma.js (WebGL) + +- **Pros**: Scales to tens of thousands of nodes at 60 FPS; good for future + cross-document visualization +- **Cons**: Optimized for force-directed layouts, weaker hierarchical + support; Reagraph is React-only (requires a React island inside Vue); + sigma.js's tree layout is immature +- **Why rejected**: Wrong regime for a single-document tree. Worth + reconsidering if/when we visualize the full corpus graph in a later release + +### Alternative 5: Mermaid + +- **Pros**: Trivial to embed, already used in docs +- **Cons**: Static rendering, no interactivity, not designed for thousands of + nodes, no per-node click/hover +- **Why rejected**: A viewer, not a visualizer. We need interactivity + +## References + +- [Neo4j integration design doc](../../design/neo4j-integration.md) §8.3 +- [Issue #186 — Neo4j integration](https://github.com/scub-france/Docling-Studio/issues/186) +- [Cytoscape.js](https://js.cytoscape.org/) +- [cytoscape-dagre](https://github.com/cytoscape/cytoscape.js-dagre) +- [vis-network](https://visjs.github.io/vis-network/docs/network/) +- [Neovis.js](https://github.com/neo4j-contrib/neovis.js) diff --git a/docs/design/neo4j-integration.md b/docs/design/neo4j-integration.md index 1965408..6090192 100644 --- a/docs/design/neo4j-integration.md +++ b/docs/design/neo4j-integration.md @@ -232,8 +232,8 @@ DETACH DELETE d - Create `(:Chunk)-[:DERIVED_FROM]->(:Element)` via source element `self_ref` - Do NOT duplicate embeddings (stay in OpenSearch, keep `embedding_ref`) - [ ] Frontend: new "Graph view" tab in debug panel - - Vue component with `cytoscape` (lighter, better layout API) - - FastAPI endpoint `/api/documents/{doc_id}/graph` returns nodes + edges around a scope (whole doc or subtree) + - Vue component with `cytoscape` (lighter, better layout API — see [ADR-001](../architecture/adrs/ADR-001-graph-visualization-library.md)) + - FastAPI endpoint `/api/documents/{doc_id}/graph` returns full nodes + edges for the document, **capped at 200 pages** (HTTP 413 beyond; pagination deferred to v0.6). The endpoint must include a `truncated: bool` flag and `node_count` / `edge_count` in the response envelope so the UI can warn the user cleanly. - View: vertical tree, colors per node type, click-to-zoom, hover details - [ ] Per-document "Graph-ready" / "RAG-ready" badge in list - [ ] README update: @@ -337,10 +337,10 @@ volumes: - Rec: OpenSearch ref (avoid duplication; OpenSearch remains source of truth for vectors). In v0.6+, consider native Neo4j vector index. 3. **Graph view UI: cytoscape or vis-network ?** - - Rec: **cytoscape** — lighter, better layout API, used by Neo4j itself. + - Decided: **Cytoscape.js** — see [ADR-001](../architecture/adrs/ADR-001-graph-visualization-library.md) for the full analysis. 4. **Graph endpoint: return full doc or paginate ?** - - Rec: full doc for v0.5 (reasonable cap at 200 pages). Pagination in v0.6. + - Decided: full doc for v0.5, **hard cap at 200 pages**. Beyond the cap, the endpoint returns HTTP 413 with a `truncated: true` flag; the UI shows "Graph too large to render — reduce scope". Pagination ships in v0.6. 5. **Error strategy**: if Neo4j is down at ingestion, fail or degrade gracefully ? - Rec: **fail fast** for v0.5 (avoid silent inconsistencies). `neo4j_required: bool` config option. From ee92e3c580f095ae7e35f09739b5a31e74b3926c Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Fri, 17 Apr 2026 15:28:28 +0200 Subject: [PATCH 3/8] =?UTF-8?q?feat(neo4j):=20Day=202=20=E2=80=94=20TreeWr?= =?UTF-8?q?iter,=20TreeReader,=20pipeline=20wiring?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Serialize a DoclingDocument to a Neo4j graph: Document + Page + Element nodes with dynamic specific labels (SectionHeader, Paragraph, Table, Figure, …), plus HAS_ROOT / PARENT_OF / NEXT / ON_PAGE edges. Replace-on- write for idempotent re-ingestion. The reader returns the verbatim document_json stored on the Document node — reconstruction from graph nodes is deferred to v0.6. Wired into AnalysisService._finalize_analysis: runs after conversion, degrades gracefully by default, fails fast when neo4j_required is set. Refs #186 --- document-parser/infra/neo4j/__init__.py | 18 +- document-parser/infra/neo4j/tree_reader.py | 66 ++++ document-parser/infra/neo4j/tree_writer.py | 337 ++++++++++++++++++ document-parser/main.py | 8 +- document-parser/services/analysis_service.py | 27 ++ .../tests/neo4j/test_tree_writer.py | 175 +++++++++ 6 files changed, 628 insertions(+), 3 deletions(-) create mode 100644 document-parser/infra/neo4j/tree_reader.py create mode 100644 document-parser/infra/neo4j/tree_writer.py create mode 100644 document-parser/tests/neo4j/test_tree_writer.py diff --git a/document-parser/infra/neo4j/__init__.py b/document-parser/infra/neo4j/__init__.py index 23be98d..5655afa 100644 --- a/document-parser/infra/neo4j/__init__.py +++ b/document-parser/infra/neo4j/__init__.py @@ -6,5 +6,21 @@ walkers between DoclingDocument and the graph model. from infra.neo4j.driver import Neo4jDriver, close_driver, get_driver from infra.neo4j.schema import bootstrap_schema +from infra.neo4j.tree_reader import ( + delete_document, + document_exists, + read_document_json, +) +from infra.neo4j.tree_writer import TreeWriteResult, write_document -__all__ = ["Neo4jDriver", "bootstrap_schema", "close_driver", "get_driver"] +__all__ = [ + "Neo4jDriver", + "TreeWriteResult", + "bootstrap_schema", + "close_driver", + "delete_document", + "document_exists", + "get_driver", + "read_document_json", + "write_document", +] diff --git a/document-parser/infra/neo4j/tree_reader.py b/document-parser/infra/neo4j/tree_reader.py new file mode 100644 index 0000000..34fd264 --- /dev/null +++ b/document-parser/infra/neo4j/tree_reader.py @@ -0,0 +1,66 @@ +"""TreeReader — fetch a DoclingDocument back from Neo4j. + +v0.5.0 implementation relies on the verbatim `document_json` property stored +on the Document node by TreeWriter. Reconstruction by walking Element nodes +is deferred to v0.6 (EnrichmentWriter prerequisite), where we may need to +rebuild the DoclingDocument after enrichments have been patched on graph +nodes directly. +""" + +from __future__ import annotations + +import logging + +from infra.neo4j.driver import Neo4jDriver + +logger = logging.getLogger(__name__) + + +async def read_document_json(neo: Neo4jDriver, doc_id: str) -> str | None: + """Return the stored DoclingDocument JSON for `doc_id`, or None if absent.""" + async with neo.driver.session(database=neo.database) as session: + result = await session.run( + "MATCH (d:Document {id: $doc_id}) RETURN d.document_json AS json", + doc_id=doc_id, + ) + record = await result.single() + if record is None: + return None + return record["json"] + + +async def document_exists(neo: Neo4jDriver, doc_id: str) -> bool: + async with neo.driver.session(database=neo.database) as session: + result = await session.run( + "MATCH (d:Document {id: $doc_id}) RETURN count(d) AS n", + doc_id=doc_id, + ) + record = await result.single() + return bool(record and record["n"] > 0) + + +async def delete_document(neo: Neo4jDriver, doc_id: str) -> int: + """Wipe everything related to a doc_id. Returns nodes removed.""" + async with neo.driver.session(database=neo.database) as session: + result = await session.run( + """ + MATCH (d:Document {id: $doc_id}) + OPTIONAL MATCH (d)-[:HAS_ROOT|HAS_CHUNK*0..]->(n) + WITH d, collect(DISTINCT n) AS children + DETACH DELETE d + WITH children + UNWIND children AS c + DETACH DELETE c + RETURN size(children) + 1 AS removed + """, + doc_id=doc_id, + ) + record = await result.single() + # Also clean up orphan elements and pages tagged with this doc_id. + await session.run( + "MATCH (e:Element {doc_id: $doc_id}) DETACH DELETE e", doc_id=doc_id + ) + await session.run( + "MATCH (p:Page {doc_id: $doc_id}) DETACH DELETE p", doc_id=doc_id + ) + return int(record["removed"]) if record else 0 diff --git a/document-parser/infra/neo4j/tree_writer.py b/document-parser/infra/neo4j/tree_writer.py new file mode 100644 index 0000000..9d93482 --- /dev/null +++ b/document-parser/infra/neo4j/tree_writer.py @@ -0,0 +1,337 @@ +"""TreeWriter — persist a DoclingDocument as a graph in Neo4j. + +v0.5.0 strategy: replace-on-write. For a given doc_id, all existing +Document/Element/Page/Chunk nodes are wiped before re-ingestion. The full +serialized `DoclingDocument` JSON is stored as a property on the Document +node so that `TreeReader` can round-trip it verbatim — reconstruction from +graph nodes is deferred to v0.6 (see docs/design/neo4j-integration.md §2). +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any + +from infra.neo4j.driver import Neo4jDriver + +logger = logging.getLogger(__name__) + + +# Docling label → specific Neo4j label. Every node also carries :Element. +_LABEL_MAP: dict[str, str] = { + "section_header": "SectionHeader", + "title": "SectionHeader", + "paragraph": "Paragraph", + "text": "Paragraph", + "list_item": "ListItem", + "list": "ListItem", + "table": "Table", + "picture": "Figure", + "formula": "Formula", + "code": "Code", + "caption": "Caption", + "footnote": "Footnote", + "page_header": "PageHeader", + "page_footer": "PageFooter", +} +_DEFAULT_LABEL = "TextElement" + + +def _element_label(docling_label: str) -> str: + return _LABEL_MAP.get(docling_label.lower(), _DEFAULT_LABEL) + + +@dataclass +class TreeWriteResult: + doc_id: str + elements_written: int + pages_written: int + + +def _iter_items(doc_data: dict[str, Any]): + """Yield every item from texts/tables/pictures/groups with its source list.""" + for key in ("texts", "tables", "pictures", "groups"): + for item in doc_data.get(key, []) or []: + yield key, item + + +def _first_prov(item: dict[str, Any]) -> tuple[int | None, list[float] | None]: + prov = item.get("prov") or [] + if not prov: + return None, None + p0 = prov[0] + bbox = p0.get("bbox") + bbox_list: list[float] | None = None + if isinstance(bbox, dict): + bbox_list = [bbox.get("l", 0.0), bbox.get("t", 0.0), bbox.get("r", 0.0), bbox.get("b", 0.0)] + elif isinstance(bbox, list): + bbox_list = list(bbox) + return p0.get("page_no"), bbox_list + + +def _parent_ref(item: dict[str, Any]) -> str | None: + parent = item.get("parent") + if isinstance(parent, dict): + return parent.get("$ref") or parent.get("cref") + return None + + +def _element_props(item: dict[str, Any], doc_id: str) -> dict[str, Any]: + page, bbox = _first_prov(item) + props: dict[str, Any] = { + "doc_id": doc_id, + "self_ref": item.get("self_ref") or "", + "docling_label": (item.get("label") or "").lower(), + "text": item.get("text") or "", + "prov_page": page, + "prov_bbox": bbox, + } + # Type-specific extras. + if "level" in item: + props["level"] = item.get("level") + if "caption" in item and isinstance(item.get("caption"), str): + props["caption"] = item.get("caption") + if item.get("data") and isinstance(item["data"], dict): + # Tables carry cell layout under data; stringify to keep the schema flat. + try: + props["cells_json"] = json.dumps(item["data"]) + except (TypeError, ValueError): + pass + return props + + +def _dfs_order(doc_data: dict[str, Any]) -> list[str]: + """Return self_refs in reading order (DFS pre-order from body).""" + by_ref: dict[str, dict[str, Any]] = {} + for _, item in _iter_items(doc_data): + ref = item.get("self_ref") + if ref: + by_ref[ref] = item + body = doc_data.get("body") or {} + order: list[str] = [] + + def walk(children: list[dict[str, Any]] | None) -> None: + if not children: + return + for ch in children: + ref = ch.get("$ref") or ch.get("cref") + if not ref: + continue + order.append(ref) + child = by_ref.get(ref) + if child: + walk(child.get("children")) + + walk(body.get("children")) + return order + + +async def write_document( + neo: Neo4jDriver, + *, + doc_id: str, + filename: str, + document_json: str, + tenant_id: str = "default", + source_uri: str | None = None, + docling_version: str | None = None, +) -> TreeWriteResult: + """Persist the full DoclingDocument tree to Neo4j. + + Idempotent: wipes any existing graph for doc_id before writing. + Fails fast (exception propagates) if Neo4j is unavailable — per design §8.5. + """ + doc_data = json.loads(document_json) + ingested_at = datetime.now(tz=timezone.utc).isoformat() + + elements: list[dict[str, Any]] = [] + for _, item in _iter_items(doc_data): + ref = item.get("self_ref") + if not ref: + continue + specific = _element_label(item.get("label") or "") + elements.append( + { + "specific_label": specific, + "parent_ref": _parent_ref(item), + **_element_props(item, doc_id), + } + ) + + pages: list[dict[str, Any]] = [] + for page_no_str, page_obj in (doc_data.get("pages") or {}).items(): + try: + page_no = int(page_no_str) + except (TypeError, ValueError): + continue + size = page_obj.get("size") or {} + pages.append( + { + "doc_id": doc_id, + "page_no": page_no, + "width": size.get("width"), + "height": size.get("height"), + } + ) + + reading_order = _dfs_order(doc_data) + + async with neo.driver.session(database=neo.database) as session: + async with await session.begin_transaction() as tx: + # 1. Wipe existing graph for this doc_id (replace strategy). + await tx.run( + "MATCH (d:Document {id: $doc_id}) " + "OPTIONAL MATCH (d)-[:HAS_ROOT|HAS_CHUNK*0..]->(n) " + "DETACH DELETE d, n", + doc_id=doc_id, + ) + # Also wipe orphan elements/chunks that may still reference this doc. + await tx.run( + "MATCH (e:Element {doc_id: $doc_id}) DETACH DELETE e", doc_id=doc_id + ) + await tx.run( + "MATCH (p:Page {doc_id: $doc_id}) DETACH DELETE p", doc_id=doc_id + ) + + # 2. Document node (carries the verbatim JSON for TreeReader). + await tx.run( + """ + CREATE (d:Document { + id: $doc_id, + title: $title, + source_uri: $source_uri, + ingested_at: datetime($ingested_at), + docling_version: $docling_version, + stages_applied: ['tree'], + last_tree_write: datetime($ingested_at), + tenant_id: $tenant_id, + document_json: $document_json + }) + """, + doc_id=doc_id, + title=filename, + source_uri=source_uri or "", + ingested_at=ingested_at, + docling_version=docling_version or "", + tenant_id=tenant_id, + document_json=document_json, + ) + + # 3. Page nodes. + if pages: + await tx.run( + "UNWIND $pages AS p " + "CREATE (:Page {doc_id: p.doc_id, page_no: p.page_no, " + "width: p.width, height: p.height})", + pages=pages, + ) + + # 4. Element nodes — use dynamic :Element: labels via APOC-free trick. + # We split by specific label so the CREATE statement is static (no APOC). + by_specific: dict[str, list[dict[str, Any]]] = {} + for e in elements: + by_specific.setdefault(e["specific_label"], []).append(e) + for specific, batch in by_specific.items(): + await tx.run( + f""" + UNWIND $batch AS e + CREATE (n:Element:{specific} {{ + doc_id: e.doc_id, + self_ref: e.self_ref, + docling_label: e.docling_label, + text: e.text, + prov_page: e.prov_page, + prov_bbox: e.prov_bbox, + level: e.level, + caption: e.caption, + cells_json: e.cells_json + }}) + """, + batch=batch, + ) + + # 5. PARENT_OF relations (tree structure). Order tracked inline. + parent_rows = [ + { + "doc_id": doc_id, + "parent_ref": e["parent_ref"], + "child_ref": e["self_ref"], + "order": idx, + } + for idx, e in enumerate(elements) + if e["parent_ref"] and e["parent_ref"] != "#/body" + ] + if parent_rows: + await tx.run( + """ + UNWIND $rows AS r + MATCH (p:Element {doc_id: r.doc_id, self_ref: r.parent_ref}) + MATCH (c:Element {doc_id: r.doc_id, self_ref: r.child_ref}) + MERGE (p)-[rel:PARENT_OF]->(c) + SET rel.order = r.order + """, + rows=parent_rows, + ) + + # 6. HAS_ROOT for top-level children of the document body. + root_rows = [ + {"doc_id": doc_id, "child_ref": e["self_ref"]} + for e in elements + if e["parent_ref"] == "#/body" + ] + if root_rows: + await tx.run( + """ + UNWIND $rows AS r + MATCH (d:Document {id: r.doc_id}) + MATCH (c:Element {doc_id: r.doc_id, self_ref: r.child_ref}) + MERGE (d)-[:HAS_ROOT]->(c) + """, + rows=root_rows, + ) + + # 7. ON_PAGE from first provenance. + on_page_rows = [ + {"doc_id": doc_id, "self_ref": e["self_ref"], "page_no": e["prov_page"]} + for e in elements + if e["prov_page"] is not None + ] + if on_page_rows: + await tx.run( + """ + UNWIND $rows AS r + MATCH (e:Element {doc_id: r.doc_id, self_ref: r.self_ref}) + MATCH (p:Page {doc_id: r.doc_id, page_no: r.page_no}) + MERGE (e)-[:ON_PAGE]->(p) + """, + rows=on_page_rows, + ) + + # 8. NEXT chain in DFS pre-order. + if len(reading_order) > 1: + pairs = [ + {"doc_id": doc_id, "a": reading_order[i], "b": reading_order[i + 1]} + for i in range(len(reading_order) - 1) + ] + await tx.run( + """ + UNWIND $pairs AS p + MATCH (a:Element {doc_id: p.doc_id, self_ref: p.a}) + MATCH (b:Element {doc_id: p.doc_id, self_ref: p.b}) + MERGE (a)-[:NEXT]->(b) + """, + pairs=pairs, + ) + + await tx.commit() + + logger.info( + "Neo4j: wrote doc %s (%d elements, %d pages)", + doc_id, + len(elements), + len(pages), + ) + return TreeWriteResult(doc_id=doc_id, elements_written=len(elements), pages_written=len(pages)) diff --git a/document-parser/main.py b/document-parser/main.py index 1c41921..911d0f4 100644 --- a/document-parser/main.py +++ b/document-parser/main.py @@ -75,6 +75,7 @@ def _build_repos() -> tuple[SqliteDocumentRepository, SqliteAnalysisRepository]: def _build_analysis_service( document_repo: SqliteDocumentRepository, analysis_repo: SqliteAnalysisRepository, + neo4j_driver=None, ) -> AnalysisService: converter = _build_converter() chunker = _build_chunker() @@ -90,6 +91,7 @@ def _build_analysis_service( conversion_timeout=settings.conversion_timeout, max_concurrent=settings.max_concurrent_analyses, config=config, + neo4j_driver=neo4j_driver, ) @@ -165,14 +167,16 @@ def _build_document_service( async def lifespan(app: FastAPI) -> AsyncIterator[None]: await init_db() document_repo, analysis_repo = _build_repos() - app.state.analysis_service = _build_analysis_service(document_repo, analysis_repo) + app.state.neo4j = await _init_neo4j() + app.state.analysis_service = _build_analysis_service( + document_repo, analysis_repo, neo4j_driver=app.state.neo4j + ) app.state.document_service = _build_document_service(document_repo, analysis_repo) ingestion_service = _build_ingestion_service() app.state.ingestion_service = ingestion_service if ingestion_service is not None: app.include_router(ingestion_router) logger.info("Ingestion router mounted") - app.state.neo4j = await _init_neo4j() logger.info("Docling Studio backend ready (engine=%s)", settings.conversion_engine) try: yield diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 9c2f4f8..301b8a5 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -69,6 +69,7 @@ class AnalysisConfig: default_table_mode: str = "accurate" batch_page_size: int = 0 + neo4j_required: bool = False # if True, ingestion fails when Neo4j write fails class AnalysisService: @@ -83,6 +84,7 @@ class AnalysisService: conversion_timeout: int = 600, max_concurrent: int = _DEFAULT_MAX_CONCURRENT, config: AnalysisConfig | None = None, + neo4j_driver=None, ): self._converter = converter self._chunker = chunker @@ -93,6 +95,7 @@ class AnalysisService: self._running_tasks: dict[str, asyncio.Task] = {} self._background_tasks: set[asyncio.Task] = set() self._config = config or AnalysisConfig() + self._neo4j = neo4j_driver async def create( self, @@ -386,8 +389,32 @@ class AnalysisService: if result.page_count: await self._document_repo.update_page_count(job.document_id, result.page_count) + await self._write_tree_to_neo4j(job, result.document_json) + logger.info("Analysis completed: %s (%d pages)", job_id, result.page_count) + async def _write_tree_to_neo4j(self, job, document_json: str | None) -> None: + """Mirror the DoclingDocument tree into Neo4j if configured. + + Silent no-op when Neo4j isn't wired in. Logs but does not fail the + analysis when the write fails, unless `config.neo4j_required` is set. + """ + if self._neo4j is None or not document_json: + return + try: + from infra.neo4j import write_document + + await write_document( + self._neo4j, + doc_id=job.document_id, + filename=job.document_filename or job.document_id, + document_json=document_json, + ) + except Exception: + logger.exception("Neo4j TreeWriter failed for doc %s", job.document_id) + if self._config.neo4j_required: + raise + async def _run_analysis_inner( self, job_id: str, diff --git a/document-parser/tests/neo4j/test_tree_writer.py b/document-parser/tests/neo4j/test_tree_writer.py new file mode 100644 index 0000000..5c74a46 --- /dev/null +++ b/document-parser/tests/neo4j/test_tree_writer.py @@ -0,0 +1,175 @@ +"""TreeWriter round-trip + structural sanity checks. + +Fixture is a hand-crafted DoclingDocument JSON with: one section containing +two paragraphs and a table, spanning two pages. Tests verify that the graph +mirrors the structure (HAS_ROOT, PARENT_OF, ON_PAGE, NEXT) and that +re-writing the same doc is an idempotent replace. +""" + +from __future__ import annotations + +import json + +from infra.neo4j import read_document_json, write_document +from infra.neo4j.schema import bootstrap_schema + + +FIXTURE = { + "name": "fixture.pdf", + "pages": { + "1": {"page_no": 1, "size": {"width": 595, "height": 842}}, + "2": {"page_no": 2, "size": {"width": 595, "height": 842}}, + }, + "body": { + "self_ref": "#/body", + "children": [ + {"$ref": "#/texts/0"}, + {"$ref": "#/texts/1"}, + {"$ref": "#/texts/2"}, + {"$ref": "#/tables/0"}, + ], + }, + "texts": [ + { + "self_ref": "#/texts/0", + "parent": {"$ref": "#/body"}, + "label": "section_header", + "text": "Introduction", + "level": 1, + "prov": [{"page_no": 1, "bbox": {"l": 10, "t": 10, "r": 100, "b": 30}}], + }, + { + "self_ref": "#/texts/1", + "parent": {"$ref": "#/body"}, + "label": "paragraph", + "text": "First paragraph on page 1.", + "prov": [{"page_no": 1, "bbox": {"l": 10, "t": 40, "r": 500, "b": 80}}], + }, + { + "self_ref": "#/texts/2", + "parent": {"$ref": "#/body"}, + "label": "paragraph", + "text": "Continued on page 2.", + "prov": [{"page_no": 2, "bbox": {"l": 10, "t": 40, "r": 500, "b": 80}}], + }, + ], + "tables": [ + { + "self_ref": "#/tables/0", + "parent": {"$ref": "#/body"}, + "label": "table", + "text": "", + "data": {"num_rows": 2, "num_cols": 2, "grid": [[1, 2], [3, 4]]}, + "prov": [{"page_no": 2, "bbox": {"l": 10, "t": 90, "r": 500, "b": 200}}], + } + ], + "pictures": [], + "groups": [], +} + + +async def _count(session, cypher: str, **params) -> int: + r = await session.run(cypher, **params) + rec = await r.single() + return int(rec["n"]) if rec else 0 + + +async def test_write_creates_expected_structure(neo4j_driver): + await bootstrap_schema(neo4j_driver) + doc_json = json.dumps(FIXTURE) + + result = await write_document( + neo4j_driver, + doc_id="doc-fixture", + filename="fixture.pdf", + document_json=doc_json, + ) + + assert result.elements_written == 4 + assert result.pages_written == 2 + + async with neo4j_driver.driver.session(database=neo4j_driver.database) as s: + assert await _count( + s, + "MATCH (d:Document {id: $id}) RETURN count(d) AS n", + id="doc-fixture", + ) == 1 + assert await _count( + s, + "MATCH (:Document {id: $id})-[:HAS_ROOT]->(e:Element) RETURN count(e) AS n", + id="doc-fixture", + ) == 4 + assert await _count( + s, + "MATCH (e:Element:SectionHeader {doc_id: $id, self_ref: '#/texts/0'}) " + "RETURN count(e) AS n", + id="doc-fixture", + ) == 1 + assert await _count( + s, + "MATCH (e:Element:Table {doc_id: $id}) RETURN count(e) AS n", + id="doc-fixture", + ) == 1 + # Reading-order chain: 3 NEXT edges for 4 elements. + assert await _count( + s, + "MATCH (a:Element {doc_id: $id})-[:NEXT]->(b:Element {doc_id: $id}) " + "RETURN count(*) AS n", + id="doc-fixture", + ) == 3 + # ON_PAGE: one per element with prov. + assert await _count( + s, + "MATCH (:Element {doc_id: $id})-[:ON_PAGE]->(:Page {doc_id: $id}) " + "RETURN count(*) AS n", + id="doc-fixture", + ) == 4 + + +async def test_rewrite_is_idempotent_replace(neo4j_driver): + await bootstrap_schema(neo4j_driver) + doc_json = json.dumps(FIXTURE) + + await write_document( + neo4j_driver, + doc_id="doc-fixture", + filename="fixture.pdf", + document_json=doc_json, + ) + # Second write with the same id must not duplicate anything. + await write_document( + neo4j_driver, + doc_id="doc-fixture", + filename="fixture.pdf", + document_json=doc_json, + ) + + async with neo4j_driver.driver.session(database=neo4j_driver.database) as s: + assert await _count( + s, "MATCH (d:Document {id: $id}) RETURN count(d) AS n", id="doc-fixture" + ) == 1 + assert await _count( + s, + "MATCH (e:Element {doc_id: $id}) RETURN count(e) AS n", + id="doc-fixture", + ) == 4 + + +async def test_reader_returns_verbatim_json(neo4j_driver): + await bootstrap_schema(neo4j_driver) + doc_json = json.dumps(FIXTURE, sort_keys=True) + await write_document( + neo4j_driver, + doc_id="doc-fixture", + filename="fixture.pdf", + document_json=doc_json, + ) + + read_back = await read_document_json(neo4j_driver, "doc-fixture") + assert read_back is not None + assert json.loads(read_back) == json.loads(doc_json) + + +async def test_reader_missing_doc_returns_none(neo4j_driver): + await bootstrap_schema(neo4j_driver) + assert await read_document_json(neo4j_driver, "no-such-doc") is None From c9359f60e1ae9696b530d3b120ca3015b6423475 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Fri, 17 Apr 2026 15:40:08 +0200 Subject: [PATCH 4/8] =?UTF-8?q?feat(neo4j):=20Day=203=20=E2=80=94=20ChunkW?= =?UTF-8?q?riter,=20graph=20API,=20GraphView,=20README?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ChunkWriter mirrors chunks into Neo4j after OpenSearch indexing, creating HAS_CHUNK edges and DERIVED_FROM back-references to the source Elements (via doc_items propagated from the local chunker). Graph API: GET /api/documents/{id}/graph returns a cytoscape-shaped payload with nodes + edges for Document / Element / Page / Chunk. Hard cap at 200 pages returns HTTP 413 per design §8.4. Frontend: new Graph tab in Studio results, rendered with Cytoscape.js + dagre layout (lazy-loaded, ~175 KB gz). Legend, node styling per element label, directional edges styled per edge type. README gains a Neo4j section with the schema, three demo Cypher queries, and env vars. Backend tests skip cleanly when the neo4j python package is not installed locally. Refs #186 --- README.md | 64 ++++ document-parser/api/graph.py | 76 ++++ document-parser/domain/value_objects.py | 9 + document-parser/infra/local_chunker.py | 12 +- document-parser/infra/neo4j/__init__.py | 5 + document-parser/infra/neo4j/chunk_writer.py | 133 +++++++ document-parser/infra/neo4j/queries.py | 213 +++++++++++ document-parser/main.py | 11 +- document-parser/services/analysis_service.py | 1 + document-parser/services/ingestion_service.py | 11 + document-parser/tests/neo4j/conftest.py | 6 +- .../tests/neo4j/test_chunk_writer.py | 113 ++++++ frontend/package-lock.json | 68 +++- frontend/package.json | 4 + frontend/src/features/analysis/graphApi.ts | 40 ++ .../src/features/analysis/ui/GraphView.vue | 341 ++++++++++++++++++ .../src/features/analysis/ui/ResultTabs.vue | 8 + frontend/src/shared/i18n.ts | 8 + 18 files changed, 1114 insertions(+), 9 deletions(-) create mode 100644 document-parser/api/graph.py create mode 100644 document-parser/infra/neo4j/chunk_writer.py create mode 100644 document-parser/infra/neo4j/queries.py create mode 100644 document-parser/tests/neo4j/test_chunk_writer.py create mode 100644 frontend/src/features/analysis/graphApi.ts create mode 100644 frontend/src/features/analysis/ui/GraphView.vue diff --git a/README.md b/README.md index eff9564..936dc10 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ Upload a PDF, configure the extraction pipeline, and visualize the results — t - **Per-page results** — right panel syncs with the current PDF page - **Chunking** — split extracted content into semantic chunks (hierarchical, hybrid, or page-based) with configurable token limits and inline editing - **Ingestion pipeline** — Docling → chunking → embedding → OpenSearch vector indexing (one-click from Studio) +- **Graph storage (Neo4j)** — full DoclingDocument tree (sections, paragraphs, tables, pages, chunks) mirrored as a graph with `PARENT_OF`, `NEXT`, `ON_PAGE`, `HAS_CHUNK`, `DERIVED_FROM` relations, with an in-app graph view powered by Cytoscape.js - **Markdown & HTML export** of extracted content - **Document management** — upload, list, delete, search, filter by indexing status - **Analysis history** — re-visit and open past analyses @@ -244,6 +245,69 @@ When ingestion is enabled, the UI shows: | `EMBEDDING_URL` | — | Embedding service endpoint (empty = ingestion disabled) | | `EMBEDDING_DIMENSION` | `384` | Vector dimension (must match embedding model) | +## Graph storage with Neo4j (opt-in) + +Docling Studio can mirror the full **DoclingDocument tree** into a [Neo4j](https://neo4j.com/) graph: sections, paragraphs, tables, figures, pages, and chunks all become first-class nodes connected by `HAS_ROOT`, `PARENT_OF`, `NEXT`, `ON_PAGE`, `HAS_CHUNK`, and `DERIVED_FROM` edges. This enables queries that are impossible with a flat chunk store — navigating a document's outline, finding all tables under a given section, or tracing a chunk back to its source elements. + +Enable Neo4j with the ingestion profile (it ships alongside OpenSearch): + +```bash +docker compose --profile ingestion \ + -f docker-compose.yml -f docker-compose.ingestion.yml \ + up --build +``` + +The Neo4j Browser is available at (user `neo4j`, password `changeme` by default). + +### Schema at a glance + +```mermaid +graph TD + D[Document] -->|HAS_ROOT| SH[SectionHeader] + D -->|HAS_CHUNK| C[Chunk] + SH -->|PARENT_OF| P[Paragraph] + SH -->|PARENT_OF| T[Table] + P -->|NEXT| T + P -->|ON_PAGE| PG[Page] + T -->|ON_PAGE| PG + C -->|DERIVED_FROM| P + C -->|DERIVED_FROM| T +``` + +### Example Cypher queries + +Find all "Methods" sections across documents (impossible in vector-only stores): + +```cypher +MATCH (d:Document)-[:HAS_ROOT]->(:Element)-[:PARENT_OF*]->(s:SectionHeader) +WHERE toLower(s.text) CONTAINS 'method' +RETURN d.title, s.text, s.level +``` + +Get the parent section and sibling elements of a chunk (context for RAG): + +```cypher +MATCH (c:Chunk {id: $chunk_id})-[:DERIVED_FROM]->(e:Element) +MATCH (e)<-[:PARENT_OF]-(parent:Element)-[:PARENT_OF]->(sibling:Element) +RETURN parent, collect(sibling) AS siblings +``` + +List all tables from documents ingested from an `invoices/` path: + +```cypher +MATCH (d:Document)-[:HAS_ROOT]->(:Element)-[:PARENT_OF*]->(t:Table) +WHERE d.source_uri CONTAINS 'invoices/' +RETURN d.title, t.caption, t.cells_json +``` + +| Variable | Default | Description | +|----------|---------|-------------| +| `NEO4J_URI` | — | Neo4j Bolt endpoint (empty = graph storage disabled) | +| `NEO4J_USER` | `neo4j` | Neo4j username | +| `NEO4J_PASSWORD` | `changeme` | Neo4j password | + +The in-app **Graph** tab (under *Results*) renders the per-document graph with [Cytoscape.js](https://js.cytoscape.org/) (see [ADR-001](docs/architecture/adrs/ADR-001-graph-visualization-library.md) for the library choice). Documents with more than **200 pages** return `HTTP 413` from `GET /api/documents/{id}/graph`; pagination ships in v0.6. + ## CI / Release GitHub Actions pipelines (see [`.github/workflows/`](.github/workflows/)): diff --git a/document-parser/api/graph.py b/document-parser/api/graph.py new file mode 100644 index 0000000..9b7b4a5 --- /dev/null +++ b/document-parser/api/graph.py @@ -0,0 +1,76 @@ +"""Graph API — returns a cytoscape-shaped view of the Neo4j graph for a doc. + +v0.5 contract: +- Returns the **full** graph for the document (see design §8.4) +- Hard cap at 200 pages; beyond that, HTTP 413 with `truncated: true` +- No pagination (ships in v0.6) +""" + +from __future__ import annotations + +import logging + +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel + +from infra.neo4j import fetch_graph + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/documents", tags=["graph"]) + +MAX_PAGES = 200 + + +class GraphNode(BaseModel): + id: str + group: str + label: str | None = None + + model_config = {"extra": "allow"} + + +class GraphEdge(BaseModel): + id: str + source: str + target: str + type: str + order: int | None = None + + +class GraphResponse(BaseModel): + doc_id: str + nodes: list[GraphNode] + edges: list[GraphEdge] + node_count: int + edge_count: int + truncated: bool + page_count: int + + +@router.get("/{doc_id}/graph", response_model=GraphResponse) +async def get_document_graph(doc_id: str, request: Request) -> GraphResponse: + neo = getattr(request.app.state, "neo4j", None) + if neo is None: + raise HTTPException(status_code=503, detail="Neo4j is not configured") + + payload = await fetch_graph(neo, doc_id, max_pages=MAX_PAGES) + if payload is None: + raise HTTPException(status_code=404, detail=f"No graph for document {doc_id}") + if payload.truncated: + raise HTTPException( + status_code=413, + detail=( + f"Graph too large: document has {payload.page_count} pages " + f"(cap {MAX_PAGES}). Pagination ships in v0.6." + ), + ) + + return GraphResponse( + doc_id=payload.doc_id, + nodes=[GraphNode(**n) for n in payload.nodes], + edges=[GraphEdge(**e) for e in payload.edges], + node_count=payload.node_count, + edge_count=payload.edge_count, + truncated=payload.truncated, + page_count=payload.page_count, + ) diff --git a/document-parser/domain/value_objects.py b/document-parser/domain/value_objects.py index dd558fc..0885852 100644 --- a/document-parser/domain/value_objects.py +++ b/document-parser/domain/value_objects.py @@ -75,6 +75,14 @@ class ChunkBbox: bbox: list[float] # [left, top, right, bottom] in TOPLEFT origin +@dataclass(frozen=True) +class ChunkDocItem: + """Source element referenced by a chunk. Enables Neo4j DERIVED_FROM edges.""" + + self_ref: str + label: str + + @dataclass(frozen=True) class ChunkResult: text: str @@ -82,3 +90,4 @@ class ChunkResult: source_page: int | None = None token_count: int = 0 bboxes: list[ChunkBbox] = field(default_factory=list) + doc_items: list[ChunkDocItem] = field(default_factory=list) diff --git a/document-parser/infra/local_chunker.py b/document-parser/infra/local_chunker.py index 670f5d0..fba7cda 100644 --- a/document-parser/infra/local_chunker.py +++ b/document-parser/infra/local_chunker.py @@ -15,7 +15,7 @@ from docling_core.transforms.chunker import HierarchicalChunker from docling_core.transforms.chunker.hybrid_chunker import HybridChunker from docling_core.types.doc.document import DoclingDocument -from domain.value_objects import ChunkBbox, ChunkingOptions, ChunkResult +from domain.value_objects import ChunkBbox, ChunkDocItem, ChunkingOptions, ChunkResult from infra.bbox import EMPTY_BBOX, to_topleft_list logger = logging.getLogger(__name__) @@ -39,9 +39,18 @@ def _chunk_sync(document_json: str, options: ChunkingOptions) -> list[ChunkResul source_page = None token_count = 0 bboxes: list[ChunkBbox] = [] + doc_items: list[ChunkDocItem] = [] if hasattr(chunk, "meta") and chunk.meta and chunk.meta.doc_items: for doc_item in chunk.meta.doc_items: + ref = getattr(doc_item, "self_ref", None) + if ref: + doc_items.append( + ChunkDocItem( + self_ref=ref, + label=str(getattr(doc_item, "label", "") or ""), + ) + ) if not hasattr(doc_item, "prov") or not doc_item.prov: continue for prov in doc_item.prov: @@ -67,6 +76,7 @@ def _chunk_sync(document_json: str, options: ChunkingOptions) -> list[ChunkResul source_page=source_page, token_count=token_count, bboxes=bboxes, + doc_items=doc_items, ) ) diff --git a/document-parser/infra/neo4j/__init__.py b/document-parser/infra/neo4j/__init__.py index 5655afa..d56a837 100644 --- a/document-parser/infra/neo4j/__init__.py +++ b/document-parser/infra/neo4j/__init__.py @@ -4,7 +4,9 @@ Provides a thin driver wrapper, idempotent schema bootstrap, and walkers between DoclingDocument and the graph model. """ +from infra.neo4j.chunk_writer import ChunkWriteResult, write_chunks from infra.neo4j.driver import Neo4jDriver, close_driver, get_driver +from infra.neo4j.queries import fetch_graph from infra.neo4j.schema import bootstrap_schema from infra.neo4j.tree_reader import ( delete_document, @@ -14,13 +16,16 @@ from infra.neo4j.tree_reader import ( from infra.neo4j.tree_writer import TreeWriteResult, write_document __all__ = [ + "ChunkWriteResult", "Neo4jDriver", "TreeWriteResult", "bootstrap_schema", "close_driver", "delete_document", "document_exists", + "fetch_graph", "get_driver", "read_document_json", + "write_chunks", "write_document", ] diff --git a/document-parser/infra/neo4j/chunk_writer.py b/document-parser/infra/neo4j/chunk_writer.py new file mode 100644 index 0000000..0b249a5 --- /dev/null +++ b/document-parser/infra/neo4j/chunk_writer.py @@ -0,0 +1,133 @@ +"""ChunkWriter — push chunk nodes and DERIVED_FROM edges to Neo4j. + +Embeddings stay in OpenSearch. Each :Chunk node carries a chunk_index so the +OpenSearch entry can be retrieved via (doc_id, chunk_index). The +`embedding_ref` property is reserved for a future vector-store id (not used +in v0.5 — OpenSearch indexes by doc_id+chunk_index already). + +When chunks carry `doc_items` provenance (list of `self_ref` strings), we +create `(:Chunk)-[:DERIVED_FROM]->(:Element)` links so that queries can go +from a chunk back to its source elements. Chunks without doc_items get no +back-links but are still persisted. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from typing import Any + +from infra.neo4j.driver import Neo4jDriver + +logger = logging.getLogger(__name__) + + +@dataclass +class ChunkWriteResult: + doc_id: str + chunks_written: int + derived_from_edges: int + + +def _chunk_id(doc_id: str, index: int) -> str: + return f"{doc_id}::chunk::{index}" + + +async def write_chunks( + neo: Neo4jDriver, + *, + doc_id: str, + chunks_json: str, +) -> ChunkWriteResult: + """Persist chunks for `doc_id`. Wipes prior chunks first (idempotent).""" + chunks: list[dict[str, Any]] = json.loads(chunks_json) + active = [c for c in chunks if not c.get("deleted")] + + chunk_rows: list[dict[str, Any]] = [] + derived_rows: list[dict[str, Any]] = [] + for idx, c in enumerate(active): + cid = _chunk_id(doc_id, idx) + chunk_rows.append( + { + "id": cid, + "doc_id": doc_id, + "text": c.get("text") or "", + "chunk_index": idx, + "token_count": c.get("tokenCount") or 0, + "embedding_ref": "", + } + ) + for item in c.get("docItems") or []: + ref = item.get("selfRef") if isinstance(item, dict) else None + if ref: + derived_rows.append({"chunk_id": cid, "doc_id": doc_id, "self_ref": ref}) + + async with neo.driver.session(database=neo.database) as session: + async with await session.begin_transaction() as tx: + # Replace existing chunks. + await tx.run( + """ + MATCH (d:Document {id: $doc_id})-[:HAS_CHUNK]->(c:Chunk) + DETACH DELETE c + """, + doc_id=doc_id, + ) + await tx.run( + "MATCH (c:Chunk {doc_id: $doc_id}) DETACH DELETE c", doc_id=doc_id + ) + + if chunk_rows: + await tx.run( + """ + MATCH (d:Document {id: $doc_id}) + UNWIND $rows AS r + CREATE (c:Chunk { + id: r.id, + doc_id: r.doc_id, + text: r.text, + chunk_index: r.chunk_index, + token_count: r.token_count, + embedding_ref: r.embedding_ref + }) + MERGE (d)-[:HAS_CHUNK]->(c) + """, + doc_id=doc_id, + rows=chunk_rows, + ) + + if derived_rows: + await tx.run( + """ + UNWIND $rows AS r + MATCH (c:Chunk {id: r.chunk_id}) + MATCH (e:Element {doc_id: r.doc_id, self_ref: r.self_ref}) + MERGE (c)-[:DERIVED_FROM]->(e) + """, + rows=derived_rows, + ) + + # Flag the Document with the new stage. + await tx.run( + """ + MATCH (d:Document {id: $doc_id}) + SET d.stages_applied = [s IN coalesce(d.stages_applied, []) WHERE s <> 'chunks'] + + ['chunks'], + d.last_chunk_write = datetime() + """, + doc_id=doc_id, + ) + + await tx.commit() + + logger.info( + "Neo4j: wrote %d chunks (%d DERIVED_FROM) for doc %s", + len(chunk_rows), + len(derived_rows), + doc_id, + ) + return ChunkWriteResult( + doc_id=doc_id, + chunks_written=len(chunk_rows), + derived_from_edges=len(derived_rows), + ) diff --git a/document-parser/infra/neo4j/queries.py b/document-parser/infra/neo4j/queries.py new file mode 100644 index 0000000..6696f28 --- /dev/null +++ b/document-parser/infra/neo4j/queries.py @@ -0,0 +1,213 @@ +"""Reusable Cypher queries — kept out of the API layer for reuse + testing.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from infra.neo4j.driver import Neo4jDriver + + +@dataclass +class GraphPayload: + doc_id: str + nodes: list[dict[str, Any]] + edges: list[dict[str, Any]] + node_count: int + edge_count: int + truncated: bool + page_count: int + + +# Full graph for one doc: Document + Elements + Pages + Chunks and their edges. +# The graph is returned as flat node + edge lists ready for cytoscape. +_FETCH_GRAPH = """ +MATCH (d:Document {id: $doc_id}) +OPTIONAL MATCH (d)-[:HAS_ROOT]->(root:Element) +OPTIONAL MATCH (e:Element {doc_id: $doc_id}) +OPTIONAL MATCH (p:Page {doc_id: $doc_id}) +OPTIONAL MATCH (c:Chunk {doc_id: $doc_id}) +WITH d, + collect(DISTINCT e) AS elements, + collect(DISTINCT p) AS pages, + collect(DISTINCT c) AS chunks +OPTIONAL MATCH (pe:Element {doc_id: $doc_id})-[r_po:PARENT_OF]->(ce:Element {doc_id: $doc_id}) +OPTIONAL MATCH (a:Element {doc_id: $doc_id})-[r_nx:NEXT]->(b:Element {doc_id: $doc_id}) +OPTIONAL MATCH (er:Element {doc_id: $doc_id})-[r_op:ON_PAGE]->(pr:Page {doc_id: $doc_id}) +OPTIONAL MATCH (d)-[r_hr:HAS_ROOT]->(rr:Element {doc_id: $doc_id}) +OPTIONAL MATCH (d)-[r_hc:HAS_CHUNK]->(rc:Chunk {doc_id: $doc_id}) +OPTIONAL MATCH (cc:Chunk {doc_id: $doc_id})-[r_df:DERIVED_FROM]->(ee:Element {doc_id: $doc_id}) +RETURN + d AS document, + elements, pages, chunks, + collect(DISTINCT {from: pe.self_ref, to: ce.self_ref, order: r_po.order, type: 'PARENT_OF'}) AS parent_edges, + collect(DISTINCT {from: a.self_ref, to: b.self_ref, type: 'NEXT'}) AS next_edges, + collect(DISTINCT {from: er.self_ref, to: pr.page_no, type: 'ON_PAGE'}) AS on_page_edges, + collect(DISTINCT {from: d.id, to: rr.self_ref, type: 'HAS_ROOT'}) AS has_root_edges, + collect(DISTINCT {from: d.id, to: rc.id, type: 'HAS_CHUNK'}) AS has_chunk_edges, + collect(DISTINCT {from: cc.id, to: ee.self_ref, type: 'DERIVED_FROM'}) AS derived_from_edges +""" + + +def _element_node(doc_id: str, e: dict[str, Any]) -> dict[str, Any]: + # Determine the specific element label: Neo4j returns it via labels(e) on the + # driver side; when we project nodes via RETURN, the driver wraps them as Node + # objects, so we convert below. + return { + "id": f"elem::{e.get('self_ref')}", + "group": "element", + "docling_label": e.get("docling_label"), + "self_ref": e.get("self_ref"), + "text": (e.get("text") or "")[:200], + "prov_page": e.get("prov_page"), + "level": e.get("level"), + "doc_id": doc_id, + } + + +def _page_node(doc_id: str, p: dict[str, Any]) -> dict[str, Any]: + return { + "id": f"page::{p.get('page_no')}", + "group": "page", + "page_no": p.get("page_no"), + "width": p.get("width"), + "height": p.get("height"), + "doc_id": doc_id, + } + + +def _chunk_node(p: dict[str, Any]) -> dict[str, Any]: + return { + "id": f"chunk::{p.get('id')}", + "group": "chunk", + "chunk_index": p.get("chunk_index"), + "text": (p.get("text") or "")[:200], + "token_count": p.get("token_count"), + } + + +def _edge_id(from_id: str, to_id: str, edge_type: str) -> str: + return f"{edge_type}::{from_id}::{to_id}" + + +async def fetch_graph( + neo: Neo4jDriver, + doc_id: str, + *, + max_pages: int = 200, +) -> GraphPayload | None: + """Return the full graph for a document, or None if the document is unknown. + + Enforces the page cap from design §8.4: beyond `max_pages`, returns a + `truncated=True` payload with empty node/edge lists so the caller can + surface a clean error (HTTP 413) to the UI. + """ + async with neo.driver.session(database=neo.database) as session: + page_count_result = await session.run( + "MATCH (p:Page {doc_id: $doc_id}) RETURN count(p) AS n", + doc_id=doc_id, + ) + pc_record = await page_count_result.single() + if pc_record is None: + return None + page_count = int(pc_record["n"]) + + exists_result = await session.run( + "MATCH (d:Document {id: $doc_id}) RETURN count(d) AS n", + doc_id=doc_id, + ) + exists_record = await exists_result.single() + if not exists_record or exists_record["n"] == 0: + return None + + if page_count > max_pages: + return GraphPayload( + doc_id=doc_id, + nodes=[], + edges=[], + node_count=0, + edge_count=0, + truncated=True, + page_count=page_count, + ) + + result = await session.run(_FETCH_GRAPH, doc_id=doc_id) + record = await result.single() + + nodes: list[dict[str, Any]] = [] + edges: list[dict[str, Any]] = [] + + if record is None: + return None + + # Document node. + doc_node = record["document"] + if doc_node is not None: + nodes.append( + { + "id": f"doc::{doc_id}", + "group": "document", + "doc_id": doc_id, + "title": doc_node.get("title"), + "stages_applied": doc_node.get("stages_applied"), + } + ) + + # Element nodes, keeping the specific label (:SectionHeader, etc.). + for e in record["elements"] or []: + if e is None: + continue + labels = [label for label in e.labels if label != "Element"] + node = _element_node(doc_id, dict(e)) + node["label"] = labels[0] if labels else "TextElement" + nodes.append(node) + + # Pages. + for p in record["pages"] or []: + if p is None: + continue + nodes.append(_page_node(doc_id, dict(p))) + + # Chunks. + for c in record["chunks"] or []: + if c is None: + continue + nodes.append(_chunk_node(dict(c))) + + # Edges — filter out rows whose from/to is null (OPTIONAL MATCH can yield them). + def _push_element_edge(e: dict[str, Any], from_prefix: str, to_prefix: str) -> None: + frm, to = e.get("from"), e.get("to") + if frm is None or to is None: + return + edges.append( + { + "id": _edge_id(f"{from_prefix}{frm}", f"{to_prefix}{to}", e["type"]), + "source": f"{from_prefix}{frm}", + "target": f"{to_prefix}{to}", + "type": e["type"], + "order": e.get("order"), + } + ) + + for e in record["parent_edges"] or []: + _push_element_edge(e, "elem::", "elem::") + for e in record["next_edges"] or []: + _push_element_edge(e, "elem::", "elem::") + for e in record["on_page_edges"] or []: + _push_element_edge(e, "elem::", "page::") + for e in record["has_root_edges"] or []: + _push_element_edge(e, "doc::", "elem::") + for e in record["has_chunk_edges"] or []: + _push_element_edge(e, "doc::", "chunk::") + for e in record["derived_from_edges"] or []: + _push_element_edge(e, "chunk::", "elem::") + + return GraphPayload( + doc_id=doc_id, + nodes=nodes, + edges=edges, + node_count=len(nodes), + edge_count=len(edges), + truncated=False, + page_count=page_count, + ) diff --git a/document-parser/main.py b/document-parser/main.py index 911d0f4..48fd426 100644 --- a/document-parser/main.py +++ b/document-parser/main.py @@ -117,7 +117,7 @@ async def _init_neo4j(): return None -def _build_ingestion_service() -> IngestionService | None: +def _build_ingestion_service(neo4j_driver=None) -> IngestionService | None: """Build the ingestion service — only if embedding + opensearch are configured.""" if not settings.embedding_url or not settings.opensearch_url: logger.info("Ingestion disabled (EMBEDDING_URL or OPENSEARCH_URL not set)") @@ -139,7 +139,7 @@ def _build_ingestion_service() -> IngestionService | None: settings.embedding_url, settings.opensearch_url, ) - return IngestionService(embedding, vector_store, config) + return IngestionService(embedding, vector_store, config, neo4j_driver=neo4j_driver) def _build_document_service( @@ -172,7 +172,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: document_repo, analysis_repo, neo4j_driver=app.state.neo4j ) app.state.document_service = _build_document_service(document_repo, analysis_repo) - ingestion_service = _build_ingestion_service() + ingestion_service = _build_ingestion_service(neo4j_driver=app.state.neo4j) app.state.ingestion_service = ingestion_service if ingestion_service is not None: app.include_router(ingestion_router) @@ -210,6 +210,11 @@ if settings.rate_limit_rpm > 0: app.include_router(documents_router) app.include_router(analyses_router) +# Graph view — mounted regardless; individual requests 503 if Neo4j is absent. +from api.graph import router as graph_router # noqa: E402 + +app.include_router(graph_router) + @app.get("/api/health", response_model=HealthResponse) async def health() -> HealthResponse: diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 301b8a5..811671c 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -44,6 +44,7 @@ def _chunk_to_dict(c: ChunkResult) -> dict: "sourcePage": c.source_page, "tokenCount": c.token_count, "bboxes": [{"page": b.page, "bbox": b.bbox} for b in c.bboxes], + "docItems": [{"selfRef": d.self_ref, "label": d.label} for d in c.doc_items], } diff --git a/document-parser/services/ingestion_service.py b/document-parser/services/ingestion_service.py index cce6b98..55a87a6 100644 --- a/document-parser/services/ingestion_service.py +++ b/document-parser/services/ingestion_service.py @@ -54,10 +54,12 @@ class IngestionService: embedding_service: EmbeddingService, vector_store: VectorStore, config: IngestionConfig | None = None, + neo4j_driver=None, ) -> None: self._embedding = embedding_service self._vector_store = vector_store self._config = config or IngestionConfig() + self._neo4j = neo4j_driver async def ensure_index(self) -> None: """Ensure the vector index exists with the correct mapping.""" @@ -139,6 +141,15 @@ class IngestionService: indexed = await self._vector_store.index_chunks(self._config.index_name, indexed_chunks) logger.info("Indexed %d/%d chunks for doc %s", indexed, len(indexed_chunks), doc_id) + # 5. Mirror chunks in Neo4j if configured (with DERIVED_FROM edges). + if self._neo4j is not None: + try: + from infra.neo4j import write_chunks + + await write_chunks(self._neo4j, doc_id=doc_id, chunks_json=chunks_json) + except Exception: + logger.exception("Neo4j ChunkWriter failed for doc %s", doc_id) + return IngestionResult( doc_id=doc_id, chunks_indexed=indexed, diff --git a/document-parser/tests/neo4j/conftest.py b/document-parser/tests/neo4j/conftest.py index 9c77d44..8b40230 100644 --- a/document-parser/tests/neo4j/conftest.py +++ b/document-parser/tests/neo4j/conftest.py @@ -11,7 +11,11 @@ import os import pytest -from infra.neo4j import close_driver, get_driver +# Skip the entire module cleanly when the neo4j driver package is absent +# (e.g. local dev without the dependency installed). +pytest.importorskip("neo4j") + +from infra.neo4j import close_driver, get_driver # noqa: E402 def _cfg() -> tuple[str, str, str]: diff --git a/document-parser/tests/neo4j/test_chunk_writer.py b/document-parser/tests/neo4j/test_chunk_writer.py new file mode 100644 index 0000000..ff9145c --- /dev/null +++ b/document-parser/tests/neo4j/test_chunk_writer.py @@ -0,0 +1,113 @@ +"""ChunkWriter creates Chunk nodes + DERIVED_FROM links. + +Builds on the tree_writer fixture — writes the tree first so that DERIVED_FROM +has Elements to link against. +""" + +from __future__ import annotations + +import json + +from infra.neo4j import fetch_graph, write_chunks, write_document +from infra.neo4j.schema import bootstrap_schema +from tests.neo4j.test_tree_writer import FIXTURE + + +CHUNKS = [ + { + "text": "Introduction. First paragraph on page 1.", + "sourcePage": 1, + "tokenCount": 8, + "docItems": [ + {"selfRef": "#/texts/0", "label": "section_header"}, + {"selfRef": "#/texts/1", "label": "paragraph"}, + ], + }, + { + "text": "Continued on page 2.", + "sourcePage": 2, + "tokenCount": 4, + "docItems": [{"selfRef": "#/texts/2", "label": "paragraph"}], + "deleted": False, + }, + # soft-deleted chunk: must be ignored + {"text": "gone", "deleted": True, "docItems": []}, +] + + +async def test_write_chunks_and_derived_from(neo4j_driver): + await bootstrap_schema(neo4j_driver) + await write_document( + neo4j_driver, + doc_id="doc-fixture", + filename="fixture.pdf", + document_json=json.dumps(FIXTURE), + ) + + result = await write_chunks( + neo4j_driver, + doc_id="doc-fixture", + chunks_json=json.dumps(CHUNKS), + ) + + assert result.chunks_written == 2 + assert result.derived_from_edges == 3 + + async with neo4j_driver.driver.session(database=neo4j_driver.database) as s: + count = await ( + await s.run( + "MATCH (:Document {id: $id})-[:HAS_CHUNK]->(c:Chunk) RETURN count(c) AS n", + id="doc-fixture", + ) + ).single() + assert count["n"] == 2 + + # First chunk derives from 2 elements, second from 1. + for idx, expected in [(0, 2), (1, 1)]: + cnt = await ( + await s.run( + "MATCH (c:Chunk {id: $cid})-[:DERIVED_FROM]->(e:Element) " + "RETURN count(e) AS n", + cid=f"doc-fixture::chunk::{idx}", + ) + ).single() + assert cnt["n"] == expected + + stages = await ( + await s.run( + "MATCH (d:Document {id: $id}) RETURN d.stages_applied AS s", id="doc-fixture" + ) + ).single() + assert "chunks" in stages["s"] + + +async def test_fetch_graph_returns_full_payload(neo4j_driver): + await bootstrap_schema(neo4j_driver) + await write_document( + neo4j_driver, + doc_id="doc-fixture", + filename="fixture.pdf", + document_json=json.dumps(FIXTURE), + ) + await write_chunks( + neo4j_driver, + doc_id="doc-fixture", + chunks_json=json.dumps(CHUNKS), + ) + + payload = await fetch_graph(neo4j_driver, "doc-fixture") + assert payload is not None + assert payload.truncated is False + assert payload.page_count == 2 + + groups = {n["group"] for n in payload.nodes} + assert groups == {"document", "element", "page", "chunk"} + + edge_types = {e["type"] for e in payload.edges} + # Every edge kind written by TreeWriter and ChunkWriter should be present. + assert {"HAS_ROOT", "PARENT_OF", "NEXT", "ON_PAGE", "HAS_CHUNK", "DERIVED_FROM"} <= edge_types + + +async def test_fetch_graph_missing_doc_returns_none(neo4j_driver): + await bootstrap_schema(neo4j_driver) + assert await fetch_graph(neo4j_driver, "no-such-doc") is None diff --git a/frontend/package-lock.json b/frontend/package-lock.json index b6e3d75..21307ad 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -1,13 +1,15 @@ { "name": "docling-studio", - "version": "0.3.1", + "version": "0.4.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "docling-studio", - "version": "0.3.1", + "version": "0.4.0", "dependencies": { + "cytoscape": "^3.30.0", + "cytoscape-dagre": "^2.5.0", "dompurify": "^3.3.3", "marked": "^17.0.4", "pinia": "^2.3.0", @@ -16,6 +18,8 @@ }, "devDependencies": { "@eslint/js": "^9.0.0", + "@types/cytoscape": "^3.21.4", + "@types/cytoscape-dagre": "^2.3.3", "@types/dompurify": "^3.2.0", "@vitejs/plugin-vue": "^6.0.5", "@vitest/mocker": "^4.1.2", @@ -1021,6 +1025,23 @@ "assertion-error": "^2.0.1" } }, + "node_modules/@types/cytoscape": { + "version": "3.21.9", + "resolved": "https://registry.npmjs.org/@types/cytoscape/-/cytoscape-3.21.9.tgz", + "integrity": "sha512-JyrG4tllI6jvuISPjHK9j2Xv/LTbnLekLke5otGStjFluIyA9JjgnvgZrSBsp8cEDpiTjwgZUZwpPv8TSBcoLw==", + "dev": true, + "license": "MIT" + }, + "node_modules/@types/cytoscape-dagre": { + "version": "2.3.4", + "resolved": "https://registry.npmjs.org/@types/cytoscape-dagre/-/cytoscape-dagre-2.3.4.tgz", + "integrity": "sha512-uOGXuPfPLFoKZaegjHl9oj4tqONNJuhUl180FiJgRZ35rVijBs6J4UP1Ah6mA6S46h+7pv4ICqpgfdC3EADZlw==", + "dev": true, + "license": "MIT", + "dependencies": { + "cytoscape": "^3.31" + } + }, "node_modules/@types/deep-eql": { "version": "4.0.2", "resolved": "https://registry.npmjs.org/@types/deep-eql/-/deep-eql-4.0.2.tgz", @@ -1824,6 +1845,37 @@ "resolved": "https://registry.npmjs.org/csstype/-/csstype-3.2.3.tgz", "integrity": "sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ==" }, + "node_modules/cytoscape": { + "version": "3.33.2", + "resolved": "https://registry.npmjs.org/cytoscape/-/cytoscape-3.33.2.tgz", + "integrity": "sha512-sj4HXd3DokGhzZAdjDejGvTPLqlt84vNFN8m7bGsOzDY5DyVcxIb2ejIXat2Iy7HxWhdT/N1oKyheJ5YdpsGuw==", + "license": "MIT", + "engines": { + "node": ">=0.10" + } + }, + "node_modules/cytoscape-dagre": { + "version": "2.5.0", + "resolved": "https://registry.npmjs.org/cytoscape-dagre/-/cytoscape-dagre-2.5.0.tgz", + "integrity": "sha512-VG2Knemmshop4kh5fpLO27rYcyUaaDkRw+6PiX4bstpB+QFt0p2oauMrsjVbUamGWQ6YNavh7x2em2uZlzV44g==", + "license": "MIT", + "dependencies": { + "dagre": "^0.8.5" + }, + "peerDependencies": { + "cytoscape": "^3.2.22" + } + }, + "node_modules/dagre": { + "version": "0.8.5", + "resolved": "https://registry.npmjs.org/dagre/-/dagre-0.8.5.tgz", + "integrity": "sha512-/aTqmnRta7x7MCCpExk7HQL2O4owCT2h8NT//9I1OQ9vt29Pa0BzSAkR5lwFUcQ7491yVi/3CXU9jQ5o0Mn2Sw==", + "license": "MIT", + "dependencies": { + "graphlib": "^2.1.8", + "lodash": "^4.17.15" + } + }, "node_modules/de-indent": { "version": "1.0.2", "resolved": "https://registry.npmjs.org/de-indent/-/de-indent-1.0.2.tgz", @@ -2252,6 +2304,15 @@ "url": "https://github.com/sponsors/sindresorhus" } }, + "node_modules/graphlib": { + "version": "2.1.8", + "resolved": "https://registry.npmjs.org/graphlib/-/graphlib-2.1.8.tgz", + "integrity": "sha512-jcLLfkpoVGmH7/InMC/1hIvOPSUh38oJtGhvrOFGzioE1DZ+0YW16RgmOJhHiuWTvGiJQ9Z1Ik43JvkRPRvE+A==", + "license": "MIT", + "dependencies": { + "lodash": "^4.17.15" + } + }, "node_modules/has-flag": { "version": "4.0.0", "resolved": "https://registry.npmjs.org/has-flag/-/has-flag-4.0.0.tgz", @@ -2401,8 +2462,7 @@ "node_modules/lodash": { "version": "4.17.23", "resolved": "https://registry.npmjs.org/lodash/-/lodash-4.17.23.tgz", - "integrity": "sha512-LgVTMpQtIopCi79SJeDiP0TfWi5CNEc/L/aRdTh3yIvmZXTnheWpKjSZhnvMl8iXbC1tFg9gdHHDMLoV7CnG+w==", - "dev": true + "integrity": "sha512-LgVTMpQtIopCi79SJeDiP0TfWi5CNEc/L/aRdTh3yIvmZXTnheWpKjSZhnvMl8iXbC1tFg9gdHHDMLoV7CnG+w==" }, "node_modules/lodash.merge": { "version": "4.6.2", diff --git a/frontend/package.json b/frontend/package.json index 3c7c581..9c10205 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -16,6 +16,8 @@ "format:check": "prettier --check src/" }, "dependencies": { + "cytoscape": "^3.30.0", + "cytoscape-dagre": "^2.5.0", "dompurify": "^3.3.3", "marked": "^17.0.4", "pinia": "^2.3.0", @@ -24,6 +26,8 @@ }, "devDependencies": { "@eslint/js": "^9.0.0", + "@types/cytoscape": "^3.21.4", + "@types/cytoscape-dagre": "^2.3.3", "@vitest/mocker": "^4.1.2", "@types/dompurify": "^3.2.0", "@vitejs/plugin-vue": "^6.0.5", diff --git a/frontend/src/features/analysis/graphApi.ts b/frontend/src/features/analysis/graphApi.ts new file mode 100644 index 0000000..e8e09fb --- /dev/null +++ b/frontend/src/features/analysis/graphApi.ts @@ -0,0 +1,40 @@ +import { apiFetch } from '../../shared/api/http' + +export interface GraphNode { + id: string + group: 'document' | 'element' | 'page' | 'chunk' + label?: string + docling_label?: string + self_ref?: string + text?: string + prov_page?: number | null + level?: number | null + page_no?: number + chunk_index?: number + title?: string + doc_id?: string + token_count?: number + [key: string]: unknown +} + +export interface GraphEdge { + id: string + source: string + target: string + type: 'HAS_ROOT' | 'PARENT_OF' | 'NEXT' | 'ON_PAGE' | 'HAS_CHUNK' | 'DERIVED_FROM' + order?: number | null +} + +export interface GraphPayload { + doc_id: string + nodes: GraphNode[] + edges: GraphEdge[] + node_count: number + edge_count: number + truncated: boolean + page_count: number +} + +export function fetchDocumentGraph(docId: string): Promise { + return apiFetch(`/api/documents/${encodeURIComponent(docId)}/graph`) +} diff --git a/frontend/src/features/analysis/ui/GraphView.vue b/frontend/src/features/analysis/ui/GraphView.vue new file mode 100644 index 0000000..925b564 --- /dev/null +++ b/frontend/src/features/analysis/ui/GraphView.vue @@ -0,0 +1,341 @@ + + + + + diff --git a/frontend/src/features/analysis/ui/ResultTabs.vue b/frontend/src/features/analysis/ui/ResultTabs.vue index afb7897..5c93e86 100644 --- a/frontend/src/features/analysis/ui/ResultTabs.vue +++ b/frontend/src/features/analysis/ui/ResultTabs.vue @@ -106,6 +106,12 @@ + + +
@@ -193,6 +199,7 @@ import { ref, computed, reactive } from 'vue' import { useAnalysisStore } from '../store' import MarkdownViewer from './MarkdownViewer.vue' import ImageGallery from './ImageGallery.vue' +import GraphView from './GraphView.vue' import { useI18n } from '../../../shared/i18n' import type { PageElement } from '../../../shared/types' @@ -223,6 +230,7 @@ const tabs = computed(() => [ { id: 'elements', label: t('results.elements') }, { id: 'markdown', label: t('results.markdown') }, { id: 'images', label: t('results.images') }, + { id: 'graph', label: t('results.graph') }, ]) const totalPages = computed(() => store.currentPages.length) diff --git a/frontend/src/shared/i18n.ts b/frontend/src/shared/i18n.ts index e6ee3fd..b74d0fa 100644 --- a/frontend/src/shared/i18n.ts +++ b/frontend/src/shared/i18n.ts @@ -81,6 +81,10 @@ const messages: Messages = { 'results.elements': 'Éléments', 'results.markdown': 'Markdown', 'results.images': 'Images', + 'results.graph': 'Graphe', + 'results.graphLoading': 'Chargement du graphe…', + 'results.graphEmpty': 'Pas encore de graphe pour ce document (activez Neo4j).', + 'results.retry': 'Réessayer', 'results.pageOf': 'Page {current} sur {total}', 'results.noElements': 'Aucun élément détecté sur cette page', 'results.noImages': 'Aucune image détectée dans ce document', @@ -253,6 +257,10 @@ const messages: Messages = { 'results.elements': 'Elements', 'results.markdown': 'Markdown', 'results.images': 'Images', + 'results.graph': 'Graph', + 'results.graphLoading': 'Loading graph…', + 'results.graphEmpty': 'No graph yet for this document (enable Neo4j).', + 'results.retry': 'Retry', 'results.pageOf': 'Page {current} of {total}', 'results.noElements': 'No elements detected on this page', 'results.noImages': 'No images detected in this document', From 5a2eaacd4d4bc086b3a4af7b9e2e2ee705d13ca4 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Mon, 20 Apr 2026 10:13:01 +0200 Subject: [PATCH 5/8] fix(neo4j): rewrite fetch_graph with CALL subqueries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previous query chained 6 OPTIONAL MATCH clauses for edges with no intervening WITH collect(), producing a cartesian product. At 6 pages (~60 elements, ~300 edges) Neo4j hit 102% CPU and hung > 5min. Rewritten with one CALL {} subquery per node/edge type: each block returns a single row with its collected list — no multiplication across types. 6-page doc now returns in 213ms (was: no return). Python reshape code (queries.py:137-210) untouched — record keys and edge map shape preserved. Refs: https://neo4j.com/developer/kb/using-subqueries-to-control-the-scope-of-aggregations/ --- document-parser/infra/neo4j/queries.py | 64 ++++++++++++++++---------- 1 file changed, 40 insertions(+), 24 deletions(-) diff --git a/document-parser/infra/neo4j/queries.py b/document-parser/infra/neo4j/queries.py index 6696f28..fdcd822 100644 --- a/document-parser/infra/neo4j/queries.py +++ b/document-parser/infra/neo4j/queries.py @@ -20,32 +20,48 @@ class GraphPayload: # Full graph for one doc: Document + Elements + Pages + Chunks and their edges. -# The graph is returned as flat node + edge lists ready for cytoscape. +# Each node/edge type is collected inside its own CALL {} subquery so every +# block contributes a single row — avoids the cartesian product that chained +# OPTIONAL MATCH on 6+ edge types would produce (hangs on multi-page docs). +# See: https://neo4j.com/developer/kb/using-subqueries-to-control-the-scope-of-aggregations/ _FETCH_GRAPH = """ MATCH (d:Document {id: $doc_id}) -OPTIONAL MATCH (d)-[:HAS_ROOT]->(root:Element) -OPTIONAL MATCH (e:Element {doc_id: $doc_id}) -OPTIONAL MATCH (p:Page {doc_id: $doc_id}) -OPTIONAL MATCH (c:Chunk {doc_id: $doc_id}) -WITH d, - collect(DISTINCT e) AS elements, - collect(DISTINCT p) AS pages, - collect(DISTINCT c) AS chunks -OPTIONAL MATCH (pe:Element {doc_id: $doc_id})-[r_po:PARENT_OF]->(ce:Element {doc_id: $doc_id}) -OPTIONAL MATCH (a:Element {doc_id: $doc_id})-[r_nx:NEXT]->(b:Element {doc_id: $doc_id}) -OPTIONAL MATCH (er:Element {doc_id: $doc_id})-[r_op:ON_PAGE]->(pr:Page {doc_id: $doc_id}) -OPTIONAL MATCH (d)-[r_hr:HAS_ROOT]->(rr:Element {doc_id: $doc_id}) -OPTIONAL MATCH (d)-[r_hc:HAS_CHUNK]->(rc:Chunk {doc_id: $doc_id}) -OPTIONAL MATCH (cc:Chunk {doc_id: $doc_id})-[r_df:DERIVED_FROM]->(ee:Element {doc_id: $doc_id}) -RETURN - d AS document, - elements, pages, chunks, - collect(DISTINCT {from: pe.self_ref, to: ce.self_ref, order: r_po.order, type: 'PARENT_OF'}) AS parent_edges, - collect(DISTINCT {from: a.self_ref, to: b.self_ref, type: 'NEXT'}) AS next_edges, - collect(DISTINCT {from: er.self_ref, to: pr.page_no, type: 'ON_PAGE'}) AS on_page_edges, - collect(DISTINCT {from: d.id, to: rr.self_ref, type: 'HAS_ROOT'}) AS has_root_edges, - collect(DISTINCT {from: d.id, to: rc.id, type: 'HAS_CHUNK'}) AS has_chunk_edges, - collect(DISTINCT {from: cc.id, to: ee.self_ref, type: 'DERIVED_FROM'}) AS derived_from_edges +CALL { WITH d MATCH (e:Element {doc_id: d.id}) RETURN collect(e) AS elements } +CALL { WITH d MATCH (p:Page {doc_id: d.id}) RETURN collect(p) AS pages } +CALL { WITH d MATCH (c:Chunk {doc_id: d.id}) RETURN collect(c) AS chunks } +CALL { + WITH d + MATCH (pe:Element {doc_id: d.id})-[r:PARENT_OF]->(ce:Element) + RETURN collect({from: pe.self_ref, to: ce.self_ref, order: r.order, type: 'PARENT_OF'}) AS parent_edges +} +CALL { + WITH d + MATCH (a:Element {doc_id: d.id})-[:NEXT]->(b:Element) + RETURN collect({from: a.self_ref, to: b.self_ref, type: 'NEXT'}) AS next_edges +} +CALL { + WITH d + MATCH (er:Element {doc_id: d.id})-[:ON_PAGE]->(pr:Page) + RETURN collect({from: er.self_ref, to: pr.page_no, type: 'ON_PAGE'}) AS on_page_edges +} +CALL { + WITH d + MATCH (d)-[:HAS_ROOT]->(rr:Element) + RETURN collect({from: d.id, to: rr.self_ref, type: 'HAS_ROOT'}) AS has_root_edges +} +CALL { + WITH d + MATCH (d)-[:HAS_CHUNK]->(rc:Chunk) + RETURN collect({from: d.id, to: rc.id, type: 'HAS_CHUNK'}) AS has_chunk_edges +} +CALL { + WITH d + MATCH (cc:Chunk {doc_id: d.id})-[:DERIVED_FROM]->(ee:Element) + RETURN collect({from: cc.id, to: ee.self_ref, type: 'DERIVED_FROM'}) AS derived_from_edges +} +RETURN d AS document, elements, pages, chunks, + parent_edges, next_edges, on_page_edges, + has_root_edges, has_chunk_edges, derived_from_edges """ From aa60fbb768afe57181acd87f988eca2c0419c614 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Mon, 20 Apr 2026 10:13:08 +0200 Subject: [PATCH 6/8] feat(ui): add Maintain step with graph visualization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move the graph view from a Verify-tab (where it sat post-analysis, off the main flow) to a dedicated Maintain step after Ingest, so the graph result is visible at the natural end of the Configure → Verify → Prepare → Ingest → Maintain pipeline. - StudioPage: new 'maintain' mode toggle + panel rendering GraphView - ResultTabs: remove obsolete graph tab - i18n: add studio.maintain (fr + en) - GraphView: fix init order — flip loading off and await nextTick before renderGraph so the canvas
is mounted when cytoscape reads its container ref (previous code bailed silently on null ref) --- .../src/features/analysis/ui/GraphView.vue | 11 ++++--- .../src/features/analysis/ui/ResultTabs.vue | 8 ----- frontend/src/pages/StudioPage.vue | 32 +++++++++++++++++-- frontend/src/shared/i18n.ts | 2 ++ 4 files changed, 38 insertions(+), 15 deletions(-) diff --git a/frontend/src/features/analysis/ui/GraphView.vue b/frontend/src/features/analysis/ui/GraphView.vue index 925b564..7589fd0 100644 --- a/frontend/src/features/analysis/ui/GraphView.vue +++ b/frontend/src/features/analysis/ui/GraphView.vue @@ -33,7 +33,7 @@