From 80c3ccba7bf52faff9089bb813c9a25f4252cb98 Mon Sep 17 00:00:00 2001 From: Eugene Yurtsev Date: Mon, 17 Mar 2025 16:45:02 -0400 Subject: [PATCH 01/20] add benchmark --- libs/langgraph/bench/__main__.py | 7 ++++++ libs/langgraph/bench/sequential.py | 39 ++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+) create mode 100644 libs/langgraph/bench/sequential.py diff --git a/libs/langgraph/bench/__main__.py b/libs/langgraph/bench/__main__.py index 677e79440..9b191f68d 100644 --- a/libs/langgraph/bench/__main__.py +++ b/libs/langgraph/bench/__main__.py @@ -7,6 +7,7 @@ from uvloop import new_event_loop from bench.fanout_to_subgraph import fanout_to_subgraph, fanout_to_subgraph_sync from bench.react_agent import react_agent +from bench.sequential import create_sequential from bench.wide_state import wide_state from langgraph.checkpoint.memory import MemorySaver from langgraph.pregel import Pregel @@ -203,6 +204,12 @@ benchmarks = ( ] }, ), + ( + "sequential_graph_200_nodes", + create_sequential(200).compile(), + create_sequential(200).compile(), + {"messages": []}, # Empty list of messages + ), ) diff --git a/libs/langgraph/bench/sequential.py b/libs/langgraph/bench/sequential.py new file mode 100644 index 000000000..1565c2044 --- /dev/null +++ b/libs/langgraph/bench/sequential.py @@ -0,0 +1,39 @@ +"""Create a sequential no-op graph consisting of a few hundred nodes.""" + +from langgraph.graph import MessagesState, StateGraph + + +def create_sequential(number_nodes) -> StateGraph: + """Create a sequential no-op graph consisting of a few hundred nodes.""" + builder = StateGraph(MessagesState) + + async def noop(state: MessagesState) -> None: + """No-op function.""" + pass + + prev_node = "__start__" + + for i in range(number_nodes): + name = f"node_{i}" + builder.add_node(name, noop) + builder.add_edge(prev_node, name) + prev_node = name + + builder.add_edge(prev_node, "__end__") + return builder + + +if __name__ == "__main__": + import asyncio + + import uvloop + + graph = create_sequential(200).compile() + input = {"messages": []} # Empty list of messages + config = {"recursion_limit": 20000000000} + + async def run(): + len([c async for c in graph.astream(input, config=config)]) + + uvloop.install() + asyncio.run(run()) From 969958695a34329b1566e51a5e01218da279d61c Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 17 Mar 2025 20:59:10 -0700 Subject: [PATCH 02/20] Add time when running directly --- libs/langgraph/bench/sequential.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/libs/langgraph/bench/sequential.py b/libs/langgraph/bench/sequential.py index 1565c2044..cceffd627 100644 --- a/libs/langgraph/bench/sequential.py +++ b/libs/langgraph/bench/sequential.py @@ -25,6 +25,7 @@ def create_sequential(number_nodes) -> StateGraph: if __name__ == "__main__": import asyncio + import time import uvloop @@ -36,4 +37,7 @@ if __name__ == "__main__": len([c async for c in graph.astream(input, config=config)]) uvloop.install() + start = time.time() asyncio.run(run()) + end = time.time() + print(f"Time taken: {end - start:.4f} seconds") From ce1077da40bc10c95d3bd6b80f8a052ef0ba5913 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 17 Mar 2025 21:04:05 -0700 Subject: [PATCH 03/20] Speed up task triggers check - Using a sentinel value is faster than raising-catching an exception --- libs/langgraph/langgraph/channels/base.py | 9 +++++++++ .../langgraph/channels/ephemeral_value.py | 16 ++++++++++------ libs/langgraph/langgraph/channels/last_value.py | 15 +++++++++++---- .../langgraph/channels/untracked_value.py | 10 +++++++--- libs/langgraph/langgraph/pregel/algo.py | 5 ++--- libs/langgraph/langgraph/pregel/io.py | 7 ++----- 6 files changed, 41 insertions(+), 21 deletions(-) diff --git a/libs/langgraph/langgraph/channels/base.py b/libs/langgraph/langgraph/channels/base.py index 4aaeb5681..c4b49c650 100644 --- a/libs/langgraph/langgraph/channels/base.py +++ b/libs/langgraph/langgraph/channels/base.py @@ -3,6 +3,7 @@ from typing import Any, Generic, Optional, Sequence, TypeVar from typing_extensions import Self +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError, InvalidUpdateError Value = TypeVar("Value") @@ -64,6 +65,14 @@ class BaseChannel(Generic[Value, Update, C], ABC): """ return False + def get_catch(self) -> Value: + """Return the current value of the channel, or MISSING if the channel + is empty. Subclasses can override to skip the EmptyChannelError check.""" + try: + return self.get() + except EmptyChannelError: + return MISSING + __all__ = [ "BaseChannel", diff --git a/libs/langgraph/langgraph/channels/ephemeral_value.py b/libs/langgraph/langgraph/channels/ephemeral_value.py index 537a8763c..4a64c6b32 100644 --- a/libs/langgraph/langgraph/channels/ephemeral_value.py +++ b/libs/langgraph/langgraph/channels/ephemeral_value.py @@ -3,6 +3,7 @@ from typing import Any, Generic, Optional, Sequence, Type from typing_extensions import Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError, InvalidUpdateError @@ -14,6 +15,7 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): def __init__(self, typ: Any, guard: bool = True) -> None: super().__init__(typ) self.guard = guard + self.value = MISSING def __eq__(self, value: object) -> bool: return isinstance(value, EphemeralValue) and value.guard == self.guard @@ -37,10 +39,10 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): def update(self, values: Sequence[Value]) -> bool: if len(values) == 0: - try: - del self.value + if self.value is not MISSING: + self.value = MISSING return True - except AttributeError: + else: return False if len(values) != 1 and self.guard: raise InvalidUpdateError( @@ -51,7 +53,9 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def get_catch(self) -> Value: + return self.value diff --git a/libs/langgraph/langgraph/channels/last_value.py b/libs/langgraph/langgraph/channels/last_value.py index 5065f4fc5..13fac983e 100644 --- a/libs/langgraph/langgraph/channels/last_value.py +++ b/libs/langgraph/langgraph/channels/last_value.py @@ -1,8 +1,9 @@ -from typing import Generic, Optional, Sequence, Type +from typing import Any, Generic, Optional, Sequence, Type from typing_extensions import Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import ( EmptyChannelError, ErrorCode, @@ -16,6 +17,10 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]): __slots__ = ("value",) + def __init__(self, typ: Any, key: str = "") -> None: + super().__init__(typ, key) + self.value = MISSING + def __eq__(self, value: object) -> bool: return isinstance(value, LastValue) @@ -50,7 +55,9 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]): return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def get_catch(self) -> Value: + return self.value diff --git a/libs/langgraph/langgraph/channels/untracked_value.py b/libs/langgraph/langgraph/channels/untracked_value.py index 9b1020710..2560d61a3 100644 --- a/libs/langgraph/langgraph/channels/untracked_value.py +++ b/libs/langgraph/langgraph/channels/untracked_value.py @@ -3,6 +3,7 @@ from typing import Generic, Optional, Sequence, Type from typing_extensions import Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError, InvalidUpdateError @@ -14,6 +15,7 @@ class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]): def __init__(self, typ: Type[Value], guard: bool = True) -> None: super().__init__(typ) self.guard = guard + self.value = MISSING def __eq__(self, value: object) -> bool: return isinstance(value, UntrackedValue) and value.guard == self.guard @@ -48,7 +50,9 @@ class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]): return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def get_catch(self) -> Value: + return self.value diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index 03b2af6f4..b2bcdcedf 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -48,6 +48,7 @@ from langgraph.constants import ( EMPTY_SEQ, ERROR, INTERRUPT, + MISSING, NO_WRITES, NS_END, NS_SEP, @@ -649,9 +650,7 @@ def prepare_single_task( if triggers := sorted( chan for chan in proc.triggers - if not isinstance( - read_channel(channels, chan, return_exception=True), EmptyChannelError - ) + if channels[chan].get_catch() is not MISSING and checkpoint["channel_versions"].get(chan, null_version) # type: ignore[operator] > seen.get(chan, null_version) ): diff --git a/libs/langgraph/langgraph/pregel/io.py b/libs/langgraph/langgraph/pregel/io.py index e9963f5c8..30e976d99 100644 --- a/libs/langgraph/langgraph/pregel/io.py +++ b/libs/langgraph/langgraph/pregel/io.py @@ -38,14 +38,11 @@ def read_channel( chan: str, *, catch: bool = True, - return_exception: bool = False, ) -> Any: try: return channels[chan].get() - except EmptyChannelError as exc: - if return_exception: - return exc - elif catch: + except EmptyChannelError: + if catch: return None else: raise From d6a457ef1d664f7c232da3a70d03f97a49f0f9d6 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 17 Mar 2025 21:26:26 -0700 Subject: [PATCH 04/20] Improve prepare_single_task trigger checks to linear complexity - Was O(n^2) due to individual channels created for every conditional edge, including the default cond edge created for Command - Now using a single channel per node for all conditional edge / command triggers, reducing to linear complexity - Improves run time on sequential(200) from 1.8s to 0.14s --- libs/langgraph/langgraph/graph/state.py | 11 ++++------- libs/langgraph/langgraph/pregel/io.py | 3 +-- libs/langgraph/tests/test_large_cases.py | 16 ++++++---------- libs/langgraph/tests/test_large_cases_async.py | 16 ++++++---------- 4 files changed, 17 insertions(+), 29 deletions(-) diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index 11df42b05..a01c3dfcb 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -821,9 +821,11 @@ class CompiledStateGraph(CompiledGraph): input_values = {k: k for k in self.builder.schemas[input_schema]} is_single_input = len(input_values) == 1 and "__root__" in input_values + branch_channel = f"branch:to:{key}" self.channels[key] = EphemeralValue(Any, guard=False) + self.channels[branch_channel] = EphemeralValue(Any, guard=False) self.nodes[key] = PregelNode( - triggers=[], + triggers=[branch_channel], # read state keys and managed values channels=(list(input_values) if is_single_input else input_values), # coerce state dict to schema class (eg. pydantic model) @@ -878,7 +880,7 @@ class CompiledStateGraph(CompiledGraph): if filtered := [p for p in packets if p != END]: writes = [ ( - ChannelWriteEntry(f"branch:{start}:{name}:{p}", start) + ChannelWriteEntry(f"branch:to:{p}", start) if not isinstance(p, Send) else p ) @@ -914,11 +916,6 @@ class CompiledStateGraph(CompiledGraph): if branch.ends else [node for node in self.builder.nodes if node != branch.then] ) - for end in ends: - if end != END: - channel_name = f"branch:{start}:{name}:{end}" - self.channels[channel_name] = EphemeralValue(Any, guard=False) - self.nodes[end].triggers.append(channel_name) # attach then subscriber if branch.then and branch.then != END: diff --git a/libs/langgraph/langgraph/pregel/io.py b/libs/langgraph/langgraph/pregel/io.py index 30e976d99..55ed413c0 100644 --- a/libs/langgraph/langgraph/pregel/io.py +++ b/libs/langgraph/langgraph/pregel/io.py @@ -14,7 +14,6 @@ from langgraph.constants import ( NULL_TASK_ID, RESUME, RETURN, - SELF, START, TAG_HIDDEN, TASKS, @@ -81,7 +80,7 @@ def map_command( if isinstance(send, Send): yield (NULL_TASK_ID, TASKS, send) elif isinstance(send, str): - yield (NULL_TASK_ID, f"branch:{START}:{SELF}:{send}", START) + yield (NULL_TASK_ID, f"branch:to:{send}", START) else: raise TypeError( f"In Command.goto, expected Send/str, got {type(send).__name__}" diff --git a/libs/langgraph/tests/test_large_cases.py b/libs/langgraph/tests/test_large_cases.py index f049e4e81..57c9b4071 100644 --- a/libs/langgraph/tests/test_large_cases.py +++ b/libs/langgraph/tests/test_large_cases.py @@ -2500,7 +2500,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None: { "langgraph_step": 2, "langgraph_node": "tools", - "langgraph_triggers": ["branch:agent:should_continue:tools"], + "langgraph_triggers": ["branch:to:tools"], "langgraph_path": (PULL, "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2559,7 +2559,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None: { "langgraph_step": 4, "langgraph_node": "tools", - "langgraph_triggers": ["branch:agent:should_continue:tools"], + "langgraph_triggers": ["branch:to:tools"], "langgraph_path": (PULL, "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2573,7 +2573,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None: { "langgraph_step": 4, "langgraph_node": "tools", - "langgraph_triggers": ["branch:agent:should_continue:tools"], + "langgraph_triggers": ["branch:to:tools"], "langgraph_path": (PULL, "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -6706,7 +6706,7 @@ def test_branch_then( "id": AnyStr(), "name": "tool_two_slow", "input": {"my_key": "value prepared", "market": "DE"}, - "triggers": ["branch:prepare:condition:tool_two_slow"], + "triggers": ["branch:to:tool_two_slow"], }, }, { @@ -10378,9 +10378,7 @@ def test_weather_subgraph( "langgraph_node": "weather_graph", "langgraph_path": [PULL, "weather_graph"], "langgraph_step": 2, - "langgraph_triggers": [ - "branch:router_node:route_after_prediction:weather_graph" - ], + "langgraph_triggers": ["branch:to:weather_graph"], "langgraph_checkpoint_ns": AnyStr("weather_graph:"), }, created_at=AnyStr(), @@ -10492,9 +10490,7 @@ def test_weather_subgraph( "langgraph_node": "weather_graph", "langgraph_path": [PULL, "weather_graph"], "langgraph_step": 2, - "langgraph_triggers": [ - "branch:router_node:route_after_prediction:weather_graph" - ], + "langgraph_triggers": ["branch:to:weather_graph"], "langgraph_checkpoint_ns": AnyStr("weather_graph:"), }, created_at=AnyStr(), diff --git a/libs/langgraph/tests/test_large_cases_async.py b/libs/langgraph/tests/test_large_cases_async.py index 250cbfe72..95fcb9646 100644 --- a/libs/langgraph/tests/test_large_cases_async.py +++ b/libs/langgraph/tests/test_large_cases_async.py @@ -2317,7 +2317,7 @@ async def test_prebuilt_tool_chat() -> None: { "langgraph_step": 2, "langgraph_node": "tools", - "langgraph_triggers": ["branch:agent:should_continue:tools"], + "langgraph_triggers": ["branch:to:tools"], "langgraph_path": ("__pregel_pull", "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2376,7 +2376,7 @@ async def test_prebuilt_tool_chat() -> None: { "langgraph_step": 4, "langgraph_node": "tools", - "langgraph_triggers": ["branch:agent:should_continue:tools"], + "langgraph_triggers": ["branch:to:tools"], "langgraph_path": ("__pregel_pull", "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2390,7 +2390,7 @@ async def test_prebuilt_tool_chat() -> None: { "langgraph_step": 4, "langgraph_node": "tools", - "langgraph_triggers": ["branch:agent:should_continue:tools"], + "langgraph_triggers": ["branch:to:tools"], "langgraph_path": ("__pregel_pull", "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -4537,7 +4537,7 @@ async def test_branch_then(checkpointer_name: str) -> None: "id": AnyStr(), "name": "tool_two_slow", "input": {"my_key": "value prepared", "market": "DE"}, - "triggers": ["branch:prepare:condition:tool_two_slow"], + "triggers": ["branch:to:tool_two_slow"], }, }, { @@ -7231,9 +7231,7 @@ async def test_weather_subgraph( "langgraph_node": "weather_graph", "langgraph_path": [PULL, "weather_graph"], "langgraph_step": 2, - "langgraph_triggers": [ - "branch:router_node:route_after_prediction:weather_graph" - ], + "langgraph_triggers": ["branch:to:weather_graph"], "langgraph_checkpoint_ns": AnyStr("weather_graph:"), }, created_at=AnyStr(), @@ -7347,9 +7345,7 @@ async def test_weather_subgraph( "langgraph_node": "weather_graph", "langgraph_path": [PULL, "weather_graph"], "langgraph_step": 2, - "langgraph_triggers": [ - "branch:router_node:route_after_prediction:weather_graph" - ], + "langgraph_triggers": ["branch:to:weather_graph"], "langgraph_checkpoint_ns": AnyStr("weather_graph:"), }, created_at=AnyStr(), From f0abf582dda828868aee496fa9659d34a9aefba3 Mon Sep 17 00:00:00 2001 From: Hamza Kyamanywa Date: Tue, 18 Mar 2025 22:25:32 +0900 Subject: [PATCH 05/20] docs: make sentence relating to how to navigate between sub graphs clearer in the docs (#3896) - fix typo / add missing word - make sentence relating to how to navigate between sub graphs clearer in the docs --- docs/docs/how-tos/command.ipynb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/docs/how-tos/command.ipynb b/docs/docs/how-tos/command.ipynb index a1f24f27b..cf0b01ae9 100644 --- a/docs/docs/how-tos/command.ipynb +++ b/docs/docs/how-tos/command.ipynb @@ -33,7 +33,7 @@ " )\n", "```\n", "\n", - "If you are using [subgraphs](#subgraphs), you might want to navigate from a node a subgraph to a different subgraph (i.e. a different node in the parent graph). To do so, you can specify `graph=Command.PARENT` in `Command`:\n", + "If you are using [subgraphs](#subgraphs), you might want to navigate from a node within a subgraph to a different subgraph (i.e. a different node in the parent graph). To do so, you can specify `graph=Command.PARENT` in `Command`:\n", "\n", "```python\n", "def my_node(state: State) -> Command[Literal[\"my_other_node\"]]:\n", From 3f241d00a3913ec5f8b21e58b6b3af9e04620128 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 06:45:41 -0700 Subject: [PATCH 06/20] Fix bench --- libs/langgraph/bench/__main__.py | 14 +++++++++++++- libs/langgraph/bench/sequential.py | 11 ++++++++--- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/libs/langgraph/bench/__main__.py b/libs/langgraph/bench/__main__.py index 9b191f68d..12d04a6ab 100644 --- a/libs/langgraph/bench/__main__.py +++ b/libs/langgraph/bench/__main__.py @@ -205,11 +205,23 @@ benchmarks = ( }, ), ( - "sequential_graph_200_nodes", + "sequential_50", + create_sequential(20).compile(), + create_sequential(20).compile(), + {"messages": []}, # Empty list of messages + ), + ( + "sequential_200", create_sequential(200).compile(), create_sequential(200).compile(), {"messages": []}, # Empty list of messages ), + # ( + # "sequential_1000", + # create_sequential(1000).compile(), + # create_sequential(1000).compile(), + # {"messages": []}, # Empty list of messages + # ), ) diff --git a/libs/langgraph/bench/sequential.py b/libs/langgraph/bench/sequential.py index cceffd627..3ab92912f 100644 --- a/libs/langgraph/bench/sequential.py +++ b/libs/langgraph/bench/sequential.py @@ -1,13 +1,18 @@ """Create a sequential no-op graph consisting of a few hundred nodes.""" from langgraph.graph import MessagesState, StateGraph +from langgraph.utils.runnable import RunnableCallable def create_sequential(number_nodes) -> StateGraph: """Create a sequential no-op graph consisting of a few hundred nodes.""" builder = StateGraph(MessagesState) - async def noop(state: MessagesState) -> None: + def noop(state: MessagesState) -> None: + """No-op function.""" + pass + + async def anoop(state: MessagesState) -> None: """No-op function.""" pass @@ -15,7 +20,7 @@ def create_sequential(number_nodes) -> StateGraph: for i in range(number_nodes): name = f"node_{i}" - builder.add_node(name, noop) + builder.add_node(name, RunnableCallable(noop, anoop)) builder.add_edge(prev_node, name) prev_node = name @@ -29,7 +34,7 @@ if __name__ == "__main__": import uvloop - graph = create_sequential(200).compile() + graph = create_sequential(2000).compile() input = {"messages": []} # Empty list of messages config = {"recursion_limit": 20000000000} From 8e829f38af9031bdeccbfcfa46c8bed7d43e7ec6 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 08:05:25 -0700 Subject: [PATCH 07/20] Smaller sizes until we merge the fixes --- libs/langgraph/bench/__main__.py | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/libs/langgraph/bench/__main__.py b/libs/langgraph/bench/__main__.py index 12d04a6ab..da6b7b6f3 100644 --- a/libs/langgraph/bench/__main__.py +++ b/libs/langgraph/bench/__main__.py @@ -205,18 +205,30 @@ benchmarks = ( }, ), ( - "sequential_50", + "sequential_20", create_sequential(20).compile(), create_sequential(20).compile(), {"messages": []}, # Empty list of messages ), ( - "sequential_200", - create_sequential(200).compile(), - create_sequential(200).compile(), + "sequential_50", + create_sequential(50).compile(), + create_sequential(50).compile(), {"messages": []}, # Empty list of messages ), # ( + # "sequential_100", + # create_sequential(100).compile(), + # create_sequential(100).compile(), + # {"messages": []}, # Empty list of messages + # ), + # ( + # "sequential_200", + # create_sequential(200).compile(), + # create_sequential(200).compile(), + # {"messages": []}, # Empty list of messages + # ), + # ( # "sequential_1000", # create_sequential(1000).compile(), # create_sequential(1000).compile(), From 47d38a302283acf676a2ad4ea9ee541e3978acf6 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 08:19:22 -0700 Subject: [PATCH 08/20] Replace get_catch w is_available --- .../langgraph/langgraph/channels/any_value.py | 23 ++++++++++++------- libs/langgraph/langgraph/channels/base.py | 14 ++++++----- libs/langgraph/langgraph/channels/binop.py | 13 +++++++---- .../channels/dynamic_barrier_value.py | 3 +++ .../langgraph/channels/ephemeral_value.py | 4 ++-- .../langgraph/channels/last_value.py | 4 ++-- .../langgraph/channels/named_barrier_value.py | 3 +++ libs/langgraph/langgraph/channels/topic.py | 3 +++ .../langgraph/channels/untracked_value.py | 4 ++-- libs/langgraph/langgraph/pregel/algo.py | 3 +-- 10 files changed, 47 insertions(+), 27 deletions(-) diff --git a/libs/langgraph/langgraph/channels/any_value.py b/libs/langgraph/langgraph/channels/any_value.py index e9dfb77d6..35452084f 100644 --- a/libs/langgraph/langgraph/channels/any_value.py +++ b/libs/langgraph/langgraph/channels/any_value.py @@ -1,8 +1,9 @@ -from typing import Generic, Optional, Sequence, Type +from typing import Any, Generic, Optional, Sequence, Type from typing_extensions import Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError @@ -12,6 +13,10 @@ class AnyValue(Generic[Value], BaseChannel[Value, Value, Value]): __slots__ = ("typ", "value") + def __init__(self, typ: Any, key: str = "") -> None: + super().__init__(typ, key) + self.value = MISSING + def __eq__(self, value: object) -> bool: return isinstance(value, AnyValue) @@ -34,17 +39,19 @@ class AnyValue(Generic[Value], BaseChannel[Value, Value, Value]): def update(self, values: Sequence[Value]) -> bool: if len(values) == 0: - try: - del self.value - return True - except AttributeError: + if self.value is MISSING: return False + else: + self.value = MISSING + return True self.value = values[-1] return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/channels/base.py b/libs/langgraph/langgraph/channels/base.py index c4b49c650..b9239be7a 100644 --- a/libs/langgraph/langgraph/channels/base.py +++ b/libs/langgraph/langgraph/channels/base.py @@ -3,7 +3,6 @@ from typing import Any, Generic, Optional, Sequence, TypeVar from typing_extensions import Self -from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError, InvalidUpdateError Value = TypeVar("Value") @@ -65,13 +64,16 @@ class BaseChannel(Generic[Value, Update, C], ABC): """ return False - def get_catch(self) -> Value: - """Return the current value of the channel, or MISSING if the channel - is empty. Subclasses can override to skip the EmptyChannelError check.""" + def is_available(self) -> bool: + """Return True if the channel is available (not empty), False otherwise. + Subclasses should override this method to provide a more efficient + implementation than calling get() and catching EmptyChannelError. + """ try: - return self.get() + self.get() + return True except EmptyChannelError: - return MISSING + return False __all__ = [ diff --git a/libs/langgraph/langgraph/channels/binop.py b/libs/langgraph/langgraph/channels/binop.py index a2360142b..413e0b91a 100644 --- a/libs/langgraph/langgraph/channels/binop.py +++ b/libs/langgraph/langgraph/channels/binop.py @@ -10,6 +10,7 @@ from typing import ( from typing_extensions import NotRequired, Required, Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError @@ -51,7 +52,7 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]): try: self.value = typ() except Exception: - pass + self.value = MISSING def __eq__(self, value: object) -> bool: return isinstance(value, BinaryOperatorAggregate) and ( @@ -81,7 +82,7 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]): def update(self, values: Sequence[Value]) -> bool: if not values: return False - if not hasattr(self, "value"): + if self.value is MISSING: self.value = values[0] values = values[1:] for value in values: @@ -89,7 +90,9 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]): return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/channels/dynamic_barrier_value.py b/libs/langgraph/langgraph/channels/dynamic_barrier_value.py index f64191e86..155c65446 100644 --- a/libs/langgraph/langgraph/channels/dynamic_barrier_value.py +++ b/libs/langgraph/langgraph/channels/dynamic_barrier_value.py @@ -85,6 +85,9 @@ class DynamicBarrierValue( raise EmptyChannelError() return None + def is_available(self) -> bool: + return self.seen == self.names + def consume(self) -> bool: if self.seen == self.names: self.seen = set() diff --git a/libs/langgraph/langgraph/channels/ephemeral_value.py b/libs/langgraph/langgraph/channels/ephemeral_value.py index 4a64c6b32..29a9a698c 100644 --- a/libs/langgraph/langgraph/channels/ephemeral_value.py +++ b/libs/langgraph/langgraph/channels/ephemeral_value.py @@ -57,5 +57,5 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): raise EmptyChannelError() return self.value - def get_catch(self) -> Value: - return self.value + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/channels/last_value.py b/libs/langgraph/langgraph/channels/last_value.py index 13fac983e..61669d390 100644 --- a/libs/langgraph/langgraph/channels/last_value.py +++ b/libs/langgraph/langgraph/channels/last_value.py @@ -59,5 +59,5 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]): raise EmptyChannelError() return self.value - def get_catch(self) -> Value: - return self.value + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/channels/named_barrier_value.py b/libs/langgraph/langgraph/channels/named_barrier_value.py index 4a1d990ca..553316e19 100644 --- a/libs/langgraph/langgraph/channels/named_barrier_value.py +++ b/libs/langgraph/langgraph/channels/named_barrier_value.py @@ -60,6 +60,9 @@ class NamedBarrierValue(Generic[Value], BaseChannel[Value, Value, set[Value]]): raise EmptyChannelError() return None + def is_available(self) -> bool: + return self.seen == self.names + def consume(self) -> bool: if self.seen == self.names: self.seen = set() diff --git a/libs/langgraph/langgraph/channels/topic.py b/libs/langgraph/langgraph/channels/topic.py index 0430343dc..91e7027f9 100644 --- a/libs/langgraph/langgraph/channels/topic.py +++ b/libs/langgraph/langgraph/channels/topic.py @@ -75,3 +75,6 @@ class Topic( return list(self.values) else: raise EmptyChannelError + + def is_available(self) -> bool: + return bool(self.values) diff --git a/libs/langgraph/langgraph/channels/untracked_value.py b/libs/langgraph/langgraph/channels/untracked_value.py index 2560d61a3..f9168131e 100644 --- a/libs/langgraph/langgraph/channels/untracked_value.py +++ b/libs/langgraph/langgraph/channels/untracked_value.py @@ -54,5 +54,5 @@ class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]): raise EmptyChannelError() return self.value - def get_catch(self) -> Value: - return self.value + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index b2bcdcedf..e8fd76a1e 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -48,7 +48,6 @@ from langgraph.constants import ( EMPTY_SEQ, ERROR, INTERRUPT, - MISSING, NO_WRITES, NS_END, NS_SEP, @@ -650,7 +649,7 @@ def prepare_single_task( if triggers := sorted( chan for chan in proc.triggers - if channels[chan].get_catch() is not MISSING + if channels[chan].is_available() and checkpoint["channel_versions"].get(chan, null_version) # type: ignore[operator] > seen.get(chan, null_version) ): From 60fc49b448c25847e3119cd0e3f9a549a01923af Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 17 Mar 2025 21:56:58 -0700 Subject: [PATCH 09/20] Speed up prepare_single_task - sequential(2000) goes from 8.4s to 4.7s - replace UUID(str).bytes with simpler str.encode() - find only the first active trigger, instead of the full list - use a dedicated function for checking active trigger --- libs/langgraph/langgraph/pregel/algo.py | 40 ++++++++++++++++++------- 1 file changed, 30 insertions(+), 10 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index ba028d7bd..614f61dc8 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -18,7 +18,6 @@ from typing import ( cast, overload, ) -from uuid import UUID from langchain_core.callbacks import Callbacks from langchain_core.callbacks.manager import AsyncParentRunManager, ParentRunManager @@ -27,6 +26,7 @@ from langchain_core.runnables.config import RunnableConfig from langgraph.channels.base import BaseChannel from langgraph.checkpoint.base import ( BaseCheckpointSaver, + ChannelVersions, Checkpoint, PendingWrite, V, @@ -432,7 +432,7 @@ def prepare_single_task( ) -> Union[None, PregelTask, PregelExecutableTask]: """Prepares a single task for the next Pregel step, given a task path, which uniquely identifies a PUSH or PULL task within the graph.""" - checkpoint_id = UUID(checkpoint["id"]).bytes + checkpoint_id = checkpoint["id"].encode() configurable = config.get(CONF, {}) parent_ns = configurable.get(CONFIG_KEY_CHECKPOINT_NS, "") @@ -641,18 +641,18 @@ def prepare_single_task( if name not in processes: return proc = processes[name] - version_type = type(next(iter(checkpoint["channel_versions"].values()), None)) + versions = checkpoint["channel_versions"] + version_type = type(next(iter(versions.values()), None)) null_version = version_type() # type: ignore[misc] if null_version is None: return - seen = checkpoint["versions_seen"].get(name, {}) # If any of the channels read by this process were updated - if triggers := sorted( - chan - for chan in proc.triggers - if channels[chan].is_available() - and checkpoint["channel_versions"].get(chan, null_version) # type: ignore[operator] - > seen.get(chan, null_version) + if triggers := _triggers( + channels, + versions, + checkpoint["versions_seen"].get(name), + null_version, + proc, ): try: val = next( @@ -761,6 +761,26 @@ def prepare_single_task( return PregelTask(task_id, name, task_path[:3]) +def _triggers( + channels: Mapping[str, BaseChannel], + versions: ChannelVersions, + seen: Optional[ChannelVersions], + null_version: V, + proc: PregelNode, +) -> Sequence[str]: + if seen is None: + for chan in proc.triggers: + if channels[chan].is_available(): + return (chan,) + else: + for chan in proc.triggers: + if channels[chan].is_available() and versions.get( + chan, null_version + ) > seen.get(chan, null_version): # type: ignore[operator] + return (chan,) + return EMPTY_SEQ + + def _scratchpad( config: RunnableConfig, pending_writes: list[PendingWrite], From 8bcdba822e4146286590203e546b2e316342991a Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 07:17:50 -0700 Subject: [PATCH 10/20] Reduce to 4.1s --- libs/langgraph/langgraph/pregel/algo.py | 34 +++++++++++++------ libs/langgraph/langgraph/pregel/loop.py | 6 ++++ .../langgraph/scheduler/kafka/executor.py | 7 +++- 3 files changed, 36 insertions(+), 11 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index 614f61dc8..1044cd1c4 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -1,3 +1,4 @@ +import binascii import itertools import sys from collections import defaultdict, deque @@ -373,6 +374,8 @@ def prepare_next_tasks( """Prepare the set of tasks that will make up the next Pregel step. This is the union of all PUSH tasks (Sends) and PULL tasks (nodes triggered by edges).""" + checkpoint_id_bytes = binascii.unhexlify(checkpoint["id"].replace("-", "")) + null_version = checkpoint_null_version(checkpoint) tasks: list[Union[PregelTask, PregelExecutableTask]] = [] # Consume pending_sends from previous step for idx, _ in enumerate(checkpoint["pending_sends"]): @@ -380,6 +383,8 @@ def prepare_next_tasks( (PUSH, idx), None, checkpoint=checkpoint, + checkpoint_id_bytes=checkpoint_id_bytes, + checkpoint_null_version=null_version, pending_writes=pending_writes, processes=processes, channels=channels, @@ -399,6 +404,8 @@ def prepare_next_tasks( (PULL, name), None, checkpoint=checkpoint, + checkpoint_id_bytes=checkpoint_id_bytes, + checkpoint_null_version=null_version, pending_writes=pending_writes, processes=processes, channels=channels, @@ -419,6 +426,8 @@ def prepare_single_task( task_id_checksum: Optional[str], *, checkpoint: Checkpoint, + checkpoint_id_bytes: bytes, + checkpoint_null_version: Optional[V], pending_writes: list[PendingWrite], processes: Mapping[str, PregelNode], channels: Mapping[str, BaseChannel], @@ -432,7 +441,6 @@ def prepare_single_task( ) -> Union[None, PregelTask, PregelExecutableTask]: """Prepares a single task for the next Pregel step, given a task path, which uniquely identifies a PUSH or PULL task within the graph.""" - checkpoint_id = checkpoint["id"].encode() configurable = config.get(CONF, {}) parent_ns = configurable.get(CONFIG_KEY_CHECKPOINT_NS, "") @@ -448,7 +456,7 @@ def prepare_single_task( triggers = [PUSH] checkpoint_ns = f"{parent_ns}{NS_SEP}{name}" if parent_ns else name task_id = _uuid5_str( - checkpoint_id, + checkpoint_id_bytes, checkpoint_ns, str(step), name, @@ -544,7 +552,7 @@ def prepare_single_task( f"{parent_ns}{NS_SEP}{packet.node}" if parent_ns else packet.node ) task_id = _uuid5_str( - checkpoint_id, + checkpoint_id_bytes, checkpoint_ns, str(step), packet.node, @@ -641,17 +649,14 @@ def prepare_single_task( if name not in processes: return proc = processes[name] - versions = checkpoint["channel_versions"] - version_type = type(next(iter(versions.values()), None)) - null_version = version_type() # type: ignore[misc] - if null_version is None: + if checkpoint_null_version is None: return # If any of the channels read by this process were updated if triggers := _triggers( channels, - versions, + checkpoint["channel_versions"], checkpoint["versions_seen"].get(name), - null_version, + checkpoint_null_version, proc, ): try: @@ -670,7 +675,7 @@ def prepare_single_task( # create task id checkpoint_ns = f"{parent_ns}{NS_SEP}{name}" if parent_ns else name task_id = _uuid5_str( - checkpoint_id, + checkpoint_id_bytes, checkpoint_ns, str(step), name, @@ -761,6 +766,15 @@ def prepare_single_task( return PregelTask(task_id, name, task_path[:3]) +def checkpoint_null_version( + checkpoint: Checkpoint, +) -> Optional[V]: + """Get the null version for the checkpoint, if available.""" + for version in checkpoint["channel_versions"].values(): + return type(version)() + return None + + def _triggers( channels: Mapping[str, BaseChannel], versions: ChannelVersions, diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index 7ab3431b7..edd69db01 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -1,4 +1,5 @@ import asyncio +import binascii import concurrent.futures from collections import defaultdict, deque from contextlib import AsyncExitStack, ExitStack @@ -79,6 +80,7 @@ from langgraph.pregel.algo import ( GetNextVersion, PregelTaskWrites, apply_writes, + checkpoint_null_version, increment, prepare_next_tasks, prepare_single_task, @@ -347,12 +349,16 @@ class PregelLoop(LoopProtocol): ): self.to_interrupt.append(task) return + checkpoint_id_bytes = binascii.unhexlify(self.checkpoint["id"].replace("-", "")) + null_version = checkpoint_null_version(self.checkpoint) if pushed := cast( Optional[PregelExecutableTask], prepare_single_task( (PUSH, task.path, write_idx, task.id, call), None, checkpoint=self.checkpoint, + checkpoint_id_bytes=checkpoint_id_bytes, + checkpoint_null_version=null_version, pending_writes=self.checkpoint_pending_writes, processes=self.nodes, channels=self.channels, diff --git a/libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py b/libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py index fa9a221d0..b8ab27674 100644 --- a/libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py +++ b/libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py @@ -1,4 +1,5 @@ import asyncio +import binascii import concurrent.futures from collections.abc import Sequence from contextlib import ( @@ -19,7 +20,7 @@ import langgraph.scheduler.kafka.serde as serde from langgraph.constants import CONFIG_KEY_DELEGATE, ERROR from langgraph.errors import CheckpointNotLatest, GraphDelegate, TaskNotFound from langgraph.pregel import Pregel -from langgraph.pregel.algo import prepare_single_task +from langgraph.pregel.algo import checkpoint_null_version, prepare_single_task from langgraph.pregel.executor import ( AsyncBackgroundExecutor, BackgroundExecutor, @@ -421,6 +422,10 @@ class KafkaExecutor(AbstractContextManager): step=saved.metadata["step"] + 1, for_execution=True, checkpointer=self.graph.checkpointer, + checkpoint_id_bytes=binascii.unhexlify( + saved.checkpoint["id"].replace("-", "") + ), + checkpoint_null_version=checkpoint_null_version(saved.checkpoint), ): # execute task, saving writes runner = PregelRunner( From 951131c8ecc40277165038459ebd07e3c64ab14d Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 08:42:12 -0700 Subject: [PATCH 11/20] Lint --- libs/langgraph/langgraph/pregel/algo.py | 11 +++++++---- libs/langgraph/langgraph/types.py | 2 +- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index 1044cd1c4..027f51ee7 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -421,6 +421,9 @@ def prepare_next_tasks( return {t.id: t for t in tasks} +PUSH_TRIGGER = (PUSH,) + + def prepare_single_task( task_path: tuple[Any, ...], task_id_checksum: Optional[str], @@ -453,7 +456,7 @@ def prepare_single_task( if name is None: raise ValueError("`call` functions must have a `__name__` attribute") # create task id - triggers = [PUSH] + triggers: Sequence[str] = PUSH_TRIGGER checkpoint_ns = f"{parent_ns}{NS_SEP}{name}" if parent_ns else name task_id = _uuid5_str( checkpoint_id_bytes, @@ -547,7 +550,7 @@ def prepare_single_task( ) return # create task id - triggers = [PUSH] + triggers = PUSH_TRIGGER checkpoint_ns = ( f"{parent_ns}{NS_SEP}{packet.node}" if parent_ns else packet.node ) @@ -788,9 +791,9 @@ def _triggers( return (chan,) else: for chan in proc.triggers: - if channels[chan].is_available() and versions.get( + if channels[chan].is_available() and versions.get( # type: ignore[operator] chan, null_version - ) > seen.get(chan, null_version): # type: ignore[operator] + ) > seen.get(chan, null_version): return (chan,) return EMPTY_SEQ diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 5efb89385..4c8cf6da4 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -149,7 +149,7 @@ class PregelExecutableTask(NamedTuple): proc: Runnable writes: deque[tuple[str, Any]] config: RunnableConfig - triggers: list[str] + triggers: Sequence[str] retry_policy: Optional[RetryPolicy] cache_policy: Optional[CachePolicy] id: str From 98b8ff904ceaa45e629f290d629af250ba49ba6c Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 09:30:29 -0700 Subject: [PATCH 12/20] Update test assertions for triggers --- libs/langgraph/tests/test_large_cases.py | 26 ++++++++--------- .../langgraph/tests/test_large_cases_async.py | 28 +++++++++---------- libs/langgraph/tests/test_pregel.py | 24 ++++++---------- libs/langgraph/tests/test_pregel_async.py | 8 +++--- 4 files changed, 39 insertions(+), 47 deletions(-) diff --git a/libs/langgraph/tests/test_large_cases.py b/libs/langgraph/tests/test_large_cases.py index 57c9b4071..a9d54b98d 100644 --- a/libs/langgraph/tests/test_large_cases.py +++ b/libs/langgraph/tests/test_large_cases.py @@ -2483,7 +2483,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None: { "langgraph_step": 1, "langgraph_node": "agent", - "langgraph_triggers": ["start:agent"], + "langgraph_triggers": ("start:agent",), "langgraph_path": (PULL, "agent"), "langgraph_checkpoint_ns": AnyStr("agent:"), "checkpoint_ns": AnyStr("agent:"), @@ -2500,7 +2500,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None: { "langgraph_step": 2, "langgraph_node": "tools", - "langgraph_triggers": ["branch:to:tools"], + "langgraph_triggers": ("branch:to:tools",), "langgraph_path": (PULL, "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2542,7 +2542,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None: { "langgraph_step": 3, "langgraph_node": "agent", - "langgraph_triggers": ["tools"], + "langgraph_triggers": ("tools",), "langgraph_path": (PULL, "agent"), "langgraph_checkpoint_ns": AnyStr("agent:"), "checkpoint_ns": AnyStr("agent:"), @@ -2559,7 +2559,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None: { "langgraph_step": 4, "langgraph_node": "tools", - "langgraph_triggers": ["branch:to:tools"], + "langgraph_triggers": ("branch:to:tools",), "langgraph_path": (PULL, "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2573,7 +2573,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None: { "langgraph_step": 4, "langgraph_node": "tools", - "langgraph_triggers": ["branch:to:tools"], + "langgraph_triggers": ("branch:to:tools",), "langgraph_path": (PULL, "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2585,7 +2585,7 @@ def test_prebuilt_tool_chat(snapshot: SnapshotAssertion) -> None: { "langgraph_step": 5, "langgraph_node": "agent", - "langgraph_triggers": ["tools"], + "langgraph_triggers": ("tools",), "langgraph_path": (PULL, "agent"), "langgraph_checkpoint_ns": AnyStr("agent:"), "checkpoint_ns": AnyStr("agent:"), @@ -5501,7 +5501,7 @@ def test_in_one_fan_out_out_one_graph_state() -> None: "id": AnyStr(), "name": "rewrite_query", "input": {"query": "what is weather in sf", "docs": []}, - "triggers": ["start:rewrite_query"], + "triggers": ("start:rewrite_query",), }, }, ), @@ -5532,7 +5532,7 @@ def test_in_one_fan_out_out_one_graph_state() -> None: "id": AnyStr(), "name": "retriever_one", "input": {"query": "query: what is weather in sf", "docs": []}, - "triggers": ["rewrite_query"], + "triggers": ("rewrite_query",), }, }, ), @@ -5546,7 +5546,7 @@ def test_in_one_fan_out_out_one_graph_state() -> None: "id": AnyStr(), "name": "retriever_two", "input": {"query": "query: what is weather in sf", "docs": []}, - "triggers": ["rewrite_query"], + "triggers": ("rewrite_query",), }, }, ), @@ -5608,7 +5608,7 @@ def test_in_one_fan_out_out_one_graph_state() -> None: "query": "query: what is weather in sf", "docs": ["doc1", "doc2", "doc3", "doc4"], }, - "triggers": ["retriever_one", "retriever_two"], + "triggers": ("retriever_one",), }, }, ), @@ -6634,7 +6634,7 @@ def test_branch_then( "id": AnyStr(), "name": "prepare", "input": {"my_key": "value", "market": "DE"}, - "triggers": ["start:prepare"], + "triggers": ("start:prepare",), }, }, { @@ -6706,7 +6706,7 @@ def test_branch_then( "id": AnyStr(), "name": "tool_two_slow", "input": {"my_key": "value prepared", "market": "DE"}, - "triggers": ["branch:to:tool_two_slow"], + "triggers": ("branch:to:tool_two_slow",), }, }, { @@ -6773,7 +6773,7 @@ def test_branch_then( "id": AnyStr(), "name": "finish", "input": {"my_key": "value prepared slow", "market": "DE"}, - "triggers": ["branch:prepare:condition::then"], + "triggers": ("branch:prepare:condition::then",), }, }, { diff --git a/libs/langgraph/tests/test_large_cases_async.py b/libs/langgraph/tests/test_large_cases_async.py index 95fcb9646..f7f11da0c 100644 --- a/libs/langgraph/tests/test_large_cases_async.py +++ b/libs/langgraph/tests/test_large_cases_async.py @@ -2300,7 +2300,7 @@ async def test_prebuilt_tool_chat() -> None: { "langgraph_step": 1, "langgraph_node": "agent", - "langgraph_triggers": ["start:agent"], + "langgraph_triggers": ("start:agent",), "langgraph_path": ("__pregel_pull", "agent"), "langgraph_checkpoint_ns": AnyStr("agent:"), "checkpoint_ns": AnyStr("agent:"), @@ -2317,7 +2317,7 @@ async def test_prebuilt_tool_chat() -> None: { "langgraph_step": 2, "langgraph_node": "tools", - "langgraph_triggers": ["branch:to:tools"], + "langgraph_triggers": ("branch:to:tools",), "langgraph_path": ("__pregel_pull", "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2359,7 +2359,7 @@ async def test_prebuilt_tool_chat() -> None: { "langgraph_step": 3, "langgraph_node": "agent", - "langgraph_triggers": ["tools"], + "langgraph_triggers": ("tools",), "langgraph_path": ("__pregel_pull", "agent"), "langgraph_checkpoint_ns": AnyStr("agent:"), "checkpoint_ns": AnyStr("agent:"), @@ -2376,7 +2376,7 @@ async def test_prebuilt_tool_chat() -> None: { "langgraph_step": 4, "langgraph_node": "tools", - "langgraph_triggers": ["branch:to:tools"], + "langgraph_triggers": ("branch:to:tools",), "langgraph_path": ("__pregel_pull", "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2390,7 +2390,7 @@ async def test_prebuilt_tool_chat() -> None: { "langgraph_step": 4, "langgraph_node": "tools", - "langgraph_triggers": ["branch:to:tools"], + "langgraph_triggers": ("branch:to:tools",), "langgraph_path": ("__pregel_pull", "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, @@ -2402,7 +2402,7 @@ async def test_prebuilt_tool_chat() -> None: { "langgraph_step": 5, "langgraph_node": "agent", - "langgraph_triggers": ["tools"], + "langgraph_triggers": ("tools",), "langgraph_path": ("__pregel_pull", "agent"), "langgraph_checkpoint_ns": AnyStr("agent:"), "checkpoint_ns": AnyStr("agent:"), @@ -3883,7 +3883,7 @@ async def test_in_one_fan_out_out_one_graph_state() -> None: "id": AnyStr(), "name": "rewrite_query", "input": {"query": "what is weather in sf", "docs": []}, - "triggers": ["start:rewrite_query"], + "triggers": ("start:rewrite_query",), }, }, ), @@ -3914,7 +3914,7 @@ async def test_in_one_fan_out_out_one_graph_state() -> None: "id": AnyStr(), "name": "retriever_one", "input": {"query": "query: what is weather in sf", "docs": []}, - "triggers": ["rewrite_query"], + "triggers": ("rewrite_query",), }, }, ), @@ -3928,7 +3928,7 @@ async def test_in_one_fan_out_out_one_graph_state() -> None: "id": AnyStr(), "name": "retriever_two", "input": {"query": "query: what is weather in sf", "docs": []}, - "triggers": ["rewrite_query"], + "triggers": ("rewrite_query",), }, }, ), @@ -3990,7 +3990,7 @@ async def test_in_one_fan_out_out_one_graph_state() -> None: "query": "query: what is weather in sf", "docs": ["doc1", "doc2", "doc3", "doc4"], }, - "triggers": ["retriever_one", "retriever_two"], + "triggers": ("retriever_one",), }, }, ), @@ -4465,7 +4465,7 @@ async def test_branch_then(checkpointer_name: str) -> None: "id": AnyStr(), "name": "prepare", "input": {"my_key": "value", "market": "DE"}, - "triggers": ["start:prepare"], + "triggers": ("start:prepare",), }, }, { @@ -4537,7 +4537,7 @@ async def test_branch_then(checkpointer_name: str) -> None: "id": AnyStr(), "name": "tool_two_slow", "input": {"my_key": "value prepared", "market": "DE"}, - "triggers": ["branch:to:tool_two_slow"], + "triggers": ("branch:to:tool_two_slow",), }, }, { @@ -4609,7 +4609,7 @@ async def test_branch_then(checkpointer_name: str) -> None: "id": AnyStr(), "name": "finish", "input": {"my_key": "value prepared slow", "market": "DE"}, - "triggers": ["branch:prepare:condition::then"], + "triggers": ("branch:prepare:condition::then",), }, }, { @@ -4778,7 +4778,7 @@ async def test_branch_then(checkpointer_name: str) -> None: "id": AnyStr(), "name": "prepare", "input": {"my_key": "value", "market": "DE"}, - "triggers": ["start:prepare"], + "triggers": ("start:prepare",), }, }, { diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index ccd899ce9..f4af98b52 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -817,7 +817,7 @@ def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None: "id": AnyStr(), "name": "one", "input": 2, - "triggers": ["input"], + "triggers": ("input",), }, }, { @@ -828,7 +828,7 @@ def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None: "id": AnyStr(), "name": "two", "input": [12], - "triggers": ["inbox"], + "triggers": ("inbox",), }, }, { @@ -863,7 +863,7 @@ def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None: "id": AnyStr(), "name": "two", "input": [3], - "triggers": ["inbox"], + "triggers": ("inbox",), }, }, { @@ -5969,9 +5969,7 @@ def test_falsy_return_from_task( "a": 5, }, "name": "graph", - "triggers": [ - "__start__", - ], + "triggers": ("__start__",), }, "step": 0, "timestamp": AnyStr(), @@ -5985,9 +5983,7 @@ def test_falsy_return_from_task( {}, ), "name": "falsy_task", - "triggers": [ - "__pregel_push", - ], + "triggers": ("__pregel_push",), }, "step": 0, "timestamp": AnyStr(), @@ -6094,9 +6090,7 @@ def test_falsy_return_from_task( "a": 5, }, "name": "graph", - "triggers": [ - "__start__", - ], + "triggers": ("__start__",), }, "step": 0, "timestamp": AnyStr(), @@ -6110,9 +6104,7 @@ def test_falsy_return_from_task( {}, ), "name": "falsy_task", - "triggers": [ - "__pregel_push", - ], + "triggers": ("__pregel_push",), }, "step": 0, "timestamp": AnyStr(), @@ -6923,7 +6915,7 @@ def test_tags_stream_mode_messages() -> None: { "langgraph_step": 1, "langgraph_node": "call_model", - "langgraph_triggers": ["start:call_model"], + "langgraph_triggers": ("start:call_model",), "langgraph_path": ("__pregel_pull", "call_model"), "langgraph_checkpoint_ns": AnyStr("call_model:"), "checkpoint_ns": AnyStr("call_model:"), diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index a42a84a28..f1bb4d8ab 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -1672,7 +1672,7 @@ async def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None: "id": AnyStr(), "name": "one", "input": 2, - "triggers": ["input"], + "triggers": ("input",), }, }, { @@ -1683,7 +1683,7 @@ async def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None: "id": AnyStr(), "name": "two", "input": [12], - "triggers": ["inbox"], + "triggers": ("inbox",), }, }, { @@ -1718,7 +1718,7 @@ async def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None: "id": AnyStr(), "name": "two", "input": [3], - "triggers": ["inbox"], + "triggers": ("inbox",), }, }, { @@ -7571,7 +7571,7 @@ async def test_tags_stream_mode_messages() -> None: { "langgraph_step": 1, "langgraph_node": "call_model", - "langgraph_triggers": ["start:call_model"], + "langgraph_triggers": ("start:call_model",), "langgraph_path": ("__pregel_pull", "call_model"), "langgraph_checkpoint_ns": AnyStr("call_model:"), "checkpoint_ns": AnyStr("call_model:"), From fa96c0ac761350fbd8c0da5f508732bfe06e24d0 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 09:34:52 -0700 Subject: [PATCH 13/20] One more --- libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py b/libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py index b8ab27674..2d442e131 100644 --- a/libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py +++ b/libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py @@ -210,6 +210,10 @@ class AsyncKafkaExecutor(AbstractAsyncContextManager): for_execution=True, checkpointer=self.graph.checkpointer, store=self.graph.store, + checkpoint_id_bytes=binascii.unhexlify( + saved.checkpoint["id"].replace("-", "") + ), + checkpoint_null_version=checkpoint_null_version(saved.checkpoint), ): # execute task, saving writes runner = PregelRunner( From 9b5549f759aa3c4d10bc9211741d91e329e4d078 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 09:43:56 -0700 Subject: [PATCH 14/20] Fix flaky assertion --- libs/langgraph/tests/test_pregel.py | 26 ++++++++++++++++++-------- 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index f4af98b52..d8d51a8a0 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -3247,14 +3247,24 @@ def test_in_one_fan_out_state_graph_waiting_edge_plus_regular( assert [ c for c in app_w_interrupt.stream({"query": "what is weather in sf"}, config) - ] == [ - {"rewrite_query": {"query": "query: what is weather in sf"}}, - {"qa": {"answer": ""}}, - {"analyzer_one": {"query": "analyzed: query: what is weather in sf"}}, - {"retriever_two": {"docs": ["doc3", "doc4"]}}, - {"retriever_one": {"docs": ["doc1", "doc2"]}}, - {"__interrupt__": ()}, - ] + ] in ( + [ + {"rewrite_query": {"query": "query: what is weather in sf"}}, + {"qa": {"answer": ""}}, + {"analyzer_one": {"query": "analyzed: query: what is weather in sf"}}, + {"retriever_two": {"docs": ["doc3", "doc4"]}}, + {"retriever_one": {"docs": ["doc1", "doc2"]}}, + {"__interrupt__": ()}, + ], + [ + {"rewrite_query": {"query": "query: what is weather in sf"}}, + {"analyzer_one": {"query": "analyzed: query: what is weather in sf"}}, + {"qa": {"answer": ""}}, + {"retriever_two": {"docs": ["doc3", "doc4"]}}, + {"retriever_one": {"docs": ["doc1", "doc2"]}}, + {"__interrupt__": ()}, + ], + ) assert [c for c in app_w_interrupt.stream(None, config)] == [ {"qa": {"answer": "doc1,doc2,doc3,doc4"}}, From 82905297fd365fda77bef12df7f8a38126a679ee Mon Sep 17 00:00:00 2001 From: Yassin Nouh <70436855+YassinNouh21@users.noreply.github.com> Date: Tue, 18 Mar 2025 18:47:17 +0200 Subject: [PATCH 15/20] docs: Add Pydantic usage examples and runtime coercion documentation (#3588) ## Description This PR enhances the state-model documentation by adding comprehensive examples for advanced Pydantic usage in LangGraph. It addresses issue #2745 regarding the need for better documentation of Pydantic schema behavior. ### Changes - Added new section on Advanced Pydantic Model Usage - Added examples for serialization behavior with nested models - Added section on runtime type coercion with examples - Added documentation for proper message type handling (BaseMessage vs AnyMessage) - Updated Pydantic error URLs to latest version ### Related Issues Closes #2745 ### Testing - All notebook cells have been executed and outputs verified - Examples demonstrate proper usage patterns - Error cases are properly documented ### Documentation The changes are documentation-focused and include: - New examples for complex Pydantic models - Runtime coercion behavior examples - Message type handling best practices ### Reviewers @eyurtsev --- docs/docs/how-tos/state-model.ipynb | 220 ++++++++++++++++++++++++++++ 1 file changed, 220 insertions(+) diff --git a/docs/docs/how-tos/state-model.ipynb b/docs/docs/how-tos/state-model.ipynb index 221a4738d..4448170cf 100644 --- a/docs/docs/how-tos/state-model.ipynb +++ b/docs/docs/how-tos/state-model.ipynb @@ -266,6 +266,226 @@ " print(\"An exception was raised because bad_node sets `a` to an integer.\")\n", " print(e)" ] + }, + { + "cell_type": "markdown", + "id": "2270bc3c", + "metadata": {}, + "source": [ + "## Multiple Nodes\n", + "\n", + "Run-time validation will also work in a multi-node graph. In the example below `bad_node` updates `a` to an integer. \n", + "\n", + "Because run-time validation occurs on **inputs**, the validation error will occur when `ok_node` is called (not when `bad_node` returns an update to the state which is inconsistent with the schema)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d832cdcc", + "metadata": {}, + "outputs": [], + "source": [ + "from langgraph.graph import StateGraph, START, END\n", + "from typing_extensions import TypedDict\n", + "\n", + "from pydantic import BaseModel\n", + "\n", + "\n", + "# The overall state of the graph (this is the public state shared across nodes)\n", + "class OverallState(BaseModel):\n", + " a: str\n", + "\n", + "\n", + "def bad_node(state: OverallState):\n", + " return {\n", + " \"a\": 123 # Invalid\n", + " }\n", + "\n", + "\n", + "def ok_node(state: OverallState):\n", + " return {\"a\": \"goodbye\"}\n", + "\n", + "\n", + "# Build the state graph\n", + "builder = StateGraph(OverallState)\n", + "builder.add_node(bad_node)\n", + "builder.add_node(ok_node)\n", + "builder.add_edge(START, \"bad_node\")\n", + "builder.add_edge(\"bad_node\", \"ok_node\")\n", + "builder.add_edge(\"ok_node\", END)\n", + "graph = builder.compile()\n", + "\n", + "# Test the graph with a valid input\n", + "try:\n", + " graph.invoke({\"a\": \"hello\"})\n", + "except Exception as e:\n", + " print(\"An exception was raised because bad_node sets `a` to an integer.\")\n", + " print(e)" + ] + }, + { + "cell_type": "markdown", + "id": "456b1f77", + "metadata": {}, + "source": [ + "## Advanced Pydantic Model Usage\n", + "\n", + "This section covers more advanced topics when using Pydantic models with LangGraph.\n", + "\n", + "### Serialization Behavior\n", + "\n", + "When using Pydantic models as state schemas, it's important to understand how serialization works, especially when:\n", + "- Passing Pydantic objects as inputs\n", + "- Receiving outputs from the graph\n", + "- Working with nested Pydantic models\n", + "\n", + "Let's see these behaviors in action:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "0e919cdc", + "metadata": {}, + "outputs": [], + "source": [ + "from langgraph.graph import StateGraph, START, END\n", + "from pydantic import BaseModel\n", + "\n", + "class NestedModel(BaseModel):\n", + " value: str\n", + "\n", + "class ComplexState(BaseModel):\n", + " text: str\n", + " count: int\n", + " nested: NestedModel\n", + "\n", + "def process_node(state: ComplexState):\n", + " # Node receives a validated Pydantic object\n", + " print(f\"Input state type: {type(state)}\")\n", + " print(f\"Nested type: {type(state.nested)}\")\n", + " \n", + " # Return a dictionary update\n", + " return {\"text\": state.text + \" processed\", \"count\": state.count + 1}\n", + "\n", + "# Build the graph\n", + "builder = StateGraph(ComplexState)\n", + "builder.add_node(\"process\", process_node)\n", + "builder.add_edge(START, \"process\")\n", + "builder.add_edge(\"process\", END)\n", + "graph = builder.compile()\n", + "\n", + "# Create a Pydantic instance for input\n", + "input_state = ComplexState(text=\"hello\", count=0, nested=NestedModel(value=\"test\"))\n", + "print(f\"Input object type: {type(input_state)}\")\n", + "\n", + "# Invoke graph with a Pydantic instance\n", + "result = graph.invoke(input_state)\n", + "print(f\"Output type: {type(result)}\")\n", + "print(f\"Output content: {result}\")\n", + "\n", + "# Convert back to Pydantic model if needed\n", + "output_model = ComplexState(**result)\n", + "print(f\"Converted back to Pydantic: {type(output_model)}\")" + ] + }, + { + "cell_type": "markdown", + "id": "f13f28ce", + "metadata": {}, + "source": [ + "### Runtime Type Coercion\n", + "\n", + "Pydantic performs runtime type coercion for certain data types. This can be helpful but also lead to unexpected behavior if you're not aware of it." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "faf59316", + "metadata": {}, + "outputs": [], + "source": [ + "from langgraph.graph import StateGraph, START, END\n", + "from pydantic import BaseModel\n", + "\n", + "class CoercionExample(BaseModel):\n", + " # Pydantic will coerce string numbers to integers\n", + " number: int\n", + " # Pydantic will parse string booleans to bool\n", + " flag: bool\n", + "\n", + "def inspect_node(state: CoercionExample):\n", + " print(f\"number: {state.number} (type: {type(state.number)})\")\n", + " print(f\"flag: {state.flag} (type: {type(state.flag)})\")\n", + " return {}\n", + "\n", + "builder = StateGraph(CoercionExample)\n", + "builder.add_node(\"inspect\", inspect_node)\n", + "builder.add_edge(START, \"inspect\")\n", + "builder.add_edge(\"inspect\", END)\n", + "graph = builder.compile()\n", + "\n", + "# Demonstrate coercion with string inputs that will be converted\n", + "result = graph.invoke({\"number\": \"42\", \"flag\": \"true\"})\n", + "\n", + "# This would fail with a validation error\n", + "try:\n", + " graph.invoke({\"number\": \"not-a-number\", \"flag\": \"true\"})\n", + "except Exception as e:\n", + " print(f\"\\nExpected validation error: {e}\")" + ] + }, + { + "cell_type": "markdown", + "id": "2844475b", + "metadata": {}, + "source": [ + "### Working with Message Models\n", + "\n", + "When working with LangChain message types in your state schema, there are important considerations for serialization. You should use `AnyMessage` (rather than `BaseMessage`) for proper serialization/deserialization when using message objects over the wire:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "bd0734b0", + "metadata": {}, + "outputs": [], + "source": [ + "from langgraph.graph import StateGraph, START, END\n", + "from pydantic import BaseModel\n", + "from langchain_core.messages import HumanMessage, AIMessage, BaseMessage\n", + "from typing import List\n", + "\n", + "class ChatState(BaseModel):\n", + " messages: List[BaseMessage] \n", + " context: str\n", + "\n", + "def add_message(state: ChatState):\n", + " return {\"messages\": state.messages + [AIMessage(content=\"Hello there!\")]}\n", + "\n", + "builder = StateGraph(ChatState)\n", + "builder.add_node(\"add_message\", add_message)\n", + "builder.add_edge(START, \"add_message\")\n", + "builder.add_edge(\"add_message\", END)\n", + "graph = builder.compile()\n", + "\n", + "# Create input with a message\n", + "initial_state = ChatState(\n", + " messages=[HumanMessage(content=\"Hi\")],\n", + " context=\"Customer support chat\"\n", + ")\n", + "\n", + "result = graph.invoke(initial_state)\n", + "print(f\"Output: {result}\")\n", + "\n", + "# Convert back to Pydantic model to see message types\n", + "output_model = ChatState(**result)\n", + "for i, msg in enumerate(output_model.messages):\n", + " print(f\"Message {i}: {type(msg).__name__} - {msg.content}\")" + ] } ], "metadata": { From 7a959f62ccaf5fd978623ed6f5e20baaaf42aad9 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 09:54:20 -0700 Subject: [PATCH 16/20] Fix assertion --- libs/langgraph/tests/test_large_cases.py | 2 +- libs/langgraph/tests/test_large_cases_async.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/langgraph/tests/test_large_cases.py b/libs/langgraph/tests/test_large_cases.py index a9d54b98d..d6637f000 100644 --- a/libs/langgraph/tests/test_large_cases.py +++ b/libs/langgraph/tests/test_large_cases.py @@ -5608,7 +5608,7 @@ def test_in_one_fan_out_out_one_graph_state() -> None: "query": "query: what is weather in sf", "docs": ["doc1", "doc2", "doc3", "doc4"], }, - "triggers": ("retriever_one",), + "triggers": (AnyStr("retriever_"),), }, }, ), diff --git a/libs/langgraph/tests/test_large_cases_async.py b/libs/langgraph/tests/test_large_cases_async.py index f7f11da0c..157e0e080 100644 --- a/libs/langgraph/tests/test_large_cases_async.py +++ b/libs/langgraph/tests/test_large_cases_async.py @@ -3990,7 +3990,7 @@ async def test_in_one_fan_out_out_one_graph_state() -> None: "query": "query: what is weather in sf", "docs": ["doc1", "doc2", "doc3", "doc4"], }, - "triggers": ("retriever_one",), + "triggers": (AnyStr("retriever_"),), }, }, ), From e7fbdeeb136bfd720a0be947b14b9ca8bba14f56 Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Tue, 18 Mar 2025 13:00:36 -0400 Subject: [PATCH 17/20] docs: fix formatting (#3901) --- docs/docs/how-tos/state-model.ipynb | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/docs/docs/how-tos/state-model.ipynb b/docs/docs/how-tos/state-model.ipynb index 4448170cf..c4fb12041 100644 --- a/docs/docs/how-tos/state-model.ipynb +++ b/docs/docs/how-tos/state-model.ipynb @@ -353,22 +353,26 @@ "from langgraph.graph import StateGraph, START, END\n", "from pydantic import BaseModel\n", "\n", + "\n", "class NestedModel(BaseModel):\n", " value: str\n", "\n", + "\n", "class ComplexState(BaseModel):\n", " text: str\n", " count: int\n", " nested: NestedModel\n", "\n", + "\n", "def process_node(state: ComplexState):\n", " # Node receives a validated Pydantic object\n", " print(f\"Input state type: {type(state)}\")\n", " print(f\"Nested type: {type(state.nested)}\")\n", - " \n", + "\n", " # Return a dictionary update\n", " return {\"text\": state.text + \" processed\", \"count\": state.count + 1}\n", "\n", + "\n", "# Build the graph\n", "builder = StateGraph(ComplexState)\n", "builder.add_node(\"process\", process_node)\n", @@ -410,17 +414,20 @@ "from langgraph.graph import StateGraph, START, END\n", "from pydantic import BaseModel\n", "\n", + "\n", "class CoercionExample(BaseModel):\n", " # Pydantic will coerce string numbers to integers\n", " number: int\n", " # Pydantic will parse string booleans to bool\n", " flag: bool\n", "\n", + "\n", "def inspect_node(state: CoercionExample):\n", " print(f\"number: {state.number} (type: {type(state.number)})\")\n", " print(f\"flag: {state.flag} (type: {type(state.flag)})\")\n", " return {}\n", "\n", + "\n", "builder = StateGraph(CoercionExample)\n", "builder.add_node(\"inspect\", inspect_node)\n", "builder.add_edge(START, \"inspect\")\n", @@ -459,13 +466,16 @@ "from langchain_core.messages import HumanMessage, AIMessage, BaseMessage\n", "from typing import List\n", "\n", + "\n", "class ChatState(BaseModel):\n", - " messages: List[BaseMessage] \n", + " messages: List[BaseMessage]\n", " context: str\n", "\n", + "\n", "def add_message(state: ChatState):\n", " return {\"messages\": state.messages + [AIMessage(content=\"Hello there!\")]}\n", "\n", + "\n", "builder = StateGraph(ChatState)\n", "builder.add_node(\"add_message\", add_message)\n", "builder.add_edge(START, \"add_message\")\n", @@ -474,8 +484,7 @@ "\n", "# Create input with a message\n", "initial_state = ChatState(\n", - " messages=[HumanMessage(content=\"Hi\")],\n", - " context=\"Customer support chat\"\n", + " messages=[HumanMessage(content=\"Hi\")], context=\"Customer support chat\"\n", ")\n", "\n", "result = graph.invoke(initial_state)\n", From 3ec95153ce6cda248b9b5e85ad24e89603ae038a Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Tue, 18 Mar 2025 15:57:40 -0400 Subject: [PATCH 18/20] ci: don't use real secrets in notebook runner (#3572) --- .github/workflows/run_notebooks.yml | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/.github/workflows/run_notebooks.yml b/.github/workflows/run_notebooks.yml index 80d2aca44..16abf71d7 100644 --- a/.github/workflows/run_notebooks.yml +++ b/.github/workflows/run_notebooks.yml @@ -57,13 +57,13 @@ jobs: env: # these won't actually be used because of the VCR cassettes # but need to set them to avoid triggering getpass() - OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} - ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} - TAVILY_API_KEY: ${{ secrets.TAVILY_API_KEY }} - LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }} - NOMIC_API_KEY: ${{ secrets.NOMIC_API_KEY }} - COHERE_API_KEY: ${{ secrets.COHERE_API_KEY }} - FIREWORKS_API_KEY: ${{ secrets.FIREWORKS_API_KEY }} + OPENAI_API_KEY: "very-secret-key" + ANTHROPIC_API_KEY: "very-secret-key" + TAVILY_API_KEY: "very-secret-key" + LANGSMITH_API_KEY: "very-secret-key" + NOMIC_API_KEY: "very-secret-key" + COHERE_API_KEY: "very-secret-key" + FIREWORKS_API_KEY: "very-secret-key" run: | if [ "${{ github.event_name }}" = "workflow_dispatch" ] || [ "${{ github.event_name }}" = "schedule" ]; then echo "Running all notebooks" From ae7dbd1fa5be2130929e24f5a8ab2051e8019760 Mon Sep 17 00:00:00 2001 From: Hamza Kyamanywa Date: Wed, 19 Mar 2025 04:58:28 +0900 Subject: [PATCH 19/20] docs: correct the word "every" (#3902) - PR fix the word "every" in the sentence "It will be called every time the LLM is called" --- docs/docs/how-tos/update-state-from-tools.ipynb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/docs/how-tos/update-state-from-tools.ipynb b/docs/docs/how-tos/update-state-from-tools.ipynb index 0a7311342..1aea00556 100644 --- a/docs/docs/how-tos/update-state-from-tools.ipynb +++ b/docs/docs/how-tos/update-state-from-tools.ipynb @@ -210,7 +210,7 @@ "id": "cbb06aea-6654-4245-91f8-af6e8f2b5377", "metadata": {}, "source": [ - "Let's now add personalization: we'll respond differently to the user based on the state values AFTER the state has been updated from the tool. To achieve this, let's define a function that will dynamically construct the system prompt based on the graph state. It will be called ever time the LLM is called and the function output will be passed to the LLM:" + "Let's now add personalization: we'll respond differently to the user based on the state values AFTER the state has been updated from the tool. To achieve this, let's define a function that will dynamically construct the system prompt based on the graph state. It will be called every time the LLM is called and the function output will be passed to the LLM:" ] }, { From e1aa1a4510f16bd222d10cac4ab16dd5fa5e2e92 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 13:11:37 -0700 Subject: [PATCH 20/20] 0.3.15 --- libs/langgraph/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/langgraph/pyproject.toml b/libs/langgraph/pyproject.toml index 7893875e8..e49391685 100644 --- a/libs/langgraph/pyproject.toml +++ b/libs/langgraph/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "langgraph" -version = "0.3.14" +version = "0.3.15" description = "Building stateful, multi-actor applications with LLMs" authors = [] license = "MIT"