Fixeslangchain-ai/langgraph#8534
`put` splits stored values in two: primitives stay inline in the
checkpoint's `channel_values`, everything else moves to
`checkpoint_blobs`, and only `_DeltaSnapshot` leaves an inline marker
behind when it moves. Stage-1 seed detection tested for that marker, so
a plain value — what a thread migrated from `BinaryOperatorAggregate`
leaves behind — was invisible to the walk.
### Effect
Migrated threads found no seed, walked to the root, and replayed every
write on every read. Values still came out correct, because replaying an
additive reducer from empty rebuilds the same list, which is why nothing
looked wrong. What was lost is early termination — the entire point of
`DeltaChannel`:
<!-- linear:table-colwidths:266,266,266 -->
| thread length | writes replayed, before | after |
| -- | -- | -- |
| 2 turns | 3 | 1 |
| 6 turns | 7 | 1 |
| 20 turns | 21 | 1 |
Read latency is flat at \~0.6ms across all three after the change.
### Approach
Stage 1 now checks both places a value can live rather than trusting the
marker. It probes `checkpoint_blobs`:
```sql
EXISTS (SELECT 1 FROM checkpoint_blobs b0
WHERE b0.thread_id = checkpoints.thread_id
AND b0.checkpoint_ns = checkpoints.checkpoint_ns
AND b0.channel = %s
AND b0.version = checkpoint -> 'channel_versions' ->> %s
AND b0.type <> 'empty') AS hb_0
```
and selects the inline value alongside it, since `None`, `str`, `int`,
`float` and `bool` stay in `channel_values` with no blob row:
```sql
checkpoint -> 'channel_values' -> %s AS inline_0
```
The blob predicate matches `checkpoint_blobs`' primary key `(thread_id,
checkpoint_ns, channel, version)` exactly, so it is one index lookup per
row per channel, bounded by the 1024-row page.
I picked reading storage over the cheaper alternative — also writing the
marker for plain values — because **that would not fix any thread
already on disk.** Existing checkpoints have no marker and there is
nowhere to add one retroactively.
The seed resolves to the blob when one exists and the inline value
otherwise. That ordering is also what keeps a genuine inline `true` — a
`bool` channel holding `True` — distinguishable from the literal `true`
marker `put` inlines for a `_DeltaSnapshot`: only the snapshot has a
blob.
`None` is deliberately not treated as a seed; a JSON null is
indistinguishable from "nothing stored" at this layer, so the walk
continues and replay from empty is correct.
Params go from two to four per channel; both callers updated.
The inline half came out of review on this PR — a blob-only probe would
have left scalar-aggregate migrations (an integer sum, say) still
replaying their full history.
### On the `type <> 'empty'` predicate
Being upfront since it isn't demonstrable with a test: `put` does not
currently produce `empty` rows on this path — `blob_versions` is
filtered to keys present in `channel_values`, so `_dump_blobs`' empty
branch is unreachable from it. I confirmed there are no `empty` rows in
a populated test database.
I kept it because stage 2 already applies the same check when resolving
the seed blob. Without it the two stages could disagree: stage 1
terminates the walk on a row stage 2 then discards, producing no seed
*and* a truncated write chain — the same failure shape this function
exists to avoid. Rationale is in the docstring so the next reader
doesn't have to ask. Happy to drop it if you'd rather not carry an
unexercised predicate.
### Tests
`libs/checkpoint-postgres/tests/test_delta_plain_value_seed.py` —
blob-stored plain-value seed, `_DeltaSnapshot` seed, a version bump with
nothing stored (which must not stop the walk short of an older real
value), inline primitives (`int`, `str`, `float`, `None`), and inline
`True` versus the snapshot marker. Each fails against the behaviour it
fixes.
Verified: postgres suite 269 passed on PG 15 and 16; delta-channel
conformance against `AsyncPostgresSaver` went from 6 of 8 to 8 of 8,
including the pre-existing `test_history_migration_plain_value_as_seed`
failure this was causing; `make lint` clean.
### Not included
I wanted a Postgres conformance runner alongside `checkpoint-sqlite`'s,
but it needs `langgraph-checkpoint-conformance` as a dev dependency and
the contributing guide asks for maintainer sign-off before adding one.
The direct tests above cover the same ground without it.
Worth flagging separately: **conformance effectively runs against**
`InMemorySaver` **only today.** `libs/checkpoint-conformance/tests/`
contains just `test_validate_memory.py`, and `checkpoint-sqlite`'s
`test_conformance_delta.py` silently skips because the package isn't
installed in its test environment (`importorskip`). Wiring it up for
sqlite and postgres is what would have caught this bug, and
langchain-ai/langgraph#8534 notes it.
Sqlite is unaffected by the bug itself — it stores `channel_values`
inline and inspects them directly. `langgraph-api` already resolves
seeds by version rather than by marker.
Follow-up to review on #8540, where a stale `# noqa: E402` slipped past
me and Sydney spotted it by eye. This turns on the rule that catches
that automatically.
`RUF100` flags a `noqa` that suppresses nothing. `sdk-py` already had it
through its blanket `RUF` selection; this adds it to the other seven
packages and clears what it finds.
### The 33 it flags, all autofixed
**Blanket `# noqa` on docstring-closing lines** (4, in
`checkpoint-postgres` and `checkpoint-sqlite`). `E501` is in
`lint.ignore` for those packages, so nothing was being suppressed:
```diff
- """ # noqa
+ """
```
**`# noqa: F821` on `anext(aiter_)`** (2). Left over from Python 3.9
support. `anext` became a builtin in 3.10, which is the floor now, so
`F821` no longer fires:
```diff
- anext(aiter_), # type: ignore[arg-type] # noqa: F821
+ anext(aiter_), # type: ignore[arg-type]
```
**Suppressions naming rules the package does not enable** (27), across
`langgraph`, `prebuilt` and `checkpoint-sqlite`: `FBT001`, `FBT002`,
`TC002`, `BLE001`, `ANN001`, `ANN002`, `ANN003`, `E501`, `F401`. Mostly
copied between packages whose rule sets differ.
### One measurement note
If you check these numbers yourself, use `--extend-select`:
```
ruff check --select RUF100 . # 81, misleading
ruff check --extend-select RUF100 . # 33, real
```
With a bare `--select`, ruff treats every other rule as disabled, so
every suppression for another rule looks unused. I quoted 81 before
catching that.
### Verified
`checkpoint-sqlite` 118 passed, `prebuilt` 284 passed, `langgraph` 1968
passed, `checkpoint-postgres` 264 passed on PG 15 and 16. `make lint`
clean in every package.
Independent of #8540 and #8537, so it can land in any order.
## Summary
- Ports the langgraph API checkpointer fix to the OSS Postgres
checkpoint.
- The `'b'` (seed-blob) branch of `_build_delta_stage2_sql`'s `UNION
ALL` was missing column aliases (`AS _kind`, `AS checkpoint_id`, `AS
task_id`, `AS idx`).
- In PostgreSQL, `UNION ALL` column names are taken from the **first**
`SELECT`. When `channels_with_chain` is empty but `channels_with_seed`
is not (e.g. very first run of a DeltaChannel graph, or
`snapshot_frequency=1`), the `'b'` branch becomes the first `SELECT`, so
columns got default names (`text`, `null`) instead of the expected
aliases — causing `KeyError: '_kind'` in
`_build_delta_channels_writes_history`.
## Test plan
- [ ] Existing `checkpoint-postgres` delta channel tests pass (`make
test` in `libs/checkpoint-postgres`)
- [ ] Manually verified fix matches the correction suggested in the
upstream API PR review comment
🤖 Generated with [Claude Code](https://claude.com/claude-code)
## Summary
- Promotes the private K-channel batched ancestor-walk to a stable
public `get_delta_channel_history` / `aget_delta_channel_history` API on
`BaseCheckpointSaver` (returns `Mapping[str, DeltaChannelHistory]`, a
TypedDict with `writes` always present and `seed` `NotRequired`)
- Removes `DELTA_SENTINEL` / `_DeltaSentinel` entirely — the saver layer
is now delta-agnostic on both write and read paths
- Reworks `DeltaChannel` snapshot cadence from "every Nth superstep" to
"every N updates to this channel," persisted in
`CheckpointMetadata.delta_updates_since_snapshot`
- Adds Postgres optimizations: paged stage-1 with cursor (1024-row
pages) and per-channel UNION ALL stage-2 (no over-fetch when channels
have different chain depths)
- Default `snapshot_frequency` becomes a positive int (default `1000`);
the previous `None` opt-out is removed
## Public API
```python
class DeltaChannelHistory(TypedDict):
writes: list[PendingWrite] # always present, possibly empty
seed: NotRequired[Any] # absent if walk reached root
def get_delta_channel_history(
self, *, config: RunnableConfig, channels: Sequence[str]
) -> Mapping[str, DeltaChannelHistory]: ...
async def aget_delta_channel_history(
self, *, config: RunnableConfig, channels: Sequence[str]
) -> Mapping[str, DeltaChannelHistory]: ...
```
`config` and `channels` are keyword-only so later additions (e.g.
`page_size`) don't shift the positional API.
The TypedDict-with-`NotRequired[seed]` shape matches the existing
checkpoint-package convention (`CheckpointMetadata` is
`TypedDict(total=False)`) — absence-via-key-omission rather than
introducing a new sentinel. Pregel translates `"seed" not in hist` to
`MISSING` on its side at consume time.
The default impl walks `get_tuple` + `parent_config` correctly but is
slow on long chains; savers that care override (`InMemorySaver`,
`PostgresSaver`).
## Sentinel removal
`DELTA_SENTINEL` and `_DeltaSentinel` are deleted entirely. The saver
layer becomes delta-agnostic:
- `DeltaChannel.checkpoint()` returns `MISSING` for non-snapshot steps;
pregel's `create_checkpoint` skips MISSING so delta channels without a
snapshot simply don't appear in `channel_values`
- `InMemorySaver.put` and Postgres `put` no longer filter sentinels
(they have nothing to filter)
- `_needs_replay` becomes `stored is MISSING`
- `DeltaChannel.from_checkpoint` accepts: `MISSING` → empty,
`_DeltaSnapshot(value)` → snapshot value, plain value → pre-migration
legacy
## Snapshot cadence
`DeltaChannel.snapshot_frequency: int` (default `1000`, positive). The
previous `None` opt-out is gone.
```python
def should_snapshot(ch_name, ch):
if force_delta_snapshot: # durability="exit"
return True
return updates_since_snapshot.get(ch_name, 0) >= ch.snapshot_frequency
```
Per-channel update counters are persisted in
`CheckpointMetadata.delta_updates_since_snapshot` (`NotRequired`,
`total=False`). The counter is incremented by `_put_checkpoint` for any
delta channel in `updated_channels` and reset to `0` by
`create_checkpoint` for channels that fire a snapshot this step.
Version-format-independent — works for `int`, `float`, and `str`
versioning schemes alike.
## Postgres optimization
Two improvements internal to the override:
**Stage-1 paged with cursor** (`LIMIT 1024` internal const, `AND
checkpoint_id < ?` for subsequent pages). The previous unpaged form
scanned every checkpoint in `(thread_id, ns)` and was pathological at
high thread depths.
**Stage-2 per-channel UNION ALL**: one `WHERE channel='X' AND
checkpoint_id = ANY(chain_X)` branch per channel plus one seed-blob
branch per channel with a seed. The previous form filtered by `channel =
ANY(channels) AND checkpoint_id = ANY(union_chain_cids)`, over-fetching
writes when channels had different chain depths (`K ×
max(chain_lengths)` vs the correct `sum(chain_lengths)`).
Both improvements stay internal to `PostgresSaver`/`AsyncPostgresSaver`;
the public contract returns a single `Mapping`.
## Benchmarks
`libs/langgraph/tests/test_delta_channel_benchmark.py`. Run via `python
libs/langgraph/tests/test_delta_channel_benchmark.py`. Postgres against
local pg:5441.
Results below trimmed to the high-signal cells. Sub-millisecond /
sub-100-turn rows omitted as warmup-bound; freq=1 omitted (chain depth =
1, nothing to optimize); peak read-time memory and Postgres storage are
flat between branches and omitted. Deep-thread reads and the
cadence-rework storage win are the load-bearing numbers.
### Postgres reads, 500 turns
| Scenario | main | branch | Δ |
|---|---:|---:|---:|
| Single-channel deep read | 17.7 ms | **6.1 ms** | **-66%** |
| Single-channel, 1000 turns | 35.0 ms | **14.3 ms** | **-59%** |
| K=3 channels, freq=50 uniform | 70.5 ms | **41.4 ms** | **-41%** |
| K=8 channels, freq=50 uniform | 214.2 ms | **139.4 ms** | **-35%** |
| K=8 channels, mixed freq (25/50/100/.../1000) | 295.6 ms | **214.4
ms** | **-27%** |
K-channel batching + paged stage-1 + per-channel UNION ALL stage-2 doing
exactly what they should at depth.
### InMemory reads, 500 turns
| Scenario | main | branch | Δ |
|---|---:|---:|---:|
| Single-channel deep read | 7.9 ms | **3.8 ms** | **-52%** |
| Single-channel, 1000 turns | 15.6 ms | **7.2 ms** | **-54%** |
| K=8 channels, freq=50 uniform | 112.3 ms | 94.6 ms | -16% |
| K=8 channels, mixed freq | 184.9 ms | **134.5 ms** | **-27%** |
### InMemory storage, 500 turns (cadence-rework win)
| Scenario | main | branch | Δ |
|---|---:|---:|---:|
| K=3, freq=50 uniform | 8.7 MB | **3.3 MB** | **-62%** |
| K=3 mixed freq | 3.8 MB | **1.3 MB** | **-66%** |
| K=8, freq=50 uniform | 23.1 MB | **8.7 MB** | **-62%** |
| K=8 mixed freq | 11.5 MB | **4.2 MB** | **-64%** |
Snapshot frequency now counts **channel updates** instead of
**supersteps**. On graphs where supersteps outpace per-channel updates
(e.g., input/end steps that don't write to channels), branch stores ~3×
fewer snapshot blobs.
### Tradeoff worth flagging
InMemory K=3 with mixed frequencies (50/200/1000) at 500 turns: **+64%
read latency** (46.6 → 76.5 ms). The mixed scenario has a channel with
`freq=1000` that goes the entire 500-turn run with no snapshot. On main,
the old superstep-counted cadence happened to fire at step=500 anyway.
New cadence gives users explicit control over walk depth via
`snapshot_frequency`. The K=8 mixed case still wins overall (-27%); this
regression is specific to the K=3 mixed shape.
Default `snapshot_frequency=1000` is the upper bound on walk depth —
it's a tunable knob.
## Tests
- New sqlite smoke test (`test_get_delta_channel_history.py`) exercises
the inherited default `BaseCheckpointSaver` impl via `SqliteSaver` /
`AsyncSqliteSaver` end-to-end with a real `DeltaChannel`-backed graph.
Sqlite uses the default unchanged — this validates the default path
actually works on a real second saver, not just on the optimized
override.
- Module-level `pytest.importorskip("langgraph.channels.delta")` guards
the test for sqlite's standalone CI environment (matches the postgres
pattern).
## Test plan
- [x] `libs/checkpoint`: 150 passed, 16 skipped
- [x] `libs/langgraph` (channels + delta migration): 41/41 (post-merge)
- [x] `libs/langgraph` (full pregel suite): 1784 passing — 6 "failures"
verified via `env -i` clean shell are local LangSmith env vars + `git
describe revision_id` polluting LangChain metadata fixtures; CI is
unaffected
- [x] `libs/checkpoint-postgres`: 40/40 saver tests + 3/3 delta channel
reconstruction tests against local Postgres
- [x] `libs/checkpoint-sqlite`: 105/105 (incl. retry-passed flake
`test_ttl_refresh`, unrelated to this PR)
- [x] Lint clean across all four libs (`ruff format`, `ruff check`,
`mypy`)
- [x] Branch-vs-main benchmarks — see results above
---------
Co-authored-by: Quanzheng Long <long@langchain.dev>
Co-authored-by: Cursor <cursoragent@cursor.com>
## Summary
Replaces the single-roundtrip `UNION ALL` DeltaChannel read with a
two-stage query that avoids fetching unused snapshot blobs, then removes
the old combined path entirely.
### Problem
`_get_channel_writes_history` used a single `UNION ALL` query that
fetched **all** checkpoint metadata, writes, and blobs for a
`(thread_id, channel)` in one shot. With `snapshot_frequency=N`, this
pulled back O(N/freq) full-size snapshot blobs even though only the
nearest one is needed to seed reconstruction. At 500 turns with
`snapshot_frequency=10`, this meant fetching ~100 complete
message-history snapshots per read.
### Solution
Two-stage read:
- **Stage 1** — lightweight scan of `checkpoints` only (no blob bytes):
walks the parent chain from the target checkpoint and stops at the first
ancestor with a snapshot, returning `chain_cids` and `seed_version`
- **Stage 2** — targeted fetch: only the writes for `chain_cids` and the
single seed blob at `seed_version`
The two-stage path is now unconditional — the old combined query and
`LG_DELTA_TWO_STAGE_QUERY` env-var gate have been removed.
### Sentinel cleanup
`DELTA_SENTINEL` is now a pure in-memory signal and is never written to
storage:
- Postgres `put()` already stripped it from `channel_values` before
writing blobs
- Memory saver `put()` now stores `"empty"` instead of serializing the
sentinel
- `EXT_DELTA_SENTINEL` (msgpack ext code 8) removed from
`JsonPlusSerializer`
- `DELTA_SENTINEL` is kept as an in-memory marker:
`DeltaChannel.checkpoint()` returns it so savers know to skip it, and
`_ChannelWritesHistory.seed` uses it to mean "no snapshot found, start
from empty"
## Performance
Benchmarked at `snapshot_frequency=10` on Postgres (`~100 tok/msg`):
| turns | old combined query | two-stage |
|------:|-------------------:|----------:|
| 50 | 6.0ms | 2.8ms (2.1x faster) |
| 100 | 10.1ms | 5.6ms (1.8x faster) |
| 500 | **216.1ms** | 15.3ms (**14x faster**) |
The old query's read time grew super-linearly with turn count because
each read fetched O(N/freq) full snapshot blobs. Two-stage keeps read
depth bounded by `snapshot_frequency` regardless of thread length.
## Test plan
- `make test` in `libs/checkpoint`, `libs/checkpoint-postgres`,
`libs/langgraph`
- Removed `test_delta_sentinel_serde_round_trip` (sentinel no longer
serializable)
- Updated `test_memory.py` — delta channel blobs stored as `"empty"`,
not serialized sentinel
- Updated `test_channels.py` — `channel_values` no longer contains
sentinel key for DeltaChannels
- Deleted `test_delta_channel_two_stage_benchmark.py` (one-stage vs
two-stage comparison; path no longer exists)
---------
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
Co-authored-by: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
## Summary
Replace f-string SQL formatting with parameterized queries to prevent
potential SQL injection in checkpoint migration code.
## Changes
Updated the migration version tracking INSERT statements in all
checkpoint saver classes to use parameterized queries instead of
f-string formatting:
- `PostgresSaver`
(libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py:100)
- `AsyncPostgresSaver`
(libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py:104-106)
- `ShallowPostgresSaver`
(libs/checkpoint-postgres/langgraph/checkpoint/postgres/shallow.py:255)
- `AsyncShallowPostgresSaver`
(libs/checkpoint-postgres/langgraph/checkpoint/postgres/shallow.py:617-619)
**Before (vulnerable to SQL injection):**
```python
cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
```
**After (using parameterized query):**
```python
cur.execute("INSERT INTO checkpoint_migrations (v) VALUES (%s)", (v,))
```
## Risk Assessment
The practical risk is low since `v` is an integer loop variable
controlled by the codebase. However, using string formatting in SQL
queries is a well-known anti-pattern that can lead to SQL injection
vulnerabilities, especially if the code is later refactored or copied to
other contexts.
## Testing
- ✅ All 216 tests passing on PostgreSQL 15 and 16
- ✅ Linting and type checking passing
- ✅ No functional changes to behavior
- **Description:** The final migration for the postgres checkpointer is
not currently idempotent. That presents problems when migrating from one
checkpointer to another or if migrations otherwise get applied twice.
This makes the final migration idempotent to avoid this problem.
- **Issue:** N/A
- **Dependencies:** N/A
- **Twitter handle:** N/A
Issue
Support for `Checkpoint.metadata.writes` was dropped in `langgraph`
v0.5.x.
In `langgraph-checkpoint-postgres` v2.0.23, metadata was serialized with
`BasePostgresSaver._dump_metadata` -> `JsonPlusSerializer.dumps` which
handles `pydantic.BaseModel`.
In v2.0.23, metadata is serialized with `psycopg.types.json.Jsonb`,
which raises `TypeError: Object of type AIMessage is not JSON
serializable` when trying to serialize `writes`.
Solution
- Add `BaseCheckpointSaver.get_serializable_checkpoint_metadata` which
pops the `writes` key.
- Log deprecation warning when strange version combinations are used
Solves https://github.com/langchain-ai/langgraph/issues/5769
---------
Co-authored-by: Alex Kondratev <56111142+soapun@users.noreply.github.com>
### Description
https://github.com/langchain-ai/langgraph/issues/6137 and
https://github.com/langchain-ai/langgraph/issues/5677 reported issues
where older checkpoints read by AsyncPostgresSaver/PostgresSaver from
`langgraph-checkpoint-postgres==2.0.19` fail to read channel values,
throwing `NoneType object is not a mapping`. This was due to a bug in
how `channel_values` is assembled:
```python
"channel_values": {
**value["checkpoint"].get("channel_values"), # <--- if channel_values doesn't exist (old checkpoint), **None errors
**self._load_blobs(value["channel_values"]),
},
```
This bug was observed for checkpoints generated by
`langgraph-checkpoint-postgres<=2.0.19`.
Fixed by providing a fallback to
`value["checkpoint"].get("channel_values")`:
```python
**value["checkpoint"],
"channel_values": {
**(
value["checkpoint"].get("channel_values") or {}
), # 'or {}' needed for backwards compat with v3 checkpoints and below, as v4 introduced channel_values key
**self._load_blobs(value["channel_values"]),
},
```
### Tests
Added test for AsyncPostgresSaver and test for PostgresSaver, using
monkeypatch to remove `channel_values` before CheckpointTuple is
assembled in `_load_checkpoint_tuple`.
### Solves
https://github.com/langchain-ai/langgraph/issues/6137 and
https://github.com/langchain-ai/langgraph/issues/5677
---------
Co-authored-by: Shahrukh Shaik <144558473+shahrukh-shaik@users.noreply.github.com>
This PR updates the dependencies in all Python packages using `uv lock
--upgrade`.
This is an automated PR created by the UV Lock Upgrade workflow.
To make tests pass:
* linting fixes
* whitespace fixes in snapshots
---------
Co-authored-by: sydney-runkle <54324534+sydney-runkle@users.noreply.github.com>
Co-authored-by: Sydney Runkle <sydneymarierunkle@gmail.com>
- Replaces checkpoint_during: bool
- checkpoint_during is deprecated but still respected
- We implement three durability modes (from least to most durable):
- "exit" - save checkpoint only when the graph exits (equivalent to
checkpoint_during=False)
- "async" - save checkpoint asynchronously while the next step executes
(the default, equivalent to old checkpoint_during=True)
- "sync" - save checkpoint synchronously before the next step starts
(new mode, slower but most durable)
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
- Channels containing primitive values don't need to be stored in separate rows in blobs table, as the overhead of a separate row will usually be higher than the size of the value
- This applies for instance to all internal channels used to manage edges, so it has a big impact just from that. It can also apply to user-managed channels depending on their values
- The same channel may switch storage between versions without any issue
Prepare langgraph-checkpoint for 0.5
- Given we have no upper bound on langgraph-checkpoint dep need to undo all changes in langgraph-checkpoint that might break previous versions of langgraph
- This has been superseded by saving the individual writes of each task through put_writes()
- Removing this speeds up checkpoint operations as it was duplicating data saved elsewhere already
- Instead store sends in a Topic channel, removing the need to fetch sends as writes against the parent checkpoint
- Remove deprecated/unused functions in langgraph-checkpoint (will require bumping min range for langgraph-checkpoint in langgraph lib)
- Implement migration of old pending sends in langgraph-checkpoint-postgres
- Ensure parent config of `checkpoint_during=False` checkpoints always points to checkpoints that were also saved
- Deletes all data associated with a thread_id
- Implemented in InMemory, Sqlite and Postgres checkpointers
Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com>
This PR adds a "shallow" version of `PostgresSaver` checkpointer that
ONLY stores the most recent checkpoint and does NOT retain any history.
It is meant to be a light-weight drop-in replacement for the
PostgresSaver that supports most of the LangGraph persistence
functionality with the exception of time travel.
Replace hardcoded database saver class names with `cls` in
`from_conn_string` factory methods to improve subclassing support
## Changes
* Replaced direct class instantiations with `cls(conn)` in
`from_conn_string` classmethods across all database implementations
* Updated both synchronous and asynchronous variants for DuckDB,
PostgreSQL, and SQLite savers
## Why
This refactor makes the database saver classes more extensible by
following Python's convention of using `cls` in class methods. This
enables proper inheritance patterns where subclasses can reuse the
factory methods without needing to override them. Previously, the
hardcoded class names would always instantiate the parent class, even
when called from a subclass.
## Testing
The change is backward compatible and doesn't alter existing
functionality. All existing tests should continue to pass as this is
purely a structural refactoring that preserves the current behavior
while improving extensibility.
## Notes
This PR addresses follow up on comments from #2518 - AsyncPostgresSaver
didn't need to be fixed but many of the other DB saver classes did.
- Initializing the store with an 'embedding config' -> this contains the
'dims' (used to create the table) and the encoder object (rn langchain
embeddings object, though that is ......)
- Call setup() -> creates the vector table.
Each document has 1 or more vectors associated with it for each json
path in the embedding config.
Would welcome critique and requests!
Leaving the params as the defaults for pgvector but open to feedback if
you think it's important to be able to more transparently configure that
in setup()
```python
from typing import TypedDict, List, Dict, Any, Optional
from langchain_openai import OpenAIEmbeddings
from langgraph.graph import StateGraph
from langgraph.store.postgres import PostgresStore
emb_config = {
"dims": 1536, # OpenAI embedding dimensions
"embed": OpenAIEmbeddings(model="text-embedding-3-small"),
"distance_type": "cosine",
}
with PostgresStore.from_conn_string(
"postgres://postgres:postgres@localhost:5441",
embedding=emb_config,
) as store:
store.setup()
# Define the state type for our graph
class State(TypedDict):
query: str
results: Optional[List[Dict[str, Any]]]
def put_stuff(state: State) -> State:
docs = [
("doc1", {"text": "red apple in kitchen"}),
("doc2", {"text": "blue car in garage"}),
("doc3", {"text": "green apple on table"}),
]
for key, value in docs:
store.put(("docs",), key, value)
def search_stuff(state: State) -> State:
"""Search for documents using vector similarity."""
results = store.search(("docs",), query=state["query"])
return {"results": results}
builder = StateGraph(State)
builder.add_node(put_stuff)
builder.add_node(search_stuff)
builder.add_edge("__start__", "put_stuff")
builder.add_edge("put_stuff", "search_stuff")
# Compile
with PostgresStore.from_conn_string(
"postgres://postgres:postgres@localhost:5441",
embedding=emb_config,
) as store:
chain = builder.compile(store=store)
result = chain.invoke({"query": "sour apple"})
# Print results
for doc in result["results"]:
print(doc.key)
print(doc.value)
print(doc.response_metadata)
```
