WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-10 11:47:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into v1

Browse Source
This commit is contained in:
Sydney Runkle
2025-07-16 18:32:07 -04:00
parent 294078adab 48dabc0538
commit c6d674cd3e
37 changed files with 1023 additions and 546 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py
+26 -10
View File
@@ -286,6 +286,7 @@ class PostgresSaver(BasePostgresSaver):
checkpoint_ns = configurable.pop("checkpoint_ns")
checkpoint_id = configurable.pop("checkpoint_id", None)
copy = checkpoint.copy()
copy["channel_values"] = copy["channel_values"].copy()
next_config = {
"configurable": {
"thread_id": thread_id,
@@ -294,16 +295,28 @@ class PostgresSaver(BasePostgresSaver):
}
}
# inline primitive values in checkpoint table
# others are stored in blobs table
blob_values = {}
for k, v in checkpoint["channel_values"].items():
if v is None or isinstance(v, (str, int, float, bool)):
pass
else:
blob_values[k] = copy["channel_values"].pop(k)
with self._cursor(pipeline=True) as cur:
cur.executemany(
self.UPSERT_CHECKPOINT_BLOBS_SQL,
self._dump_blobs(
thread_id,
checkpoint_ns,
copy.pop("channel_values"), # type: ignore[misc]
new_versions,
),
)
if blob_versions := {
k: v for k, v in new_versions.items() if k in blob_values
}:
cur.executemany(
self.UPSERT_CHECKPOINT_BLOBS_SQL,
self._dump_blobs(
thread_id,
checkpoint_ns,
blob_values,
blob_versions,
),
)
cur.execute(
self.UPSERT_CHECKPOINTS_SQL,
(
@@ -436,7 +449,10 @@ class PostgresSaver(BasePostgresSaver):
},
{
**value["checkpoint"],
"channel_values": self._load_blobs(value["channel_values"]),
"channel_values": {
**value["checkpoint"].get("channel_values"),
**self._load_blobs(value["channel_values"]),
},
},
value["metadata"],
(
libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py
+27 -11
View File
@@ -243,6 +243,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
checkpoint_id = configurable.pop("checkpoint_id", None)
copy = checkpoint.copy()
copy["channel_values"] = copy["channel_values"].copy()
next_config = {
"configurable": {
"thread_id": thread_id,
@@ -251,17 +252,29 @@ class AsyncPostgresSaver(BasePostgresSaver):
}
}
# inline primitive values in checkpoint table
# others are stored in blobs table
blob_values = {}
for k, v in checkpoint["channel_values"].items():
if v is None or isinstance(v, (str, int, float, bool)):
pass
else:
blob_values[k] = copy["channel_values"].pop(k)
async with self._cursor(pipeline=True) as cur:
await cur.executemany(
self.UPSERT_CHECKPOINT_BLOBS_SQL,
await asyncio.to_thread(
self._dump_blobs,
thread_id,
checkpoint_ns,
copy.pop("channel_values"), # type: ignore[misc]
new_versions,
),
)
if blob_versions := {
k: v for k, v in new_versions.items() if k in blob_values
}:
await cur.executemany(
self.UPSERT_CHECKPOINT_BLOBS_SQL,
await asyncio.to_thread(
self._dump_blobs,
thread_id,
checkpoint_ns,
blob_values,
blob_versions,
),
)
await cur.execute(
self.UPSERT_CHECKPOINTS_SQL,
(
@@ -395,7 +408,10 @@ class AsyncPostgresSaver(BasePostgresSaver):
},
{
**value["checkpoint"],
"channel_values": self._load_blobs(value["channel_values"]),
"channel_values": {
**value["checkpoint"].get("channel_values"),
**self._load_blobs(value["channel_values"]),
},
},
value["metadata"],
(