PregelRunner: in-superstep task executor

源码版本1.2.9

Responsibilities

PregelRunner does one thing: actually run the batch of PregelExecutableTask prepared by PregelLoop.tick, collect their writes, and hand them back to PregelLoop. It does not care what the graph looks like, does not care about the checkpoint, does not care how channels reduce — it only handles the three things "concurrent execution + failure handling + writes persisted".

Its position is very narrow: inside the main loop of Pregel.astream (sync main loop:2964-2984), right after while loop.tick() returns True, comes for _ in runner.tick([t for t in loop.tasks.values() if not t.writes], ...):, and after that finishes, loop.after_tick(). In other words, the runner's input is "the tasks computed by tick that have not yet written writes", and its output is "those tasks' writes are now in checkpoint_pending_writes" — with no second-layer buffer in between.

PregelRunner itself is very thin: two public methods tick (tick:176) and commit (commit:574), plus an async atick. The heavy lifting is all in the two low-level callers _call / _acall (_call:700) / (_acall:789) — they wrap the node function into a schedulable future, handle retry policy, sub-task Send fan-out, and streaming chunk callbacks, and ultimately write writes back to the task object.

Design motivation

Why is the runner a generator instead of a normal method?

• Streaming yield: the runner takes the generator form for _ in runner.tick(...) (yield return type:188); each completed task yields once and hands control back to the upper layer, which can immediately forward streaming chunks to the client instead of waiting for the whole superstep to finish. This is the foundation of LangGraph streaming.
• Single-task fast path: when len(tasks) == 1 and timeout is None and get_waiter is None, a synchronous direct call is taken without spinning up a thread pool (fast path:203-254). Most single-branch agent scenarios hit this path, avoiding thread-pool overhead.
• commit as a Future callback: FuturesDict.on_done (on_done:116) is triggered when each future completes, and internally it is weakref.WeakMethod(self.commit) — commit is a per-task write-back hook, not a batch wrap-up. This means writes enter pending_writes in completion order, and the checkpoint can be written as early as possible.
• Error handler routing: _should_route_to_error_handler (_should_route_to_error_handler:171-174) checks whether a failed node has an error_handler node configured. On a hit, the exception id is added to _handled_exception_ids and a handler task is scheduled in place of the original task — so a failure does not immediately panic, but flows through another node for wrap-up.
• _should_stop_others cancellation: when any task raises a non-GraphBubbleUp exception, the runner cancels all other in-flight tasks in the same batch (_should_stop_others:616-634) — enforcing the within-superstep semantics of "either all succeed, or enter the handler". GraphInterrupt is explicitly excluded, because interrupts are not failures.

Key files

• class PregelRunner:135-138 — class definition; the docstring states its responsibility in one line: execute tasks, commit writes, yield control, interrupt other tasks when needed.
• __init__:140-169 — holds weakrefs to submit / put_writes, the node_error_handler_map, the schedule_error_handler / aschedule_error_handler callbacks, and _handled_exception_ids — a cross-tick exception dedup set.
• FuturesDict:75-134 — a custom dict subclass; the on_done callback fires commit per future completion; the should_stop field holds the _should_stop_others partial.
• tick signature:176-188 — takes Iterable[PregelExecutableTask], returns Iterator[None]; the signature exposes the generator semantics.
• fast path:203-254 — runs synchronously when single task + no timeout + no waiter; schedules an error handler on failure as needed.
• schedule tasks:259-276 — multi-task path: each task gets a future, scheduled via self.submit() (which is actually PregelLoop.submit).
• concurrent wait loop:282-323 — a concurrent.futures.wait(FIRST_COMPLETED) loop; each completed task triggers a commit, emits output, and schedules a handler task if needed.
• commit method:574-613 — per-task wrap-up: cancelled writes ERROR, GraphInterrupt writes INTERRUPT, ordinary exception writes ERROR + ERROR_SOURCE_NODE, normal completion writes task.writes + a NO_WRITES marker.
• _should_stop_others:616-634 — decides whether to cancel other tasks; explicitly excludes GraphBubbleUp and already-handled exceptions.
• _panic_or_proceed:650-697 — wrap-up function; cancels all in-flight futures, merges multiple GraphInterrupts into one, and raises TimeoutError on timeout.
• _call:700-787 — low-level wrapper for node calls: retry, stream chunk emit, Send fan-out via schedule_task callback, accumulates task.writes.
• _should_route_to_error_handler:171-174 — returns True when task.name in self.node_error_handler_map; error-handler nodes themselves cannot be routed (prevents recursion).

Data flow

The start of tick constructs a FuturesDict, an enhanced dict that automatically calls commit when each future completes:

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

This comes from tick head:176-199. Note that the first yield immediately hands control back — the upper layer's first iteration of for _ in runner.tick(...) is just "starting"; actual task execution happens in subsequent iterations. This is a simplified form of generator coroutine, avoiding the introduction of the async keyword into the sync path.

commit is the per-task wrap-up function that lands a task's writes into pending_writes:

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)

This comes from commit:574-613. put_writes is a weakref to PregelLoop.put_writes (put_writes:415); it stuffs writes into checkpoint_pending_writes and starts a checkpointer future to persist them. Note several special writes:

• CancelledError -> writes (ERROR, exception), so that after_tick wrap-up counts the task as "failed but recorded" without panicking.
• GraphInterrupt -> writes (INTERRUPT, ...) + any RESUME writes; the interrupt's resume value is persisted alongside the interrupt so the next resume reads them together.
• Ordinary exception + configured error handler -> writes an additional (ERROR_SOURCE_NODE, task.name) marker; this marker is scanned by PregelLoop._resume_error_handlers_if_applicable (ERROR_SOURCE_NODE scan:775) on resume to schedule the corresponding handler node.
• Normal completion with no writes -> a (NO_WRITES, None) marker is added, preventing prepare_next_tasks from treating it as "not yet run" (NO_WRITES marker:609-611).

The runner's execution flow can be drawn like this:

Boundaries and failure modes

• _handled_exception_ids persists across ticks: this set in __init__ is instance-level, not per-tick (_handled_exception_ids:169). Once an exception object id enters the set, _should_stop_others will no longer treat it as a failure, preventing the original exception from being re-raised after the error handler completes.
• GraphBubbleUp is not a failure: commit explicitly passes on GraphBubbleUp and writes nothing (GraphBubbleUp pass:592-594). Such exceptions (e.g. ParentCommand) are signals to the parent graph, not real errors.
• NO_WRITES prevents duplicate task execution: when a task completes normally but writes nothing, a (NO_WRITES, None) is still appended (NO_WRITES:609-611). Otherwise, prepare_next_tasks after after_tick would see that this task has no writes, assume it never ran, and run it again on resume.
• schedule_error_handler may return None: the handler node itself may not be configured or cannot be constructed, in which case the task takes the normal panic path (handler optional:230-248). The ERROR_SOURCE_NODE write in commit coexists with this path; _resume_error_handlers_if_applicable will retry scheduling on resume.
• A timeout does not raise GraphInterrupt: _panic_or_proceed's if inflight: raise timeout_exc_cls("Timed out") (timeout:691-697) — a timeout is a real exception and bubbles out of Pregel, unlike GraphInterrupt which is swallowed.
• Traceback trimming: before reraise, frames in the EXCLUDED_FRAME_FNAMES list are skipped (tb trim:241-247), filtering out langgraph internal stack frames so the traceback the user sees points directly at their node code.

Summary

PregelRunner is a thin shell: the generator tick schedules concurrent execution + per-task commit lands writes into pending_writes, with error-handler routing and cancellation in between. Understanding it means understanding "how tasks actually run within a superstep". See /pregel/loop for the loop driver on the upper layer, /pregel/algo for the three algorithm functions prepare_next_tasks / apply_writes / should_interrupt, /pregel/pregel for overall assembly, and /stream/run-stream for how streaming output is produced between runner yields.

See official docs: LangGraph docs · README.