WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-17 21:25:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

linting and test fixes

Browse Source
This commit is contained in:
Sydney Runkle
2025-04-22 17:10:54 -07:00
parent a8b70cb23c
commit 85e73653f9
5 changed files with 170 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/pregel/__init__.py
+24 -16
View File
@@ -2806,19 +2806,23 @@ class Pregel(PregelProtocol):
**kwargs,
):
if stream_mode == "values":
if isinstance(chunk, dict):
if (ints := chunk.get(INTERRUPT)) is not None:
interrupts.extend(ints)
latest = chunk
if (
isinstance(chunk, dict)
and (ints := chunk.get(INTERRUPT)) is not None
):
interrupts.extend(ints)
else:
latest = chunk
else:
chunks.append(chunk)
if stream_mode == "values":
if interrupts:
return {
**latest,
INTERRUPT: interrupts,
}
return (
{**latest, INTERRUPT: interrupts}
if isinstance(latest, dict)
else {INTERRUPT: interrupts}
)
return latest
else:
return chunks
@@ -2871,19 +2875,23 @@ class Pregel(PregelProtocol):
**kwargs,
):
if stream_mode == "values":
if isinstance(chunk, dict):
if (ints := chunk.get(INTERRUPT)) is not None:
interrupts.extend(ints)
latest = chunk
if (
isinstance(chunk, dict)
and (ints := chunk.get(INTERRUPT)) is not None
):
interrupts.extend(ints)
else:
latest = chunk
else:
chunks.append(chunk)
if stream_mode == "values":
if interrupts:
return {
**latest,
INTERRUPT: interrupts,
}
return (
{**latest, INTERRUPT: interrupts}
if isinstance(latest, dict)
else {INTERRUPT: interrupts}
)
return latest
else:
return chunks
libs/langgraph/langgraph/types.py
+4 -7
View File
@@ -1,7 +1,5 @@
import dataclasses
import hashlib
import sys
import uuid
from collections import deque
from collections.abc import Hashable, Sequence
from typing import (
@@ -21,6 +19,7 @@ from typing import (
from langchain_core.runnables import Runnable, RunnableConfig
from typing_extensions import Self
from xxhash import xxh3_128_hexdigest
from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata
from langgraph.utils.fields import get_update_as_tuples
@@ -146,10 +145,9 @@ class Interrupt:
@property
def interrupt_id(self) -> str:
"""Generate a unique ID for the interrupt based on its namespace."""
identifier = (
uuid.uuid4().bytes if self.ns is None else "".join(self.ns).encode()
)
return hashlib.sha256(identifier).hexdigest()
if self.ns is None:
return "placeholder-id"
return xxh3_128_hexdigest("".join(self.ns).encode())
class StateUpdate(NamedTuple):
@@ -486,7 +484,6 @@ def interrupt(value: Any) -> Any:
GraphInterrupt: On the first invocation within the node, halts execution and surfaces the provided value to the client.
"""
from langgraph.constants import (
CONF,
CONFIG_KEY_CHECKPOINT_NS,
CONFIG_KEY_SCRATCHPAD,
CONFIG_KEY_SEND,
libs/langgraph/tests/test_large_cases.py
+35 -9
View File
@@ -5666,6 +5666,9 @@ def test_dynamic_interrupt(
) == {
"my_key": "value",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -5712,6 +5715,9 @@ def test_dynamic_interrupt(
assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
"my_key": "value ⛰️",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
if "shallow" not in checkpointer_name:
@@ -5840,6 +5846,9 @@ def test_copy_checkpoint(
) == {
"my_key": "value one",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -5890,6 +5899,9 @@ def test_copy_checkpoint(
assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
"my_key": "value ⛰️ one",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
if "shallow" not in checkpointer_name:
assert [c.metadata for c in tool_two.checkpointer.list(thread1)] == [
@@ -6040,6 +6052,13 @@ def test_dynamic_interrupt_subgraph(
) == {
"my_key": "value",
"market": "DE",
"__interrupt__": [
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:"), AnyStr("do:")],
)
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -6086,6 +6105,13 @@ def test_dynamic_interrupt_subgraph(
assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
"my_key": "value ⛰️",
"market": "DE",
"__interrupt__": [
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:"), AnyStr("do:")],
)
],
}
if "shallow" not in checkpointer_name:
@@ -7318,15 +7344,15 @@ def test_send_dedupe_on_resume(
graph = builder.compile(checkpointer=checkpointer)
thread1 = {"configurable": {"thread_id": "1"}}
assert graph.invoke(["0"], thread1, checkpoint_during=checkpoint_during) == [
"0",
"1",
"3.1",
"2|Command(goto=Send(node='2', arg=3))",
"2|Command(goto=Send(node='flaky', arg=4))",
"3",
"2|3",
]
assert graph.invoke(["0"], thread1, checkpoint_during=checkpoint_during) == {
"__interrupt__": [
Interrupt(
value="Bahh",
resumable=False,
ns=None,
),
],
}
assert builder.nodes["2"].runnable.func.ticks == 3
assert builder.nodes["flaky"].runnable.func.ticks == 1
# check state
libs/langgraph/tests/test_pregel.py
+36 -3
View File
@@ -3703,7 +3703,16 @@ def test_subgraph_checkpoint_true_interrupt(
assert graph.invoke(
{"foo": "foo"}, config, checkpoint_during=checkpoint_during
) == {"foo": "hi! foo"}
) == {
"foo": "hi! foo",
"__interrupt__": [
Interrupt(
value="Provide baz value",
resumable=True,
ns=[AnyStr("node_2"), AnyStr("subgraph_node_1:")],
)
],
}
assert graph.get_state(config, subgraphs=True).tasks[0].state.values == {
"bar": "hi! foo"
}
@@ -5444,7 +5453,7 @@ def test_interrupt_task_functional(
config = {"configurable": {"thread_id": "1"}}
# First run, interrupted at bar
assert not graph.invoke({"a": ""}, config)
graph.invoke({"a": ""}, config)
# Resume with an answer
res = graph.invoke(Command(resume="bar"), config)
assert res == {"a": "foobar"}
@@ -7334,10 +7343,27 @@ def test_interrupt_subgraph_reenter_checkpointer_true(
)
config = {"configurable": {"thread_id": "1"}}
assert parent.invoke({"foo": "", "counter": 0}, config) == {"foo": "", "counter": 0}
assert parent.invoke({"foo": "", "counter": 0}, config) == {
"foo": "",
"counter": 0,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
assert parent.invoke(Command(resume="bar"), config) == {
"foo": "subgraph_2",
"counter": 1,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
assert parent.invoke(Command(resume="qux"), config) == {
"foo": "subgraph_2|parent",
@@ -7362,6 +7388,13 @@ def test_interrupt_subgraph_reenter_checkpointer_true(
assert parent.invoke({"foo": "meow", "counter": 0}, config) == {
"foo": "meow",
"counter": 0,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
# confirm that we preserve the state values from the previous invocation
assert bar_values == [None, "barbaz", "quxbaz"]
libs/langgraph/tests/test_pregel_async.py
+71 -12
View File
@@ -531,6 +531,9 @@ async def test_dynamic_interrupt(checkpointer_name: str) -> None:
) == {
"my_key": "value",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -713,6 +716,13 @@ async def test_dynamic_interrupt_subgraph(checkpointer_name: str) -> None:
) == {
"my_key": "value",
"market": "DE",
"__interrupt__": [
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:"), AnyStr("do:")],
)
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -903,6 +913,9 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
) == {
"my_key": "value one",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -964,6 +977,13 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
) == {
"my_key": "value ⛰️ one",
"market": "DE",
"__interrupt__": [
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:")],
)
],
}
if "shallow" not in checkpointer_name:
@@ -1108,13 +1128,29 @@ async def test_node_not_cancelled_on_other_node_interrupted(
# writes from "awhile" are applied to last chunk
assert await graph.ainvoke({"hello": "world"}, thread) == {
"hello": "world again"
"hello": "world again",
"__interrupt__": [
Interrupt(
value="I am bad",
resumable=True,
ns=[AnyStr("bad:")],
)
],
}
assert not inner_task_cancelled
assert awhiles == 1
assert await graph.ainvoke(None, thread, debug=True) == {"hello": "world again"}
assert await graph.ainvoke(None, thread, debug=True) == {
"hello": "world again",
"__interrupt__": [
Interrupt(
value="I am bad",
resumable=True,
ns=[AnyStr("bad:")],
)
],
}
assert not inner_task_cancelled
assert awhiles == 1
@@ -2795,15 +2831,15 @@ async def test_send_dedupe_on_resume(
thread1 = {"configurable": {"thread_id": "1"}}
assert await graph.ainvoke(
["0"], thread1, checkpoint_during=checkpoint_during
) == [
"0",
"1",
"3.1",
"2|Command(goto=Send(node='2', arg=3))",
"2|Command(goto=Send(node='flaky', arg=4))",
"3",
"2|3",
]
) == {
"__interrupt__": [
Interrupt(
value="Bahh",
resumable=False,
ns=None,
),
],
}
assert builder.nodes["2"].runnable.func.ticks == 3
assert builder.nodes["flaky"].runnable.func.ticks == 1
# resume execution
@@ -5555,7 +5591,16 @@ async def test_subgraph_checkpoint_true_interrupt(
assert await graph.ainvoke(
{"foo": "foo"}, config, checkpoint_during=checkpoint_during
) == {"foo": "hi! foo"}
) == {
"foo": "hi! foo",
"__interrupt__": [
Interrupt(
value="Provide baz value",
resumable=True,
ns=[AnyStr("node_2"), AnyStr("subgraph_node_1:")],
)
],
}
assert (await graph.aget_state(config, subgraphs=True)).tasks[
0
].state.values == {"bar": "hi! foo"}
@@ -8104,6 +8149,13 @@ async def test_interrupt_subgraph_reenter_checkpointer_true(
assert await parent.ainvoke({"foo": "", "counter": 0}, config) == {
"foo": "",
"counter": 0,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
assert await parent.ainvoke(Command(resume="bar"), config) == {
"foo": "subgraph_2",
@@ -8132,6 +8184,13 @@ async def test_interrupt_subgraph_reenter_checkpointer_true(
assert await parent.ainvoke({"foo": "meow", "counter": 0}, config) == {
"foo": "meow",
"counter": 0,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
# confirm that we preserve the state values from the previous invocation
assert bar_values == [None, "barbaz", "quxbaz"]