Pregel-Algorithmusschicht: Interrupt, Zurückschreiben, Dispatch

源码版本1.2.9

Verantwortung

_algo.py ist der «mathematische Kern» von PregelLoop und PregelRunner. Die Loop ist dafür zuständig, «wie viele Runden zu drehen sind»; der Runner dafür, «wie Tasks nebenläufig ausgeführt werden»; und algo übernimmt «die Semantik jedes Schritts»: wann ein Interrupt nötig ist, wie die von Knoten zurückgeschriebenen Writes in Kanäle (channel) eingefügt werden, welche Knoten im nächsten Schritt auszulösen sind. Es hat keinen Zustand, kein IO — ausschließlich reine Funktionen, die als Eingabe Checkpoint + Kanäle + Tasks nehmen und als Ausgabe eine neue Kanalversionstabelle / eine updated_channels-Menge / ein neues Task-Dictionary liefern.

Die drei Kernfunktionen entsprechen den drei Aufrufabschnitten von PregelLoop.tick + after_tick: should_interrupt(should_interrupt:155) wird im mittleren tick-Teil aufgerufen und entscheidet, ob vor der Ausführung ein GraphInterrupt geworfen wird; apply_writes(apply_writes:232) wird am Anfang von after_tick aufgerufen, fügt die Writes aller Tasks dieses Schritts in die Kanäle ein und gibt updated_channels zurück; prepare_next_tasks(prepare_next_tasks:392) wird am Anfang von tick aufgerufen und berechnet anhand des aktuellen Checkpoints, welche Tasks in diesem Schritt auszuführen sind.

Anders gesagt: Diese drei Funktionen entscheiden, «was Pregel in jedem Schritt der BSP-Schleife sieht, tut und wie die Kanäle danach aussehen». Wer sie verstanden hat, hat die «Zustandsänderungsregeln» von Pregel verstanden.

Entwurfsmotivation

Warum die Algorithmuslogik als reine Funktionen aufteilen?

• Testbarkeit: should_interrupt / apply_writes / prepare_next_tasks sind reine Funktionen; Eingaben sind ein Checkpoint-Dict plus ein Channel-Mapping, Ausgaben Datenstrukturen. Sie lassen sich ohne Pregels IO, Submit und Checkpointer isoliert unittesten — genau das ist die Form vieler Tests in libs/langgraph/tests/unit/.
• Gemeinsam für Sync/Async: PregelLoop hat die Unterklassen SyncPregelLoop und AsyncPregelLoop, aber die algo-Schicht ist komplett nebenwirkungsfrei; beide Unterklassen teilen sich dieselbe Logik(tick calls prepare_next_tasks:612). Andernfalls müsste man die Algorithmen jeweils für Sync und Async pflegen und Bugs zweimal reparieren.
• updated_channels-Optimierung: apply_writes gibt die updated_channels-Menge zurück; prepare_next_tasks empfängt sie als Hint und überspringt «alle Knoten scannen, um Trigger zu prüfen»(updated_channels hint:475-486). Bei großen Graphen senkt dieser Schritt das O(N×M)-Trigger-Scannen auf O(updated×triggered) und ist die wichtigste Performance-Optimierung.
• should_interrupt nutzt versions_seen für Dedup: versions_seen des INTERRUPT-Kanals(versions_seen INTERRUPT:163) notiert die Kanalversionsnummern, die beim letzten Interrupt gesehen wurden. Nur wenn es «seit dem letzten Interrupt neue Kanal-Updates gab», wird ein erneuter Interrupt erwogen — das vermeitet die Endlosschleife aus wiederholt ausgelöstem Interrupt, dann Resume, dann erneutem Interrupt.
• Finish-Signal am Ende von apply_writes: Wenn bump_step and updated_channels.isdisjoint(trigger_to_nodes) gilt, wird channels[chan].finish() aufgerufen(finish:336-342) und allen Kanälen signalisiert, «dies ist der letzte Superstep». Permanente Kanäle (wie LastValue) nutzen diese Gelegenheit, sich als nicht verfügbar zu markieren, sodass prepare_next_tasks keinen Knoten mehr auslöst — das ist der «natürliche Tod» des Graphen.

Schlüsseldateien

• should_interrupt:155-185 — Prüft, ob der Graph unterbrochen werden soll; schaut zuerst auf versions_seen[INTERRUPT], um zu beurteilen, ob es neue Updates gibt, dann ob die getriggerten Tasks in der interrupt_nodes-Liste stehen.
• any_updates_since_prev_interrupt:161-168 — Vergleicht Kanalversionsnummern, um zu beurteilen, «ob es seit dem letzten Interrupt Updates gab»; das ist der Kern der Interrupt-Dedup.
• interrupt_nodes filter:170-184 — interrupt_nodes == "*" bedeutet, dass alle nicht-hidden Tasks zutreffen; sonst wird nach task.name in interrupt_nodes gefiltert.
• apply_writes:232-345 — Fügt die Writes der Tasks dieses Schritts in die Kanäle ein und gibt die updated_channels-Menge zurück.
• bump_step:256-259 — Wenn ein Task triggers hat, gilt bump_step=True; das bedeutet, dass dieser Schritt Kanäle verbraucht und den Step-Zähler vorantreibt.
• update seen versions:262-269 — Jeder Task notiert die Versionsnummern seiner gelesenen Trigger-Kanäle in versions_seen; beim nächsten _triggers-Check heißt das «gab es seit der letzten Ausführung ein neues Update».
• consume channels:284-292 — Ruft channels[chan].consume() auf, um die in diesem Schritt gelesenen Kanäle als konsumiert zu markieren; das ist die Implementierung der PULL-Kanal-Semantik «einmal lesen dann leeren».
• apply writes to channels:315-323 — channels[chan].update(vals) fügt die Writes tatsächlich in den Kanal ein; nur is_available()-Kanäle kommen in updated_channels.
• bump_step notify:326-333 — Bei bump_step=True wird auch Kanälen, die in diesem Schritt nicht aktualisiert wurden, ein update(EMPTY_SEQ) gegeben, um ihre Versionsnummern voranzutreiben — Knoten, die diese Kanäle abonnieren aber in diesem Schritt nicht getriggert wurden, werden beim nächsten tick nicht fälschlich ausgelöst.
• finish signal:336-342 — Wenn kein Knoten durch die Updates dieses Schritts getriggert wird, wird an alle Kanäle ein finish()-Signal gesendet; permanente Kanäle werden dadurch unavailable, der Graph endet natürlich.
• prepare_next_tasks:392-513 — Berechnet, welche Tasks im nächsten Schritt laufen; PUSH (aus Send-Fan-out) und PULL (aus Kanten-Triggern) kommen hier zusammen.
• updated_channels optimization:475-486 — Nutzt updated_channels + trigger_to_nodes des vorherigen Schritts, um die Menge der getriggerten Knoten rückzufragen und den Full-Table-Scan zu überspringen.
• prepare_single_task:524 — Konstruktion eines einzelnen Tasks; PUSH geht über prepare_push_task_functional / prepare_push_task_send, PULL über _triggers-Prüfung + _proc_input für die Eingabe.
• _triggers:1260-1277 — Prüft, ob der Trigger-Kanal eines Knotens «eine ungelesene neue Version hat»; das ist die Kernbedingung der PULL-Dispatch.
• local_read:188-224 — Eingang, über den Knotenfunktionen Kanalwerte lesen; unterstützt fresh=True, um die lokale Sicht «gerade selbst geschrieben, aber noch nicht in den Kanal eingefügt» zu lesen — wird für die Write-then-Read-Semantik der bedingten Kante (conditional edge) verwendet.

Datenfluss

should_interrupt ist die einfachste der drei; die Logik lautet «neues Update + aktueller Task trifft Liste»:

def should_interrupt(
    checkpoint: Checkpoint,
    interrupt_nodes: All | Sequence[str],
    tasks: Iterable[PregelExecutableTask],
) -> list[PregelExecutableTask]:
    """Check if the graph should be interrupted based on current state."""
    version_type = type(next(iter(checkpoint["channel_versions"].values()), None))
    null_version = version_type()  # type: ignore[misc]
    seen = checkpoint["versions_seen"].get(INTERRUPT, {})
    # interrupt if any channel has been updated since last interrupt
    any_updates_since_prev_interrupt = any(
        version > seen.get(chan, null_version)  # type: ignore[operator]
        for chan, version in checkpoint["channel_versions"].items()
    )
    # and any triggered node is in interrupt_nodes list
    return (
        [task for task in tasks if ...]
        if any_updates_since_prev_interrupt
        else []
    )

Das stammt aus should_interrupt:155-185. Entscheidend ist der spezielle Eintrag versions_seen[INTERRUPT] — er ist nicht die Versionsnummer, die ein Knoten gelesen hat, sondern «die Momentaufnahme aller Kanalversionsnummern zum Zeitpunkt des letzten Interrupts». any_updates_since_prev_interrupt vergleicht genau diese Momentaufnahme mit den aktuellen channel_versions; sobald eine beliebige Kanalversionsnummer höher ist, gilt «es gibt etwas Neues», und erst dann darf die Interrupt-Entscheidung einsetzen. So wird nach einem Resume verhindert, dass selbst dann sofort wieder ein Interrupt kommt, wenn der nächste tick denselben Knoten auslöst — es muss erst eine echte neue Schreibung in den Kanal erfolgen.

apply_writes ist der Kern der Abschlussphase; zuerst wird versions_seen aktualisiert, dann die gelesenen Kanäle konsumiert, schließlich der neue Wert geschrieben:

# update seen versions
for task in tasks:
    checkpoint["versions_seen"].setdefault(task.name, {}).update(
        {
            chan: checkpoint["channel_versions"][chan]
            for chan in task.triggers
            if chan in checkpoint["channel_versions"]
        }
    )

# Consume all channels that were read
for chan in {
    chan
    for task in tasks
    for chan in task.triggers
    if chan not in RESERVED and chan in channels
}:
    if channels[chan].consume() and next_version is not None:
        checkpoint["channel_versions"][chan] = next_version

# Apply writes to channels
updated_channels: set[str] = set()
for chan, vals in pending_writes_by_channel.items():
    if chan in channels:
        if channels[chan].update(vals) and next_version is not None:
            checkpoint["channel_versions"][chan] = next_version
            if channels[chan].is_available():
                updated_channels.add(chan)

Das stammt aus apply_writes 核心:262-323. Die drei Abschnitte sind semantisch klar:

1. versions_seen aktualisieren — Für jeden Task wird die aktuelle Trigger-Kanal-Versionsnummer notiert, als Beleg dafür, dass «dieser Knoten diese Kanalversion gesehen hat». Beim nächsten _triggers-Check wird gegen zukünftige Versionsnummern verglichen; nur bei einem Update wird erneut getriggert.
2. consume() der gelesenen Kanäle — Kanäle vom Typ Topic, die «einmal lesen dann leeren», leeren in consume() ihren internen Buffer und behalten nur den Versionsnummern-Fortschritt. Das ist der Grund, warum nach dem Konsumieren einer per Send-Fan-out zugestellten Nachricht andere Knoten nicht erneut getriggert werden.
3. update(vals) schreibt neue Werte — Alle Writes aller Tasks dieses Schritts an einen Kanal werden gesammelt und in einem Rutsch zusammengeführt, was die Merge-Logik der Reducer / LastValue-Kanäle auslöst. Nur is_available()-Kanäle (noch Wert und nicht finish()) kommen in updated_channels — nach finish() wird die Versionsnummer des Kanals zwar fortgeschrieben, aber er triggert keinen Knoten mehr.

prepare_next_tasks wird am Anfang von tick aufgerufen und entscheidet, welche Tasks in diesem Schritt laufen:

# Consume pending tasks
tasks_channel = cast(Topic[Send] | None, channels.get(TASKS))
if tasks_channel and tasks_channel.is_available():
    for idx, _ in enumerate(tasks_channel.get()):
        if task := prepare_single_task(
            (PUSH, idx), None, ..., for_execution=for_execution, ...
        ):
            tasks.append(task)

# This section is an optimization that allows which nodes will be active
# during the next step.
if updated_channels and trigger_to_nodes:
    triggered_nodes: set[str] = set()
    for channel in updated_channels:
        if node_ids := trigger_to_nodes.get(channel):
            triggered_nodes.update(node_ids)
    candidate_nodes: Iterable[str] = sorted(triggered_nodes)
elif not checkpoint["channel_versions"]:
    candidate_nodes = ()
else:
    candidate_nodes = processes.keys()

for name in candidate_nodes:
    if task := prepare_single_task((PULL, name), None, ..., for_execution=for_execution, ...):
        tasks.append(task)
return {t.id: t for t in tasks}

Das stammt aus prepare_next_tasks 核心:441-513. Zwei Task-Typen fließen hier zusammen:

• PUSH-Task: Holt Send-Objekte aus dem TASKS-Kanal; diese wurden im vorherigen Schritt per return Send(...) fan-out, und jedes Send entspricht einem PUSH-Task.
• PULL-Task: Fragt über updated_channels und trigger_to_nodes zurück — dieses Mapping ist zur Kompilierzeit berechnet («Kanal → Liste der abonnierenden Knoten»); die Vereinigungsmenge liefert direkt die in diesem Schritt zu triggernden Kandidatenknoten. sorted sorgt für eine deterministische Task-Reihenfolge, die das Ausführungsergebnis nicht beeinflusst, aber das Checkpoint-Replay sehr wohl.

Jeder Kandidatenknoten durchläuft dann eine zweite Bestätigung über _triggers(_triggers call:606-612) — nur wenn «der Kanal eine neue Versionsnummer hat und der Knoten diese Version noch nicht gesehen hat», wird tatsächlich ein Task erzeugt. Dieser Schritt ist entscheidend für die Dedup von PULL-Knoten: Selbst wenn updated_channels den Trigger-Kanal eines Knotens enthält, wird der Task nicht neu geschedult, wenn versions_seen[name] diese Version bereits notiert hat.

Die Position der algo-Schicht innerhalb des Pregel-Zyklus:

Grenzen und Fehler

• Wenn versions_seen[INTERRUPT] nicht aktualisiert wird, entsteht eine Endlosschleife: should_interrupt gibt nur dann eine nicht-leere Liste zurück, wenn any_updates_since_prev_interrupt wahr ist; die Aktualisierung von versions_seen[INTERRUPT] geschieht in der impliziten Logik von _put_checkpoint nach apply_writes. Wenn nach einem Interrupt beim Resume die seen-Version des INTERRUPT-Kanals nicht korrekt fortgeschrieben wird, wird dasselbe Update wiederholt als «neu» eingestuft, was zu einer Interrupt-Resume-Interrupt-Endlosschleife führt(versions_seen INTERRUPT:163-168).
• bump_step ist das einzige Signal für den Step-Fortschritt: Wenn any(t.triggers for t in tasks) falsch ist, wird nicht gebumpt — das ist die Semantik «Null-Task schreibt nur Kanäle, treibt aber keinen Step voran»(bump_step:259). Null-Tasks werden verwendet, um Input-Writes vor tick zu injizieren, und dürfen den Step-Zähler nicht springen lassen.
• finish() wird nur gerufen, wenn «kein Knoten getriggert wurde»: bump_step and updated_channels.isdisjoint(trigger_to_nodes)(finish condition:336) — solange in updated_channels irgendein von einem Knoten abonnierter Kanal enthalten ist, wird nicht gefinisht. Das bedeutet: finish wird erst beim «Tod des Graphen im letzten Schritt» gesendet, nicht in jedem Schritt.
• Schreiben in unbekannte Kanäle erzeugt nur Warnung, keinen Fehler: Wenn in pending_writes_by_channel ein Task in einen Kanal schreibt, der nicht in channels existiert, wird nur logger.warning aufgerufen(unknown channel warn:310-313). Das dient der Fehlertoleranz — z. B. wenn ein Knoten in einen bereits gefinishten Kanal schreibt, sollte das den gesamten Graphen nicht zum Absturz bringen.
• local_read mit fresh=True kopiert den Kanal: Beim fresh-Lesen wird für jeden Kanal channels[k].copy() aufgerufen(fresh read copy:213-219) und die Writes dieses Tasks auf der Kopie angewendet — so kann ein Knoten einer bedingten Kante (conditional edge) seine eigene, gerade geschriebene, aber noch nicht per apply_writes übernommene lokale Sicht lesen, ohne die globale Sicht anderer Tasks zu beeinflussen.
• Hash-Algorithmus der Task-ID in prepare_single_task: Bei checkpoint["v"] > 1 wird _xxhash_str verwendet, sonst _uuid5_str(task_id_func:550) — das ist Kompatibilität zwischen Checkpoint-Versionen; alte Checkpoints, die Task-IDs per uuid5 erzeugt haben, müssen weiterhin replayed werden können.

Zusammenfassung

Die drei reinen Funktionen should_interrupt / apply_writes / prepare_next_tasks in _algo.py extrahieren die semantische Schicht von Pregel sauber: wann unterbrochen wird, wie Writes zusammengeführt werden, wer im nächsten Schritt läuft. Sie haben weder Zustand noch IO und werden von PregelLoop.tick + after_tick aufgerufen, um den gesamten BSP-Zyklus zu treiben. Die Schleifentreiberschicht steht in /pregel/loop; die Ausführungsschicht in /pregel/runner; das Gesamtbau in /pregel/pregel; die Semantik der Methoden update / consume / finish / is_available der Kanäle in /channel/base-channel.

Siehe offizielle Dokumentation: LangGraph-Dokumentation · README.