Graphiti add_episode: Parameter, Indizes und die Fehler aus meinen Graphen

add_episode ist der Schreibaufruf von Graphiti. Er nimmt eine Episode, also einen Text, ein JSON-Dokument oder einige Chatzeilen, lässt das Sprachmodell Entitäten und Fakten extrahieren, gleicht sie mit dem Bestand im Graphen ab und speichert jeden Fakt mit dem Zeitpunkt, ab dem er gilt. Vor dem ersten Aufruf gehört einmal build_indices_and_constraints().

Zuletzt geprüft: September 2026, am Quellcode von graphiti-core 0.30.2 und an den Graphen, die ich auf FalkorDB und auf einer eingebetteten LadybugDB betreibe.

add_episode auf einen Blick

Die empfohlene Einrichtung: ein Treiber pro Gruppe, Indizes zuerst

Die Einrichtung, die sich in meinen Graphen bewährt hat: den Treiber für die Gruppe anlegen, in die Sie schreiben, die Indizes einmal aufbauen und dann add_episode für jede Episode aufrufen.

import asyncio
from datetime import datetime, timezone

from graphiti_core import Graphiti
from graphiti_core.driver.falkordb_driver import FalkorDriver
from graphiti_core.nodes import EpisodeType


async def main():
    driver = FalkorDriver(host="localhost", port=6379, database="acme")
    # Graphiti nutzt standardmäßig OpenAI; für alles andere llm_client und embedder übergeben.
    graphiti = Graphiti(graph_driver=driver)
    await graphiti.build_indices_and_constraints()

    result = await graphiti.add_episode(
        name="ticket-4711",
        episode_body="Acme hat den Go-live vom 1. Oktober auf den 15. Oktober verschoben.",
        source=EpisodeType.text,
        source_description="Support-Ticket",
        reference_time=datetime(2026, 9, 17, 9, 30, tzinfo=timezone.utc),
        group_id="acme",
    )
    print(len(result.nodes), "Entitäten,", len(result.edges), "Fakten")

    await graphiti.close()


asyncio.run(main())

Tragen die database des Treibers und die group_id denselben Namen, bleiben Schreiben und Suchen in einem Graphen. Was passiert, wenn sie auseinanderlaufen, beschreiben die nächsten Abschnitte.

build_indices_and_constraints: einmal, vor dem ersten add_episode

build_indices_and_constraints() legt die Range-Indizes an und die vier Volltext-Indizes, auf denen die Suche läuft: node_name_and_summary, edge_name_and_fact, episode_content und community_name.

Rufen Sie es selbst auf. Der FalkorDB-Treiber plant den Aufbau nur dann selbst ein, wenn er in einer laufenden Event-Loop entsteht. Ein Treiber, der auf Modulebene vor asyncio.run angelegt wird, baut also nichts. Und wenn add_episode zum ersten Mal in eine neue Gruppe schreibt, startet der Treiber, den es für den Graphen dieser Gruppe anlegt, seinen eigenen Indexaufbau im Hintergrund, gleichzeitig mit Ihren ersten Schreibvorgängen. Ein abgewarteter Aufruf pro Graph beim Start deckt beide Fälle ab.

Ihn zu wiederholen schadet nicht: Auf FalkorDB wird ein vorhandener Index protokolliert und übersprungen, und meine Ingest-Skripte rufen ihn zu Beginn jedes Laufs auf. delete_existing=True löscht die Indizes und baut sie neu auf, was bei einem großen Graphen dauert.

Die Parameter von add_episode, die zählen

add_episode gibt ein AddEpisodeResults zurück, mit episode, nodes (Entitäten), edges (Fakten), episodic_edges, communities und community_edges.

EpisodeType: text, json oder message

Jeder Typ hat einen eigenen Extraktions-Prompt, die Wahl verändert also, was das Modell herauszieht.

import json
from datetime import datetime, timezone

from graphiti_core.nodes import EpisodeType

# Chat- oder Meetingzeilen: eine Zeile "Sprecher: Text" pro Beitrag.
await graphiti.add_episode(
    name="standup-2026-09-17",
    episode_body="Anna: Der Acme-Import hängt am SSO.\nBen: Ich frage heute bei deren IT nach.",
    source=EpisodeType.message,
    source_description="Team-Chat",
    reference_time=datetime.now(timezone.utc),
    group_id="acme",
)

# Ein Datensatz aus einem anderen System: ein JSON-String, kein dict.
await graphiti.add_episode(
    name="crm-acme",
    episode_body=json.dumps({"account": "Acme", "stage": "pilot", "owner": "Anna"}),
    source=EpisodeType.json,
    source_description="CRM-Kundendatensatz",
    reference_time=datetime.now(timezone.utc),
    group_id="acme",
)

message ist voreingestellt. Ein Dokument ohne source= läuft deshalb durch den Chat-Prompt, der nach Sprechern sucht. Setzen Sie EpisodeType.text für Dokumente, Tickets und Wikiseiten.

group_id auf FalkorDB: Geschrieben wird in den Graphen der Gruppe, gesucht nicht

Auf Neo4j ist group_id eine Eigenschaft innerhalb einer Datenbank, und search(..., group_ids=["acme"]) filtert darauf. Auf FalkorDB ist sie ein eigener Graph: add_episode richtet eine Kopie des Treibers auf einen Graphen aus, der nach der Gruppe benannt ist. Ohne group_id landen Schreibvorgänge auf FalkorDB in der Gruppe _ von default_db.

Graphiti.search liest über den Treiber der Instanz. Schreiben Sie die Gruppe acme aus einer Instanz, deren Treiber auf default_db zeigt, kommt eine Suche nach acme auf dieser Instanz leer zurück, ohne Fehlermeldung. Zwei Auswege: den Treiber mit database="acme" anlegen, wie in der Einrichtung oben, oder über einen Treiber suchen, der auf die Gruppe geklont ist.

acme = graphiti.driver.clone(database="acme")
edges = await graphiti.search("Wann geht Acme live?", group_ids=["acme"], driver=acme)
for edge in edges:
    print(edge.fact, edge.valid_at, edge.invalid_at)

Meine eigene kg-Kommandozeile geht den zweiten Weg: eine gemeinsame Graphiti-Instanz und für jeden Lesezugriff ein auf die Domäne geklonter Treiber.

add_episode parallel: was gehalten hat und was brach

Mein erstes Ingest-Skript wartete eine Episode nach der anderen ab: etwa 2 Minuten pro Eintrag, knapp 7 Stunden für 200. Dieselben Einträge mit je einem add_episode-Aufruf hinter einem Semaphor brauchten etwa 10 Sekunden pro Eintrag und insgesamt rund 35 Minuten.

sem = asyncio.Semaphore(8)


async def ingest(item):
    async with sem:
        return await graphiti.add_episode(**item)


results = await asyncio.gather(*(ingest(item) for item in items))

Ich ließ 14 gleichzeitig laufen, passend zu den 14 Modell-Workern dahinter. Richten Sie die Zahl nach dem Ratenlimit Ihres Modells. Drei Grenzen gelten:

• Verschiedene Gruppen gleichzeitig brauchen graphiti-core 0.30.2 oder neuer. Bis 0.30.1 stellte add_episode den gemeinsamen Treiber auf den Graphen der Gruppe um. Ein paralleler Aufruf für eine andere Gruppe konnte seine Schreibvorgänge so im falschen Graphen ablegen, ohne Fehlermeldung (Issue 1676). Ab 0.30.2 bekommt jeder Aufruf einen eigenen Treiber. Auf älteren Versionen gilt: eine Gruppe pro Prozess.
• Eine eingebettete LadybugDB nimmt einen Schreiber. Auf LadybugDB, die ich über Graphitis Kuzu-Treiber betreibe, scheitert ein zweiter schreibender Prozess mit IO exception: Could not set lock on file und ein zweiter Schreiber im selben Prozess mit Cannot start a new write transaction in the system. Only one write transaction at a time. Lassen Sie die Erzeuger in eine Warteschlange schreiben und einen einzigen Anwender die Einträge übernehmen.
• SEMAPHORE_LIMIT ist ein eigener Regler. Er begrenzt die parallelen Modell- und Datenbankaufrufe innerhalb jedes add_episode: voreingestellt 20 in graphiti-core und 10 im offiziellen MCP-Server. Senken Sie ihn, sobald Ihr Anbieter 429-Fehler zurückgibt.

add_episode_bulk: der Dedup-Schritt, der brach

add_episode_bulk nimmt eine Liste von RawEpisode-Objekten und löst Dubletten über den ganzen Stapel in weniger, größeren Modellaufrufen auf. In meinen Graphen lieferte dieser Dedup-Prompt mit einem kleinen Modell das JSON-Schema statt der Daten, und der Lauf brach beim ersten Dedup-Aufruf mit einem pydantic ValidationError ab: NodeResolutions.entity_resolutions fehlte.

Einzelne add_episode-Aufrufe hinter einem Semaphor gleichen jede Episode für sich mit dem Graphen ab und laufen seitdem sauber. Den Stapelaufruf sollten Sie nur mit einem Modell versuchen, das strukturierte Ausgaben zuverlässig füllt, und zuerst an einer Stichprobe.

Fehler aus meinen Graphen und wie ich sie behoben habe

Die FalkorDB-Einstellungen aus den beiden Zeilen oben, so wie sie auf meinem Server laufen:

services:
  falkordb:
    image: falkordb/falkordb:latest
    environment:
      FALKORDB_ARGS: "TIMEOUT 600000"
    volumes:
      - falkordb_data:/var/lib/falkordb/data

volumes:
  falkordb_data:

Nachweisen, dass jede Episode angekommen ist

add_episode gibt zurück, was es geschrieben hat. Protokollieren Sie len(result.nodes) und len(result.edges) für jede Episode und führen Sie einen Cursor über das, was hineingegangen ist. Ein abgebrochener Lauf setzt dann dort fort, wo er stand, statt dieselben Episoden doppelt zu schreiben. Mein Ingest schreibt seinen Cursor alle 10 Episoden.

Die Grundidee dahinter beschreibt der Leitfaden zum temporalen Wissensgraphen. Die Entscheidungen für den Produktivbetrieb, also Backends, Modelle, Embeddings und Versionen, beschreibe ich auf Englisch in Graphiti in production. Graphiti Local kapselt die Leseseite in sechs reine Lese-Werkzeuge über MCP und eine kg-Kommandozeile, mit einem Schreibpfad, der auf eine Person wartet. Eine Kontextschicht dieser Art auf Ihrem eigenen Tenant beschreibt die Knowledge-Graph-Beratung.

Graphiti Local ist ein unabhängiges Community-Projekt auf Basis von Graphiti. Es steht in keiner Verbindung zu Zep und wird von Zep weder unterstützt noch empfohlen. Versionen und Voreinstellungen mit Stand 18. September 2026.

Änderungen

• 18. September 2026: Erste Fassung, geprüft am Quellcode von graphiti-core 0.30.2.