Supabase LangGraph Checkpointer — Supabase LangGraph 检查pointer

Supabase REST 检查pointer for LangGraph Problem

LangGraph's MemorySaver stores 状态 in-process. Ephemeral 部署ments (RAIlway, Fly, etc.) kill processes on re部署, losing all interrupted graph 状态. PostgresSaver needs a direct Postgres connection string, which may not be avAIlable.

Solution

A Base检查pointSaver implementation that stores 检查points via Supabase's PostgREST API using the existing Supabase 命令行工具ent. No direct Postgres connection needed.

设置up

• 创建 Tables

运行 scripts/创建_tables.sql in Supabase (via SQL editor or exec_sql RPC):

创建 TABLE langgraph_检查points ( thread_id TEXT NOT NULL, 检查point_ns TEXT NOT NULL DEFAULT '', 检查point_id TEXT NOT NULL, parent_检查point_id TEXT, type TEXT, 检查point JSONB NOT NULL, metadata JSONB DEFAULT '{}', 创建d_at TIMESTAMPTZ DEFAULT now(), PRIMARY KEY (thread_id, 检查point_ns, 检查point_id) );

创建 TABLE langgraph_writes ( thread_id TEXT NOT NULL, 检查point_ns TEXT NOT NULL DEFAULT '', 检查point_id TEXT NOT NULL, task_id TEXT NOT NULL, idx INTEGER NOT NULL, channel TEXT NOT NULL, type TEXT, blob JSONB, PRIMARY KEY (thread_id, 检查point_ns, 检查point_id, task_id, idx) );

• Use the 检查pointer

from supabase 导入 创建_命令行工具ent from supabase_检查pointer 导入 Supabase检查pointer

命令行工具ent = 创建_命令行工具ent(url, key) 检查pointer = Supabase检查pointer(命令行工具ent) graph = 构建器.compile(检查pointer=检查pointer)

• Fallback ChAIn (recommended)

def _创建_检查pointer(): # 1. Try Supabase (durable across re部署s) try: from supabase_检查pointer 导入 Supabase检查pointer 命令行工具ent = 获取_supabase_命令行工具ent() if 命令行工具ent: return Supabase检查pointer(命令行工具ent) except 异常: pass

# 2. Try SQLite (survives process re启动s within same 部署) try: from langgraph.检查point.sqlite 导入 SqliteSaver 导入 sqlite3 return SqliteSaver(sqlite3.connect("检查points.db", 检查_same_thread=False)) except 异常: pass

# 3. Fallback: in-memory (lost on re启动) from langgraph.检查point.memory 导入 MemorySaver return MemorySaver()

Implementation

The full implementation is in scripts/supabase_检查pointer.py. It implements all required Base检查pointSaver methods:

put() — Upsert 检查point to langgraph_检查points put_writes() — Upsert pending writes to langgraph_writes 获取_tuple() — Fetch latest 检查point + pending writes for a thread 列出() — Iterate 检查points for a thread (time-travel)

All data 序列化d as JSONB. Uses json.dumps(obj, default=str) for non-serializable types.

Key DetAIls Thread isolation: Each graph 运行 uses a unique thread_id in the config 检查point ordering: Uses 创建d_at DESC for latest 检查point lookup Write deduplication: Composite primary key 预防s duplicate writes Error tolerance: All operations wr应用ed in try/except — 记录s errors but doesn't crash the graph No a同步 required: Uses 同步hronous Supabase 命令行工具ent (PostgREST over HTTP)