Bulk Operations

Efficiently ingest multiple episodes at once with bulk operations for faster knowledge graph construction

Bulk operations allow you to add multiple episodes to your knowledge graph in a single operation, dramatically improving throughput and reducing API costs.

Why Use Bulk Operations?

Performance

Process multiple episodes in parallel with shared context

Cost Efficiency

Reduce LLM API calls through batched processing

Deduplication

Better entity deduplication across episodes

Consistency

Atomic operations for data integrity

Basic Bulk Ingestion

Use add_episode_bulk() to add multiple episodes:

from graphiti_core import Graphiti
from graphiti_core.utils.bulk_utils import RawEpisode
from graphiti_core.nodes import EpisodeType
from datetime import datetime, timezone

# Create list of episodes
episodes = [
    RawEpisode(
        name="Episode 1",
        content="Alice is a software engineer at Google.",
        source=EpisodeType.text,
        source_description="Team bio",
        reference_time=datetime.now(timezone.utc)
    ),
    RawEpisode(
        name="Episode 2",
        content="Bob is a product manager at Microsoft.",
        source=EpisodeType.text,
        source_description="Team bio",
        reference_time=datetime.now(timezone.utc)
    ),
    RawEpisode(
        name="Episode 3",
        content="Alice and Bob worked together at Amazon.",
        source=EpisodeType.text,
        source_description="Team bio",
        reference_time=datetime.now(timezone.utc)
    )
]

# Initialize Graphiti
graphiti = Graphiti(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password"
)

# Add episodes in bulk
result = await graphiti.add_episode_bulk(bulk_episodes=episodes)

print(f"Episodes added: {len(result.episodes)}")
print(f"Entities extracted: {len(result.nodes)}")
print(f"Relationships extracted: {len(result.edges)}")

RawEpisode Structure

The RawEpisode model defines episodes for bulk processing:

from graphiti_core.utils.bulk_utils import RawEpisode
from graphiti_core.nodes import EpisodeType
from datetime import datetime, timezone

episode = RawEpisode(
    name="Unique episode identifier",
    content="Episode content as string",
    source=EpisodeType.text,  # text, json, or message
    source_description="Where this data came from",
    reference_time=datetime.now(timezone.utc),
    uuid=None  # Optional: auto-generated if not provided
)

Parameter	Type	Required	Description
name	str	Yes	Unique episode name
content	str	Yes	Episode content (text or JSON string)
source	EpisodeType	Yes	Type of episode
source_description	str	Yes	Source description
reference_time	datetime	Yes	When this information was valid
uuid	str	No	Custom UUID (auto-generated if omitted)

Processing JSON Data

Convert JSON objects to strings for bulk ingestion:

import json
from graphiti_core.utils.bulk_utils import RawEpisode
from graphiti_core.nodes import EpisodeType
from datetime import datetime, timezone

# Load product data
products = [
    {"name": "Laptop", "price": 1299, "category": "Electronics"},
    {"name": "Phone", "price": 899, "category": "Electronics"},
    {"name": "Headphones", "price": 199, "category": "Accessories"}
]

# Create bulk episodes
episodes = [
    RawEpisode(
        name=f"Product {i}",
        content=json.dumps(product),
        source=EpisodeType.json,
        source_description="Product catalog",
        reference_time=datetime.now(timezone.utc)
    )
    for i, product in enumerate(products)
]

# Ingest
result = await graphiti.add_episode_bulk(bulk_episodes=episodes)

Loading from Files

Read episodes from external files:

import json
from pathlib import Path
from graphiti_core.utils.bulk_utils import RawEpisode
from graphiti_core.nodes import EpisodeType
from datetime import datetime, timezone

# Load from JSON file
with open("data/products.json") as f:
    data = json.load(f)

episodes = [
    RawEpisode(
        name=f"Product {i}",
        content=json.dumps(item),
        source=EpisodeType.json,
        source_description="Product database export",
        reference_time=datetime.now(timezone.utc)
    )
    for i, item in enumerate(data["products"])
]

result = await graphiti.add_episode_bulk(bulk_episodes=episodes)

Bulk Parameters

The add_episode_bulk() method supports the same parameters as add_episode():

from pydantic import BaseModel, Field

# Define custom entities
class Product(BaseModel):
    """A commercial product."""
    category: str | None = Field(description="Product category")
    price: float | None = Field(description="Price in USD")

class Category(BaseModel):
    """A product category."""
    department: str | None = Field(description="Department name")

# Define relationships
class BelongsTo(BaseModel):
    """Product belongs to category."""
    pass

entity_types = {"Product": Product, "Category": Category}
edge_types = {"BELONGS_TO": BelongsTo}
edge_type_map = {("Product", "Category"): ["BELONGS_TO"]}

result = await graphiti.add_episode_bulk(
    bulk_episodes=episodes,
    group_id="product_catalog",
    entity_types=entity_types,
    edge_types=edge_types,
    edge_type_map=edge_type_map
)

Parameter	Type	Description
bulk_episodes	list[RawEpisode]	Episodes to process
group_id	str	Graph partition identifier
entity_types	dict	Custom entity type definitions
excluded_entity_types	list[str]	Entity types to exclude
edge_types	dict	Custom relationship definitions
edge_type_map	dict	Allowed relationships by entity type
custom_extraction_instructions	str	Additional instructions for extraction
saga	str | SagaNode	Saga to associate episodes with

Bulk Results

The add_episode_bulk() method returns an AddBulkEpisodeResults object:

result = await graphiti.add_episode_bulk(bulk_episodes=episodes)

print(f"Episodes: {len(result.episodes)}")
print(f"Entities: {len(result.nodes)}")
print(f"Relationships: {len(result.edges)}")
print(f"Episodic edges: {len(result.episodic_edges)}")

# Access specific results
for episode in result.episodes:
    print(f"Episode: {episode.name} ({episode.uuid})")

for node in result.nodes:
    print(f"Entity: {node.name} - {node.summary}")
    print(f"  Labels: {node.labels}")
    print(f"  Attributes: {node.attributes}")

for edge in result.edges:
    print(f"Relationship: {edge.fact}")
    print(f"  Type: {edge.name}")

Bulk vs Sequential Processing

Bulk Processing

Advantages:

• Faster overall throughput
• Better entity deduplication
• Reduced API calls
• More efficient embeddings

Disadvantages:

• No edge invalidation
• No date extraction
• All-or-nothing operation

Best for:

• Initial data loading
• Batch imports
• Historical data

Sequential Processing

Advantages:

• Edge invalidation
• Temporal reasoning
• Incremental updates
• Better error handling

Disadvantages:

• Slower throughput
• More API calls
• Higher costs

Best for:

• Real-time updates
• Incremental ingestion
• Complex temporal data

Example: E-commerce Bulk Ingest

import json
from graphiti_core import Graphiti
from graphiti_core.utils.bulk_utils import RawEpisode
from graphiti_core.nodes import EpisodeType
from datetime import datetime, timezone
from pathlib import Path

async def ingest_product_catalog():
    # Load product data
    with open("data/products.json") as f:
        products = json.load(f)["products"]
    
    # Create episodes
    episodes = [
        RawEpisode(
            name=f"Product {i}",
            content=json.dumps(product),
            source=EpisodeType.json,
            source_description="Product catalog",
            reference_time=datetime.now(timezone.utc)
        )
        for i, product in enumerate(products)
    ]
    
    # Initialize Graphiti
    graphiti = Graphiti(
        uri="bolt://localhost:7687",
        user="neo4j",
        password="password"
    )
    
    try:
        # Build indices (first time only)
        await graphiti.build_indices_and_constraints()
        
        # Bulk ingest
        result = await graphiti.add_episode_bulk(
            bulk_episodes=episodes,
            group_id="product_catalog"
        )
        
        print(f"✓ Imported {len(result.episodes)} products")
        print(f"✓ Extracted {len(result.nodes)} entities")
        print(f"✓ Created {len(result.edges)} relationships")
        
    finally:
        await graphiti.close()

# Run
import asyncio
asyncio.run(ingest_product_catalog())

Chunking Large Datasets

For very large datasets, process in chunks:

from graphiti_core import Graphiti
from graphiti_core.utils.bulk_utils import RawEpisode

def chunk_list(items, chunk_size):
    """Split list into chunks."""
    for i in range(0, len(items), chunk_size):
        yield items[i:i + chunk_size]

async def ingest_large_dataset(all_episodes: list[RawEpisode]):
    graphiti = Graphiti(
        uri="bolt://localhost:7687",
        user="neo4j",
        password="password"
    )
    
    try:
        # Process in chunks of 100
        for i, chunk in enumerate(chunk_list(all_episodes, 100)):
            print(f"Processing chunk {i+1}...")
            result = await graphiti.add_episode_bulk(bulk_episodes=chunk)
            print(f"  Added {len(result.episodes)} episodes")
    finally:
        await graphiti.close()

Sagas with Bulk Operations

Associate bulk episodes with a saga:

from graphiti_core.utils.bulk_utils import RawEpisode
from graphiti_core.nodes import EpisodeType
from datetime import datetime, timezone

# Create sequential episodes
episodes = [
    RawEpisode(
        name=f"Message {i}",
        content=f"Conversation message {i}",
        source=EpisodeType.message,
        source_description="Chat transcript",
        reference_time=datetime.now(timezone.utc)
    )
    for i in range(10)
]

# Add to saga
result = await graphiti.add_episode_bulk(
    bulk_episodes=episodes,
    saga="Customer Support Chat 123"  # Creates or uses existing saga
)

# Episodes are linked in a chain
print(f"Saga episodes: {len(result.episodes)}")

Sagas connect episodes with:

• HAS_EPISODE edges from saga to each episode
• NEXT_EPISODE edges linking consecutive episodes

Performance Optimization

1. Batch size

   Process 50-200 episodes per batch for optimal performance

2. Parallel processing

   Split large datasets into chunks and process sequentially

3. Group IDs

   Use different group_id values to partition data

4. Monitor memory

   Large batches increase memory usage - adjust based on available RAM

Error Handling

try:
    result = await graphiti.add_episode_bulk(bulk_episodes=episodes)
    print(f"Success: {len(result.episodes)} episodes")
except Exception as e:
    print(f"Bulk ingestion failed: {e}")
    # Consider processing episodes individually to identify failures
    for episode in episodes:
        try:
            await graphiti.add_episode(
                name=episode.name,
                episode_body=episode.content,
                source=episode.source,
                source_description=episode.source_description,
                reference_time=episode.reference_time
            )
        except Exception as ep_error:
            print(f"Failed episode {episode.name}: {ep_error}")

Custom Extraction Instructions

Provide additional context for entity extraction:

result = await graphiti.add_episode_bulk(
    bulk_episodes=episodes,
    custom_extraction_instructions="""
    Focus on extracting product names, prices, and categories.
    Pay special attention to product hierarchies and relationships.
    Extract manufacturer information when available.
    """
)

Best Practices

Batch Size

Process 50-200 episodes per batch for optimal performance and memory usage

Error Recovery

Implement chunking and retry logic for large datasets

Consistent Schemas

Use the same entity types across all episodes in a batch

Monitor Progress

Log results after each batch to track progress

Limitations

Bulk operations do not support:

• Edge invalidation - Old relationships are not automatically invalidated
• Date extraction - Temporal information is not extracted from content
• Community updates - Use build_communities() separately if needed

For temporal data with evolving relationships, use sequential add_episode() calls instead of bulk operations.

Next Steps

Adding Episodes

Learn about sequential episode addition

Custom Entities

Define entity types for bulk ingestion

Searching

Search your bulk-loaded knowledge graph