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

Remove local_write utility (#4751)

Browse Source 
- The validation isn't worth the cost of having to pass list of nodes to task config
This commit is contained in:
Nuno Campos
2025-05-19 15:49:17 -04:00
committed by GitHub
parent d825e39df9
commit 46a9d3159d
2 changed files with 5 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/pregel/__init__.py
+2 -11
View File
@@ -85,7 +85,6 @@ from langgraph.pregel.algo import (
PregelTaskWrites,
apply_writes,
local_read,
local_write,
prepare_next_tasks,
)
from langgraph.pregel.call import identifier
@@ -1684,11 +1683,7 @@ class Pregel(PregelProtocol):
run_name=self.name + "UpdateState",
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
self.nodes.keys(),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,
@@ -2111,11 +2106,7 @@ class Pregel(PregelProtocol):
run_name=self.name + "UpdateState",
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
self.nodes.keys(),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,
libs/langgraph/langgraph/pregel/algo.py
+3 -32
View File
@@ -64,7 +64,6 @@ from langgraph.constants import (
TASKS,
Send,
)
from langgraph.errors import InvalidUpdateError
from langgraph.managed.base import ManagedValueMapping
from langgraph.pregel.call import get_runnable_for_task, identifier
from langgraph.pregel.io import read_channels
@@ -212,22 +211,6 @@ def local_read(
return values
def local_write(
commit: Callable[[Sequence[tuple[str, Any]]], None],
process_keys: Iterable[str],
writes: Sequence[tuple[str, Any]],
) -> None:
"""Function injected under CONFIG_KEY_SEND in task config, to write to channels.
Validates writes and forwards them to `commit` function."""
for chan, value in writes:
if chan in (PUSH, TASKS) and value is not None:
if not isinstance(value, Send):
raise InvalidUpdateError(f"Expected Send, got {value}")
if value.node not in process_keys:
raise InvalidUpdateError(f"Invalid node name {value.node} in packet")
commit(writes)
def increment(current: Optional[int], channel: BaseChannel) -> int:
"""Default channel versioning function, increments the current int version."""
return current + 1 if current is not None else 1
@@ -626,11 +609,7 @@ def prepare_single_task(
configurable={
CONFIG_KEY_TASK_ID: task_id,
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
processes.keys(),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,
@@ -750,11 +729,7 @@ def prepare_single_task(
configurable={
CONFIG_KEY_TASK_ID: task_id,
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
processes.keys(),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,
@@ -888,11 +863,7 @@ def prepare_single_task(
configurable={
CONFIG_KEY_TASK_ID: task_id,
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
tuple(processes.keys()),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,