Project 9: Graphiti Framework Integration

Integrate your components with Zep’s Graphiti framework to build a production-grade temporal knowledge graph with the three-tier memory architecture (episodic, semantic, community).

Quick Reference

Attribute	Value
Difficulty	Level 4: Expert
Time Estimate	2-3 weeks (30-40 hours)
Language	Python
Prerequisites	Projects 1-8, Docker, production Python patterns
Key Topics	Graphiti architecture, FalkorDB, three-tier memory, temporal edges, group conversations

1. Learning Objectives

By completing this project, you will:

Understand Graphiti’s architecture and design decisions.
Set up FalkorDB for graph storage.
Build the three-tier memory architecture with Graphiti.
Implement temporal edge management.
Handle multi-user group conversations.

2. Theoretical Foundation

2.1 Core Concepts

Graphiti: Zep’s open-source framework for building temporal knowledge graphs. Uses FalkorDB (Redis-based graph).
Three-Tier Architecture:
1. Episodic: Raw conversation episodes with timestamps
2. Semantic: Extracted entities and relationships (nodes/edges)
3. Community: Clustered summaries for efficient retrieval
FalkorDB: Redis-based graph database with Cypher support. In-memory for speed, persistent to disk.
Temporal Edges: Relationships with valid_from, valid_to, and created_at timestamps.
Episode-to-Entity Pipeline: Episodes → Entity extraction → Entity resolution → Graph update.

2.2 Why This Matters

Graphiti represents state-of-the-art in AI memory systems:

Handles conversation → knowledge graph automatically
Temporal reasoning built-in
Community summaries for efficient retrieval
Battle-tested in production at Zep

2.3 Common Misconceptions

“It’s just a wrapper around Neo4j.” Graphiti has its own architecture with FalkorDB.
“Install and it works.” Requires careful integration and tuning.
“Replaces your custom code.” You integrate your extractors/resolvers with Graphiti.

2.4 ASCII Diagram: Graphiti Architecture

GRAPHITI THREE-TIER ARCHITECTURE
════════════════════════════════════════════════════════════════

INPUT
─────
│ Episode │ → Raw conversation with timestamp
│         │   {"messages": [...], "timestamp": "2024-12-15"}
└─────────┘

          │
          ▼
┌─────────────────────────────────────────────────────────────┐
│                    GRAPHITI CORE                             │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌─────────────────────────────────────────────────────┐    │
│  │              EPISODE PROCESSOR                       │    │
│  │                                                      │    │
│  │  1. Store raw episode                               │    │
│  │  2. Extract entities (LLM)                          │    │
│  │  3. Extract relationships (LLM)                     │    │
│  │  4. Resolve entities (dedup)                        │    │
│  │  5. Update graph with temporal edges                │    │
│  └─────────────────────────────────────────────────────┘    │
│                          │                                   │
│                          ▼                                   │
│  ┌─────────────────────────────────────────────────────┐    │
│  │                   GRAPH UPDATER                      │    │
│  │                                                      │    │
│  │  • Create/update nodes (entities)                   │    │
│  │  • Create edges with temporal properties            │    │
│  │  • Handle edge invalidation (valid_to)              │    │
│  │  • Maintain provenance (episode → entities)         │    │
│  └─────────────────────────────────────────────────────┘    │
│                          │                                   │
│                          ▼                                   │
│  ┌─────────────────────────────────────────────────────┐    │
│  │              COMMUNITY BUILDER                       │    │
│  │                                                      │    │
│  │  • Run Leiden clustering periodically               │    │
│  │  • Generate community summaries                     │    │
│  │  • Build hierarchy for multi-level access           │    │
│  └─────────────────────────────────────────────────────┘    │
│                                                              │
└─────────────────────────────────────────────────────────────┘
          │
          ▼
┌─────────────────────────────────────────────────────────────┐
│                      FALKORDB                                │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  EPISODIC LAYER                                              │
│  ───────────────                                             │
│  (Episode {id, timestamp, content, embedding})               │
│                                                              │
│  SEMANTIC LAYER                                              │
│  ──────────────                                              │
│  (Entity {name, type, properties, embedding})                │
│       │                                                      │
│  ─[RELATIONSHIP {valid_from, valid_to, episode_id}]─▶       │
│       │                                                      │
│  (Entity)                                                    │
│                                                              │
│  COMMUNITY LAYER                                             │
│  ───────────────                                             │
│  (Community {id, level, summary, embedding})                 │
│       │                                                      │
│  ─[CONTAINS]─▶ (Entity)                                     │
│                                                              │
└─────────────────────────────────────────────────────────────┘


TEMPORAL EDGE EXAMPLE
═════════════════════

Episode at T1: "Alice works at Acme"
┌─────────┐                      ┌─────────┐
│  Alice  │──[WORKS_AT]────────▶│  Acme   │
└─────────┘   valid_from: T1     └─────────┘
              valid_to: NULL
              episode: ep_001

Episode at T2: "Alice left Acme, joined TechCo"
┌─────────┐                      ┌─────────┐
│  Alice  │──[WORKS_AT]────────▶│  Acme   │
└─────────┘   valid_from: T1     └─────────┘
    │         valid_to: T2  ◄── INVALIDATED
    │
    └────────[WORKS_AT]────────▶┌─────────┐
              valid_from: T2     │ TechCo  │
              valid_to: NULL     └─────────┘
              episode: ep_002

3. Project Specification

3.1 What You Will Build

A complete integration with Graphiti that:

Sets up FalkorDB storage
Processes episodes through the Graphiti pipeline
Builds community summaries
Provides temporal retrieval APIs

3.2 Functional Requirements

Initialize Graphiti: graph = Graphiti(driver, llm_client)
Add episode: graph.add_episode(messages, timestamp) → ProcessedEpisode
Query current: graph.search(query) → List[Entity]
Query temporal: graph.search(query, at_time=datetime) → List[Entity]
Get entity history: graph.get_history(entity_id) → List[Edge]
Rebuild communities: graph.rebuild_communities() → None
Query communities: graph.search_communities(query) → List[Community]

3.3 Example Usage / Output

from graphiti_core import Graphiti
from datetime import datetime

# Initialize
graphiti = Graphiti(
    uri="bolt://localhost:6379",
    user="default",
    password="",
    llm_client=llm_client
)

# Add episode
episode = await graphiti.add_episode(
    name="project_discussion",
    episode_body="Alice mentioned she's been working on the API redesign with Bob. They're using FastAPI and planning to launch in Q1.",
    source_description="team_meeting",
    reference_time=datetime(2024, 12, 15, 10, 0, 0)
)

print(f"Processed episode: {episode.uuid}")
print(f"Extracted entities: {[e.name for e in episode.entities]}")
# Extracted entities: ['Alice', 'Bob', 'API redesign', 'FastAPI', 'Q1']
print(f"Created edges: {len(episode.edges)}")
# Created edges: 4

# Search current state
results = await graphiti.search("Who is working on API projects?")
for node in results.nodes:
    print(f"  {node.name}: {node.summary}")
# Alice: Engineer working on API redesign
# Bob: Engineer collaborating with Alice on API redesign

# Search at specific time
past_results = await graphiti.search(
    "Who is working on API projects?",
    at_time=datetime(2024, 11, 1)  # Before this episode
)
# Returns empty or different results based on past state

# Get entity timeline
history = await graphiti.get_entity_history("alice_uuid")
for edge in history:
    print(f"{edge.valid_from} - {edge.valid_to}: {edge.relation} → {edge.target}")
# 2024-12-15 - None: WORKS_ON → API redesign
# 2024-06-01 - 2024-12-01: WORKS_ON → Legacy System

# Search via community summaries
communities = await graphiti.search_communities("API development team")
for comm in communities:
    print(f"Community: {comm.summary[:100]}...")

4. Solution Architecture

4.1 High-Level Design

┌────────────────┐     ┌────────────────┐     ┌────────────────┐
│  Your App      │────▶│   Graphiti     │────▶│   FalkorDB     │
│                │     │   Core         │     │                │
└────────────────┘     └────────────────┘     └────────────────┘
                              │
                       ┌──────┴──────┐
                       │             │
                       ▼             ▼
                ┌───────────┐ ┌───────────┐
                │  LLM      │ │ Embedding │
                │  Client   │ │  Model    │
                └───────────┘ └───────────┘

4.2 Key Components

Component	Responsibility	Technology
Graphiti Core	Episode processing orchestration	graphiti-core library
FalkorDB	Graph storage with Cypher	Docker + Redis
LLM Client	Entity/relationship extraction	OpenAI/Anthropic
Embedder	Vector embeddings for search	sentence-transformers
Community Builder	Leiden + summarization	Built into Graphiti

4.3 Data Model (Graphiti Native)

# Graphiti's internal models (for reference)

class EpisodeNode:
    uuid: str
    name: str
    content: str
    source_description: str
    created_at: datetime
    valid_at: datetime  # When episode occurred
    embedding: list[float]

class EntityNode:
    uuid: str
    name: str
    entity_type: str
    summary: str
    created_at: datetime
    embedding: list[float]

class EntityEdge:
    uuid: str
    source_node_uuid: str
    target_node_uuid: str
    relation: str  # e.g., "WORKS_ON", "KNOWS"
    fact: str  # Natural language fact
    valid_from: datetime
    valid_to: datetime | None  # NULL = currently valid
    created_at: datetime
    episode_uuid: str  # Provenance
    embedding: list[float]

5. Implementation Guide

5.1 Development Environment Setup

# Create project
mkdir graphiti-integration && cd graphiti-integration
python -m venv .venv && source .venv/bin/activate

# Install Graphiti
pip install graphiti-core

# Start FalkorDB
docker run -d --name falkordb -p 6379:6379 falkordb/falkordb:latest

# Verify connection
python -c "from falkordb import FalkorDB; db = FalkorDB(); print(db.list_graphs())"

5.2 Project Structure

graphiti-integration/
├── src/
│   ├── config.py          # FalkorDB and LLM config
│   ├── client.py          # Graphiti wrapper
│   ├── processors.py      # Custom extractors (optional)
│   ├── retrieval.py       # Search utilities
│   └── cli.py             # CLI interface
├── tests/
│   ├── test_ingestion.py
│   └── test_retrieval.py
├── docker-compose.yml     # FalkorDB + app
└── README.md

5.3 Implementation Phases

Phase 1: Basic Setup (8-10h)

Goals:

FalkorDB running and accessible
Graphiti initialized and connected
First episode processed

Tasks:

Set up Docker Compose for FalkorDB
Configure Graphiti with LLM client
Process test episode
Verify graph contents in FalkorDB

Checkpoint: Episode creates entities and edges in graph.

Phase 2: Search and Retrieval (10-12h)

Goals:

Semantic search working
Temporal queries working
Entity history retrieval

Tasks:

Test Graphiti’s built-in search
Implement temporal filtering
Build entity history view
Add CLI for exploration

Checkpoint: Can query “what did Alice work on in Q3?”

Phase 3: Communities and Production (10-12h)

Goals:

Community detection running
Summaries generated
Production-ready setup

Tasks:

Trigger community rebuild
Test community-based search
Add error handling and retries
Document configuration options

Checkpoint: Full three-tier architecture operational.

6. Testing Strategy

6.1 Test Categories

Category	Purpose	Examples
Unit	Test configuration	FalkorDB connection
Integration	Test episode flow	Episode → entities → edges
E2E	Test full retrieval	Question → answer with provenance

6.2 Critical Test Cases

Entity extraction: Episode mentions entities that get created
Edge temporality: New info invalidates old edges
Search accuracy: Query finds relevant entities
Community coverage: All entities belong to communities

7. Common Pitfalls & Debugging

Pitfall	Symptom	Solution
FalkorDB not running	Connection refused	Check Docker container status
LLM extraction fails	No entities created	Check LLM API key and model
Duplicate entities	Same person appears twice	Check entity resolution config
Communities empty	No summaries	Trigger manual rebuild

Debugging Tools:

# Connect to FalkorDB CLI
docker exec -it falkordb redis-cli

# List graphs
GRAPH.LIST

# Query graph
GRAPH.QUERY graphiti "MATCH (n) RETURN n LIMIT 10"

8. Extensions & Challenges

8.1 Beginner Extensions

Add custom entity types
Build admin dashboard

8.2 Intermediate Extensions

Add group conversation support
Implement custom extractors

8.3 Advanced Extensions

Add streaming episode processing
Implement cross-graph federation

9. Real-World Connections

9.1 Industry Applications

Zep Cloud: Production Graphiti deployment
AI Agents: Long-term memory for assistants
Enterprise Knowledge: Company-wide memory graphs

9.2 Interview Relevance

Explain three-tier memory architecture
Discuss temporal edge invalidation
Describe community-based retrieval

10. Resources

10.1 Essential Reading

Graphiti Documentation — https://github.com/getzep/graphiti
FalkorDB Documentation — https://docs.falkordb.com
Zep Blog Posts — Architecture deep dives

Previous: Project 8 (Community Detection)
Next: Project 10 (Mem0g Memory Layer)

11. Self-Assessment Checklist

I can set up FalkorDB with Docker
I understand Graphiti’s three-tier architecture
I can process episodes and verify graph state
I know how temporal edges are invalidated

12. Submission / Completion Criteria

Minimum Viable Completion:

FalkorDB running
Episodes processing successfully
Basic search working

Full Completion:

Temporal queries working
Community summaries generated
CLI for exploration

Excellence:

Custom extractors integrated
Production error handling
Performance benchmarks

Project 9: Graphiti Framework Integration

Quick Reference

1. Learning Objectives

2. Theoretical Foundation

2.1 Core Concepts

2.2 Why This Matters

2.3 Common Misconceptions

2.4 ASCII Diagram: Graphiti Architecture

3. Project Specification

3.1 What You Will Build

3.2 Functional Requirements

3.3 Example Usage / Output

4. Solution Architecture

4.1 High-Level Design

4.2 Key Components

4.3 Data Model (Graphiti Native)

5. Implementation Guide

5.1 Development Environment Setup

5.2 Project Structure

5.3 Implementation Phases

Phase 1: Basic Setup (8-10h)

Phase 2: Search and Retrieval (10-12h)

Phase 3: Communities and Production (10-12h)

6. Testing Strategy

6.1 Test Categories

6.2 Critical Test Cases

7. Common Pitfalls & Debugging

8. Extensions & Challenges

8.1 Beginner Extensions

8.2 Intermediate Extensions

8.3 Advanced Extensions

9. Real-World Connections

9.1 Industry Applications

9.2 Interview Relevance

10. Resources

10.1 Essential Reading

10.2 Related Projects

11. Self-Assessment Checklist

12. Submission / Completion Criteria