Implementation:Langchain ai Langgraph EphemeralValue Channel
| 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
- Repository: Langchain_ai_Langgraph
- File: libs/langgraph/langgraph/channels/ephemeral_value.py
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