WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-28 10:49:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1552 from langchain-ai/nc/30aug/sort-pending-writes-on-fetch

Browse Source 
postgres/sqlite: Sort pending writes when fetching from db
This commit is contained in:
Nuno Campos
2024-08-30 14:12:54 -07:00
committed by GitHub
parent 770c092b57 e744db169a
commit 13cf9b7ec5
3 changed files with 6 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py
+2 -2
View File
@@ -76,14 +76,14 @@ select
) as channel_values,
(
select
array_agg(array[cw.task_id::text::bytea, cw.channel::bytea, cw.type::bytea, cw.blob])
array_agg(array[cw.task_id::text::bytea, cw.channel::bytea, cw.type::bytea, cw.blob] order by cw.task_id, cw.idx)
from checkpoint_writes cw
where cw.thread_id = checkpoints.thread_id
and cw.checkpoint_ns = checkpoints.checkpoint_ns
and cw.checkpoint_id = checkpoints.checkpoint_id
) as pending_writes,
(
select array_agg(array[cw.type::bytea, cw.blob])
select array_agg(array[cw.type::bytea, cw.blob] order by cw.idx)
from checkpoint_writes cw
where cw.thread_id = checkpoints.thread_id
and cw.checkpoint_ns = checkpoints.checkpoint_ns
libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
+2 -2
View File
@@ -244,7 +244,7 @@ class SqliteSaver(BaseCheckpointSaver):
}
# find any pending writes
cur.execute(
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
(
str(config["configurable"]["thread_id"]),
checkpoint_ns,
@@ -331,7 +331,7 @@ class SqliteSaver(BaseCheckpointSaver):
metadata,
) in cur:
wcur.execute(
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
(thread_id, checkpoint_ns, checkpoint_id),
)
yield CheckpointTuple(
libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py
+2 -2
View File
@@ -278,7 +278,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver):
}
# find any pending writes
await cur.execute(
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
(
str(config["configurable"]["thread_id"]),
checkpoint_ns,
@@ -348,7 +348,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver):
metadata,
) in cur:
await wcur.execute(
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
(thread_id, checkpoint_ns, checkpoint_id),
)
yield CheckpointTuple(