Fix Slow Dedup on Large Knowledge Graphs
What we're building: A drop-in dedup optimization that keeps chunk ingestion under 5 minutes even with 100K+ entities.
Prerequisites:
- Python 3.10+
-
graphiti-coreinstalled - A running knowledge graph with 50K+ entities (or a test graph)
The Problem
Graphiti's dedup step scans all existing entities for each new chunk. At 50K-100K entities, that's millions of similarity comparisons per ingestion. The fix: pre-filter candidates before running expensive embeddings.
Step 1: Profile Your Current Bottleneck
First, confirm where time goes. Add timing around the dedup step:
import time
from graphiti_core import Graphiti
graph = Graphiti(
url="neo4j://localhost:7687",
api_key="your-key",
entity_embedding_model="text-embedding-3-small",
dedup_embedding_model="text-embedding-3-small",
)
start = time.perf_counter()
# Your existing ingestion code
graph.add_episode(episode, group_id="rfc-adr")
print(f"Dedup took: {time.perf_counter() - start:.1f}s")
If this exceeds 60s, you're hitting the scaling wall.
Step 2: Add Semantic Pre-Filtering
The trick: use cheap keyword matching to narrow candidates before embedding comparison.
from graphiti_core.utils.maintenance import dedup_episode
from graphiti_core.llm_client import LLMClient
from graphiti_core.embedder import EntityEmbedder
class FastDedupGraphiti(Graphiti):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._cached_entities = None
self._cache_ts = 0
async def _get_candidates(self, episode, max_candidates=500):
"""Pre-filter entities by keyword overlap."""
# Refresh cache every 5 min
if time.time() - self._cache_ts > 300:
self._cached_entities = await self.get_all_entities()
self._cache_ts = time.time()
# Extract keywords from episode text
keywords = set(
word.lower() for word in episode.text.split()
if len(word) > 4 and word.isalnum()
)[:50]
# Score entities by keyword overlap
scored = []
for entity in self._cached_entities:
entity_words = set(entity.name.lower().split())
overlap = len(keywords & entity_words)
if overlap > 0:
scored.append((overlap, entity))
# Sort by overlap, take top N
scored.sort(reverse=True, key=lambda x: x[0])
return [e for _, e in scored[:max_candidates]]
Step 3: Override the Dedup Method
Now hook into the dedup pipeline:
async def dedup_episode_fast(self, episode, group_id):
"""Faster dedup using pre-filtered candidates."""
candidates = await self._get_candidates(episode)
if not candidates:
return
# Only run expensive embedding comparison on filtered set
await dedup_episode(
episode=episode,
entities=candidates, # ← was: all entities
llm_client=self.llm_client,
embedder=self.entity_embedder,
group_id=group_id,
)
Step 4: Patch the Ingestion Pipeline
import asyncio
from graphiti_core.utils.maintenance import extract_episode_entities
async def ingest_with_fast_dedup(graph, episode, group_id):
# Extract entities first (this stays the same)
entities = await extract_episode_entities(
episode=episode,
llm_client=graph.llm_client,
embedder=graph.entity_embedder,
group_id=group_id,
)
# Fast dedup instead of the default
await graph.dedup_episode_fast(episode, group_id)
# Link entities to episode
await graph.link_episode(episode, entities, group_id)
Step 5: Batch Processing for Large Runs
For the RFC-ADR seed run, process in batches with progress tracking:
async def seed_batch(graph, episodes, group_id, batch_size=10):
for i in range(0, len(episodes), batch_size):
batch = episodes[i:i + batch_size]
# Process batch concurrently
await asyncio.gather(*[
ingest_with_fast_dedup(graph, ep, group_id) for ep in batch
])
# Check timing
elapsed = time.time() - start_time
avg = elapsed / (i + len(batch))
print(f"Batch {i//batch_size + 1}: {avg:.1f}s/episode")
# Graceful stop if we're exceeding limits
if avg > 45:
print("WARNING: Average exceeds 45s, consider increasing batch_size")
Adding Observability
You got the speed fix. Now make failures visible — this is where TracePilot comes in:
pip install tracepilot-sdk
from tracepilot_sdk import TracePilot
tp = TracePilot("tp_live_YOUR_KEY")
# Wrap your ingestion
@tp.trace("graph-ingestion")
async def monitored_ingest(graph, episode, group_id):
with tp.span("dedup-phase"):
await graph.dedup_episode_fast(episode, group_id)
tp.log_metric("candidates", len(graph._cached_entities))
with tp.span("embedding-phase"):
result = await ingest_with_fast_dedup(graph, episode, group_id)
tp.log_metric("entities_linked", len(result))
return result
One line change. Now every ingestion shows you:
- How many candidates survived pre-filtering
- Exact time per phase
- Which episodes exceed your 5-min budget
Next Steps
- Tune the keyword threshold — start with 4-char minimum, adjust based on your domain
- Add incremental entity caching — only re-embed entities modified since last run
- Consider vector index — if you have >200K entities, add a proper vector index (HNSW) instead of brute-force filtering
Expected result: With pre-filtering, you'll see 10-50x fewer embedding comparisons. The RFC-ADR run should drop from 5+ minutes to under 60 seconds per chunk.
You got this. Ship it, measure it, and let the dashboard show you what's next.
Debugging AI agents shouldn't feel like reading The Matrix.
Join other engineers who are building reliable autonomous workflows in our community: TracePilot Discord
Top comments (0)