> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/getzep/graphiti/llms.txt
> Use this file to discover all available pages before exploring further.

# 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?

<CardGroup cols={2}>
  <Card title="Performance" icon="gauge-high">
    Process multiple episodes in parallel with shared context
  </Card>

  <Card title="Cost Efficiency" icon="coins">
    Reduce LLM API calls through batched processing
  </Card>

  <Card title="Deduplication" icon="compress">
    Better entity deduplication across episodes
  </Card>

  <Card title="Consistency" icon="check">
    Atomic operations for data integrity
  </Card>
</CardGroup>

## Basic Bulk Ingestion

Use `add_episode_bulk()` to add multiple episodes:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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()`:

```python theme={null}
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:

```python theme={null}
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

<Tabs>
  <Tab title="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
  </Tab>

  <Tab title="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
  </Tab>
</Tabs>

## Example: E-commerce Bulk Ingest

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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

<Steps>
  <Step title="Batch size">
    Process 50-200 episodes per batch for optimal performance
  </Step>

  <Step title="Parallel processing">
    Split large datasets into chunks and process sequentially
  </Step>

  <Step title="Group IDs">
    Use different `group_id` values to partition data
  </Step>

  <Step title="Monitor memory">
    Large batches increase memory usage - adjust based on available RAM
  </Step>
</Steps>

## Error Handling

```python theme={null}
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:

```python theme={null}
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

<CardGroup cols={2}>
  <Card title="Batch Size" icon="layer-group">
    Process 50-200 episodes per batch for optimal performance and memory usage
  </Card>

  <Card title="Error Recovery" icon="life-ring">
    Implement chunking and retry logic for large datasets
  </Card>

  <Card title="Consistent Schemas" icon="shapes">
    Use the same entity types across all episodes in a batch
  </Card>

  <Card title="Monitor Progress" icon="chart-line">
    Log results after each batch to track progress
  </Card>
</CardGroup>

## 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

<Info>
  For temporal data with evolving relationships, use sequential `add_episode()` calls instead of bulk operations.
</Info>

## Next Steps

<CardGroup cols={2}>
  <Card title="Adding Episodes" icon="plus" href="/guides/adding-episodes">
    Learn about sequential episode addition
  </Card>

  <Card title="Custom Entities" icon="shapes" href="/guides/custom-entities">
    Define entity types for bulk ingestion
  </Card>

  <Card title="Searching" icon="magnifying-glass" href="/guides/searching">
    Search your bulk-loaded knowledge graph
  </Card>
</CardGroup>