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

push

Browse Source
This commit is contained in:
Sydney Runkle
2026-03-05 09:51:24 -08:00
parent 26d279a0ac
commit eb2f09e321
1 changed files with 14 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/pregel/_loop.py
+14 -28
View File
@@ -645,6 +645,20 @@ class PregelLoop:
configurable.get(CONFIG_KEY_RESUMING, input_signals_resume)
)
# When replaying from a specific checkpoint, drop cached RESUME
# writes so that interrupt() calls re-fire instead of returning
# stale values. But if a resume value is being provided (e.g.
# Command(resume=...) or CONFIG_KEY_RESUMING), keep them —
# multi-interrupt scenarios need previously resolved values preserved.
if self.is_replaying:
is_resume_with_value = (
isinstance(self.input, Command) and self.input.resume is not None
) or configurable.get(CONFIG_KEY_RESUMING, False)
if not is_resume_with_value:
self.checkpoint_pending_writes = [
w for w in self.checkpoint_pending_writes if w[1] != RESUME
]
# map command to writes
if isinstance(self.input, Command):
if (resume := self.input.resume) is not None:
@@ -1131,20 +1145,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
if saved.pending_writes is not None
else []
)
# When replaying from a specific checkpoint, drop cached RESUME
# writes so that interrupt() calls re-fire instead of returning
# stale values. But if a resume value is being provided (e.g.
# Command(resume=...) on a specific checkpoint), keep them —
# multi-interrupt scenarios need previously resolved values preserved.
if self.is_replaying:
has_resume_value = (
isinstance(self.input, Command) and self.input.resume is not None
) or self.config.get(CONF, {}).get(CONFIG_KEY_RESUMING, False)
if not has_resume_value:
self.checkpoint_pending_writes = [
w for w in self.checkpoint_pending_writes if w[1] != RESUME
]
self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
self.channels, self.managed = channels_from_checkpoint(
self.specs, self.checkpoint
@@ -1331,20 +1331,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
if saved.pending_writes is not None
else []
)
# When replaying from a specific checkpoint, drop cached RESUME
# writes so that interrupt() calls re-fire instead of returning
# stale values. But if a resume value is being provided (e.g.
# Command(resume=...) on a specific checkpoint), keep them —
# multi-interrupt scenarios need previously resolved values preserved.
if self.is_replaying:
has_resume_value = (
isinstance(self.input, Command) and self.input.resume is not None
) or self.config.get(CONF, {}).get(CONFIG_KEY_RESUMING, False)
if not has_resume_value:
self.checkpoint_pending_writes = [
w for w in self.checkpoint_pending_writes if w[1] != RESUME
]
self.submit = await self.stack.enter_async_context(
AsyncBackgroundExecutor(self.config)
)