PregelRunner: Task-Ausführer innerhalb eines Supersteps

源码版本1.2.9

Verantwortung

PregelRunner macht eine Sache: die von PregelLoop.tick vorbereitete Charge von PregelExecutableTask tatsächlich laufen lassen und ihre Writes einsammeln und an PregelLoop zurückgeben. Er kümmert sich nicht darum, wie der Graph aussieht, nicht um Checkpoints, nicht darum, wie Kanäle reduzieren — nur um die drei Dinge «nebenläufige Ausführung + Fehlerbehandlung + Writes persistieren».

Seine Position ist sehr eng: Im Hauptzyklus von Pregel.astream(sync main loop:2964-2984) folgt nach while loop.tick() True direkt for _ in runner.tick([t for t in loop.tasks.values() if not t.writes], ...):, und nach dem Lauf kommt loop.after_tick(). Eingabe des Runners sind also «die von tick berechneten, noch keine Writes geschrieben habenden Tasks», Ausgabe «diese Tasks haben ihre Writes in checkpoint_pending_writes abgelegt»; dazwischen gibt es keinen zweiten Puffer.

PregelRunner selbst ist sehr dünn: zwei öffentliche Methoden tick(tick:176) und commit(commit:574), zusätzlich die asynchrone Variante atick. Die eigentliche Arbeit steckt in den beiden unterliegenden Aufrufern _call / _acall(_call:700)/(_acall:789) — sie wickeln die Knotenfunktion in eine dispatchbare Future ein, behandeln Retry-Policy, Send-Fan-out von Teiltasks und Streaming-Chunk-Callbacks und schreiben die Writes schließlich in das Task-Objekt zurück.

Entwurfsmotivation

Warum ist der Runner als Generator statt als normale Methode gebaut?

• Streaming-yield: Der Runner ist in der Generator-Form for _ in runner.tick(...)(yield return type:188); nach jedem fertigen Task wird einmal yielded und die Steuerung an die obere Ebene zurückgegeben — so kann die obere Ebene Streaming-Chunks sofort an den Client schicken, statt auf den gesamten Superstep zu warten. Das ist die Grundlage von LangGraphs Streaming.
• Single-Task-Fast-Path: Wenn len(tasks) == 1 and timeout is None and get_waiter is None, wird direkt synchron aufgerufen und kein Thread-Pool gestartet(fast path:203-254). In den meisten Single-Branch-Agent-Szenarien trifft dieser Pfad zu und erspart die Thread-Pool-Kosten.
• commit als Future-Callback: FuturesDict.on_done(on_done:116) wird beim Abschließen jeder Future getriggert; intern ist es weakref.WeakMethod(self.commit) — commit ist ein Per-Task-Hook zum Persistieren von Writes, kein Batch-Abschluss. So ist die Reihenfolge, in der Writes in pending_writes einfließen,天然 nach Abschlussreihenfolge, und auch der Checkpoint kann so früh wie möglich geschrieben werden.
• Error-Handler-Routing: _should_route_to_error_handler(_should_route_to_error_handler:171-174) prüft, ob für einen fehlgeschlagenen Knoten ein error_handler-Knoten konfiguriert ist. Bei Treffer wird die Exception-ID in _handled_exception_ids aufgenommen und ein Handler-Task statt des ursprünglichen Tasks geschedult — so führt ein Fehler nicht sofort zu Panic, sondern geht durch einen anderen Knoten in den Abschluss.
• _should_stop_others-Abbruchmechanismus: Sobald ein Task eine Nicht-GraphBubbleUp-Exception wirft, bricht der Runner die anderen in-flight-Tasks desselben Batches ab(_should_stop_others:616-634) — das garantiert die Semantik «in demselben Superstep entweder alle erfolgreich oder in den Handler» innerhalb des Schritts. GraphInterrupt wird explizit ausgenommen, da ein Interrupt kein Fehler ist.

Schlüsseldateien

• class PregelRunner:135-138 — Klassendefinition; ein Docstring-Zeile nennt die Verantwortung: Task ausführen, Writes committen, Steuerung yielden, andere Tasks bei Bedarf abbrechen.
• __init__:140-169 — Hält weakrefs auf submit / put_writes, node_error_handler_map, schedule_error_handler/aschedule_error_handler-Callbacks sowie _handled_exception_ids, die übergreifende Dedup-Menge für Exceptions über mehrere tick hinweg.
• FuturesDict:75-134 — Eigenes dict-Subclass; der on_done-Callback triggert bei Abschluss jeder Future commit; das Feld should_stop hält ein _should_stop_others-Partial.
• tick signature:176-188 — Eingabe Iterable[PregelExecutableTask], Rückgabe Iterator[None]; die Signatur macht die Generator-Semantik sichtbar.
• fast path:203-254 — Bei Single-Task + kein Timeout + kein Waiters wird direkt synchron ausgeführt; bei Fehler wird bei Bedarf ein Error-Handler geschedult.
• schedule tasks:259-276 — Multi-Task-Pfad: Für jeden Task wird eine Future gestartet und über self.submit() (tatsächlich PregelLoop.submit) geschedult.
• concurrent wait loop:282-323 — concurrent.futures.wait(FIRST_COMPLETED)-Schleife; nach jedem fertigen Task wird commit aufgerufen, Output emittet und bei Bedarf ein Handler-Task gestartet.
• commit method:574-613 — Per-Task-Abschluss: Cancelled schreibt ERROR, GraphInterrupt schreibt INTERRUPT, normale Exception schreibt ERROR + ERROR_SOURCE_NODE, regulär schreibt task.writes + NO_WRITES-Marker.
• _should_stop_others:616-634 — Prüft, ob andere Tasks abzubrechen sind; schließt GraphBubbleUp und bereits behandelte Exceptions explizit aus.
• _panic_or_proceed:650-697 — Abschlussfunktion: bricht alle in-flight-Futures ab, fasst mehrere GraphInterrupt zu einem zusammen und wirft bei Timeout TimeoutError.
• _call:700-787 — Untere Hülle des Knotenaufrufs: Retry, Stream-Chunk-Emit, Send-Fan-out-Callback schedule_task, akkumuliert task.writes.
• _should_route_to_error_handler:171-174 — Gibt True zurück, wenn task.name in self.node_error_handler_map; ein Error-Handler-Knoten selbst darf nicht weitergeleitet werden (verhindert Rekursion).

Datenfluss

Am Anfang von tick wird eine FuturesDict konstruiert — ein erweitertes dict, das bei Abschluss jeder Future automatisch commit aufruft:

def tick(
    self,
    tasks: Iterable[PregelExecutableTask],
    *,
    reraise: bool = True,
    timeout: float | None = None,
    retry_policy: Sequence[RetryPolicy] | None = None,
    get_waiter: Callable[[], concurrent.futures.Future[None]] | None = None,
    schedule_task: Callable[
        [PregelExecutableTask, int, Call | None],
        PregelExecutableTask | None,
    ],
) -> Iterator[None]:
    tasks = tuple(tasks)
    futures = FuturesDict(
        callback=weakref.WeakMethod(self.commit),
        event=threading.Event(),
        should_stop=partial(
            _should_stop_others, handled_exception_ids=self._handled_exception_ids
        ),
        future_type=concurrent.futures.Future,
    )
    # give control back to the caller
    yield

Das stammt aus tick 头部:176-199. Beachten Sie das erste yield — es gibt die Steuerung sofort zurück; die erste Iteration des äußeren for _ in runner.tick(...) ist nur der «Start», die eigentliche Task-Ausführung geschieht in späteren Iterationen. Das ist eine vereinfachte Form der Generator-Koroutine, um das async-Schlüsselwort im synchronen Pfad zu vermeiden.

commit ist die Per-Task-Abschlussfunktion, die die Writes eines Tasks in pending_writes einträgt:

def commit(
    self,
    task: PregelExecutableTask,
    exception: BaseException | None,
) -> None:
    if isinstance(exception, asyncio.CancelledError):
        task.writes.append((ERROR, exception))
        self.put_writes()(task.id, task.writes)
    elif exception:
        if isinstance(exception, GraphInterrupt):
            if exception.args[0]:
                writes = [(INTERRUPT, exception.args[0])]
                if resumes := [w for w in task.writes if w[0] == RESUME]:
                    writes.extend(resumes)
                self.put_writes()(task.id, writes)
        elif isinstance(exception, GraphBubbleUp):
            pass
        else:
            task.writes.append((ERROR, exception))
            if self._should_route_to_error_handler(task) and not isinstance(
                exception, GraphBubbleUp
            ):
                task.writes.append((ERROR_SOURCE_NODE, task.name))
                self._handled_exception_ids.add(id(exception))
            self.put_writes()(task.id, task.writes)
    else:
        if self.node_finished and (
            task.config is None or TAG_HIDDEN not in task.config.get("tags", [])
        ):
            self.node_finished(task.name)
        if not task.writes:
            task.writes.append((NO_WRITES, None))
        self.put_writes()(task.id, task.writes)

Das stammt aus commit:574-613. put_writes ist eine weakref auf PregelLoop.put_writes(put_writes:415); es steckt die Writes in checkpoint_pending_writes und startet die Future des Checkpointers zur Persistenz. Beachten Sie die verschiedenen Sonder-Writes:

• CancelledError → schreibt (ERROR, exception), sodass after_tick beim Abschluss den Task als «fehlgeschlagen, aber aufgezeichnet» behandelt, ohne Panic.
• GraphInterrupt → schreibt (INTERRUPT, ...) + eventuelle RESUME-Writes; der Wiederaufnahmewert des Interrupts muss zusammen mit dem Interrupt persistiert werden, damit er beim nächsten Resume zusammen ausgelesen wird.
• Normale Exception + konfigurierter Error-Handler → schreibt zusätzlich (ERROR_SOURCE_NODE, task.name); diese Markierung wird von PregelLoop._resume_error_handlers_if_applicable aufgegriffen(ERROR_SOURCE_NODE scan:775), um bei der Wiederaufnahme den passenden Handler-Knoten zu schedulen.
• Normale Beendigung ohne schreibbare Writes → ergänzt ein (NO_WRITES, None), damit prepare_next_tasks ihn nicht als «noch nie gelaufen» auffasst(NO_WRITES marker:609-611).

Der Ausfluss des gesamten Runners lässt sich so darstellen:

Grenzen und Fehler

• _handled_exception_ids ist über tick hinweg persistent: Diese Menge wird in __init__ auf Instanzebene angelegt, nicht pro tick(_handled_exception_ids:169). Sobald die id eines Exception-Objekts in der Menge ist, wird sie von _should_stop_others nicht mehr als Fehler eingestuft; das verhindert, dass nach dem Lauf eines Error-Handlers die ursprüngliche Exception erneut geworfen wird.
• GraphBubbleUp zählt nicht als Fehler: commit macht bei GraphBubbleUp explizit pass und schreibt keinerlei Writes(GraphBubbleUp pass:592-594). Solche Exceptions (z. B. ParentCommand) signalisieren dem Eltern-Graphen und sind keine echten Fehler.
• NO_WRITES verhindert doppelte Task-Ausführung: Beendet ein Task normal, ohne etwas zu schreiben, wird trotzdem ein (NO_WRITES, None) ergänzt(NO_WRITES:609-611). Andernfalls würde prepare_next_tasks nach after_tick feststellen, dass dieser Task keine Writes hat, ihn für «nie gelaufen» halten und beim Resume erneut ausführen.
• schedule_error_handler kann None zurückgeben: Auch ein Handler-Knoten selbst kann fehlen oder nicht konstruierbar sein; in diesem Fall geht der Task den normalen Panic-Pfad(handler optional:230-248). Der ERROR_SOURCE_NODE-Write in commit koexistiert mit diesem Pfad; _resume_error_handlers_if_applicable versucht es bei der Wiederaufnahme erneut.
• Timeout wirft keinen GraphInterrupt: _panic_or_proceed hat if inflight: raise timeout_exc_cls("Timed out")(timeout:691-697) — Timeout ist eine echte Exception, die aus Pregel herausblubbert, anders als GraphInterrupt, der verschluckt wird.
• Traceback-Beschneidung: Vor dem Reraise werden Frames aus der Liste EXCLUDED_FRAME_FNAMES übersprungen(tb trim:241-247), um interne LangGraph-Stackframes herauszufiltern, sodass der für den Nutzer sichtbare Traceback direkt auf seinen Knotencode zeigt.

Zusammenfassung

PregelRunner ist eine dünne Hülle: Der Generator tick schedult die nebenläufige Ausführung, und der Per-Task-commit schreibt die Writes in pending_writes; zwischengeschaltet sind Error-Handler-Routing und Abbruchmechanismus. Wer das verstanden hat, hat verstanden, «wie innerhalb eines Supersteps Tasks tatsächlich zum Laufen kommen». Die darüberliegende PregelLoop-Schleife steht in /pregel/loop; die drei Algorithmusfunktionen prepare_next_tasks / apply_writes / should_interrupt in /pregel/algo; das Gesamtbau in /pregel/pregel; wie die Stream-Ausgabe zwischen den Runner-yields verarbeitet wird, steht in /stream/run-stream.

Siehe offizielle Dokumentation: LangGraph-Dokumentation · README.