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 EphemeralValue Channel

From Leeroopedia
Revision as of 11:26, 16 February 2026 by Admin (talk | contribs) (Auto-imported from implementations/Langchain_ai_Langgraph_EphemeralValue_Channel.md)
(diff) ← Older revision | Latest revision (diff) | Newer revision → (diff)
Knowledge Sources
Domains Channels, State
Last Updated 2026-02-11 16:00 GMT

Overview

A channel that stores a value for exactly one step and automatically clears it in the next step, with an optional guard that enforces single-writer semantics.

Description

The EphemeralValue channel extends `BaseChannel` to provide transient, single-step value storage. When a value is written during step N, it is available for reading during step N but is automatically cleared when `update()` is called with an empty sequence at the beginning of step N+1. This makes it ideal for passing one-time signals or intermediate results between nodes within a single execution step without persisting them across steps.

The channel supports an optional `guard` parameter (defaulting to `True`) that enforces single-writer semantics. When `guard=True`, the `update()` method raises `InvalidUpdateError` if more than one value is written in the same step, preventing ambiguous state. Setting `guard=False` relaxes this constraint, storing the last value in the sequence when multiple writes occur. This mirrors the guard behavior found in `LastValue` but applied to an ephemeral context.

Like other channels, `EphemeralValue` uses the `MISSING` sentinel to represent the empty state and raises `EmptyChannelError` from `get()` when no value is present. Checkpointing preserves the current value (including `MISSING`) so that execution can resume correctly after interruption. The `from_checkpoint()` method restores the value if the checkpoint contains one, preserving the ephemeral channel's state across checkpoint boundaries.

Usage

Use `EphemeralValue` for channels that carry transient signals such as error flags, one-time notifications, or intermediate computation results that should not persist beyond the current step. It is particularly useful in graphs where a node needs to communicate a value to a downstream node within the same step but that value should not appear in subsequent steps.

Code Reference

Source Location

Signature

class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]):
    def __init__(self, typ: Any, guard: bool = True) -> 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.ephemeral_value import EphemeralValue

I/O Contract

Constructor Parameters

Parameter Type Required Description
typ `Any` Yes The type of value stored in the channel
guard `bool` No If `True` (default), raises `InvalidUpdateError` when more than one value is written per step

update

Input Guard Output Description
Empty sequence (had value) N/A `True` Clears channel to `MISSING`
Empty sequence (was MISSING) N/A `False` No change
Single value N/A `True` Stores the value
Multiple values `True` Raises `InvalidUpdateError` Guard prevents multi-write
Multiple values `False` `True` Stores the last value

get

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

Usage Examples

from langgraph.channels.ephemeral_value import EphemeralValue
from langgraph.errors import EmptyChannelError, InvalidUpdateError

# Create an ephemeral channel with guard
channel = EphemeralValue(str, guard=True)

# Write a value (simulating step N)
channel.update(["hello"])
assert channel.get() == "hello"
assert channel.is_available()

# Next step: empty update clears the value
channel.update([])
assert not channel.is_available()

try:
    channel.get()
except EmptyChannelError:
    print("Value cleared after step")

# Guard prevents multiple writers
channel_guarded = EphemeralValue(int, guard=True)
try:
    channel_guarded.update([1, 2])
except InvalidUpdateError:
    print("Cannot write multiple values with guard=True")

# Without guard, last value wins
channel_unguarded = EphemeralValue(int, guard=False)
channel_unguarded.update([1, 2, 3])
assert channel_unguarded.get() == 3

Related Pages

Page Connections

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