Corporate knowledge graphs break in production the moment you treat them as a batch artifact.
I ran into this building 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:
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.
"""
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
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:
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.
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:
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"]
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"]
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:
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.