Graphiti

Overview

The Graphiti class is the main entry point for building and managing real-time, temporally-aware knowledge graphs. It provides methods for adding episodes, searching, building communities, and managing the graph structure.

Constructor

Graphiti(
    uri: str | None = None,
    user: str | None = None,
    password: str | None = None,
    llm_client: LLMClient | None = None,
    embedder: EmbedderClient | None = None,
    cross_encoder: CrossEncoderClient | None = None,
    store_raw_episode_content: bool = True,
    graph_driver: GraphDriver | None = None,
    max_coroutines: int | None = None,
    tracer: Tracer | None = None,
    trace_span_prefix: str = 'graphiti',
)

Initialize a Graphiti instance with database connection and client configurations.

str | None

The URI of the Neo4j database. Required when graph_driver is None.

str | None

The username for authenticating with the Neo4j database.

str | None

The password for authenticating with the Neo4j database.

LLMClient | None

An instance of LLMClient for natural language processing tasks. If not provided, a default OpenAIClient will be initialized.

EmbedderClient | None

An instance of EmbedderClient for embedding tasks. If not provided, a default OpenAIEmbedder will be initialized.

CrossEncoderClient | None

An instance of CrossEncoderClient for reranking tasks. If not provided, a default OpenAIRerankerClient will be initialized.

bool

default:"True"

Whether to store the raw content of episodes.

GraphDriver | None

An instance of GraphDriver for database operations. If not provided, a default Neo4jDriver will be initialized.

int | None

The maximum number of concurrent operations allowed. Overrides SEMAPHORE_LIMIT set in the environment. If not set, the Graphiti default is used.

Tracer | None

An OpenTelemetry tracer instance for distributed tracing. If not provided, tracing is disabled (no-op).

str

default:"graphiti"

Prefix to prepend to all span names.

Example

from graphiti_core import Graphiti

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

Methods

add_episode

async def add_episode(
    name: str,
    episode_body: str,
    source_description: str,
    reference_time: datetime,
    source: EpisodeType = EpisodeType.message,
    group_id: str | None = None,
    uuid: str | None = None,
    update_communities: bool = False,
    entity_types: dict[str, type[BaseModel]] | None = None,
    excluded_entity_types: list[str] | None = None,
    previous_episode_uuids: list[str] | None = None,
    edge_types: dict[str, type[BaseModel]] | None = None,
    edge_type_map: dict[tuple[str, str], list[str]] | None = None,
    custom_extraction_instructions: str | None = None,
    saga: str | SagaNode | None = None,
    saga_previous_episode_uuid: str | None = None,
) -> AddEpisodeResults

Process an episode and update the graph with extracted entities and relationships.

str

required

The name of the episode.

str

required

The content of the episode.

str

required

A description of the episode’s source.

datetime

required

The reference time for the episode.

EpisodeType

default:"EpisodeType.message"

The type of the episode. Options: EpisodeType.message, EpisodeType.json, EpisodeType.text.

str | None

An id for the graph partition the episode is a part of.

str | None

Optional uuid of the episode.

bool

default:"False"

Whether to update communities with new node information.

dict[str, type[BaseModel]] | None

Dictionary mapping entity type names to their Pydantic model definitions.

list[str] | None

List of entity type names to exclude from the graph. Entities classified into these types will not be added to the graph. Can include ‘Entity’ to exclude the default entity type.

list[str] | None

List of episode uuids to use as the previous episodes. If not provided, the most recent episodes by created_at date will be used.

dict[str, type[BaseModel]] | None

Dictionary mapping edge type names to their Pydantic model definitions.

dict[tuple[str, str], list[str]] | None

Mapping of (source_type, target_type) tuples to allowed edge type names.

str | None

Custom extraction instructions string to be included in the extract entities and extract edges prompts. This allows for additional instructions or context to guide the extraction process.

str | SagaNode | None

Either a saga name (str) or a SagaNode object to associate this episode with. If a string is provided and a saga with this name already exists in the group, the episode will be added to it. Otherwise, a new saga will be created. Sagas are connected to episodes via HAS_EPISODE edges, and consecutive episodes are linked via NEXT_EPISODE edges.

str | None

UUID of the previous episode in the saga. If provided, skips the database query to find the most recent episode. Useful for efficiently adding multiple episodes to the same saga in sequence. The returned AddEpisodeResults.episode.uuid can be passed as this parameter for the next episode.

EpisodicNode

The created or updated episodic node.

list[EpisodicEdge]

List of episodic edges connecting entities to the episode.

list[EntityNode]

List of entity nodes extracted from the episode.

list[EntityEdge]

List of entity edges (relationships) extracted from the episode.

list[CommunityNode]

List of community nodes (only if update_communities=True).

list[CommunityEdge]

List of community edges (only if update_communities=True).

Example

from datetime import datetime
from graphiti_core.nodes import EpisodeType

result = await graphiti.add_episode(
    name="User Conversation",
    episode_body="user: I love pizza\nassistant: Pizza is delicious!",
    source_description="Chat conversation",
    reference_time=datetime.now(),
    source=EpisodeType.message,
    group_id="user_123"
)

print(f"Extracted {len(result.nodes)} entities")
print(f"Extracted {len(result.edges)} relationships")

add_episode_bulk

async def add_episode_bulk(
    bulk_episodes: list[RawEpisode],
    group_id: str | None = None,
    entity_types: dict[str, type[BaseModel]] | None = None,
    excluded_entity_types: list[str] | None = None,
    edge_types: dict[str, type[BaseModel]] | None = None,
    edge_type_map: dict[tuple[str, str], list[str]] | None = None,
    custom_extraction_instructions: str | None = None,
    saga: str | SagaNode | None = None,
) -> AddBulkEpisodeResults

Process multiple episodes in bulk and update the graph.

list[RawEpisode]

required

A list of RawEpisode objects to be processed and added to the graph. Each RawEpisode contains: name, content, source_description, source, reference_time, and optional uuid.

str | None

An id for the graph partition the episode is a part of.

dict[str, type[BaseModel]] | None

Dictionary mapping entity type names to Pydantic models.

list[str] | None

List of entity type names to exclude from extraction.

dict[str, type[BaseModel]] | None

Dictionary mapping edge type names to Pydantic models.

dict[tuple[str, str], list[str]] | None

Mapping of (source_type, target_type) to allowed edge types.

str | None

Custom extraction instructions string to be included in the extract entities and extract edges prompts.

str | SagaNode | None

Either a saga name (str) or a SagaNode object to associate all episodes with. If a string is provided and a saga with this name already exists in the group, the episodes will be added to it. Otherwise, a new saga will be created.

list[EpisodicNode]

List of created episodic nodes.

list[EpisodicEdge]

List of episodic edges.

list[EntityNode]

List of extracted entity nodes.

list[EntityEdge]

List of extracted entity edges.

list[CommunityNode]

List of community nodes (empty in bulk operations).

list[CommunityEdge]

List of community edges (empty in bulk operations).

Example

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

episodes = [
    RawEpisode(
        name="Episode 1",
        content="user: Hello",
        source_description="Chat",
        source=EpisodeType.message,
        reference_time=datetime.now()
    ),
    RawEpisode(
        name="Episode 2",
        content="user: How are you?",
        source_description="Chat",
        source=EpisodeType.message,
        reference_time=datetime.now()
    )
]

result = await graphiti.add_episode_bulk(episodes, group_id="user_123")
print(f"Processed {len(result.episodes)} episodes")

search

async def search(
    query: str,
    center_node_uuid: str | None = None,
    group_ids: list[str] | None = None,
    num_results: int = 10,
    search_filter: SearchFilters | None = None,
    driver: GraphDriver | None = None,
) -> list[EntityEdge]

Perform a hybrid search on the knowledge graph.

str

required

The search query string.

str | None

Facts will be reranked based on proximity to this node.

list[str] | None

The graph partitions to return data from.

int

default:"10"

The maximum number of results to return.

SearchFilters | None

Filters to apply to the search.

GraphDriver | None

The graph driver to use. If not provided, uses the default driver.

list[EntityEdge]

List of EntityEdge objects that are relevant to the search query.

Example

edges = await graphiti.search(
    query="What does the user like?",
    group_ids=["user_123"],
    num_results=5
)

for edge in edges:
    print(f"{edge.source_node_uuid} -> {edge.target_node_uuid}: {edge.fact}")

search_

async def search_(
    query: str,
    config: SearchConfig = COMBINED_HYBRID_SEARCH_CROSS_ENCODER,
    group_ids: list[str] | None = None,
    center_node_uuid: str | None = None,
    bfs_origin_node_uuids: list[str] | None = None,
    search_filter: SearchFilters | None = None,
    driver: GraphDriver | None = None,
) -> SearchResults

Advanced search method that returns Graph objects (nodes and edges) with configurable search strategies and rerankers.

str

required

The search query string.

SearchConfig

default:"COMBINED_HYBRID_SEARCH_CROSS_ENCODER"

Search configuration specifying search methods and rerankers. See search_config_recipes for preset configurations.

list[str] | None

The graph partitions to return data from.

str | None

Center node for node distance reranking.

list[str] | None

Origin nodes for breadth-first search.

SearchFilters | None

Filters to apply to the search.

GraphDriver | None

The graph driver to use.

list[EntityEdge]

List of relevant entity edges.

list[float]

Reranker scores for edges.

list[EntityNode]

List of relevant entity nodes.

list[float]

Reranker scores for nodes.

list[EpisodicNode]

List of relevant episodes.

list[float]

Reranker scores for episodes.

list[CommunityNode]

List of relevant communities.

list[float]

Reranker scores for communities.

Example

from graphiti_core.search.search_config_recipes import COMBINED_HYBRID_SEARCH_CROSS_ENCODER

results = await graphiti.search_(
    query="What does the user like?",
    config=COMBINED_HYBRID_SEARCH_CROSS_ENCODER,
    group_ids=["user_123"]
)

print(f"Found {len(results.edges)} edges")
print(f"Found {len(results.nodes)} nodes")
print(f"Found {len(results.episodes)} episodes")

build_communities

async def build_communities(
    group_ids: list[str] | None = None,
    driver: GraphDriver | None = None,
) -> tuple[list[CommunityNode], list[CommunityEdge]]

Use a community clustering algorithm to find communities of nodes and create community nodes summarizing the content.

list[str] | None

Create communities only for the listed group_ids. If blank, the entire graph will be used.

GraphDriver | None

The graph driver to use.

list[CommunityNode]

List of created community nodes.

list[CommunityEdge]

List of edges connecting entities to communities.

Example

community_nodes, community_edges = await graphiti.build_communities(
    group_ids=["user_123"]
)

print(f"Created {len(community_nodes)} communities")

retrieve_episodes

async def retrieve_episodes(
    reference_time: datetime,
    last_n: int = EPISODE_WINDOW_LEN,
    group_ids: list[str] | None = None,
    source: EpisodeType | None = None,
    driver: GraphDriver | None = None,
    saga: str | None = None,
) -> list[EpisodicNode]

Retrieve the last n episodic nodes from the graph.

datetime

required

The reference time to retrieve episodes before.

int

default:"EPISODE_WINDOW_LEN"

The number of episodes to retrieve.

list[str] | None

The group ids to return data from.

EpisodeType | None

Filter episodes by source type.

GraphDriver | None

The graph driver to use.

str | None

If provided, only retrieve episodes that belong to the saga with this name.

list[EpisodicNode]

List of the most recent EpisodicNode objects.

build_indices_and_constraints

async def build_indices_and_constraints(
    delete_existing: bool = False
)

Build indices and constraints in the Neo4j database to optimize query performance and ensure data integrity.

bool

default:"False"

Whether to clear existing indices before creating new ones.

Example

await graphiti.build_indices_and_constraints()

add_triplet

async def add_triplet(
    source_node: EntityNode,
    edge: EntityEdge,
    target_node: EntityNode
) -> AddTripletResults

Add a single triplet (source node, edge, target node) to the graph.

EntityNode

required

The source entity node.

EntityEdge

required

The edge connecting the nodes.

EntityNode

required

The target entity node.

list[EntityNode]

List of saved nodes (source and target).

list[EntityEdge]

List of saved edges.

remove_episode

async def remove_episode(
    episode_uuid: str
)

Remove an episode and its associated edges and nodes from the graph.

str

required

The UUID of the episode to remove.

Example

await graphiti.remove_episode("episode-uuid-123")

get_nodes_and_edges_by_episode

async def get_nodes_and_edges_by_episode(
    episode_uuids: list[str]
) -> SearchResults

Retrieve all nodes and edges associated with specific episodes.

list[str]

required

List of episode UUIDs to retrieve data for.

list[EntityNode]

List of entity nodes mentioned in the episodes.

list[EntityEdge]

List of entity edges from the episodes.

close

async def close()

Close the connection to the Neo4j database. This should be called when the Graphiti instance is no longer needed.

Example

try:
    # Use graphiti
    await graphiti.add_episode(...)
finally:
    await graphiti.close()

Properties

token_tracker

@property
def token_tracker

Access the LLM client’s token usage tracker. Returns: TokenUsageTracker that can be used to:

Get token usage by prompt type: tracker.get_usage()
Get total token usage: tracker.get_total_usage()
Print a formatted summary: tracker.print_summary()
Reset tracking: tracker.reset()

Example

# Add some episodes
await graphiti.add_episode(...)

# Check token usage
usage = graphiti.token_tracker.get_total_usage()
print(f"Total tokens used: {usage['total_tokens']}")

# Print detailed summary
graphiti.token_tracker.print_summary()

Core

Data Models

Drivers

LLM Clients

Embedders

Overview

Constructor

Example

Methods

add_episode

Example

add_episode_bulk

Example

search

Example

search_

Example

build_communities

Example

retrieve_episodes

build_indices_and_constraints

Example

add_triplet

remove_episode

Example

get_nodes_and_edges_by_episode

close

Example

Properties

token_tracker

Example

​Overview

​Constructor

​Example

​Methods

​add_episode

​Example

​add_episode_bulk

​Example

​search

​Example

​search_

​Example

​build_communities

​Example

​retrieve_episodes

​build_indices_and_constraints

​Example

​add_triplet

​remove_episode

​Example

​get_nodes_and_edges_by_episode

​close

​Example

​Properties

​token_tracker

​Example

Overview

Constructor

Example

Methods

add_episode

Example

add_episode_bulk

Example

search

Example

search_

Example

build_communities

Example

retrieve_episodes

build_indices_and_constraints

Example

add_triplet

remove_episode

Example

get_nodes_and_edges_by_episode

close

Example

Properties

token_tracker

Example