WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-06 17:57:49 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: extract _track helper for available_channels sync in apply_writes

Browse Source 
Replace 5 repeated inline blocks that sync available_channels with
a local _track() helper that checks is_available() and updates the
set, returning the availability bool for callers that also need to
update updated_channels.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
John Kennedy
2026-03-12 00:03:48 +00:00
co-authored by Claude Opus 4.6
parent 49e78713cc
commit b24d88f54a
2 changed files with 20 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/pregel/_algo.py
+20 -28
View File
@@ -267,6 +267,18 @@ def apply_writes(
None,
)
# Sync available_channels with channel's actual availability state.
# Returns True if the channel is available (for callers that also need
# to update updated_channels).
def _track(chan: str) -> bool:
avail = channels[chan].is_available()
if available_channels is not None:
if avail:
available_channels.add(chan)
else:
available_channels.discard(chan)
return avail
# Consume all channels that were read
for chan in {
chan
@@ -276,11 +288,7 @@ def apply_writes(
}:
if channels[chan].consume() and next_version is not None:
checkpoint["channel_versions"][chan] = next_version
if available_channels is not None:
if channels[chan].is_available():
available_channels.add(chan)
else:
available_channels.discard(chan)
_track(chan)
# Group writes by channel
pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)
@@ -302,17 +310,10 @@ def apply_writes(
if channels[chan].update(vals) and next_version is not None:
checkpoint["channel_versions"][chan] = next_version
# unavailable channels can't trigger tasks, so don't add them
if channels[chan].is_available():
if _track(chan):
updated_channels.add(chan)
if available_channels is not None:
available_channels.add(chan)
elif available_channels is not None:
available_channels.discard(chan)
elif available_channels is not None:
if channels[chan].is_available():
available_channels.add(chan)
else:
available_channels.discard(chan)
else:
_track(chan)
# Channels that weren't updated in this step are notified of a new step
if bump_step:
@@ -329,10 +330,8 @@ def apply_writes(
if channels[chan].update(EMPTY_SEQ) and next_version is not None:
checkpoint["channel_versions"][chan] = next_version
# unavailable channels can't trigger tasks, so don't add them
if channels[chan].is_available():
if _track(chan):
updated_channels.add(chan)
elif available_channels is not None:
available_channels.discard(chan)
# If this is (tentatively) the last superstep, notify all channels of finish
if bump_step and updated_channels.isdisjoint(trigger_to_nodes):
@@ -340,17 +339,10 @@ def apply_writes(
if channels[chan].finish() and next_version is not None:
checkpoint["channel_versions"][chan] = next_version
# unavailable channels can't trigger tasks, so don't add them
if channels[chan].is_available():
if _track(chan):
updated_channels.add(chan)
if available_channels is not None:
available_channels.add(chan)
elif available_channels is not None:
available_channels.discard(chan)
elif available_channels is not None:
if channels[chan].is_available():
available_channels.add(chan)
else:
available_channels.discard(chan)
else:
_track(chan)
# Return managed values writes to be applied externally
return updated_channels
libs/langgraph/langgraph/pregel/_loop.py
-1
View File
@@ -180,7 +180,6 @@ class PregelLoop:
_migrate_checkpoint: Callable[[Checkpoint], None] | None
submit: Submit
channels: Mapping[str, BaseChannel]
_has_untracked_channels: bool
_available_channels: set[str]
managed: ManagedValueMapping
checkpoint: Checkpoint