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

# Graphiti

> Main Graphiti class for building temporally-aware knowledge graphs

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

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

<ParamField path="uri" type="str | None">
  The URI of the Neo4j database. Required when `graph_driver` is None.
</ParamField>

<ParamField path="user" type="str | None">
  The username for authenticating with the Neo4j database.
</ParamField>

<ParamField path="password" type="str | None">
  The password for authenticating with the Neo4j database.
</ParamField>

<ParamField path="llm_client" type="LLMClient | None">
  An instance of LLMClient for natural language processing tasks. If not provided, a default OpenAIClient will be initialized.
</ParamField>

<ParamField path="embedder" type="EmbedderClient | None">
  An instance of EmbedderClient for embedding tasks. If not provided, a default OpenAIEmbedder will be initialized.
</ParamField>

<ParamField path="cross_encoder" type="CrossEncoderClient | None">
  An instance of CrossEncoderClient for reranking tasks. If not provided, a default OpenAIRerankerClient will be initialized.
</ParamField>

<ParamField path="store_raw_episode_content" type="bool" default="True">
  Whether to store the raw content of episodes.
</ParamField>

<ParamField path="graph_driver" type="GraphDriver | None">
  An instance of GraphDriver for database operations. If not provided, a default Neo4jDriver will be initialized.
</ParamField>

<ParamField path="max_coroutines" type="int | None">
  The maximum number of concurrent operations allowed. Overrides SEMAPHORE\_LIMIT set in the environment. If not set, the Graphiti default is used.
</ParamField>

<ParamField path="tracer" type="Tracer | None">
  An OpenTelemetry tracer instance for distributed tracing. If not provided, tracing is disabled (no-op).
</ParamField>

<ParamField path="trace_span_prefix" type="str" default="graphiti">
  Prefix to prepend to all span names.
</ParamField>

### Example

```python theme={null}
from graphiti_core import Graphiti

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

## Methods

### add\_episode

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

<ParamField path="name" type="str" required>
  The name of the episode.
</ParamField>

<ParamField path="episode_body" type="str" required>
  The content of the episode.
</ParamField>

<ParamField path="source_description" type="str" required>
  A description of the episode's source.
</ParamField>

<ParamField path="reference_time" type="datetime" required>
  The reference time for the episode.
</ParamField>

<ParamField path="source" type="EpisodeType" default="EpisodeType.message">
  The type of the episode. Options: `EpisodeType.message`, `EpisodeType.json`, `EpisodeType.text`.
</ParamField>

<ParamField path="group_id" type="str | None">
  An id for the graph partition the episode is a part of.
</ParamField>

<ParamField path="uuid" type="str | None">
  Optional uuid of the episode.
</ParamField>

<ParamField path="update_communities" type="bool" default="False">
  Whether to update communities with new node information.
</ParamField>

<ParamField path="entity_types" type="dict[str, type[BaseModel]] | None">
  Dictionary mapping entity type names to their Pydantic model definitions.
</ParamField>

<ParamField path="excluded_entity_types" type="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.
</ParamField>

<ParamField path="previous_episode_uuids" 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.
</ParamField>

<ParamField path="edge_types" type="dict[str, type[BaseModel]] | None">
  Dictionary mapping edge type names to their Pydantic model definitions.
</ParamField>

<ParamField path="edge_type_map" type="dict[tuple[str, str], list[str]] | None">
  Mapping of (source\_type, target\_type) tuples to allowed edge type names.
</ParamField>

<ParamField path="custom_extraction_instructions" type="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.
</ParamField>

<ParamField path="saga" type="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.
</ParamField>

<ParamField path="saga_previous_episode_uuid" type="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.
</ParamField>

<ResponseField name="episode" type="EpisodicNode">
  The created or updated episodic node.
</ResponseField>

<ResponseField name="episodic_edges" type="list[EpisodicEdge]">
  List of episodic edges connecting entities to the episode.
</ResponseField>

<ResponseField name="nodes" type="list[EntityNode]">
  List of entity nodes extracted from the episode.
</ResponseField>

<ResponseField name="edges" type="list[EntityEdge]">
  List of entity edges (relationships) extracted from the episode.
</ResponseField>

<ResponseField name="communities" type="list[CommunityNode]">
  List of community nodes (only if update\_communities=True).
</ResponseField>

<ResponseField name="community_edges" type="list[CommunityEdge]">
  List of community edges (only if update\_communities=True).
</ResponseField>

#### Example

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

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

<ParamField path="bulk_episodes" type="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.
</ParamField>

<ParamField path="group_id" type="str | None">
  An id for the graph partition the episode is a part of.
</ParamField>

<ParamField path="entity_types" type="dict[str, type[BaseModel]] | None">
  Dictionary mapping entity type names to Pydantic models.
</ParamField>

<ParamField path="excluded_entity_types" type="list[str] | None">
  List of entity type names to exclude from extraction.
</ParamField>

<ParamField path="edge_types" type="dict[str, type[BaseModel]] | None">
  Dictionary mapping edge type names to Pydantic models.
</ParamField>

<ParamField path="edge_type_map" type="dict[tuple[str, str], list[str]] | None">
  Mapping of (source\_type, target\_type) to allowed edge types.
</ParamField>

<ParamField path="custom_extraction_instructions" type="str | None">
  Custom extraction instructions string to be included in the extract entities and extract edges prompts.
</ParamField>

<ParamField path="saga" type="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.
</ParamField>

<ResponseField name="episodes" type="list[EpisodicNode]">
  List of created episodic nodes.
</ResponseField>

<ResponseField name="episodic_edges" type="list[EpisodicEdge]">
  List of episodic edges.
</ResponseField>

<ResponseField name="nodes" type="list[EntityNode]">
  List of extracted entity nodes.
</ResponseField>

<ResponseField name="edges" type="list[EntityEdge]">
  List of extracted entity edges.
</ResponseField>

<ResponseField name="communities" type="list[CommunityNode]">
  List of community nodes (empty in bulk operations).
</ResponseField>

<ResponseField name="community_edges" type="list[CommunityEdge]">
  List of community edges (empty in bulk operations).
</ResponseField>

#### Example

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

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

<ParamField path="query" type="str" required>
  The search query string.
</ParamField>

<ParamField path="center_node_uuid" type="str | None">
  Facts will be reranked based on proximity to this node.
</ParamField>

<ParamField path="group_ids" type="list[str] | None">
  The graph partitions to return data from.
</ParamField>

<ParamField path="num_results" type="int" default="10">
  The maximum number of results to return.
</ParamField>

<ParamField path="search_filter" type="SearchFilters | None">
  Filters to apply to the search.
</ParamField>

<ParamField path="driver" type="GraphDriver | None">
  The graph driver to use. If not provided, uses the default driver.
</ParamField>

<ResponseField name="edges" type="list[EntityEdge]">
  List of EntityEdge objects that are relevant to the search query.
</ResponseField>

#### Example

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

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

<ParamField path="query" type="str" required>
  The search query string.
</ParamField>

<ParamField path="config" type="SearchConfig" default="COMBINED_HYBRID_SEARCH_CROSS_ENCODER">
  Search configuration specifying search methods and rerankers. See search\_config\_recipes for preset configurations.
</ParamField>

<ParamField path="group_ids" type="list[str] | None">
  The graph partitions to return data from.
</ParamField>

<ParamField path="center_node_uuid" type="str | None">
  Center node for node distance reranking.
</ParamField>

<ParamField path="bfs_origin_node_uuids" type="list[str] | None">
  Origin nodes for breadth-first search.
</ParamField>

<ParamField path="search_filter" type="SearchFilters | None">
  Filters to apply to the search.
</ParamField>

<ParamField path="driver" type="GraphDriver | None">
  The graph driver to use.
</ParamField>

<ResponseField name="edges" type="list[EntityEdge]">
  List of relevant entity edges.
</ResponseField>

<ResponseField name="edge_reranker_scores" type="list[float]">
  Reranker scores for edges.
</ResponseField>

<ResponseField name="nodes" type="list[EntityNode]">
  List of relevant entity nodes.
</ResponseField>

<ResponseField name="node_reranker_scores" type="list[float]">
  Reranker scores for nodes.
</ResponseField>

<ResponseField name="episodes" type="list[EpisodicNode]">
  List of relevant episodes.
</ResponseField>

<ResponseField name="episode_reranker_scores" type="list[float]">
  Reranker scores for episodes.
</ResponseField>

<ResponseField name="communities" type="list[CommunityNode]">
  List of relevant communities.
</ResponseField>

<ResponseField name="community_reranker_scores" type="list[float]">
  Reranker scores for communities.
</ResponseField>

#### Example

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

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

<ParamField path="group_ids" type="list[str] | None">
  Create communities only for the listed group\_ids. If blank, the entire graph will be used.
</ParamField>

<ParamField path="driver" type="GraphDriver | None">
  The graph driver to use.
</ParamField>

<ResponseField name="community_nodes" type="list[CommunityNode]">
  List of created community nodes.
</ResponseField>

<ResponseField name="community_edges" type="list[CommunityEdge]">
  List of edges connecting entities to communities.
</ResponseField>

#### Example

```python theme={null}
community_nodes, community_edges = await graphiti.build_communities(
    group_ids=["user_123"]
)

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

### retrieve\_episodes

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

<ParamField path="reference_time" type="datetime" required>
  The reference time to retrieve episodes before.
</ParamField>

<ParamField path="last_n" type="int" default="EPISODE_WINDOW_LEN">
  The number of episodes to retrieve.
</ParamField>

<ParamField path="group_ids" type="list[str] | None">
  The group ids to return data from.
</ParamField>

<ParamField path="source" type="EpisodeType | None">
  Filter episodes by source type.
</ParamField>

<ParamField path="driver" type="GraphDriver | None">
  The graph driver to use.
</ParamField>

<ParamField path="saga" type="str | None">
  If provided, only retrieve episodes that belong to the saga with this name.
</ParamField>

<ResponseField name="episodes" type="list[EpisodicNode]">
  List of the most recent EpisodicNode objects.
</ResponseField>

### build\_indices\_and\_constraints

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

<ParamField path="delete_existing" type="bool" default="False">
  Whether to clear existing indices before creating new ones.
</ParamField>

#### Example

```python theme={null}
await graphiti.build_indices_and_constraints()
```

### add\_triplet

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

<ParamField path="source_node" type="EntityNode" required>
  The source entity node.
</ParamField>

<ParamField path="edge" type="EntityEdge" required>
  The edge connecting the nodes.
</ParamField>

<ParamField path="target_node" type="EntityNode" required>
  The target entity node.
</ParamField>

<ResponseField name="nodes" type="list[EntityNode]">
  List of saved nodes (source and target).
</ResponseField>

<ResponseField name="edges" type="list[EntityEdge]">
  List of saved edges.
</ResponseField>

### remove\_episode

```python theme={null}
async def remove_episode(
    episode_uuid: str
)
```

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

<ParamField path="episode_uuid" type="str" required>
  The UUID of the episode to remove.
</ParamField>

#### Example

```python theme={null}
await graphiti.remove_episode("episode-uuid-123")
```

### get\_nodes\_and\_edges\_by\_episode

```python theme={null}
async def get_nodes_and_edges_by_episode(
    episode_uuids: list[str]
) -> SearchResults
```

Retrieve all nodes and edges associated with specific episodes.

<ParamField path="episode_uuids" type="list[str]" required>
  List of episode UUIDs to retrieve data for.
</ParamField>

<ResponseField name="nodes" type="list[EntityNode]">
  List of entity nodes mentioned in the episodes.
</ResponseField>

<ResponseField name="edges" type="list[EntityEdge]">
  List of entity edges from the episodes.
</ResponseField>

### close

```python theme={null}
async def close()
```

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

#### Example

```python theme={null}
try:
    # Use graphiti
    await graphiti.add_episode(...)
finally:
    await graphiti.close()
```

## Properties

### token\_tracker

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

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