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-core installed
• 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

1. Tune the keyword threshold — start with 4-char minimum, adjust based on your domain
2. Add incremental entity caching — only re-embed entities modified since last run
3. 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