Graphiti Tutorial: Temporal Knowledge Graphs for AI Agents with Bi-Temporal Edges and Hybrid Search

By Fadi Updated May 28, 2026 14 min read Knowledge Graphs

TL;DR — Graphiti: temporal knowledge graphs for AI agents. Graphiti GitHub: "Temporal context graphs that track how facts change over time, maintain provenance, support ontology. Entities with evolving summaries, facts with temporal validity windows, episodes as provenance. Hybrid retrieval: semantic + BM25 + graph traversal. Purpose-built for agents on evolving data." Quickstart: "add_episode with text/JSON, auto-extracts entities/relationships, creates embeddings, tracks temporal. search() hybrid: semantic + BM25 + graph. Graph-aware reranking with center_node_uuid." Episodes: "Pipeline: extract → context retrieval → entity extraction → dedup → edge extraction → edge resolution/invalidation. Bi-temporal: valid_at + created_at. add_episode for invalidation, add_episode_bulk for batch." API: "search() returns EntityEdge, search_() returns Graph with configurable strategies. MCP server for AI assistants." Learn more with knowledge graph basics, build from documents, GraphRAG tutorial, and agent memory.

Graphiti GitHub introduces the concept: "Graphiti is a framework for building and querying temporal context graphs for AI agents. Unlike static knowledge graphs, Graphiti's context graphs track how facts change over time, maintain provenance to source data, and support both prescribed and learned ontology — making them purpose-built for agents operating on evolving, real-world data."

Graphiti Architecture

flowchart TD subgraph Input["Episode Input"] Text["Text Episode
EpisodeType.text
unstructured text
articles, transcripts"] Message["Message Episode
EpisodeType.message
multi-turn conversations
chat history"] JSON["JSON Episode
EpisodeType.json
structured data
records, catalogs"] end subgraph Pipeline["Episode Processing Pipeline"] Step1["1. Extract
Parse episode content"] Step2["2. Context Retrieval
Previous episodes
for entity resolution"] Step3["3. Entity Extraction
LLM extracts entities
optional custom types
via Pydantic models"] Step4["4. Node Dedup
Resolve same entity
across episodes"] Step5["5. Edge Extraction
LLM extracts relationships
triplets: Entity→Rel→Entity"] Step6["6. Edge Resolution
& Invalidation
contradict old edges
expire old, create new"] end subgraph Graph["Knowledge Graph Components"] Entities["Entities (Nodes)
people, products, policies
summaries evolve over time
custom types via Pydantic"] Facts["Facts (Edges)
triplets with temporal windows
valid_at: when fact became true
created_at: when ingested
invalidated when contradicted"] Episodes["Episodes (Provenance)
raw data as ingested
EpisodicNode: content, source
MENTIONS edges to entities
every fact traces back here"] Ontology["Custom Types
developer-defined
entity and edge types
Pydantic models
prescribed or learned"] end subgraph Search["Hybrid Search"] Semantic["Semantic
embedding similarity
vector search"] BM25["BM25
keyword matching
full-text search"] Graph["Graph Traversal
entity neighborhoods
multi-hop paths"] Rerank["Graph Reranking
center_node_uuid
graph distance scoring
contextual results"] end subgraph Agent["AI Agent Memory"] Context["Structured Context
replaces flat chat history
evolving knowledge graph
query across time"] MCP["MCP Server
AI assistants interact
episode management
entity management
hybrid search"] Zep["Zep Production
context graphs at scale
governed retrieval
low-latency assembly
production deployments"] end Text --> Step1 Message --> Step1 JSON --> Step1 Step1 --> Step2 Step2 --> Step3 Step3 --> Step4 Step4 --> Step5 Step5 --> Step6 Step3 --> Entities Step5 --> Facts Step1 --> Episodes Step3 --> Ontology Entities --> Semantic Facts --> Graph Episodes --> BM25 Semantic --> Rerank BM25 --> Rerank Graph --> Rerank Rerank --> Context Context --> MCP MCP --> Zep style Input fill:#4169E1,color:#fff style Pipeline fill:#39FF14,color:#000 style Graph fill:#2D1B69,color:#fff style Search fill:#FF6B6B,color:#fff

Graphiti vs Static Knowledge Graphs

Feature	Static KG	Graphiti
Temporal tracking	❌ Timeless facts	✅ Bi-temporal (valid_at + created_at)
Fact evolution	❌ No evolution	✅ Summaries evolve over time
Edge invalidation	❌ No contradiction handling	✅ Old edges expired when contradicted
Provenance	❌ No traceability	✅ Episodes trace every fact to source
Entity resolution	❌ Manual	✅ Automatic across episodes
Ontology	Optional	✅ Prescribed + learned, Pydantic types
Query across time	❌	✅ What was true then vs now
Agent memory	❌ Flat chat history	✅ Structured evolving context
Search	Cypher/vector	✅ Hybrid: semantic + BM25 + graph
Best for	Static domains	Evolving real-world data, agents

Implementation

from dataclasses import dataclass
from typing import Optional
from enum import Enum

class EpisodeType(Enum):
    TEXT = "text"
    MESSAGE = "message"
    JSON = "json"

@dataclass
class GraphitiTutorialGuide:
    """Graphiti tutorial implementation guide."""

    def get_setup(self) -> str:
        """Graphiti setup and initialization."""
        return (
            "# === GRAPHITI SETUP ===\n"
            "# pip install graphiti-core\n"
            "\n"
            "from graphiti import Graphiti\n"
            "from graphiti.entities\n"
            "    import EpisodeType\n"
            "from datetime import datetime,\n"
            "    timezone\n"
            "import json\n"
            "\n"
            "# 1. CONNECT TO NEO4J\n"
            "graphiti = Graphiti(\n"
            "    neo4j_uri=\n"
            "      'bolt://localhost:7687',\n"
            "    neo4j_user='neo4j',\n"
            "    neo4j_password='password',\n"
            ")\n"
            "# Also supports: FalkorDB,\n"
            "#   Kuzu, Amazon Neptune\n"
            "\n"
            "# 2. INITIALIZE INDICES\n"
            "await graphiti.build_indices()\n"
            "await graphiti.build_constraints()\n"
            "\n"
            "# 3. READY TO ADD EPISODES"
        )

    def get_add_episodes(self) -> str:
        """Add episodes to Graphiti graph."""
        return (
            "# === ADD EPISODES ===\n"
            "\n"
            "# Text episode\n"
            "await graphiti.add_episode(\n"
            "    name='Tech Article',\n"
            "    episode_body=(\n"
            "        'MIT researchers unveiled'\n"
            "        ' ClimateNet, an AI system'\n"
            "        ' for predicting climate'\n"
            "        ' patterns with unprecedented'\n"
            "        ' accuracy.'),\n"
            "    source=EpisodeType.text,\n"
            "    source_description=\n"
            "        'Technology magazine',\n"
            "    reference_time=datetime(\n"
            "        2023, 11, 15, 9, 30),\n"
            ")\n"
            "# Graphiti auto-extracts:\n"
            "#   Entities: MIT, ClimateNet\n"
            "#   Relations: MIT -> developed\n"
            "    -> ClimateNet\n"
            "#   Creates EpisodicNode\n"
            "#   MENTIONS edges to entities\n"
            "\n"
            "# JSON episode\n"
            "await graphiti.add_episode(\n"
            "    name='Product Update',\n"
            "    episode_body=json.dumps({\n"
            "        'name': 'Gavin Newsom',\n"
            "        'position': 'Governor',\n"
            "        'state': 'California',\n"
            "    }),\n"
            "    source=EpisodeType.json,\n"
            "    source_description=\n"
            "        'structured data',\n"
            "    reference_time=datetime.\n"
            "        now(timezone.utc),\n"
            ")\n"
            "\n"
            "# Message episode (chat)\n"
            "await graphiti.add_episode(\n"
            "    name='User Chat',\n"
            "    episode_body=(\n"
            "        'Alice: I started working'\n"
            "        ' at Acme Corp.\\n'\n"
            "        'Bob: Great! What role?\\n'\n"
            "        'Alice: Senior Engineer.'),\n"
            "    source=EpisodeType.message,\n"
            "    source_description=\n"
            "        'chat conversation',\n"
            "    reference_time=datetime.\n"
            "        now(timezone.utc),\n"
            ")\n"
            "\n"
            "# BULK INGESTION\n"
            "# (no edge invalidation)\n"
            "from graphiti import RawEpisode\n"
            "bulk_episodes = [\n"
            "    RawEpisode(\n"
            "        name=f'Item {i}',\n"
            "        content=json.dumps(item),\n"
            "        source=EpisodeType.json,\n"
            "        source_description=\n"
            "            'catalog',\n"
            "        reference_time=\n"
            "            datetime.now(),\n"
            "    )\n"
            "    for i, item in enumerate(\n"
            "        product_data)\n"
            "]\n"
            "await graphiti.\n"
            "    add_episode_bulk(\n"
            "    bulk_episodes)\n"
            "# NOTE: bulk does NOT perform\n"
            "#   edge invalidation. Use\n"
            "#   add_episode for incremental\n"
            "#   updates with contradictions."
        )

    def get_search(self) -> str:
        """Search Graphiti knowledge graph."""
        return (
            "# === SEARCH ===\n"
            "\n"
            "# 1. HYBRID SEARCH\n"
            "#   (semantic + BM25 + graph)\n"
            "edges = await graphiti.search(\n"
            "    query='Who works at'\n"
            "      ' Acme Corp?',\n"
            "    num_results=10,\n"
            ")\n"
            "for edge in edges:\n"
            "    print(f'{edge.name}: '\n"
            "      f'{edge.fact}')\n"
            "# Returns EntityEdge list\n"
            "#   with facts and temporal\n"
            "#   validity windows\n"
            "\n"
            "# 2. ADVANCED SEARCH\n"
            "#   (configurable strategies)\n"
            "from graphiti.search\n"
            "    import SearchConfig,\n"
            "    COMBINED_HYBRID_SEARCH_\n"
            "      CROSS_ENCODER\n"
            "\n"
            "results = await graphiti.\n"
            "    search_(\n"
            "    query='What changed about'\n"
            "      ' Alice recently?',\n"
            "    config=\n"
            "      COMBINED_HYBRID_SEARCH_\n"
            "        CROSS_ENCODER,\n"
            "    num_results=10,\n"
            ")\n"
            "# Returns Graph objects\n"
            "#   (nodes + edges)\n"
            "\n"
            "# 3. GRAPH-AWARE RERANKING\n"
            "edges = await graphiti.search(\n"
            "    query='Alice career',\n"
            "    center_node_uuid=\n"
            "      alice_node_uuid,\n"
            "    num_results=5,\n"
            ")\n"
            "# Reranks by graph distance\n"
            "#   from center node\n"
            "\n"
            "# 4. GROUP FILTERING\n"
            "edges = await graphiti.search(\n"
            "    query='products',\n"
            "    group_ids=['catalog_2026'],\n"
            "    num_results=10,\n"
            ")\n"
            "# Search within specific\n"
            "#   partition/group\n"
            "\n"
            "# 5. EPISODE RETRIEVAL\n"
            "episodes = await graphiti.\n"
            "    retrieve_episodes(\n"
            "    query='ClimateNet',\n"
            "    num_results=5,\n"
            ")\n"
            "# Returns source episodes\n"
            "#   for provenance tracking"
        )

    def get_custom_types(self) -> str:
        """Custom entity and edge types via Pydantic."""
        return (
            "# === CUSTOM ONTOLOGY ===\n"
            "from pydantic import BaseModel\n"
            "from graphiti import Graphiti\n"
            "\n"
            "# Define custom entity types\n"
            "class Person(BaseModel):\n"
            "    name: str\n"
            "    role: str | None = None\n"
            "    department: str | None = None\n"
            "\n"
            "class Organization(BaseModel):\n"
            "    name: str\n"
            "    industry: str | None = None\n"
            "    founded: int | None = None\n"
            "\n"
            "class Project(BaseModel):\n"
            "    name: str\n"
            "    status: str | None = None\n"
            "    deadline: str | None = None\n"
            "\n"
            "# Define custom edge types\n"
            "class WorksAt(BaseModel):\n"
            "    since: str | None = None\n"
            "    role: str | None = None\n"
            "\n"
            "class AssignedTo(BaseModel):\n"
            "    assigned_date: str | None = None\n"
            "    priority: str | None = None\n"
            "\n"
            "# Add episode with custom types\n"
            "await graphiti.add_episode(\n"
            "    name='HR Update',\n"
            "    episode_body=(\n"
            "        'Alice joined Acme Corp'\n"
            "        ' as Senior Engineer in'\n"
            "        ' the AI team. She was'\n"
            "        ' assigned to Project'\n"
            "        ' Phoenix.'),\n"
            "    source=EpisodeType.text,\n"
            "    source_description='HR',\n"
            "    reference_time=datetime.\n"
            "        now(timezone.utc),\n"
            "    entity_types={\n"
            "        'Person': Person,\n"
            "        'Organization':\n"
            "            Organization,\n"
            "        'Project': Project,\n"
            "    },\n"
            "    edge_types={\n"
            "        'WorksAt': WorksAt,\n"
            "        'AssignedTo': AssignedTo,\n"
            "    },\n"
            "    # edge_type_map defines\n"
            "    #   which edges connect\n"
            "    #   which entity types\n"
            "    edge_type_map={\n"
            "        ('Person',\n"
            "         'Organization'):\n"
            "            ['WorksAt'],\n"
            "        ('Person',\n"
            "         'Project'):\n"
            "            ['AssignedTo'],\n"
            "    },\n"
            ")"
        )

    def get_component_overview(self) -> dict:
        """Graphiti component overview."""
        return {
            "entities": {
                "what": "Nodes: people, products, policies, concepts",
                "temporal": "Summaries evolve over time",
                "custom": "Developer-defined via Pydantic models",
                "resolution": "Automatic deduplication across episodes",
            },
            "facts": {
                "what": "Edges: triplets (Entity → Relationship → Entity)",
                "temporal": "Bi-temporal: valid_at + created_at",
                "invalidation": "Old edges expired when contradicted",
                "provenance": "Every fact traces back to episode",
            },
            "episodes": {
                "what": "Raw data as ingested — ground truth stream",
                "types": "text, message (chat), json (structured)",
                "node": "EpisodicNode: content, source, valid_at, created_at",
                "edges": "MENTIONS edges connect episodes to entities",
                "bulk": "add_episode_bulk for batch (no invalidation)",
            },
            "ontology": {
                "what": "Custom entity and edge types",
                "how": "Pydantic models for type safety",
                "modes": "Prescribed (developer-defined) or learned",
                "edge_type_map": "Defines which edges connect which entity types",
            },
        }

Graphiti Tutorial Checklist

[ ] Graphiti: temporal context graph framework for AI agents — tracks how facts change over time
[ ] Unlike static KGs: bi-temporal edges, provenance, entity resolution, edge invalidation, evolving summaries
[ ] Components: Entities (nodes), Facts (edges), Episodes (provenance), Custom Types (ontology)
[ ] Entities: people, products, policies — summaries evolve over time, custom types via Pydantic
[ ] Facts: triplets with temporal validity windows — valid_at (when true) + created_at (when ingested)
[ ] Episodes: raw data as ingested — EpisodicNode with content, source, valid_at, created_at
[ ] Episodes: MENTIONS edges connect episodes to extracted entities for provenance
[ ] Custom Types: developer-defined entity and edge types via Pydantic models
[ ] Ontology modes: prescribed (developer-defined) or learned
[ ] edge_type_map: defines which edges connect which entity types
[ ] Episode types: text (unstructured), message (multi-turn chat), json (structured data)
[ ] Episode processing pipeline: extract → context retrieval → entity extraction → node dedup → edge extraction → edge resolution & invalidation
[ ] Context retrieval: previous episodes provide context for entity resolution
[ ] Entity extraction: LLM analyzes episode, optional custom types
[ ] Node deduplication: resolves same entity across episodes automatically
[ ] Edge extraction: LLM extracts relationship triplets
[ ] Edge resolution & invalidation: old edges expired when contradicted by new facts
[ ] reference_time: sets valid_at — grounded to when facts occurred, not when processed
[ ] add_episode: performs edge invalidation — use for incremental updates with contradictions
[ ] add_episode_bulk: parallel batch ingestion — does NOT perform edge invalidation
[ ] add_episode_bulk: use only for populating empty graphs or when invalidation not needed
[ ] Graphiti supports: Neo4j 5.26, FalkorDB 1.1.2, Amazon Neptune, Kuzu
[ ] Setup: Graphiti(neo4j_uri, neo4j_user, neo4j_password), build_indices(), build_constraints()
[ ] search(): hybrid search — semantic embeddings + BM25 keyword + graph traversal
[ ] search(): returns list of EntityEdge with facts and temporal validity
[ ] search_(): advanced search with configurable strategies and rerankers
[ ] search_(): returns Graph objects (nodes and edges)
[ ] COMBINED_HYBRID_SEARCH_CROSS_ENCODER: default search config
[ ] center_node_uuid: graph-aware reranking by graph distance from center node
[ ] group_ids: partition filtering for searching within specific groups
[ ] retrieve_episodes: retrieve source episodes for provenance tracking
[ ] get_nodes_and_edges_by_episode: get graph data for specific episode
[ ] MCP server: AI assistants interact with Graphiti via MCP protocol
[ ] MCP server: episode management, entity management, hybrid search, group management
[ ] Zep: production-scale context graph management, governed retrieval, low-latency assembly
[ ] Agent memory: replaces flat chat history with structured, evolving knowledge graph
[ ] Agent memory: query across time, meaning, and relationships
[ ] Agent memory: continuously integrates user interactions and enterprise data
[ ] Custom extraction instructions: additional context to guide extraction prompts
[ ] Saga: associate episodes with named sagas, linked via HAS_EPISODE and NEXT_EPISODE edges
[ ] Python 3.10+ required
[ ] Read knowledge graph basics for concepts
[ ] Read build from documents for extraction
[ ] Read GraphRAG tutorial for comparison
[ ] Read agent memory for memory patterns
[ ] Test: episodes correctly extract entities and relationships
[ ] Test: entity resolution deduplicates same entity across episodes
[ ] Test: edge invalidation expires old facts when contradicted
[ ] Test: hybrid search returns relevant results with temporal context
[ ] Test: graph-aware reranking improves result relevance
[ ] Test: custom types constrain extraction to defined schema
[ ] Test: bulk ingestion performance for large datasets
[ ] Document graph database choice, episode types, custom ontology, search strategy, agent integration

FAQ

What is Graphiti and how does it differ from static knowledge graphs?

Graphiti builds temporal context graphs that track how facts change over time, with bi-temporal edges and provenance. Graphiti GitHub: "Graphiti is a framework for building and querying temporal context graphs for AI agents. Unlike static knowledge graphs, Graphiti context graphs track how facts change over time, maintain provenance to source data, and support prescribed and learned ontology — purpose-built for agents operating on evolving, real-world data. Continuously integrates user interactions, structured and unstructured enterprise data, and external information into a coherent, queryable graph." Components: (1) Entities (nodes): people, products, policies with summaries that evolve over time. (2) Facts/Relationships (edges): triplets with temporal validity windows. (3) Episodes (provenance): raw data as ingested — every derived fact traces back here. (4) Custom Types (ontology): developer-defined entity and edge types via Pydantic models. Differences: (1) Static KG: facts are timeless, no provenance, no evolution. (2) Graphiti: bi-temporal edges track valid_at and created_at, edge invalidation when facts change, episodes preserve provenance, entity resolution across episodes.

How do you add episodes to a Graphiti knowledge graph?

Use add_episode with text or JSON content, source type, reference_time. Graphiti Quickstart: "Episodes are primary units of information. Text or structured JSON, automatically processed to extract entities and relationships. graphiti.add_episode(name, episode_body, source, source_description, reference_time). Graphiti automatically extracts entities, identifies relationships, creates embeddings, tracks temporal information." Graphiti Episodes: "Episode processing pipeline: (1) Extract entities, (2) Context retrieval from previous episodes, (3) Entity extraction via LLM, (4) Node deduplication, (5) Edge extraction, (6) Edge resolution and invalidation. EpisodicNode stores name, content, source, valid_at, created_at, entity_edges, group_id. MENTIONS edges connect episodes to entities." Adding Episodes: "EpisodeType.text for text, EpisodeType.message for multi-turn conversations, EpisodeType.json for structured data. add_episode_bulk for large datasets — parallel processing but no edge invalidation." Usage: (1) EpisodeType.text: unstructured text. (2) EpisodeType.message: multi-turn chat. (3) EpisodeType.json: structured data. (4) reference_time: when content occurred. (5) add_episode_bulk: batch ingestion, no invalidation.

How does Graphiti handle temporal facts and edge invalidation?

Bi-temporal edges track valid_at (when fact became true) and created_at (when ingested); edge invalidation marks old facts as expired. Graphiti GitHub: "Facts/Relationships (edges): triplets with temporal validity windows. Entities with summaries that evolve over time. Build context graphs that evolve with every interaction — tracking what is true now and what was true before." Graphiti Episodes: "Edge Resolution and Invalidation: when new episode contradicts existing edge, old edge is invalidated (expired) and new edge created. reference_time sets valid_at — grounded to when facts occurred, not when processed. add_episode performs edge invalidation; add_episode_bulk does NOT — use add_episode for incremental updates with contradiction handling." Temporal: (1) valid_at: when the fact became true (reference_time). (2) created_at: when ingested into Graphiti. (3) Edge invalidation: old edge expired when contradicted. (4) Query across time: what was true then vs now. (5) Provenance: every fact traces to episode. (6) Use add_episode (not bulk) for contradiction handling.

How do you search a Graphiti knowledge graph?

Use search() for hybrid retrieval combining semantic, BM25, and graph traversal; search_() for advanced configurable strategies. Graphiti Quickstart: "search() performs hybrid search combining semantic similarity using embeddings, BM25 keyword matching, and graph-based retrieval. Graph-aware reranking with center_node_uuid for contextual results." Graphiti API: "search(query, center_node_uuid, group_ids, num_results, search_filter) returns list of EntityEdge. search_(query, config, group_ids, center_node_uuid, bfs_origin_node_uuids, search_filter) returns Graph objects (nodes and edges) with configurable search strategies and rerankers. COMBINED_HYBRID_SEARCH_CROSS_ENCODER default config. retrieve_episodes for episode retrieval. get_nodes_and_edges_by_episode for provenance." Search: (1) search(): hybrid (semantic + BM25 + graph), returns EntityEdge list. (2) search_(): advanced, configurable strategies, returns Graph (nodes + edges). (3) center_node_uuid: graph-aware reranking. (4) COMBINED_HYBRID_SEARCH_CROSS_ENCODER: default config. (5) group_ids: partition filtering.

How do you use Graphiti for AI agent memory?

Graphiti provides temporal context graphs as structured agent memory, replacing flat chat history with evolving knowledge. Graphiti GitHub: "Give agents rich, structured context instead of flat document chunks or raw chat history. Query across time, meaning, and relationships with hybrid retrieval. Open-source temporal context graph engine at the core of Zep context infrastructure for AI agents. Zep manages context graphs at scale, providing governed, low-latency context retrieval and assembly for production agent deployments. MCP server allows AI assistants to interact with Graphiti via MCP protocol — episode management, entity management, semantic and hybrid search, group management." Agent memory: (1) Episodes: ingest user interactions as they happen. (2) Entity resolution: recognizes same entity across episodes. (3) Temporal: tracks what was true then vs now. (4) Hybrid search: semantic + keyword + graph for context retrieval. (5) MCP server: AI assistants interact via MCP. (6) Zep: production-scale context graph management. (7) Replaces flat chat history with structured, evolving knowledge graph.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →