Zep Cloud (Lab 04) is Graphiti plus hosting plus a wrapper API. The graph engine itself is open source. Run it against your own Neo4j and you keep every byte of memory in-house, with the same valid_at / invalid_at semantics. The trade is you operate Neo4j. When this finishes you have a local Graphiti instance, two facts where one supersedes the other, and a query that distinguishes "currently true" from "was true last Tuesday".
You outgrew the free tier, OR your data cannot leave your network (regulated industries, customer PII rules, on-prem only), OR you want to write your own query layer over the graph instead of going through Zep's API. The semantics are identical to Lab 04 because Zep Cloud is Graphiti underneath. The package is graphiti-core on PyPI, MIT-licensed, maintained by the Zep team.
Graphiti needs a graph DB. Neo4j 5.x is the documented backend. Single docker run. Bolt on 7687, browser UI on 7474 if you want to poke at the graph visually after.
docker run -d --name vcn33-neo4j \ -p 7474:7474 -p 7687:7687 \ -e NEO4J_AUTH=neo4j/vcn33password \ -e NEO4J_PLUGINS='["apoc"]' \ neo4j:5
Wait ~10 seconds for the DB to boot. Sanity check:
docker logs vcn33-neo4j 2>&1 | grep -i "started" # expected: a line containing "Started." indicating Neo4j is ready
Open http://localhost:7474 in a browser if you want the visual confirmation. Username neo4j, password vcn33password.
One pip line. Graphiti calls out to an LLM for entity / edge extraction; OpenAI is the default.
pip install graphiti-core
# macOS / Linux export OPENAI_API_KEY="sk-..." export NEO4J_URI="bolt://localhost:7687" export NEO4J_USER="neo4j" export NEO4J_PASSWORD="vcn33password" # Windows PowerShell $env:OPENAI_API_KEY = "sk-..." $env:NEO4J_URI = "bolt://localhost:7687" $env:NEO4J_USER = "neo4j" $env:NEO4J_PASSWORD = "vcn33password"
Same pattern as Lab 04: plant npm, sleep, plant pnpm. Watch the graph supersede the older edge.
import asyncio
import os
import sys
from datetime import datetime, timezone
sys.stdout.reconfigure(encoding="utf-8")
from graphiti_core import Graphiti
from graphiti_core.nodes import EpisodeType
async def main() -> None:
g = Graphiti(
uri=os.environ["NEO4J_URI"],
user=os.environ["NEO4J_USER"],
password=os.environ["NEO4J_PASSWORD"],
)
# one-time: builds the schema indexes if missing.
await g.build_indices_and_constraints()
group = "vcn33-lab08"
# week 1: npm
await g.add_episode(
name="week-1-pm",
episode_body="Rayyan uses npm for everything.",
source=EpisodeType.message,
source_description="user statement",
reference_time=datetime(2026, 5, 4, tzinfo=timezone.utc),
group_id=group,
)
# week 2: switched
await g.add_episode(
name="week-2-pm",
episode_body="Rayyan just switched from npm to pnpm.",
source=EpisodeType.message,
source_description="user statement",
reference_time=datetime(2026, 5, 11, tzinfo=timezone.utc),
group_id=group,
)
print("planted 2 episodes; extraction runs synchronously in this call")
# query: which package manager does the user use?
results = await g.search(
query="what package manager does the user use?",
group_ids=[group],
num_results=5,
)
print("\n--- SEARCH RESULTS ---")
for r in results:
# results are typed; field names mirror Lab 04.
print(f" fact: {getattr(r, 'fact', r)}")
print(f" valid_at: {getattr(r, 'valid_at', None)}")
print(f" invalid_at: {getattr(r, 'invalid_at', None)}")
print()
await g.close()
if __name__ == "__main__":
asyncio.run(main())
Run it:
python graphiti_demo.py
planted 2 episodes; extraction runs synchronously in this call
--- SEARCH RESULTS ---
fact: Rayyan uses pnpm
valid_at: 2026-05-11 00:00:00+00:00
invalid_at: None
fact: Rayyan uses npm
valid_at: 2026-05-04 00:00:00+00:00
invalid_at: 2026-05-11 00:00:00+00:00
The real win: ask "what did the user use last week?" and the graph filters by valid_at <= T < invalid_at. Append this to the script and rerun (or wrap as a second async function).
# ... inside main(), after the first search() block:
from datetime import datetime, timezone
as_of = datetime(2026, 5, 7, tzinfo=timezone.utc) # mid-week, before the switch
results = await g.search(
query="what package manager does the user use?",
group_ids=[group],
num_results=5,
)
print(f"\n--- AS OF {as_of.isoformat()} ---")
for r in results:
v_at = getattr(r, "valid_at", None)
inv_at = getattr(r, "invalid_at", None)
# client-side temporal filter; some Graphiti versions expose this
# on .search() directly as a kwarg. Either path is equivalent.
if v_at is None or v_at > as_of:
continue
if inv_at is not None and as_of >= inv_at:
continue
print(f" active on that date: {getattr(r, 'fact', r)}")
--- AS OF 2026-05-07T00:00:00+00:00 --- active on that date: Rayyan uses npm
You ran the same graph engine that backs Zep Cloud, on hardware you control, with the same temporal semantics. Two real costs: Neo4j is a heavier dependency than "import a Python package" (it is a JVM process eating ~700MB), and the entity extraction calls cost OpenAI tokens. Zep Cloud bundles both away in exchange for cents per write and lock-in.
When you cleanup, stop and remove the container:
docker stop vcn33-neo4j && docker rm vcn33-neo4j
Graphiti's defaults assume OpenAI but the LLM client is pluggable. You can wire Claude (Anthropic), Gemini, or a local Ollama model by passing a custom client. The graph engine itself does not care which model extracts the entities; it only cares that the extracted JSON matches the expected schema. For local-only deployments this is the path to zero cloud dependencies (Neo4j local + Ollama local + Graphiti local).
Neo4j container exits immediately. Port 7687 is already taken (often by another local Neo4j or a stale container). docker ps -a to find it; remove the old one, or remap to a different host port (-p 7688:7687) and update NEO4J_URI to match.
"ServiceUnavailable: Failed to establish connection". Neo4j is still warming up. Wait 15-20s after docker run and retry. docker logs vcn33-neo4j shows a "Started." line when it is ready.
First call hangs for ~30s. Graphiti runs entity extraction inline on add_episode(). The first call also compiles indexes if missing. After that, writes settle in a couple of seconds.
Field names differ on output. The Graphiti SDK has churned a bit; .fact / .valid_at / .invalid_at are the canonical EntityEdge fields but the result object wrapping them has rotated names ("results", "edges", "matches"). The code above uses getattr defensively to survive minor SDK drift.