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',