Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Langchain ai Langgraph AnyValue Channel

From Leeroopedia
Knowledge Sources
Domains Channels, State
Last Updated 2026-02-11 16:00 GMT

Overview

A channel implementation that stores the last value received, assuming all values written in the same step are equivalent, without performing guard validation.

Description

The AnyValue channel extends `BaseChannel` to provide a simple last-write-wins storage mechanism. When multiple values are written to the channel within a single step, `AnyValue` stores only the last value in the sequence without raising an error or performing any equality checks. This differs from `LastValue` (which enforces a single-writer constraint) and makes `AnyValue` suitable for channels where multiple nodes may write the same logical value.

The channel uses the `MISSING` sentinel from `langgraph._internal._typing` to distinguish between an empty (never-written) state and a state holding an actual value (including `None`). When `update()` is called with an empty sequence, the channel transitions back to the `MISSING` state if it previously held a value, or remains empty if it was already `MISSING`. The `get()` method raises `EmptyChannelError` when the channel is in the `MISSING` state.

Checkpointing is fully supported: `checkpoint()` returns the current value (which may be `MISSING`), and `from_checkpoint()` restores the channel to the checkpointed state. The `copy()` method provides an efficient shallow copy that preserves the current value and type information. The `is_available()` method provides a fast boolean check without the overhead of exception handling.

Usage

Use `AnyValue` for internal graph channels where multiple nodes may produce the same value in a single step and you want to accept any of them without raising a conflict error. It is commonly used for system-level channels in the Pregel execution engine where the value is deterministic regardless of which node wrote it.

Code Reference

Source Location

Signature

class AnyValue(Generic[Value], BaseChannel[Value, Value, Value]):
    def __init__(self, typ: Any, key: str = "") -> None: ...

    @property
    def ValueType(self) -> type[Value]: ...
    @property
    def UpdateType(self) -> type[Value]: ...

    def copy(self) -> Self: ...
    def from_checkpoint(self, checkpoint: Value) -> Self: ...
    def update(self, values: Sequence[Value]) -> bool: ...
    def get(self) -> Value: ...
    def is_available(self) -> bool: ...
    def checkpoint(self) -> Value: ...

Import

from langgraph.channels.any_value import AnyValue

I/O Contract

Constructor Parameters

Parameter Type Required Description
typ `Any` Yes The type of value stored in the channel
key `str` No Channel key name (default: `""`)

update

Input Output Description
`values: Sequence[Value]` (non-empty) `True` Stores the last value in the sequence
`values: Sequence[Value]` (empty, was set) `True` Clears to `MISSING` state
`values: Sequence[Value]` (empty, was MISSING) `False` No change

get

State Behavior
Value present Returns the stored value
`MISSING` Raises `EmptyChannelError`

Checkpoint Round-Trip

Method Description
`checkpoint()` Returns current value (may be `MISSING` sentinel)
`from_checkpoint(val)` Creates new channel; restores value if not `MISSING`

Usage Examples

from langgraph.channels.any_value import AnyValue
from langgraph.errors import EmptyChannelError

# Create a channel for string values
channel = AnyValue(str, key="status")

# Initially empty
assert not channel.is_available()

# Update with values - stores the last one
channel.update(["processing", "complete"])
assert channel.get() == "complete"
assert channel.is_available()

# Checkpoint and restore
cp = channel.checkpoint()
restored = channel.from_checkpoint(cp)
assert restored.get() == "complete"

# Empty update clears the channel
channel.update([])
try:
    channel.get()
except EmptyChannelError:
    print("Channel is now empty")

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment