Compare commits

...

14 commits

Author SHA1 Message Date
Pier-Jean Malandrino
e4c53f1809 test(chunking): add doc_items field to ChunkResult serialization test
The `doc_items` field was added to `ChunkResult` earlier in the
feature branch (used by ChunkWriter for DERIVED_FROM edges), but the
test fixture was never updated. CI caught it now that the PR is open.

Fixes: tests/test_chunking.py::TestChunkResult::test_serializable
2026-04-20 10:26:05 +02:00
Pier-Jean Malandrino
5bc98ee483 chore(lint): fix ruff violations on Neo4j files
Pre-existing ruff violations surfaced by CI on PR #188:

- TC001: move runtime `Neo4jDriver` imports into `TYPE_CHECKING`
  blocks (queries.py, chunk_writer.py, schema.py, tree_reader.py,
  tree_writer.py)
- SIM117: combine nested `async with` in chunk_writer.write_chunks
  and tree_writer.write_document
- SIM105: replace try/except/pass with `contextlib.suppress` in
  tree_writer._element_props
- F401: remove unused `Neo4jDriver` import in main._init_neo4j
- RUF100: remove unused `# noqa: E402` in tests/neo4j/conftest.py
- I001: sort imports in tests/neo4j/test_chunk_writer.py and
  test_tree_writer.py

Zero behaviour change. `ruff check .` now passes cleanly.
2026-04-20 10:22:33 +02:00
Pier-Jean Malandrino
aa60fbb768 feat(ui): add Maintain step with graph visualization
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 <div> is mounted when cytoscape reads its
  container ref (previous code bailed silently on null ref)
2026-04-20 10:13:18 +02:00
Pier-Jean Malandrino
5a2eaacd4d fix(neo4j): rewrite fetch_graph with CALL subqueries
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/
2026-04-20 10:13:18 +02:00
Pier-Jean Malandrino
c9359f60e1 feat(neo4j): Day 3 — ChunkWriter, graph API, GraphView, README
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
2026-04-20 10:13:18 +02:00
Pier-Jean Malandrino
ee92e3c580 feat(neo4j): Day 2 — TreeWriter, TreeReader, pipeline wiring
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
2026-04-20 10:13:18 +02:00
Pier-Jean Malandrino
3474390688 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
2026-04-20 10:13:18 +02:00
Pier-Jean Malandrino
dfbca40730 feat(neo4j): Day 1 — compose service, driver, schema bootstrap
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
2026-04-20 10:13:18 +02:00
Pier-Jean Malandrino
358e575f0f
Merge pull request #185 from scub-france/feature/remote-chunking
feat: enable chunking in remote (Docling Serve) mode
2026-04-19 20:20:14 +02:00
Pier-Jean Malandrino
3bdc4cec50 feat: enable chunking in remote (Docling Serve) mode
Hybrid approach: reuse LocalChunker to chunk the DoclingDocument JSON
returned by Serve, so chunking works identically in both local and
remote modes without calling Serve's chunk endpoint.

Backend:
- _build_chunker() always returns LocalChunker (remove engine guard)
- Use docling-core[chunking] extra for required dependencies
- Skip client-side batching in remote mode (Serve manages its own
  resources, and batching discards document_json needed for chunking)
- Fix Serve form fields: remove generate_page_images (not a Serve
  field), use repeated form keys for to_formats and page_range
- Log Serve error response body on 4xx/5xx for diagnosis
- Fix FastAPI 204 DELETE routes missing response_model=None

Frontend:
- Update chunking feature flag to enable Prepare UI in remote mode

Closes #51
2026-04-16 15:11:14 +02:00
Pier-Jean Malandrino
0bffe6e7d4
Merge pull request #183 from scub-france/feat/centralize-magic-numbers-arch-tests
refactor: centralize magic numbers + arch tests (#168, #177)
2026-04-16 10:51:48 +02:00
Pier-Jean Malandrino
987d43735d fix(ci): install pytestarch in backend tests job (#177)
CI was missing pytestarch dependency, causing test_architecture.py to fail
at collection time. Switch to requirements-test.txt which includes all
test dependencies.
2026-04-16 08:32:07 +00:00
Pier-Jean Malandrino
7a76d2efbd feat: add hexagonal architecture tests with pytestarch (#177)
- Create tests/test_architecture.py with 20 automated rules:
  inter-layer dependency checks (domain, services, api, infra, persistence),
  external dependency constraints (fastapi, sqlalchemy, httpx, opensearchpy),
  and port convention enforcement (Protocol only in domain.ports)
- Add requirements-test.txt with pytestarch dependency
- Fix persistence.database importing infra.settings (read DB_PATH from env directly)
2026-04-16 08:32:07 +00:00
Pier-Jean Malandrino
f2436290c5 refactor: centralize magic numbers for page dimensions, limits, and timeout (#168)
- Move DEFAULT_PAGE_WIDTH/HEIGHT to domain/value_objects.py and import in both converters
- Add opensearch_default_limit to Settings (configurable via OPENSEARCH_DEFAULT_LIMIT env var)
- Pass settings.conversion_timeout to ServeConverter, removing independent _DEFAULT_TIMEOUT
- Update OpenSearchStore to accept default_limit from Settings via constructor
2026-04-16 08:32:07 +00:00
50 changed files with 3090 additions and 59 deletions

View file

@ -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

View file

@ -29,7 +29,7 @@ jobs:
with:
python-version: "3.12"
cache: pip
cache-dependency-path: document-parser/requirements.txt
cache-dependency-path: document-parser/requirements-test.txt
- name: Install system dependencies
run: sudo apt-get update && sudo apt-get install -y --no-install-recommends poppler-utils
@ -37,8 +37,8 @@ jobs:
- name: Install Python dependencies
run: |
pip install --upgrade pip
pip install -r requirements.txt
pip install pytest pytest-asyncio httpx ruff
pip install -r requirements-test.txt
pip install httpx ruff
- name: Lint
run: ruff check .

View file

@ -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 <http://localhost:7474> (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/)):

View file

@ -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:

View file

@ -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:

View file

@ -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)

View file

@ -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<string>, // ["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:
- 34 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 — 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:
- "Graph storage with Neo4j" section
- Schema diagram (Mermaid or image)
- 23 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 ?**
- 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 ?**
- 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.
---
## 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
- [ ] 23 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.*

View file

@ -155,7 +155,7 @@ async def delete_chunk(job_id: str, chunk_index: int, service: ServiceDep) -> li
]
@router.delete("/{job_id}", status_code=204)
@router.delete("/{job_id}", status_code=204, response_model=None)
async def delete_analysis(job_id: str, service: ServiceDep) -> None:
"""Delete an analysis job."""
deleted = await service.delete(job_id)

View file

@ -85,7 +85,7 @@ async def get_document(doc_id: str, service: ServiceDep) -> DocumentResponse:
return _to_response(doc)
@router.delete("/{doc_id}", status_code=204)
@router.delete("/{doc_id}", status_code=204, response_model=None)
async def delete_document(doc_id: str, service: ServiceDep) -> None:
"""Delete a document and its file."""
deleted = await service.delete(doc_id)

View file

@ -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,
)

View file

@ -74,7 +74,7 @@ async def ingest_analysis(
)
@router.delete("/{doc_id}", status_code=204)
@router.delete("/{doc_id}", status_code=204, response_model=None)
async def delete_ingested_document(doc_id: str, ingestion: IngestionDep) -> None:
"""Delete all indexed chunks for a document."""
await ingestion.delete_document(doc_id)

View file

@ -8,6 +8,10 @@ from __future__ import annotations
from dataclasses import dataclass, field
# US Letter page dimensions (points) — fallback when page size is unknown
DEFAULT_PAGE_WIDTH: float = 612.0
DEFAULT_PAGE_HEIGHT: float = 792.0
@dataclass(frozen=True)
class PageElement:
@ -71,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
@ -78,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)

View file

@ -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,
)
)

View file

@ -36,6 +36,8 @@ from docling_core.types.doc import (
)
from domain.value_objects import (
DEFAULT_PAGE_HEIGHT,
DEFAULT_PAGE_WIDTH,
ConversionOptions,
ConversionResult,
PageDetail,
@ -50,10 +52,6 @@ logger = logging.getLogger(__name__)
# Uses a timeout to prevent a frozen conversion from blocking all others.
_converter_lock = threading.Lock()
# US Letter page dimensions (points) — fallback when page size is unknown
_DEFAULT_PAGE_WIDTH = 612.0
_DEFAULT_PAGE_HEIGHT = 792.0
# Default converter (lazy-init on first request)
_default_converter: DoclingConverter | None = None
@ -175,11 +173,11 @@ def _process_content_item(
logger.warning(
"Page %d not found in document metadata — using US Letter fallback (%sx%s pt)",
page_no,
_DEFAULT_PAGE_WIDTH,
_DEFAULT_PAGE_HEIGHT,
DEFAULT_PAGE_WIDTH,
DEFAULT_PAGE_HEIGHT,
)
pages[page_no] = PageDetail(
page_number=page_no, width=_DEFAULT_PAGE_WIDTH, height=_DEFAULT_PAGE_HEIGHT
page_number=page_no, width=DEFAULT_PAGE_WIDTH, height=DEFAULT_PAGE_HEIGHT
)
page_height = pages[page_no].height
@ -248,10 +246,10 @@ def _convert_sync(
pages_detail = [
PageDetail(
page_number=i + 1,
width=doc.pages[i + 1].size.width if (i + 1) in doc.pages else _DEFAULT_PAGE_WIDTH,
width=doc.pages[i + 1].size.width if (i + 1) in doc.pages else DEFAULT_PAGE_WIDTH,
height=doc.pages[i + 1].size.height
if (i + 1) in doc.pages
else _DEFAULT_PAGE_HEIGHT,
else DEFAULT_PAGE_HEIGHT,
)
for i in range(page_count)
]

View file

@ -0,0 +1,31 @@
"""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.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,
document_exists,
read_document_json,
)
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",
]

View file

@ -0,0 +1,134 @@
"""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 TYPE_CHECKING, Any
if TYPE_CHECKING:
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,
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),
)

View file

@ -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")

View file

@ -0,0 +1,230 @@
"""Reusable Cypher queries — kept out of the API layer for reuse + testing."""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
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.
# 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})
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
"""
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,
)

View file

@ -0,0 +1,50 @@
"""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 typing import TYPE_CHECKING
if TYPE_CHECKING:
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),
)

View file

@ -0,0 +1,64 @@
"""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 typing import TYPE_CHECKING
if TYPE_CHECKING:
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

View file

@ -0,0 +1,335 @@
"""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 contextlib
import json
import logging
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
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.
with contextlib.suppress(TypeError, ValueError):
props["cells_json"] = json.dumps(item["data"])
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=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,
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:<specific> 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))

View file

@ -69,13 +69,14 @@ class OpenSearchStore:
verify_certs: Whether to verify TLS certificates.
"""
def __init__(self, url: str, *, verify_certs: bool = False) -> None:
def __init__(self, url: str, *, verify_certs: bool = False, default_limit: int = 1000) -> None:
self._client = AsyncOpenSearch(
hosts=[url],
use_ssl=url.startswith("https"),
verify_certs=verify_certs,
ssl_show_warn=False,
)
self._default_limit = default_limit
# -- lifecycle -------------------------------------------------------------
@ -147,9 +148,11 @@ class OpenSearchStore:
index_name: str,
doc_id: str,
*,
limit: int = 1000,
limit: int | None = None,
) -> list[SearchResult]:
"""Retrieve all indexed chunks for a document, ordered by chunk_index."""
if limit is None:
limit = self._default_limit
resp = await self._client.search(
index=index_name,
body={

View file

@ -21,6 +21,8 @@ import httpx
from docling_core.types.doc.base import BoundingBox, CoordOrigin
from domain.value_objects import (
DEFAULT_PAGE_HEIGHT,
DEFAULT_PAGE_WIDTH,
ConversionOptions,
ConversionResult,
PageDetail,
@ -31,7 +33,6 @@ from infra.bbox import to_topleft_list
logger = logging.getLogger(__name__)
_API_PREFIX = "/v1"
_DEFAULT_TIMEOUT = 600.0
# Docling Serve label → our element type
_LABEL_MAP = {
@ -60,7 +61,7 @@ class ServeConverter:
self,
base_url: str,
api_key: str | None = None,
timeout: float = _DEFAULT_TIMEOUT,
timeout: float = 600.0,
):
self._base_url = base_url.rstrip("/")
self._api_key = api_key
@ -95,6 +96,13 @@ class ServeConverter:
headers=self._headers(),
)
if response.status_code >= 400:
logger.error(
"Docling Serve error %d: %s (form_data=%s)",
response.status_code,
response.text[:500],
{k: v for k, v in form_data.items()},
)
response.raise_for_status()
result_data = response.json()
@ -121,8 +129,12 @@ def _build_form_data(
) -> dict[str, str | list[str]]:
"""Build form fields matching Docling Serve's multipart form contract.
Array fields (to_formats) are sent as lists httpx encodes them as
repeated form keys (to_formats=md&to_formats=html&to_formats=json).
Serve uses FastAPI's ``Form()`` parsing — list/tuple fields are sent
as **repeated form keys** (httpx encodes Python lists this way
automatically: ``to_formats=md&to_formats=html&to_formats=json``).
Note: ``generate_page_images`` is a PdfPipelineOptions field, NOT a
ConvertDocumentsOptions field sending it causes a 422.
"""
data: dict[str, str | list[str]] = {
"to_formats": ["md", "html", "json"],
@ -134,11 +146,12 @@ def _build_form_data(
"do_picture_classification": str(options.do_picture_classification).lower(),
"do_picture_description": str(options.do_picture_description).lower(),
"include_images": str(options.generate_picture_images).lower(),
"generate_page_images": str(options.generate_page_images).lower(),
"images_scale": str(options.images_scale),
}
if page_range is not None:
data["page_range"] = f"{page_range[0]}-{page_range[1]}"
# Serve expects page_range as two repeated form fields:
# page_range=1&page_range=10
data["page_range"] = [str(page_range[0]), str(page_range[1])]
return data
@ -192,8 +205,8 @@ def _extract_pages_from_docling_document(doc: dict) -> list[PageDetail]:
size = page_data.get("size", {})
pages_dict[page_no] = PageDetail(
page_number=page_no,
width=size.get("width", 612.0),
height=size.get("height", 792.0),
width=size.get("width", DEFAULT_PAGE_WIDTH),
height=size.get("height", DEFAULT_PAGE_HEIGHT),
)
# Process all element arrays
@ -220,8 +233,8 @@ def _add_element(item: dict, pages: dict[int, PageDetail]) -> None:
if page_no not in pages:
pages[page_no] = PageDetail(
page_number=page_no,
width=612.0,
height=792.0,
width=DEFAULT_PAGE_WIDTH,
height=DEFAULT_PAGE_HEIGHT,
)
bbox_data = prov.get("bbox", {})

View file

@ -25,6 +25,10 @@ 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"
db_path: str = "./data/docling_studio.db"
@ -54,6 +58,10 @@ class Settings:
errors.append(f"rate_limit_rpm must be >= 0 (got {self.rate_limit_rpm})")
if self.batch_page_size < 0:
errors.append(f"batch_page_size must be >= 0 (got {self.batch_page_size})")
if self.opensearch_default_limit < 1:
errors.append(
f"opensearch_default_limit must be >= 1 (got {self.opensearch_default_limit})"
)
if self.embedding_dimension < 1:
errors.append(f"embedding_dimension must be >= 1 (got {self.embedding_dimension})")
if self.default_table_mode not in ("accurate", "fast"):
@ -97,6 +105,10 @@ 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"),
db_path=os.environ.get("DB_PATH", "./data/docling_studio.db"),

View file

@ -47,6 +47,7 @@ def _build_converter():
return ServeConverter(
base_url=settings.docling_serve_url,
api_key=settings.docling_serve_api_key,
timeout=settings.conversion_timeout,
)
else:
from infra.local_converter import LocalConverter
@ -56,12 +57,15 @@ def _build_converter():
def _build_chunker():
"""Build the chunker adapter — only available in local mode."""
if settings.conversion_engine == "local":
from infra.local_chunker import LocalChunker
"""Build the chunker adapter.
return LocalChunker()
return None
Uses LocalChunker in all modes in remote mode it chunks the
DoclingDocument JSON returned by Docling Serve, so docling-core
(lightweight) is the only local dependency needed.
"""
from infra.local_chunker import LocalChunker
return LocalChunker()
def _build_repos() -> tuple[SqliteDocumentRepository, SqliteAnalysisRepository]:
@ -71,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()
@ -86,10 +91,33 @@ def _build_analysis_service(
conversion_timeout=settings.conversion_timeout,
max_concurrent=settings.max_concurrent_analyses,
config=config,
neo4j_driver=neo4j_driver,
)
def _build_ingestion_service() -> IngestionService | None:
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 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(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)")
@ -99,7 +127,10 @@ def _build_ingestion_service() -> IngestionService | None:
from infra.opensearch_store import OpenSearchStore
embedding = EmbeddingClient(settings.embedding_url)
vector_store = OpenSearchStore(settings.opensearch_url)
vector_store = OpenSearchStore(
settings.opensearch_url,
default_limit=settings.opensearch_default_limit,
)
config = IngestionConfig(
embedding_dimension=settings.embedding_dimension,
)
@ -108,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(
@ -136,15 +167,24 @@ 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()
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)
logger.info("Ingestion router mounted")
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(
@ -170,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:

View file

@ -9,11 +9,9 @@ from contextlib import asynccontextmanager
import aiosqlite
from infra.settings import settings
logger = logging.getLogger(__name__)
DB_PATH = settings.db_path
DB_PATH = os.environ.get("DB_PATH", "./data/docling_studio.db")
_SCHEMA = """
CREATE TABLE IF NOT EXISTS documents (

View file

@ -0,0 +1,4 @@
-r requirements.txt
pytest>=8.0.0,<9.0.0
pytest-asyncio>=0.23.0,<1.0.0
pytestarch>=2.0.0,<3.0.0

View file

@ -1,4 +1,4 @@
docling-core>=2.0.0,<3.0.0
docling-core[chunking]>=2.0.0,<3.0.0
fastapi>=0.115.0,<1.0.0
uvicorn[standard]>=0.32.0,<1.0.0
python-multipart>=0.0.12
@ -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

View file

@ -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],
}
@ -69,6 +70,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 +85,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 +96,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,
@ -324,11 +328,18 @@ class AnalysisService:
file_path: str,
options: ConversionOptions,
) -> ConversionResult | None:
"""Run batched or single conversion. Returns None if the job was deleted mid-batch."""
"""Run batched or single conversion. Returns None if the job was deleted mid-batch.
Batching is only used for local mode it limits memory usage when
Docling runs in-process. In remote mode the Serve instance manages
its own resources, and batching would discard document_json (needed
for chunking).
"""
total_pages = _count_pdf_pages(file_path)
batch_size = self._config.batch_page_size
is_remote = self._is_remote_converter()
if batch_size > 0 and total_pages > batch_size:
if batch_size > 0 and total_pages > batch_size and not is_remote:
return await self._run_batched_conversion(
job_id, file_path, options, total_pages, batch_size
)
@ -337,6 +348,15 @@ class AnalysisService:
timeout=self._conversion_timeout,
)
def _is_remote_converter(self) -> bool:
"""Check if the converter is a remote (Serve) adapter."""
try:
from infra.serve_converter import ServeConverter
return isinstance(self._converter, ServeConverter)
except ImportError:
return False
async def _finalize_analysis(
self,
job_id: str,
@ -370,8 +390,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,

View file

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

View file

View file

@ -0,0 +1,40 @@
"""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
# 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
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()

View file

@ -0,0 +1,111 @@
"""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

View file

@ -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

View file

@ -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

View file

@ -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"

View file

@ -0,0 +1,196 @@
"""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

View file

@ -0,0 +1,208 @@
"""Hexagonal architecture tests — enforce layer dependency rules.
Uses pytestarch for inter-layer dependency rules and ast-based import
scanning for external (third-party) dependency constraints.
Rules enforced:
- domain -> no import from api, services, infra, persistence
- services -> no import from api, infra, persistence
- api -> no import from infra, persistence
- infra -> no import from api, services
- persistence -> no import from api, services, infra
- domain -> no import of fastapi, sqlalchemy, httpx, opensearchpy
- services -> no import of fastapi
"""
from __future__ import annotations
import ast
from pathlib import Path
import pytest
from pytestarch import Rule, get_evaluable_architecture
# ---------------------------------------------------------------------------
# pytestarch evaluable (project root = document-parser/)
# ---------------------------------------------------------------------------
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
# pytestarch uses the directory name as module prefix when given absolute paths.
# We use the directory name to build qualified module references.
_PREFIX = _PROJECT_ROOT.name # "document-parser"
_evaluable = get_evaluable_architecture(str(_PROJECT_ROOT), str(_PROJECT_ROOT))
def _mod(layer: str) -> str:
"""Return the fully-qualified pytestarch module name for a layer."""
return f"{_PREFIX}.{layer}"
# ---------------------------------------------------------------------------
# Helper: collect top-level imports from all .py files in a package
# ---------------------------------------------------------------------------
def _collect_imports(package: str) -> set[str]:
"""Return the set of top-level module names imported by *package*."""
pkg_path = Path(_PROJECT_ROOT) / package
imports: set[str] = set()
for py_file in pkg_path.rglob("*.py"):
tree = ast.parse(py_file.read_text(), filename=str(py_file))
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
imports.add(alias.name.split(".")[0])
elif isinstance(node, ast.ImportFrom) and node.module:
imports.add(node.module.split(".")[0])
return imports
# ---------------------------------------------------------------------------
# Inter-layer dependency rules (pytestarch)
# ---------------------------------------------------------------------------
class TestDomainLayerIsolation:
"""domain must not depend on any other layer."""
@pytest.mark.parametrize("forbidden", ["api", "services", "infra", "persistence"])
def test_domain_does_not_import(self, forbidden: str):
rule = (
Rule()
.modules_that()
.are_sub_modules_of(_mod("domain"))
.should_not()
.import_modules_that()
.are_sub_modules_of(_mod(forbidden))
)
rule.assert_applies(_evaluable)
class TestServicesLayerIsolation:
"""services may import domain only."""
@pytest.mark.parametrize("forbidden", ["api", "infra", "persistence"])
def test_services_does_not_import(self, forbidden: str):
rule = (
Rule()
.modules_that()
.are_sub_modules_of(_mod("services"))
.should_not()
.import_modules_that()
.are_sub_modules_of(_mod(forbidden))
)
rule.assert_applies(_evaluable)
class TestApiLayerIsolation:
"""api may import services and domain, but not infra or persistence."""
@pytest.mark.parametrize("forbidden", ["infra", "persistence"])
def test_api_does_not_import(self, forbidden: str):
rule = (
Rule()
.modules_that()
.are_sub_modules_of(_mod("api"))
.should_not()
.import_modules_that()
.are_sub_modules_of(_mod(forbidden))
)
rule.assert_applies(_evaluable)
class TestInfraLayerIsolation:
"""infra may import domain (ports), but not api or services."""
@pytest.mark.parametrize("forbidden", ["api", "services"])
def test_infra_does_not_import(self, forbidden: str):
rule = (
Rule()
.modules_that()
.are_sub_modules_of(_mod("infra"))
.should_not()
.import_modules_that()
.are_sub_modules_of(_mod(forbidden))
)
rule.assert_applies(_evaluable)
class TestPersistenceLayerIsolation:
"""persistence may import domain, but not api, services, or infra."""
@pytest.mark.parametrize("forbidden", ["api", "services", "infra"])
def test_persistence_does_not_import(self, forbidden: str):
rule = (
Rule()
.modules_that()
.are_sub_modules_of(_mod("persistence"))
.should_not()
.import_modules_that()
.are_sub_modules_of(_mod(forbidden))
)
rule.assert_applies(_evaluable)
# ---------------------------------------------------------------------------
# External dependency rules (ast-based)
# ---------------------------------------------------------------------------
_DOMAIN_FORBIDDEN_EXTERNALS = {"fastapi", "sqlalchemy", "httpx", "opensearchpy"}
_SERVICES_FORBIDDEN_EXTERNALS = {"fastapi"}
class TestDomainExternalDependencies:
"""domain must not import infrastructure-specific third-party libraries."""
@pytest.mark.parametrize("lib", sorted(_DOMAIN_FORBIDDEN_EXTERNALS))
def test_domain_does_not_import_external(self, lib: str):
imports = _collect_imports("domain")
assert lib not in imports, f"domain imports forbidden external library '{lib}'"
class TestServicesExternalDependencies:
"""services must not import web-framework libraries."""
@pytest.mark.parametrize("lib", sorted(_SERVICES_FORBIDDEN_EXTERNALS))
def test_services_does_not_import_external(self, lib: str):
imports = _collect_imports("services")
assert lib not in imports, f"services imports forbidden external library '{lib}'"
# ---------------------------------------------------------------------------
# Convention: ports live exclusively in domain.ports
# ---------------------------------------------------------------------------
class TestPortConvention:
"""Protocol definitions (ports) must live in domain.ports only."""
def test_no_protocol_outside_domain_ports(self):
"""No Protocol subclass should be defined outside domain/ports.py."""
ports_file = Path(_PROJECT_ROOT) / "domain" / "ports.py"
for py_file in Path(_PROJECT_ROOT).rglob("*.py"):
if py_file == ports_file:
continue
# Skip test files and __pycache__
if "tests" in py_file.parts or "__pycache__" in py_file.parts:
continue
tree = ast.parse(py_file.read_text(), filename=str(py_file))
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
for base in node.bases:
base_name = _get_name(base)
if base_name == "Protocol":
pytest.fail(
f"Protocol '{node.name}' defined in {py_file.relative_to(_PROJECT_ROOT)}"
f" — ports must live in domain/ports.py"
)
def _get_name(node: ast.expr) -> str:
"""Extract a simple name from an AST expression node."""
if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
return node.attr
return ""

View file

@ -66,6 +66,7 @@ class TestChunkResult:
"source_page": 1,
"token_count": 10,
"bboxes": [],
"doc_items": [],
}
@ -465,3 +466,68 @@ class TestRechunkEndpoint:
},
)
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# Remote chunking path — hybrid local chunking from Serve's document_json
# ---------------------------------------------------------------------------
class TestRemoteChunkingPath:
"""Verify that chunking works on document_json produced by Serve (remote mode)."""
@pytest.mark.asyncio
async def test_rechunk_with_serve_document_json(self):
"""AnalysisService.rechunk() works with a LocalChunker even in remote mode."""
from infra.local_chunker import LocalChunker
from services.analysis_service import AnalysisService
chunker = LocalChunker()
analysis_repo = AsyncMock()
document_repo = AsyncMock()
converter = AsyncMock() # ServeConverter mock — not used for rechunking
service = AnalysisService(
converter=converter,
analysis_repo=analysis_repo,
document_repo=document_repo,
chunker=chunker,
)
# Simulate a completed job with document_json from Serve
job = AnalysisJob(id="j-remote", document_id="d1")
job.mark_running()
job.mark_completed(
markdown="# Title\nParagraph text here.",
html="<h1>Title</h1><p>Paragraph text here.</p>",
pages_json="[]",
document_json=json.dumps({
"schema_name": "DoclingDocument",
"version": "1.0.0",
"name": "test",
"origin": {
"mimetype": "application/pdf",
"filename": "test.pdf",
"binary_hash": 0,
},
"furniture": {"self_ref": "#/furniture", "children": [], "content_layer": "furniture"},
"body": {"self_ref": "#/body", "children": [], "content_layer": "body"},
"groups": [],
"texts": [],
"pictures": [],
"tables": [],
"key_value_items": [],
"form_items": [],
"pages": {},
}),
)
analysis_repo.find_by_id = AsyncMock(return_value=job)
analysis_repo.update_chunks = AsyncMock(return_value=True)
chunks = await service.rechunk(
"j-remote",
{"chunker_type": "hybrid", "max_tokens": 512},
)
assert isinstance(chunks, list)
analysis_repo.update_chunks.assert_called_once()

View file

@ -39,7 +39,6 @@ class TestBuildFormData:
assert data["do_picture_classification"] == "false"
assert data["do_picture_description"] == "false"
assert data["include_images"] == "false"
assert data["generate_page_images"] == "false"
assert data["images_scale"] == "1.0"
assert set(data["to_formats"]) == {"md", "html", "json"}
@ -56,9 +55,14 @@ class TestBuildFormData:
assert data["images_scale"] == "2.0"
assert data["include_images"] == "true"
def test_page_range_included_when_set(self):
def test_no_generate_page_images_field(self):
"""generate_page_images is a PdfPipelineOptions field, not a Serve field."""
data = _build_form_data(ConversionOptions())
assert "generate_page_images" not in data
def test_page_range_as_repeated_fields(self):
data = _build_form_data(ConversionOptions(), page_range=(11, 20))
assert data["page_range"] == "11-20"
assert data["page_range"] == ["11", "20"]
def test_page_range_absent_when_none(self):
data = _build_form_data(ConversionOptions())
@ -402,11 +406,12 @@ class TestServeConverterConvert:
assert len(result.pages[0].elements) == 1
assert result.pages[0].elements[0].type == "title"
# Verify form fields sent as dict with list for repeated keys
# Verify form fields sent correctly
call_kwargs = mock_client.post.call_args
sent_data = call_kwargs.kwargs.get("data", {})
assert sent_data["do_ocr"] == "true"
assert set(sent_data["to_formats"]) == {"md", "html", "json"}
assert "generate_page_images" not in sent_data
@pytest.mark.asyncio
async def test_http_error_raises(self, tmp_path):
@ -414,6 +419,8 @@ class TestServeConverterConvert:
test_file.write_bytes(b"%PDF-1.4 fake content")
mock_response = MagicMock()
mock_response.status_code = 500
mock_response.text = "Internal Server Error"
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
"Server Error",
request=MagicMock(),
@ -510,3 +517,17 @@ class TestConverterWiring:
converter = _build_converter()
assert isinstance(converter, ServeConverter)
assert converter._api_key == "my-key"
def test_remote_engine_builds_chunker(self):
"""Chunker must be available in remote mode (hybrid local chunking)."""
from infra.local_chunker import LocalChunker
from infra.settings import Settings
with patch(
"main.settings",
Settings(conversion_engine="remote", docling_serve_url="http://serve:5001"),
):
from main import _build_chunker
chunker = _build_chunker()
assert isinstance(chunker, LocalChunker)

View file

@ -19,6 +19,7 @@ class TestSettingsDefaults:
assert s.max_page_count == 0
assert s.max_file_size_mb == 50
assert s.batch_page_size == 0
assert s.opensearch_default_limit == 1000
assert s.upload_dir == "./uploads"
assert s.db_path == "./data/docling_studio.db"
assert "http://localhost:3000" in s.cors_origins
@ -103,6 +104,12 @@ class TestSettingsValidation:
with pytest.raises(ValueError, match="lock_timeout must be > 0"):
Settings(lock_timeout=0)
def test_zero_opensearch_default_limit_rejected(self):
import pytest
with pytest.raises(ValueError, match="opensearch_default_limit must be >= 1"):
Settings(opensearch_default_limit=0)
def test_invalid_table_mode_rejected(self):
import pytest
@ -146,6 +153,7 @@ class TestSettingsFromEnv:
monkeypatch.setenv("MAX_PAGE_COUNT", "20")
monkeypatch.setenv("MAX_FILE_SIZE_MB", "100")
monkeypatch.setenv("BATCH_PAGE_SIZE", "15")
monkeypatch.setenv("OPENSEARCH_DEFAULT_LIMIT", "500")
monkeypatch.setenv("UPLOAD_DIR", "/data/uploads")
monkeypatch.setenv("DB_PATH", "/data/test.db")
monkeypatch.setenv("CORS_ORIGINS", "http://a.com, http://b.com")
@ -163,6 +171,7 @@ class TestSettingsFromEnv:
assert s.max_page_count == 20
assert s.max_file_size_mb == 100
assert s.batch_page_size == 15
assert s.opensearch_default_limit == 500
assert s.upload_dir == "/data/uploads"
assert s.db_path == "/data/test.db"
assert s.cors_origins == ["http://a.com", "http://b.com"]

View file

@ -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",

View file

@ -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",

View file

@ -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<GraphPayload> {
return apiFetch<GraphPayload>(`/api/documents/${encodeURIComponent(docId)}/graph`)
}

View file

@ -0,0 +1,342 @@
<template>
<div class="graph-view" data-e2e="graph-view">
<div v-if="loading" class="graph-placeholder">
<div class="spinner-large" />
<span>{{ t('results.graphLoading') }}</span>
</div>
<div v-else-if="error" class="graph-placeholder error" data-e2e="graph-error">
<span>{{ error }}</span>
<button class="retry-btn" @click="load">{{ t('results.retry') }}</button>
</div>
<div v-else-if="empty" class="graph-placeholder">
<span>{{ t('results.graphEmpty') }}</span>
</div>
<template v-else>
<div class="graph-toolbar">
<span class="graph-stats">
{{ payload?.node_count }} nodes · {{ payload?.edge_count }} edges ·
{{ payload?.page_count }} pages
</span>
<span class="graph-legend">
<span class="legend-chip legend-document">Document</span>
<span class="legend-chip legend-section">Section</span>
<span class="legend-chip legend-paragraph">Paragraph</span>
<span class="legend-chip legend-table">Table</span>
<span class="legend-chip legend-figure">Figure</span>
<span class="legend-chip legend-page">Page</span>
<span class="legend-chip legend-chunk">Chunk</span>
</span>
</div>
<div ref="containerRef" class="graph-canvas" data-e2e="graph-canvas" />
</template>
</div>
</template>
<script setup lang="ts">
import { onMounted, onBeforeUnmount, ref, watch, nextTick } from 'vue'
import { useI18n } from '../../../shared/i18n'
import { fetchDocumentGraph, type GraphPayload } from '../graphApi'
const props = defineProps<{ docId: string | null }>()
const { t } = useI18n()
const containerRef = ref<HTMLDivElement | null>(null)
const payload = ref<GraphPayload | null>(null)
const loading = ref(false)
const error = ref<string | null>(null)
const empty = ref(false)
// eslint-disable-next-line @typescript-eslint/no-explicit-any
let cy: any | null = null
const NODE_COLORS: Record<string, string> = {
document: '#1E293B',
SectionHeader: '#F97316',
Paragraph: '#3B82F6',
TextElement: '#3B82F6',
Table: '#8B5CF6',
Figure: '#22C55E',
ListItem: '#06B6D4',
Formula: '#EC4899',
Code: '#14B8A6',
Caption: '#EAB308',
Page: '#94A3B8',
Chunk: '#DC2626',
}
function nodeColor(n: GraphPayload['nodes'][number]): string {
if (n.group === 'document') return NODE_COLORS.document
if (n.group === 'page') return NODE_COLORS.Page
if (n.group === 'chunk') return NODE_COLORS.Chunk
return NODE_COLORS[n.label || 'TextElement'] || NODE_COLORS.TextElement
}
function nodeLabel(n: GraphPayload['nodes'][number]): string {
if (n.group === 'document') return n.title || n.id
if (n.group === 'page') return `p.${n.page_no}`
if (n.group === 'chunk') return `chunk #${n.chunk_index}`
const txt = (n.text || '').slice(0, 40)
return txt || n.label || n.docling_label || n.self_ref || n.id
}
async function load(): Promise<void> {
if (!props.docId) {
empty.value = true
return
}
loading.value = true
error.value = null
empty.value = false
try {
payload.value = await fetchDocumentGraph(props.docId)
if (!payload.value.nodes.length) {
empty.value = true
return
}
// Flip loading off so the canvas <div> mounts, then wait a tick before init.
loading.value = false
await nextTick()
await renderGraph()
} catch (e) {
error.value = (e as Error).message || 'Failed to load graph'
console.error('Failed to load graph', e)
} finally {
loading.value = false
}
}
async function renderGraph(): Promise<void> {
if (!containerRef.value || !payload.value) return
// Dynamic import keeps cytoscape out of the main chunk.
const [{ default: cytoscape }, { default: dagre }] = await Promise.all([
import('cytoscape'),
import('cytoscape-dagre'),
])
// eslint-disable-next-line @typescript-eslint/no-explicit-any
;(cytoscape as any).use(dagre)
if (cy) {
cy.destroy()
cy = null
}
const elements = [
...payload.value.nodes.map((n) => ({
data: {
id: n.id,
label: nodeLabel(n),
bg: nodeColor(n),
group: n.group,
raw: n,
},
})),
...payload.value.edges.map((e) => ({
data: {
id: e.id,
source: e.source,
target: e.target,
type: e.type,
},
})),
]
cy = cytoscape({
container: containerRef.value,
elements,
style: [
{
selector: 'node',
style: {
'background-color': 'data(bg)',
label: 'data(label)',
color: '#0F172A',
'font-size': 10,
'text-wrap': 'ellipsis',
'text-max-width': '140px',
'text-valign': 'center',
'text-halign': 'center',
width: 28,
height: 28,
'border-width': 1,
'border-color': '#0F172A',
},
},
{
selector: 'node[group = "document"]',
style: { shape: 'round-rectangle', width: 60, height: 36, color: '#F8FAFC' },
},
{
selector: 'node[group = "page"]',
style: { shape: 'round-rectangle', width: 40, height: 24 },
},
{
selector: 'node[group = "chunk"]',
style: { shape: 'diamond', color: '#F8FAFC' },
},
{
selector: 'edge',
style: {
width: 1,
'line-color': '#94A3B8',
'target-arrow-color': '#94A3B8',
'target-arrow-shape': 'triangle',
'curve-style': 'bezier',
'font-size': 8,
color: '#64748B',
},
},
{
selector: 'edge[type = "PARENT_OF"]',
style: { 'line-color': '#1E293B', 'target-arrow-color': '#1E293B', width: 1.5 },
},
{
selector: 'edge[type = "NEXT"]',
style: { 'line-style': 'dashed', 'line-color': '#64748B' },
},
{
selector: 'edge[type = "ON_PAGE"]',
style: { 'line-color': '#CBD5E1', width: 1 },
},
{
selector: 'edge[type = "DERIVED_FROM"]',
style: { 'line-color': '#DC2626', 'target-arrow-color': '#DC2626' },
},
],
// eslint-disable-next-line @typescript-eslint/no-explicit-any
layout: {
name: 'dagre',
rankDir: 'TB',
nodeSep: 30,
edgeSep: 10,
rankSep: 40,
} as any,
wheelSensitivity: 0.15,
})
}
function disposeGraph(): void {
if (cy) {
cy.destroy()
cy = null
}
}
onMounted(load)
onBeforeUnmount(disposeGraph)
watch(
() => props.docId,
() => {
disposeGraph()
load()
},
)
</script>
<style scoped>
.graph-view {
display: flex;
flex-direction: column;
height: 100%;
overflow: hidden;
}
.graph-toolbar {
display: flex;
justify-content: space-between;
align-items: center;
padding: 8px 12px;
border-bottom: 1px solid var(--border);
gap: 12px;
flex-wrap: wrap;
}
.graph-stats {
font-family: 'IBM Plex Mono', monospace;
font-size: 11px;
color: var(--text-muted);
}
.graph-legend {
display: flex;
gap: 6px;
flex-wrap: wrap;
}
.legend-chip {
font-size: 10px;
font-weight: 600;
padding: 2px 8px;
border-radius: 10px;
color: #f8fafc;
}
.legend-document {
background: #1e293b;
}
.legend-section {
background: #f97316;
}
.legend-paragraph {
background: #3b82f6;
}
.legend-table {
background: #8b5cf6;
}
.legend-figure {
background: #22c55e;
}
.legend-page {
background: #94a3b8;
color: #0f172a;
}
.legend-chunk {
background: #dc2626;
}
.graph-canvas {
flex: 1;
min-height: 0;
background: var(--bg);
}
.graph-placeholder {
display: flex;
flex-direction: column;
align-items: center;
justify-content: center;
height: 100%;
gap: 12px;
color: var(--text-muted);
font-size: 14px;
padding: 32px;
text-align: center;
}
.graph-placeholder.error {
color: var(--error);
}
.retry-btn {
background: var(--accent);
color: white;
border: none;
padding: 6px 16px;
border-radius: var(--radius-sm);
cursor: pointer;
font-size: 12px;
}
.spinner-large {
width: 32px;
height: 32px;
border: 3px solid var(--border-light);
border-top-color: var(--accent);
border-radius: 50%;
animation: spin 0.8s linear infinite;
}
@keyframes spin {
to {
transform: rotate(360deg);
}
}
</style>

View file

@ -29,12 +29,12 @@ describe('useFeatureFlagStore', () => {
expect(store.isEnabled('chunking')).toBe(true)
})
it('disables chunking when engine is remote', async () => {
it('enables chunking when engine is remote', async () => {
mockApiFetch.mockResolvedValue({ status: 'ok', engine: 'remote' })
const store = useFeatureFlagStore()
await store.load()
expect(store.engine).toBe('remote')
expect(store.isEnabled('chunking')).toBe(false)
expect(store.isEnabled('chunking')).toBe(true)
})
it('enables disclaimer when deploymentMode is huggingface', async () => {

View file

@ -32,7 +32,7 @@ interface FeatureFlagContext {
const featureRegistry: Record<FeatureFlag, FeatureFlagDef> = {
chunking: {
description: 'Document chunking for RAG preparation',
isEnabled: (ctx) => ctx.engine === 'local',
isEnabled: (ctx) => ctx.engine !== null,
},
disclaimer: {
description: 'Show shared-instance disclaimer banner',

View file

@ -23,6 +23,6 @@ describe('useFeatureFlag', () => {
expect(flag.value).toBe(true)
store.$patch({ engine: 'remote' })
expect(flag.value).toBe(false)
expect(flag.value).toBe(true)
})
})

View file

@ -84,6 +84,27 @@
</svg>
{{ t('studio.ingest') }}
</button>
<button
v-if="chunkingEnabled && ingestionEnabled && ingestionStore.available"
class="toggle-btn"
data-e2e="toggle-btn maintain-btn"
:class="{ active: mode === 'maintain' }"
@click="mode = 'maintain'"
:disabled="!analysisStore.currentAnalysis"
>
<svg class="toggle-icon" viewBox="0 0 20 20" fill="currentColor">
<path
d="M10 3.5a6.5 6.5 0 100 13 6.5 6.5 0 000-13zM6 10a4 4 0 118 0 4 4 0 01-8 0zm4-2a2 2 0 100 4 2 2 0 000-4z"
/>
<path
d="M10 1v2M10 17v2M1 10h2M17 10h2M3.5 3.5l1.4 1.4M15.1 15.1l1.4 1.4M3.5 16.5l1.4-1.4M15.1 4.9l1.4-1.4"
stroke="currentColor"
stroke-width="1.5"
fill="none"
/>
</svg>
{{ t('studio.maintain') }}
</button>
</div>
</div>
<div class="topbar-actions">
@ -475,6 +496,11 @@
:chunk-count="analysisStore.currentChunks?.length ?? 0"
/>
</div>
<!-- MAINTAIN MODE -->
<div v-if="mode === 'maintain'" class="maintain-panel">
<GraphView :doc-id="analysisStore.currentAnalysis?.documentId ?? null" />
</div>
</div>
</div>
</div>
@ -489,6 +515,7 @@ import { useIngestionStore } from '../features/ingestion/store'
import { DocumentUpload, DocumentList } from '../features/document/index'
import { ResultTabs } from '../features/analysis/index'
import BboxOverlay from '../features/analysis/ui/BboxOverlay.vue'
import GraphView from '../features/analysis/ui/GraphView.vue'
import { ChunkPanel } from '../features/chunking'
import { IngestPanel } from '../features/ingestion'
import { useFeatureFlag } from '../features/feature-flags'
@ -1404,10 +1431,11 @@ onBeforeUnmount(() => {
padding-top: 16px;
}
/* Verify / Prepare / Ingest panels */
/* Verify / Prepare / Ingest / Maintain panels */
.verify-panel,
.prepare-panel,
.ingest-panel-wrapper {
.ingest-panel-wrapper,
.maintain-panel {
height: 100%;
overflow: hidden;
display: flex;

View file

@ -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',
@ -110,6 +114,7 @@ const messages: Messages = {
// Chunking
'studio.prepare': 'Préparer',
'studio.ingest': 'Ingérer',
'studio.maintain': 'Maintenir',
'chunking.settings': 'Chunking',
'chunking.chunkerType': 'Type de chunker',
'chunking.maxTokens': 'Tokens max',
@ -253,6 +258,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',
@ -279,6 +288,7 @@ const messages: Messages = {
'studio.prepare': 'Prepare',
'studio.ingest': 'Ingest',
'studio.maintain': 'Maintain',
'chunking.settings': 'Chunking',
'chunking.chunkerType': 'Chunker type',
'chunking.maxTokens': 'Max tokens',