PregelLoop: Superstep-Schleifentreiber

源码版本1.2.9

Verantwortung

PregelLoop ist der «Taktgeber» der LangGraph-Laufzeit. Pregel selbst ist nur für das Zusammenbauen der Ressourcen und das Bereitstellen der Eingänge invoke / astream zuständig; was den Graphen Runde für Runde tatsächlich treibt, ist eine PregelLoop-Instanz: Sie hält den aktuellen Checkpoint (checkpoint), die Kanal-Menge (channel), pending_writes, den step-Zähler, die status-Zustandsmaschine sowie ein tasks-Dictionary — all das ist der gesamte Kontext innerhalb eines einzelnen Superstep (superstep). Darauf baut die äußere Form while loop.tick(): ... loop.after_tick() auf.

Ihre beiden Kernmethoden sind tick(tick:599) und after_tick(after_tick:683). Ersteres übernimmt das «vor dem Lauf»: prüfen, ob die Schrittzahlgrenze überschritten ist, prepare_next_tasks aufrufen, um die in diesem Schritt auszuführenden Knoten in self.tasks zu laden, Interrupts prüfen, drain_requested behandeln und die von einer vorherigen Wiederaufnahme übrig gebliebenen pending_writes wieder an erfolgreiche Knoten zurückheften. Letzteres übernimmt das «nach dem Lauf»: alle task.writes einsammeln, apply_writes aufrufen, um sie in die Kanäle zu übernehmen, Pending leeren, Checkpoint schreiben und schließlich entscheiden, ob ein interrupt_after nötig ist. Die beiden flankieren den BSP-Zyklus «Eingabe lesen → dispatchen → ausführen → zurückschreiben → persistieren» vorne und hinten.

Anders gesagt: PregelLoop führt Knoten nicht selbst aus — das ist die Aufgabe von PregelRunner; es entscheidet nur, «wer in diesem Schritt drankommt, wie nach dem Lauf aufgeräumt wird und ob es einen weiteren Schritt braucht».

Entwurfsmotivation

Warum die Schleifenlogik in tick und after_tick aufteilen statt mit einer einzigen step()-Methode zu erledigen?

Ausführung an den Aufrufer übergeben: Nachdem tick True zurückgegeben hat, treibt der Aufrufer (Implementierungskörper von Pregel.astream, siehe sync main loop:2964-2984) selbst runner.tick, um diese Task-Charge abzuarbeiten, und ruft danach loop.after_tick auf. So muss PregelLoop keine Referenz auf PregelRunner halten; beide Objekte arbeiten lose gekoppelt zusammen, statt dass eines im anderen schachtelt.
Explizite Zustandsmaschine: Das status-Feld hat sieben Ausprägungen: "input" / "pending" / "done" / "draining" / "interrupt_before" / "interrupt_after" / "out_of_steps"(status Literal:256-264). Jeder Zustand entspricht einem Austrittsgrund; externer Code kann daran entscheiden, ob resume, raise oder break nötig ist, ohne die Rückgabewert-Semantik raten zu müssen.
Klare Wiederaufnahme-Semantik: is_replaying ist nur beim ersten tick wahr — es bedeutet, dass dieser Schritt das «Replay» der Tasks ist, die vor dem letzten Interrupt bereits fertig waren; sobald after_tick durchlaufen ist, wird es ausgeschaltet(is_replaying = False:716). Zusammen mit _reapply_writes_to_succeeded_nodes, das die im Checkpoint verbliebenen erfolgreichen Writes an In-Memory-Tasks zurückheftet, werden erfolgreiche Knoten nicht erneut ausgeführt, während fehlgeschlagene in den Error-Handler gehen.
Drain als Eingriff von außen: RunControl.drain_requested wird vom Stream-Konsumenten gesetzt (z. B. wenn der Client mittendrin anhalten möchte); tick tritt bei Erkennung in den draining-Zustand ein und gibt False zurück, sodass die äußere while-Schleife natürlich endet, statt Knoten mittendrin hart abzuwürgen.
Exit-mode delta writes und Checkpoint-Zeitplanung: _delta_write_futs(_delta_write_futs:207) sammelt die put_writes-Futures aller Delta-Kanäle; _checkpointer_put_after_previous muss sie erst drainen, bevor der nächste Checkpoint geschrieben wird — das garantiert die kausale Reihenfolge «Writes erzeugen den Checkpoint, nicht der Checkpoint überdeckt die Writes».

Schlüsseldateien

class PregelLoop:158 — Klassendefinition; Felder sind fast alle nackte Attribute ohne Property-Verpackung, damit die Unterklassen SyncPregelLoop / AsyncPregelLoop direkt überschreiben können.
status Literal:256-264 — Sieben-Zustände-Enumeration; "out_of_steps" wird bei Überschreitung der Schrittzahlgrenze ausgelöst, "draining" bei externem Stoppwunsch, "done" bei normalem Ende.
tick method:599-681 — Die gesamte Logik eines vollständigen tick: Schrittzahlprüfung, prepare_next_tasks, Done-Entscheidung bei leeren Tasks, Drain, _reapply_writes_to_succeeded_nodes, should_interrupt.
out_of_steps:607-609 — Tritt in den out_of_steps-Zustand ein und gibt False zurück, wenn self.step > self.stop; die äußere while-Schleife beendet sich.
done branch:653-655 — Setzt den done-Zustand, wenn self.tasks leer ist — das ist der «natürliche Tod» des Graphen.
draining branch:657-659 — Setzt den draining-Zustand, wenn control.drain_requested wahr ist.
reapply + resume handlers:662-664 — Heftet nach der Wiederaufnahme zuerst die pending writes an erfolgreiche Knoten zurück und ruft dann _resume_error_handlers_if_applicable auf, um für fehlgeschlagene Knoten Handler-Tasks vorzubereiten.
interrupt_before:667-671 — Ruft should_interrupt auf, um zu prüfen, ob Knoten in der interrupt_before-Liste liegen; bei Treffer wird GraphInterrupt geworfen.
after_tick method:683-726 — Writes einsammeln → apply_writes → values emitten → Pending leeren → _put_checkpoint → interrupt_after-Prüfung.
_reapply_writes_to_succeeded_nodes:736-749 — Heftet pending writes an In-Memory-Tasks zurück, überspringt aber die vier Steuersignale ERROR / ERROR_SOURCE_NODE / INTERRUPT / RESUME.
_resume_error_handlers_if_applicable:751-816 — Scannt die ERROR_SOURCE_NODE-Markierungen, konstruiert für fehlgeschlagene Knoten Error-Handler-Tasks und lässt den Runner die ursprünglichen Tasks überspringen und direkt die Handler ausführen.
put_writes:415-508 — Eingang für die von Tasks produzierten Writes; dedupliziert Steuerkanäle, akkumuliert Null-Tasks und startet die put_writes-Future des Checkpointers.
SyncPregelLoop:1469 / AsyncPregelLoop:1722 — Die beiden Unterklassen; überschreiben die synchron/asynchron unterschiedlichen Teile wie __enter__/__exit__, accept_push, schedule_error_handler.

Datenfluss

Der Eingang von tick prüft zuerst die Schrittzahlgrenze und ruft dann prepare_next_tasks auf, um zu berechnen, «welche Tasks im nächsten Schritt laufen». Dieser Schritt ist der Samen des gesamten Supersteps — alle nachfolgenden Aktionen verarbeiten diese Task-Charge.

python

def tick(self) -> bool:
    """Execute a single iteration of the Pregel loop.

    Returns:
        True if more iterations are needed.
    """

    # check if iteration limit is reached
    if self.step > self.stop:
        self.status = "out_of_steps"
        return False

    # prepare next tasks
    self.tasks = prepare_next_tasks(
        self.checkpoint,
        self.checkpoint_pending_writes,
        self.nodes,
        self.channels,
        self.managed,
        self.config,
        self.step,
        self.stop,
        for_execution=True,
        manager=self.manager,
        store=self.store,
        checkpointer=self.checkpointer,
        trigger_to_nodes=self.trigger_to_nodes,
        updated_channels=self.updated_channels,
        retry_policy=self.retry_policy,
        cache_policy=self.cache_policy,
    )

Das stammt aus tick 头部:599-629. self.step > self.stop ist die Austrittsbedingung «Schrittzahlgrenze überschritten» — stop wird in PregelLoop.__init__ auf recursion_limit gesetzt; wenn die Rekursionstiefe überschritten ist, endet man bei out_of_steps. prepare_next_tasks nimmt den aktuellen Checkpoint, pending_writes, die Knoten- und Kanaltabellen und berechnet, welche Tasks in diesem Schritt auszuführen sind; sie werden in self.tasks gesteckt. for_execution=True bedeutet, dass diese Tasks wirklich laufen sollen (nicht ein Dry-Run für Streaming-Vorschau).

Es folgen drei parallele Austrittsbedingungen — leere Tasks / Drain / Wiederaufnahme mit zurückgehefteten Writes — und danach should_interrupt:

python

# if no more tasks, we're done
if not self.tasks:
    self.status = "done"
    return False

if self.control is not None and self.control.drain_requested:
    self.status = "draining"
    return False

# if there are pending writes from a previous loop, apply them
if not self.is_replaying and self.checkpoint_pending_writes:
    self._reapply_writes_to_succeeded_nodes(self.tasks)
    self._resume_error_handlers_if_applicable()

# before execution, check if we should interrupt
if self.interrupt_before and should_interrupt(
    self.checkpoint, self.interrupt_before, self.tasks.values()
):
    self.status = "interrupt_before"
    raise GraphInterrupt()

Das stammt aus tick 中段:652-671. Die interrupt_before-Prüfung erfolgt erst nach der Wiederaufnahme mit pending_writes — denn bei der Wiederaufnahme spiegeln die Kanalversionsnummern bereits die letzten Schreibvorgänge, sodass should_interrupt korrekt beurteilen kann, «ob es seit dem letzten Interrupt neue Updates gab».

after_tick wird aufgerufen, nachdem der Runner alle Tasks abgearbeitet hat, und übernimmt den Abschluss:

python

def after_tick(self) -> None:
    # finish superstep
    writes = [w for t in self.tasks.values() for w in t.writes]
    self._delta_channels_with_overwrite.update(
        ch
        for ch, v in writes
        if isinstance(self.specs.get(ch), DeltaChannel) and _get_overwrite(v)[0]
    )
    # all tasks have finished
    self.updated_channels = apply_writes(
        self.checkpoint,
        self.channels,
        self.tasks.values(),
        self.checkpointer_get_next_version,
        self.trigger_to_nodes,
    )

Das stammt aus after_tick 头部:683-698. Zuerst werden alle task.writes eingeebnet und nebenbei erfasst, welche Delta-Kanäle ein Overwrite erlitten haben (relevant für den Startwert beim Sparse Replay); dann ruft apply_writes diese Writes tatsächlich in die Kanäle ein. Der Rückgabewert updated_channels wird im nächsten Schritt von prepare_next_tasks genutzt, um das «Finden des als nächstes auszulösenden Knotens» zu beschleunigen. Am Ende von after_tick wird zudem checkpoint_pending_writes geleert, is_replaying auf False gesetzt, der Checkpoint über _put_checkpoint({"source": "loop"}) geschrieben und schließlich die interrupt_after-Prüfung ausgeführt(after_tick 尾部:714-724).

Der Lebenszyklus eines ganzen Supersteps lässt sich so darstellen:

Grenzen und Fehler

out_of_steps ist kein Fehler: step > stop gibt direkt False zurück; status wird auf out_of_steps gesetzt, die äußere while-Schleife endet natürlich. Pregel entscheidet anhand von status, ob eine RecursionError geworfen oder still abgeschlossen wird — das ist eine «Soft-Limit»-Semantik, kein Ausnahmepfad(out_of_steps:607-609).
draining darf keine Lifecycle-Events emitten: _emit_graph_lifecycle_event lehnt Aufrufe im draining-Zustand explizit ab(draining guard:384-385), da Drain ein aktiver Benutzerstopp ist und nicht als interrupt-/resume-Semantik zählt.
Bei Wiederaufnahme müssen Steuersignale in pending writes übersprungen werden: _reapply_writes_to_succeeded_nodes muss ERROR / ERROR_SOURCE_NODE / INTERRUPT / RESUME überspringen(skip control signals:746-747), da sonst fehlgeschlagene Tasks als erfolgreich behandelt und bereits gelaufene Handler erneut getriggert würden.
Error-Handler bei der Wiederaufnahme neu schedulen: _resume_error_handlers_if_applicable wirkt nur auf Knoten mit error_handler_node(handler_node check:791-793); fehlgeschlagene Knoten ohne Handler bleiben unangetastet und gehen über _should_stop_others des Runners den Panic-Pfad.
put_writes-Akkumulationssemantik für Null-Tasks: Writes mit NULL_TASK_ID werden nicht überschrieben, sondern akkumuliert(null task accumulate:422-431), verwendet für Input-Writes, die «zu keinem Knoten gehören, aber in den Checkpoint müssen».
Futures der Delta-Kanäle müssen vor dem nächsten Checkpoint flüssig gemacht werden: _delta_write_futs sammelt die put_writes-Futures aller Delta-Kanäle; vor dem Schreiben des nächsten Checkpoints werden sie zuerst drainet(_delta_write_futs:201-207) — umgekehrte Reihenfolge würde beim Sparse Replay die Writes verbergen, die den Checkpoint erzeugt haben.

Zusammenfassung

PregelLoop ist ein zustandsmaschinengetriebener Superstep-Taktgeber, der die drei Phasen «Tasks vorbereiten → laufen lassen → aufräumen» um PregelRunner herum aufspannt, ohne selbst die Ausführung anzufassen. Wer tick / after_tick, die sieben status-Zustände sowie den Wiederaufnahmemechanismus aus is_replaying + _reapply_writes_to_succeeded_nodes verstanden hat, hat die Hauptachse des LangGraph-Laufzeitmodells erfasst. Wie der Runner die Tasks tatsächlich zum Laufen bringt, steht in /pregel/runner; die internen Details der Algorithmusfunktionen prepare_next_tasks / apply_writes / should_interrupt in /pregel/algo; wie das Ganze zusammengesetzt wird, in /pregel/pregel. Die Schreibzurück-Semantik der Kanäle steht in /channel/base-channel.

Siehe offizielle Dokumentation: LangGraph-Dokumentation · README.

PregelLoop: Superstep-Schleifentreiber ​

Verantwortung ​

Entwurfsmotivation ​

Schlüsseldateien ​

Datenfluss ​

Grenzen und Fehler ​

Zusammenfassung ​

PregelLoop: Superstep-Schleifentreiber

Verantwortung

Entwurfsmotivation

Schlüsseldateien

Datenfluss

Grenzen und Fehler

Zusammenfassung