The shape of a side-by-side
If the graph and the vector store see different text, every comparison is tainted. The ingest route writes to both in one call, with the graph as the primary and the vector store as best-effort. If the graph write fails, nothing is indexed. If the vector write fails, the graph still succeeds and you get a warning in the response.
@router.post("/ingest", response_model=IngestResponse)
async def ingest(request: IngestRequest):
"""Extract entities + relationships and write them to Neo4j."""
try:
counts = get_ingestion_pipeline().ingest_text(request.text, source=request.source)
except GraphStoreUnavailable as e:
raise _neo4j_unreachable(e)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ingestion failed: {e}")
vector_chunks = 0
vector_error: Optional[str] = None
if request.also_index_vector:
try:
vector_chunks = get_vector_store().index_text(request.text, source=request.source)
except Exception as e:
# Vector indexing is best-effort for the baseline comparison.
# The graph side already succeeded, so we surface a warning
# instead of failing the whole ingest.
vector_error = str(e)
return IngestResponse(
nodes_created=counts.get("nodes_created", 0),
relationships_created=counts.get("relationships_created", 0),
source=request.source,
vector_chunks_indexed=vector_chunks,
vector_warning=vector_error,
)The graph is authoritative. The vector store is a comparison baseline. That ordering is deliberate: a failed vector index should not block a successful graph ingest, but a failed graph ingest should abort the whole request because there is nothing useful to compare against.
Fair concern. The response body always includes vector_chunks_indexed, so you can tell at a glance whether both stores got the text. For production, you would either make this strictly transactional or log failures to an alerting system. For a workshop, best-effort plus a visible warning is the honest middle ground.
Quiz: Quiz
Loading practice…