Write to Neo4j

LLMGraphTransformer hands back GraphDocument objects. langchain-neo4j knows how to write them. The GraphStore wrapper in this project adds two things on top: lazy connection so the API boots even when Neo4j is down, and a clear error when a route that needs the graph is hit before the container is ready.

graph_store.py
python
from langchain_neo4j import Neo4jGraph

class GraphStore:
    def connect(self):
        """Connect lazily. Raises GraphStoreUnavailable if Neo4j is down."""
        if self._graph is not None:
            return self._graph
        try:
            self._graph = Neo4jGraph(
                url=self.uri,
                username=self.user,
                password=self.password,
                database=self.database,
            )
            # Touch the connection so failures surface now, not later.
            self._graph.query("RETURN 1 AS ok")
            return self._graph
        except Exception as e:
            self._graph = None
            raise GraphStoreUnavailable(
                f"Neo4j unreachable at {self.uri}. "
                f"Start it with `make neo4j-up`. Underlying error: {e}"
            ) from e

The try-then-touch pattern is important. If you only construct the Neo4jGraph object, you will not notice auth failures or DNS problems until the first real query. Running RETURN 1 forces every connectivity issue to surface at connect time.

graph_store.py
python
def add_graph_documents(self, graph_documents, include_source=True):
    """Persist LangChain GraphDocuments into Neo4j and report counts."""
    graph = self.connect()
    nodes = sum(len(gd.nodes) for gd in graph_documents)
    rels = sum(len(gd.relationships) for gd in graph_documents)
    graph.add_graph_documents(graph_documents, include_source=include_source)
    try:
        graph.refresh_schema()
    except Exception:
        pass
    return {"nodes_created": nodes, "relationships_created": rels}

include_source=True adds a Document node for each source chunk and links every extracted entity back to it. That is what lets you trace a Cypher answer back to the paragraph it came from, which matters when a user asks "where did you get that".

Because half the endpoints do not need the graph. Health checks, static routes, and the vector-only baseline all work without Neo4j. If the API refuses to boot when Neo4j is down, you lose observability when you need it most. Lazy connection plus a clear 503 on graph-dependent routes is the right tradeoff.

Validation checklist: Ingest your first document

Loading practice…

Checkpoint: Ingestion checkpoint

Loading practice…