Nebenläufige Ausführung: Future-Orchestrierung bei @task

源码版本1.2.9

Verantwortung

Die Laufzeitsemantik von @task wird vom PregelRunner übernommen — der Task-Aufruf selbst wird nicht sofort ausgeführt, sondern Funktion und Argumente gehen an den aktuellen runner, der entscheidet, wann und in welchem Thread / welcher Coroutine er läuft. Diese Schicht ist für drei Dinge zuständig: Task-Aufrufe in PUSH-Tasks übersetzen und in eine Queue legen, mehrere Tasks innerhalb desselben Superstep (superstep) gleichzeitig ausführen und die Ergebnisse über ein future an den Aufrufer zurückgeben. PregelRunner (PregelRunner:135) ist das Zentrum dieser Orchestrierung.

Die Position liegt unterhalb von PregelLoop und oberhalb von BackgroundExecutor / AsyncBackgroundExecutor. loop.tick berechnet, welche Tasks in diesem Schritt laufen sollen, und übergibt sie an runner.tick / atick (PregelRunner.tick:176); der runner reicht jeden Task per submit an den executor weiter, verfolgt alle in-flight futures in einem FuturesDict und schreibt, sobald sie done sind, die writes zurück und leitet Ausnahmen an den Fehlerbehandlungs-Knoten weiter.

Entwurfsmotivation

Task-Aufruf ist PUSH-Task: Wenn du im entrypoint-Rumpf task_a(); task_b() schreibst, brauchen diese beiden Tasks keine explizite Kante — _call_with_options holt den CONFIG_KEY_CALL-Callback aus dem config (das ist das vom runner injizierte _call / _acall), verpackt den Task in eine Call-Datenstruktur und schiebt ihn per schedule_task in den TASKS-Kanal (schedule PUSH task:713). Die nächste Runde prepare_next_tasks findet ihn von selbst.
Einheitliche Abstraktion für synchron und asynchron: SyncAsyncFuture (SyncAsyncFuture:253) ist zugleich ein concurrent.futures.Future und ein Objekt mit __await__ — sein __await__ ist ein einzelnes yield, das die Kontrolle an den äußeren runner abgibt. In einer synchronen entrypoint blockiert task().result() bis das Ergebnis da ist; in einer asynchronen entrypoint gibt await task() die Coroutine ab; beide nutzen dasselbe future-Objekt.
next_tick-Scheduling: Ein Unter-Task wird beim Schedulen nicht sofort ausgeführt, sondern mit __next_tick__=True (next_tick:768) dem executor mitgegeben — «erst im nächsten tick laufen». So werden zuerst die writes des aktuellen tick committet und stream-Events ausgegeben, bevor der Unter-Task beginnt. Die Stream-Reihenfolge bleibt damit konsistent mit der Aufrufreihenfolge im Code.
Reentrant-Deduplizierung: Wird derselbe Task mehrfach aufgerufen, sucht _call zuerst in futures nach einem future mit derselben id und verwendet es erneut. Ist der Task schon abgeschlossen (next_task.writes nicht leer), wird direkt aus den writes der RETURN-Wert geholt und ein bereits abgeschlossenes future zurückgegeben (dedup already ran:724) — das verhindert, dass Unter-Tasks beim Wiederholen eines Eltern-Tasks erneut ausgeführt werden.
max_concurrency-Semaphor: Im asynchronen Pfad liest AsyncBackgroundExecutor config["max_concurrency"] und erzeugt ein asyncio.Semaphore; jede Task-Coroutine wird in gated(semaphore, coro) gehüllt (gated:214), um die Obergrenze gleichzeitig in-flight befindlicher Tasks innerhalb eines Graphen zu begrenzen und zu verhindern, dass LLM-Aufrufe den Event-Loop überlasten.

Schlüsseldateien

PregelRunner:135 — Nebenläufigkeits-Executor; hält submit / put_writes / node_error_handler_map / schedule_error_handler.
PregelRunner.tick:176 — Führt synchron einen Batch von Tasks aus; eine concurrent.futures.wait(FIRST_COMPLETED)-Schleife holt done futures und committet writes.
PregelRunner.atick:360 — Asynchrone Variante; asyncio.wait ersetzt das synchrone wait, Iterator-Semantik bleibt gleich.
_call:700 — Der CONFIG_KEY_CALL-Callback im synchronen Pfad; übersetzt Task-Aufrufe in PUSH-Tasks, dedupliziert und verkettet futures.
_acall:789 — Asynchroner Pfad-Callback; gleiche Semantik, verarbeitet Coroutines.
schedule PUSH task:713 — Schiebt die Call-Datenstruktur per schedule_task (tatsächlich loop.accept_push / aaccept_push) in den TASKS-Kanal.
SyncAsyncFuture:253 — Brückenobjekt, das zugleich ein sync Future und ein async awaitable ist; __await__ gibt mit einem einzelnen yield die Kontrolle ab.
_call_with_options:276 — Implementierung von _TaskFunction.__call__; holt CONFIG_KEY_CALL aus dem config und ruft es auf.
BackgroundExecutor:40 — Synchroner Threadpool-Context-Manager; submit verwendet copy_context(), um contextvar im Unter-Thread zu vererben.
AsyncBackgroundExecutor:122 — Asynchroner Context-Manager auf Basis von asyncio.get_running_loop(); optionales max_concurrency-Semaphor.
gated:214 — async def gated(semaphore, coro): erst acquire() dann release(); legt die Nebenläufigkeitsobergrenze um die Task-Coroutine.

Datenfluss

_call ist der Kern des Task-Aufrufs; es übersetzt «in der entrypoint-Funktion task_fn(arg) aufrufen» in «im runner einen PUSH-Task schedulen»:

python

# schedule the next task, if the callback returns one
if next_task := schedule_task(
    task(),
    scratchpad.call_counter(),
    Call(
        func,
        input,
        retry_policy=retry_policy,
        cache_policy=cache_policy,
        callbacks=callbacks,
        timeout=timeout,
    ),
):
    if fut := next(
        (
            f
            for f, t in list(futures().items())
            if t is not None and t == next_task.id
        ),
        None,
    ):
        # if the parent task was retried,
        # the next task might already be running
        pass
    elif next_task.writes:
        # if it already ran, return the result
        fut = concurrent.futures.Future()
        ret = next((v for c, v in next_task.writes if c == RETURN), MISSING)
        # ...
    else:
        # schedule the next task
        fut = submit()(
            run_with_retry,
            next_task,
            retry_policy,
            configurable={
                CONFIG_KEY_CALL: partial(
                    _call,
                    weakref.ref(next_task),
                    futures=futures,
                    # ...
                ),
            },
            __reraise_on_exit__=False,
            __next_tick__=True,
        )
        SKIP_RERAISE_SET.add(fut)
        futures()[fut] = next_task

Die drei Zweige nacheinander: (1) Task mit derselben id läuft bereits — future wiederverwenden; (2) Task ist schon gelaufen (writes nicht leer) — aus writes den RETURN-Wert holen und ein abgeschlossenes future konstruieren; (3) neuer Task — submit() gibt run_with_retry an den BackgroundExecutor oder AsyncBackgroundExecutor ab und injiziert das rekursive _call als CONFIG_KEY_CALL in den config des Unter-Tasks, sodass ein task()-Aufruf innerhalb des Unter-Tasks denselben Scheduling-Pfad nimmt.

Der Kern von PregelRunner.tick selbst ist ein FuturesDict plus eine concurrent.futures.wait(FIRST_COMPLETED)-Schleife (wait loop:281). Sobald ein future done ist, wird der zugehörige Task herausgeholt und per commit werden seine writes in den loop zurückgeschrieben; falls der Task einen Fehlerbehandlungs-Knoten konfiguriert hat, wird ein weiterer error-handler-Task per submit eingereiht. Die Schleife läuft, bis alle in-flight futures erledigt sind (oder nur noch der get_waiter-Platzhalter-future übrig ist), und gibt die Kontrolle an loop.after_tick zurück, wo die Kanäle geschrieben werden.

Grenzen und Fehler

Sync-Kontext mit async-Task wirft Fehler: _call prüft am Anfang if inspect.iscoroutinefunction(func): raise RuntimeError (sync async mismatch:716) — in einer synchronen entrypoint ist kein await möglich, und der asynchrone Task kann nicht laufen.
Ausnahmen von Unter-Tasks blieben zum Eltern-Task durch: SKIP_RERAISE_SET.add(fut) markiert das future des Unter-Tasks mit «nicht am Ende des tick re-raisen»; die Ausnahme wird per chainFuture an das Eltern-future gekettet und wird erst beim await des Eltern-Tasks sichtbar (skip reraise:781). So folgt die Fehlerweitergabe der Aufrufhierarchie, statt vom runner direkt geworfen zu werden und den gesamten tick abzubrechen.
__next_tick__ ist im synchronen Modus verzögert: BackgroundExecutor.submit verwendet next_tick(ctx.run, fn, ...) (next_tick sync:66), um die Task in den nächsten tick einzureihen; im asynchronen executor ist __next_tick__ ein noop, weil asyncio-Tasks ohnehin asynchron schedulen.
max_concurrency betrifft nicht die Knoten-Nebenläufigkeit: Das Semaphor des AsyncBackgroundExecutor schränkt nur Task-Coroutines ein (gated), nicht wie viele Knoten PregelRunner.tick gleichzeitig schedul darf — Knoten-Nebenläufigkeit wird durch prepare_next_tasks bestimmt, das Semaphor drosselt nur (semaphore:153).
Unter-Task wird bei Wiederholung des Eltern-Tasks nicht erneut ausgeführt: Wenn der Eltern-Task wiederholt wird, läuft _call in den Zweig «dedup already ran» und gibt entweder das bestehende future zurück oder holt das Ergebnis aus writes (dedup already ran:724) — idempotent.
contextvar wird vererbt: BackgroundExecutor.submit verwendet copy_context() (copy_context:62), damit der Unter-Thread die contextvar des Eltern-Threads (RunnableConfig usw.) übernimmt und der Unter-Task nicht ohne config dasteht.

Zusammenfassung

Die Nebenläufigkeit von @task stammt aus der future-Orchestrierung des PregelRunner; die zentralen Abstraktionen sind SyncAsyncFuture als Brücke zwischen synchron/asynchron, _call / _acall als Übersetzung von Task-Aufruf in PUSH-Task sowie BackgroundExecutor / AsyncBackgroundExecutor als Ausführungsthreads bzw. -coroutinen. Scheduling-Details der PUSH-Tasks stehen in /subgraph/send-command; die Semantik von Tasks in entrypoint in /func/entrypoint-task. Siehe offizielle Dokumentation: LangGraph 文档 · README。

Nebenläufige Ausführung: Future-Orchestrierung bei @task ​

Verantwortung ​

Entwurfsmotivation ​

Schlüsseldateien ​

Datenfluss ​

Grenzen und Fehler ​

Zusammenfassung ​