# Incremental Graph Updates for Corporate Knowledge Graphs: Three Problems Batch Pipelines Can't Solve > Source: > Published: 2026-08-04 02:38:14+00:00 Corporate knowledge graphs break in production the moment you treat them as a batch artifact. I ran into this building [2asy.ai](https://2asy.ai), a Graph-RAG system for East Asian corporate intelligence. The initial architecture was standard: ingest a document set, run NER and entity resolution across the batch, write the graph. Clean, deterministic, fast to prototype. Then the first update cycle hit. The batch pipeline handled entity resolution using blocking keys across the entire document corpus. With incremental updates, you only have the incoming batch. The blocking keys no longer reach the existing graph nodes. In practice this means: a new trade document arrives with "Samsung Elec. Co., Ltd." The existing graph has "Samsung Electronics Co., Ltd." as a canonical node with 847 existing edges. Without live-graph entity resolution, you insert a second node and start building a disconnected duplicate. Here is the resolution-at-boundary pattern that works: ``` python from dataclasses import dataclass from typing import Optional import numpy as np @dataclass class EntityCandidate: raw_name: str normalized_name: str language: str doc_id: str embedding: Optional[np.ndarray] = None def resolve_against_live_graph( candidate: EntityCandidate, graph_client, # neo4j.GraphDatabase or similar embed_fn, threshold: float = 0.85, ) -> Optional[str]: """ Run entity resolution against existing graph nodes before inserting. Returns the existing node ID if a match is found, else None. """ # Step 1: Blocking — retrieve candidate neighbors by name prefix blocking_query = """ MATCH (e:Entity) WHERE e.blocking_key STARTS WITH $prefix RETURN e.id AS id, e.canonical_name AS name, e.name_embedding AS emb LIMIT 50 """ prefix = candidate.normalized_name[:4].upper() neighbors = graph_client.run(blocking_query, prefix=prefix).data() if not neighbors: return None # Step 2: Embedding similarity against blocked candidates if candidate.embedding is None: candidate.embedding = embed_fn(candidate.normalized_name) best_score = 0.0 best_id = None for row in neighbors: if row["emb"] is None: continue existing_emb = np.array(row["emb"]) score = float(np.dot(candidate.embedding, existing_emb) / (np.linalg.norm(candidate.embedding) * np.linalg.norm(existing_emb) + 1e-9)) if score > best_score: best_score = score best_id = row["id"] return best_id if best_score >= threshold else None ``` If `resolve_against_live_graph` returns a node ID, you merge onto that node. If it returns None, you insert as new. The full entity resolution run happens at the boundary, not after insertion. A company relocates its headquarters. The new filing contains the correct address. The old record is already in the graph. Naive append gives you both — which means every Graph-RAG query touching that entity gets conflicting facts. The LLM has to guess which one is current. The fix is to model each property value as a time-bounded assertion rather than a mutable attribute: ``` python from datetime import date from neo4j import GraphDatabase def upsert_fact_with_supersession( driver: GraphDatabase, entity_id: str, property_name: str, new_value: str, valid_from: date, source_doc_id: str, ) -> None: """ Write a new fact, terminating any currently-open assertion for the same property. Old assertions are retained for audit; only valid_to is set. """ with driver.session() as session: session.run(""" // Terminate the previous open assertion for this property MATCH (e:Entity {id: $entity_id})-[:HAS_FACT]->(f:Fact) WHERE f.property_name = $property_name AND f.valid_to IS NULL SET f.valid_to = date($valid_from) - duration('P1D') """, entity_id=entity_id, property_name=property_name, valid_from=str(valid_from)) session.run(""" // Insert the new assertion with open valid_to MATCH (e:Entity {id: $entity_id}) CREATE (e)-[:HAS_FACT]->(f:Fact { property_name: $property_name, value: $new_value, valid_from: date($valid_from), valid_to: null, source_doc_id: $source_doc_id }) """, entity_id=entity_id, property_name=property_name, new_value=new_value, valid_from=str(valid_from), source_doc_id=source_doc_id) ``` Queries then specify a point-in-time context: ``` // Get headquarters address as of 2024-03-01 MATCH (e:Entity {id: $entity_id})-[:HAS_FACT]->(f:Fact) WHERE f.property_name = 'headquarters_address' AND f.valid_from <= date('2024-03-01') AND (f.valid_to IS NULL OR f.valid_to >= date('2024-03-01')) RETURN f.value AS address ``` This is the pattern that makes Graph-RAG `as-of` queries reliable. Without it, you are relying on insertion order, which is neither guaranteed nor queryable. When two entity nodes merge (because `resolve_against_live_graph` found a match), the incoming node's edges need to be transferred to the surviving node. If your update pipeline does not model this explicitly, you end up with a partially disconnected graph where relationships point to deprecated node IDs. ``` python def merge_nodes_and_transfer_edges( driver: GraphDatabase, surviving_id: str, deprecated_id: str, ) -> int: """ Transfer all edges from deprecated node to surviving node, then mark deprecated node as merged. Returns the number of edges transferred. """ with driver.session() as session: # Transfer outgoing edges result = session.run(""" MATCH (deprecated:Entity {id: $deprecated_id})-[r]->(target) WHERE NOT target.id = $surviving_id MATCH (surviving:Entity {id: $surviving_id}) CALL apoc.refactor.from(r, surviving) YIELD input, output RETURN count(*) AS transferred """, deprecated_id=deprecated_id, surviving_id=surviving_id) outgoing = result.single()["transferred"] # Transfer incoming edges result = session.run(""" MATCH (source)-[r]->(deprecated:Entity {id: $deprecated_id}) WHERE NOT source.id = $surviving_id MATCH (surviving:Entity {id: $surviving_id}) CALL apoc.refactor.to(r, surviving) YIELD input, output RETURN count(*) AS transferred """, deprecated_id=deprecated_id, surviving_id=surviving_id) incoming = result.single()["transferred"] # Mark deprecated node session.run(""" MATCH (e:Entity {id: $deprecated_id}) SET e.merged_into = $surviving_id, e.merged_at = datetime() """, deprecated_id=deprecated_id, surviving_id=surviving_id) return outgoing + incoming ``` The `merged_into` property lets you build a forwarding index. Any query that arrives with a deprecated node ID can be redirected to the surviving node without a full graph scan. After a merge, the direct neighbors of the deprecated node may have consistency issues — edges that now point through the merged node incorrectly, or facts that reference the deprecated entity in their source context. The lightweight fix is a dirty-neighbor queue: ``` python def flag_dirty_neighbors( driver: GraphDatabase, node_id: str, reason: str, ) -> list[str]: """ Flag the direct neighbors of a node for consistency recheck in the next ingestion cycle. """ with driver.session() as session: result = session.run(""" MATCH (n:Entity {id: $node_id})-[*1]-(neighbor:Entity) WHERE NOT neighbor.id = $node_id SET neighbor.dirty = true, neighbor.dirty_reason = $reason, neighbor.dirty_at = datetime() RETURN collect(DISTINCT neighbor.id) AS neighbor_ids """, node_id=node_id, reason=reason) return result.single()["neighbor_ids"] ``` At the start of each ingestion cycle, you pull all dirty neighbors and run a consistency check before processing new documents. This handles cascade propagation without a full graph traversal on every update. Batch rebuild works in a prototype. Production knowledge graphs need an explicit update model from the start: Adding these after the fact means rebuilding the schema and reprocessing historical documents. The three problems above are not edge cases — they are the production steady state for any graph that receives continuous updates. *I work on entity resolution and agentic infrastructure at er-api.hannune.ai. The Graph-RAG system this is based on runs against East Asian corporate registries where transliteration variance and entity deduplication are the hard problems.*