Adding Episodes

Episodes are the primary units of information in Graphiti. Each episode represents a discrete piece of content that Graphiti processes to extract entities and relationships.

Episode Types

Graphiti supports three episode types:

EpisodeType.text - Plain text content like meeting notes or articles
EpisodeType.json - Structured JSON data
EpisodeType.message - Conversation messages with speaker patterns

Adding a Single Episode

Import dependencies

from datetime import datetime, timezone
from graphiti_core import Graphiti
from graphiti_core.nodes import EpisodeType

Initialize Graphiti

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

Add a text episode

await graphiti.add_episode(
    name="Meeting Notes Q4",
    episode_body="Alice reviewed the mobile app redesign progress. Bob presented the engineering timeline.",
    source=EpisodeType.text,
    source_description="Q4 planning meeting",
    reference_time=datetime.now(timezone.utc)
)

The reference_time parameter specifies when this information was valid, enabling temporal queries.

Adding JSON Episodes

For structured data, use EpisodeType.json and serialize your data:

import json

product_data = {
    "name": "Wool Runners",
    "material": "SuperLight Wool",
    "color": "Dark Grey",
    "technology": "SuperLight Foam"
}

await graphiti.add_episode(
    name="Product Catalog Entry",
    episode_body=json.dumps(product_data),
    source=EpisodeType.json,
    source_description="Product database",
    reference_time=datetime.now(timezone.utc)
)

Adding Message Episodes

For conversations, use EpisodeType.message:

conversation = """User: I need help setting up a development environment.
Assistant: I can help with that. Which cloud provider are you using?
User: We're using AWS with PostgreSQL and Redis."""

await graphiti.add_episode(
    name="Support Chat 2024-01-15",
    episode_body=conversation,
    source=EpisodeType.message,
    source_description="Customer support conversation",
    reference_time=datetime.now(timezone.utc)
)

Episode Parameters

Parameter	Type	Required	Description
`name`	`str`	Yes	Unique identifier for the episode
`episode_body`	`str`	Yes	The content (plain text or JSON string)
`source`	`EpisodeType`	Yes	Type of episode (text, json, or message)
`source_description`	`str`	Yes	Description of where this data came from
`reference_time`	`datetime`	Yes	When this information was valid
`group_id`	`str`	No	Graph partition identifier
`uuid`	`str`	No	Custom UUID (auto-generated if not provided)
`entity_types`	`dict`	No	Custom entity type definitions
`edge_types`	`dict`	No	Custom relationship type definitions
`update_communities`	`bool`	No	Whether to update community summaries

Advanced Options

Group IDs for Multi-Tenancy

Use group_id to partition your graph for different users or contexts:

await graphiti.add_episode(
    name="User Session",
    episode_body="User completed checkout process",
    source=EpisodeType.text,
    source_description="Analytics event",
    reference_time=datetime.now(timezone.utc),
    group_id="user_12345"  # Isolate data per user
)

Custom Entity Types

Define custom entity types to extract domain-specific information:

from pydantic import BaseModel, Field

class Person(BaseModel):
    """A human person"""
    first_name: str | None = Field(description="First name")
    last_name: str | None = Field(description="Last name")
    occupation: str | None = Field(description="Work occupation")

await graphiti.add_episode(
    name="Team Meeting",
    episode_body="Sarah Johnson, the lead engineer, presented the roadmap.",
    source=EpisodeType.text,
    source_description="Meeting notes",
    reference_time=datetime.now(timezone.utc),
    entity_types={"Person": Person}
)

See the Custom Entities guide for more details.

Updating Communities

Communities are clusters of related entities. Enable automatic community updates:

result = await graphiti.add_episode(
    name="Episode 5",
    episode_body="Content about related topics",
    source=EpisodeType.text,
    source_description="Source",
    reference_time=datetime.now(timezone.utc),
    update_communities=True  # Regenerate community summaries
)

print(f"Communities updated: {len(result.communities)}")

Episode Results

The add_episode method returns an AddEpisodeResults object:

result = await graphiti.add_episode(...)

print(f"Episode UUID: {result.episode.uuid}")
print(f"Entities extracted: {len(result.nodes)}")
print(f"Relationships extracted: {len(result.edges)}")
print(f"Communities: {len(result.communities)}")

# Access extracted entities
for node in result.nodes:
    print(f"Entity: {node.name} - {node.summary}")

# Access relationships
for edge in result.edges:
    print(f"Fact: {edge.fact}")

Best Practices

Sequential Processing

Add episodes sequentially, not in parallel. Each episode builds on the context of previous ones.

Descriptive Names

Use clear, unique names for episodes to make them easy to identify later.

Accurate Timestamps

Set reference_time to when the information was valid, not when you’re processing it.

Source Descriptions

Provide meaningful source descriptions to track data provenance.

Dense Content Handling

Graphiti automatically detects and chunks dense content (like cost reports with many entities):

# This dense JSON will be automatically chunked
aws_costs = {
    "report_type": "AWS Cost Breakdown",
    "services": [
        {"name": "Amazon S3", "cost": 2487.97},
        {"name": "Amazon RDS", "cost": 1071.74},
        # ... many more services
    ]
}

await graphiti.add_episode(
    name="AWS Cost Report",
    episode_body=json.dumps(aws_costs),
    source=EpisodeType.json,
    source_description="Monthly cost breakdown",
    reference_time=datetime.now(timezone.utc)
)
# Graphiti automatically chunks this into multiple LLM calls

Chunking is controlled by environment variables:

CHUNK_MIN_TOKENS - Minimum tokens before considering chunking (default: 1000)
CHUNK_DENSITY_THRESHOLD - Entity density threshold (default: 0.15)
CHUNK_TOKEN_SIZE - Target size per chunk (default: 3000)

Error Handling

try:
    result = await graphiti.add_episode(
        name="Example",
        episode_body="Content here",
        source=EpisodeType.text,
        source_description="Example source",
        reference_time=datetime.now(timezone.utc)
    )
except Exception as e:
    print(f"Failed to add episode: {e}")
    # Handle error appropriately

Next Steps

Bulk Operations

Learn how to add multiple episodes efficiently

Searching

Query your knowledge graph to retrieve information

Custom Entities

Define domain-specific entity types

Get Started

Core Concepts

Guides

Integrations

Advanced

Episode Types

Adding a Single Episode

Adding JSON Episodes

Adding Message Episodes

Episode Parameters

Advanced Options

Group IDs for Multi-Tenancy

Custom Entity Types

Updating Communities

Episode Results

Best Practices

Sequential Processing

Descriptive Names

Accurate Timestamps

Source Descriptions

Dense Content Handling

Error Handling

Next Steps

Bulk Operations

Searching

Custom Entities

Build docs developers (and LLMs) love

Get Started

Core Concepts

Guides

Integrations

Advanced

​Episode Types

​Adding a Single Episode

​Adding JSON Episodes

​Adding Message Episodes

​Episode Parameters

​Advanced Options

​Group IDs for Multi-Tenancy

​Custom Entity Types

​Updating Communities

​Episode Results

​Best Practices

Sequential Processing

Descriptive Names

Accurate Timestamps

Source Descriptions

​Dense Content Handling

​Error Handling

​Next Steps

Bulk Operations

Searching

Custom Entities

Build docs developers (and LLMs) love

Episode Types

Adding a Single Episode

Adding JSON Episodes

Adding Message Episodes

Episode Parameters

Advanced Options

Group IDs for Multi-Tenancy

Custom Entity Types

Updating Communities

Episode Results

Best Practices

Dense Content Handling

Error Handling

Next Steps