Custom Checkpointer / MemorySaver - is there a comprehensive saver with a Test file to use as a guide? (LangGraph JS) #103

New Issue

@jacoblee93 commented on GitHub (Oct 3, 2024):

Have overhauling docs/reference examples as one of the next things I'll look at - stay tuned!

@jacoblee93 commented on GitHub (Oct 3, 2024): Have overhauling docs/reference examples as one of the next things I'll look at - stay tuned!

yindo commented

@hgoona commented on GitHub (Oct 3, 2024):

@jacoblee93 looking forward to this, but would it be at all possible to get your review of my (draft) adapter for SurrealDB?

Here's what I have so far:
sdbSaver.ts
sdbSaver.test.ts - tested with Vitest
☝🏾these are both based off the Sqlite saver (where I also looked other "savers?" to try fill in the blanks.

NOTES:
The SdbSaver is currently failing these tests:

❯ SdbSaver - should save and retrieve checkpoints correctly (8)
     ✅ confirm TestSdb - is connected
     ✅ get undefined checkpoint
     ✅ save first checkpoint
     ✅ add some writes
     ❌ get first checkpoint tuple
     ❌ save second checkpoint
     ✅ list checkpoints
     ➖ KEEP_TEST_DB_ALIVE

CURRENT UNDERSTANDING OF "THE SAVER??":

db storage has 2 tables: checkpoints,writes
What is writes? What does it do? Is it a temporary file or permanent thing to be stored in the DB?
If "state" is stored within checkpoints, are messages (AI, Human, Tool Call, Tool Messages) also stored within checkpoints? <<< if Yes, shouldn't these messages be a separate table of unique messages that is then referenced by the specific checkpoint:record to avoid duplication of identical data that would otherwise be copied inside each consecutive checkpoints record that is part of the same "thread" ? Therefore, when a thread is branched, the next checkpoint can reference the existing messages from before the "branching-point" and append new messages for states that continue the threads beyond the "branching-point".
👆🏾at least this is how I was planning to maintain history before "Checkpointers" were created by LangChain.
Also, what is the checkpoint_ns ??

@hgoona commented on GitHub (Oct 3, 2024): @jacoblee93 looking forward to this, but would it be at all possible to get your review of my (draft) adapter for SurrealDB? Here's what I have so far: `sdbSaver.ts` `sdbSaver.test.ts` - tested with Vitest ☝🏾these are both based off the Sqlite saver (where I also looked other "savers?" to try fill in the blanks. NOTES: The SdbSaver is currently failing these tests: ``` ❯ SdbSaver - should save and retrieve checkpoints correctly (8) ✅ confirm TestSdb - is connected ✅ get undefined checkpoint ✅ save first checkpoint ✅ add some writes ❌ get first checkpoint tuple ❌ save second checkpoint ✅ list checkpoints ➖ KEEP_TEST_DB_ALIVE ``` CURRENT UNDERSTANDING OF "THE SAVER??": 1. db storage has 2 tables: `checkpoints`,`writes` 2. What is `writes`? What does it do? Is it a temporary file or permanent thing to be stored in the DB? 3. If "state" is stored within `checkpoints`, are `messages` (AI, Human, Tool Call, Tool Messages) also stored within `checkpoints`? <<< if Yes, shouldn't these messages be a separate table of unique `messages` that is then referenced by the specific checkpoint:record to avoid duplication of identical data that would otherwise be copied inside each consecutive `checkpoints` record that is part of the same "thread" ? Therefore, when a thread is branched, the next `checkpoint` can reference the existing `messages` from before the "branching-point" and append new `messages` for states that continue the threads beyond the "branching-point". 👆🏾at least this is how I was planning to maintain history before "Checkpointers" were created by LangChain. 4. Also, what is the `checkpoint_ns` ??

yindo commented

@hgoona commented on GitHub (Oct 3, 2024):

WORK-IN-PROGRESS for SdbSaver:
I have kept the SqliteSaver code commented out as a direct reference to hopeful come up with an equivalent query in SurrealDB🤞🏾 - I've attempted to NOT deviate from the SqliteSaver query structure as much as possible (given my current lack of clarity on the above).

// sdbSaver.ts

// import Database, { Database as DatabaseType } from "better-sqlite3";
import { debugLog } from "$lib/debugLog";
import type { SdbQueryClass } from "$lib/server/sdbUtils";
// import { SdbQuery } from "$lib/server/sdbUtils";
import type { RunnableConfig } from "@langchain/core/runnables";
import {
	BaseCheckpointSaver,
	type Checkpoint,
	type CheckpointListOptions,
	type CheckpointTuple,
	type SerializerProtocol,
	type PendingWrite,
	type CheckpointMetadata,
} from "@langchain/langgraph-checkpoint";
// import type Surreal from "surrealdb";

interface CheckpointRow {
	checkpoint: string;
	metadata: string;
	parent_checkpoint_id?: string;
	thread_id: string;
	checkpoint_id: string;
	checkpoint_ns?: string;
	cpType?: string; //fka: 'type'
}

interface WritesRow {
	thread_id: string;
	checkpoint_ns: string;
	checkpoint_id: string;
	task_id: string;
	idx: number;
	channel: string;
	cpType?: string; //fka: 'type'
	cpValue?: string; //fka: 'value
}


// DEBUGS
const debug_checkpointer = true
// const debug_checkpointer = false

// const debug_output = true
// // const debug_output = false

const serviceName = "Memory5b_sdb"

/**
 * Based on SqliteSaver
 */
export class SdbSaver extends BaseCheckpointSaver {
	// constructor(serde?: SerializerProtocol) {
	// 	super(serde);
	// }

	// public db: Surreal;

	// constructor(db: Surreal, serde?: SerializerProtocol) {
	public db: SdbQueryClass;

	constructor(db: SdbQueryClass, serde?: SerializerProtocol) {
		super(serde);
		this.db = db;
	}
	// //   constructor(db: DatabaseType, serde?: SerializerProtocol) {
	// //     super(serde);
	// //     this.db = db;
	// //     this.isSetup = false;
	// //   }

	//   static fromConnString(connStringOrLocalPath: string): SqliteSaver {
	//     return new SqliteSaver(new Database(connStringOrLocalPath));
	//   }

	//   protected setup(): void {
	//     if (this.isSetup) {
	//       return;
	//     }

	//     this.db.pragma("journal_mode=WAL");
	//     this.db.exec(`
	// CREATE TABLE IF NOT EXISTS checkpoints (
	//   thread_id TEXT NOT NULL,
	//   checkpoint_ns TEXT NOT NULL DEFAULT '',
	//   checkpoint_id TEXT NOT NULL,
	//   parent_checkpoint_id TEXT,
	//   type TEXT,
	//   checkpoint BLOB,
	//   metadata BLOB,
	//   PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
	// );`);
	//     this.db.exec(`
	// CREATE TABLE IF NOT EXISTS writes (
	//   thread_id TEXT NOT NULL,
	//   checkpoint_ns TEXT NOT NULL DEFAULT '',
	//   checkpoint_id TEXT NOT NULL,
	//   task_id TEXT NOT NULL,
	//   idx INTEGER NOT NULL,
	//   channel TEXT NOT NULL,
	//   type TEXT,
	//   value BLOB,
	//   PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
	// );`);

	//     this.isSetup = true;
	//   }


	// NOTE
	//   PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
	//   PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)

	async getTuple(config: RunnableConfig): Promise<CheckpointTuple | undefined> {
		debugLog(debug_checkpointer, `${serviceName} getTuple▶`)

		const {
			thread_id,
			// checkpoint_ns = "", // why is this here??
			checkpoint_ns, // why is this here??
			checkpoint_id,
		} = config.configurable ?? {};

		console.log("getTuple:", {
			thread_id,
			checkpoint_ns,
			checkpoint_id
		});

		// let row: CheckpointRow;
		let rowSDB: CheckpointRow;

		if (checkpoint_id) {
			console.log("getTuple▶1");
			// row = this.db
			// 	.prepare(
			// 		`SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?`
			// 	)
			// 	.get(thread_id, checkpoint_ns, checkpoint_id) as CheckpointRow;

			const query = /*surql*/ `
					SELECT * FROM checkpoints
					WHERE thread_id = $thread_id
					AND checkpoint_ns = $checkpoint_ns
					AND checkpoint_id = $checkpoint_id
				`
			// rowSDB = await SdbQuery.query(query, { thread_id, checkpoint_ns, checkpoint_id }) as CheckpointRow
			rowSDB = await this.db.query(query, { thread_id, checkpoint_ns, checkpoint_id }) as CheckpointRow
		} else { // no checkpoint_id
			console.log("getTuple▶2");
			// row = this.db
			// 	.prepare(
			// 		`SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1`
			// 	)
			// 	.get(thread_id, checkpoint_ns) as CheckpointRow;

			const query = /*surql*/ `
				SELECT * FROM checkpoints
				WHERE thread_id = $thread_id
				AND checkpoint_ns = $checkpoint_ns
				ORDER BY checkpoint_id DESC 
				LIMIT 1
			`
			rowSDB = await this.db.query(query, { thread_id, checkpoint_ns }) as CheckpointRow
		}

		console.log("getTuple:", { rowSDB });

		// if (row === undefined) {
		if (rowSDB === undefined) {
			return undefined;
		}

		let finalConfig = config;
		if (!checkpoint_id) {
			console.log("getTuple▶3");

			finalConfig = {
				configurable: {
					// thread_id: row.thread_id,
					thread_id: rowSDB.thread_id,
					checkpoint_ns: checkpoint_ns ? checkpoint_ns : "", //prevent returning 'undefined'
					// checkpoint_id: row.checkpoint_id,
					checkpoint_id: rowSDB.checkpoint_id,
				},
			};
		}
		if (
			finalConfig.configurable?.thread_id === undefined ||
			finalConfig.configurable?.checkpoint_id === undefined
		) {
			console.log("getTuple▶4");

			throw new Error("Missing thread_id or checkpoint_id");
		}
		console.log("getTuple▶5");

		// find any pending writes
		// const pendingWritesRows = this.db
		// 	.prepare(
		// 		`SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?`
		// 	)
		// 	.all(
		// 		finalConfig.configurable.thread_id.toString(),
		// 		checkpoint_ns,
		// 		finalConfig.configurable.checkpoint_id.toString()
		// 	) as WritesRow[];


		const query = /*surql*/ `
			SELECT * FROM writes
			WHERE thread_id = $thread_id
			AND checkpoint_ns = $checkpoint_ns
			AND checkpoint_id = $checkpoint_id
		`
		const bindings = {
			thread_id: finalConfig.configurable.thread_id.toString(),
			checkpoint_ns,
			checkpoint_id: finalConfig.configurable.checkpoint_id.toString()
		}

		const pendingWritesRowsSDB = await this.db.query(query, {
			bindings
		}, true) as WritesRow[]

		const pendingWrites = await Promise.all(
			// pendingWritesRows.map(async (row) => {
			pendingWritesRowsSDB.map(async (rowSDB) => {
				return [
					// row.task_id,
					// row.channel,
					// await this.serde.loadsTyped(row.type ?? "json", row.value ?? ""),
					rowSDB.task_id,
					rowSDB.channel,
					await this.serde.loadsTyped(rowSDB.cpType ?? "json", rowSDB.cpValue ?? ""),
				] as [string, string, unknown];
			})
		);

		console.log("getTuple▶6");

		// const checkpointSerde = (await this.serde.loadsTyped(
		// 	// row.type ?? "json",
		// 	// row.checkpoint
		// 	rowSDB.cpType ?? "json",
		// 	rowSDB.checkpoint
		// )) as Checkpoint

		// const metadataSerde = (await this.serde.loadsTyped(
		// 	// row.type ?? "json",
		// 	// row.metadata
		// 	rowSDB.cpType ?? "json",
		// 	rowSDB.metadata
		// )) as CheckpointMetadata

		let checkpointSerde: Checkpoint = {} as Checkpoint
		let metadataSerde

		try {
			checkpointSerde = await this.serde.loadsTyped(
				rowSDB.cpType ?? "json",
				rowSDB.checkpoint ?? "{}"
			) as Checkpoint

			metadataSerde = (await this.serde.loadsTyped(
				rowSDB.cpType ?? "json",
				rowSDB.metadata ?? "{}"
			)) as CheckpointMetadata
		} catch (error) {
			console.error(error)
			// handle the error as needed
		}

		const finalOutput: CheckpointTuple = {
			config: finalConfig,
			checkpoint: checkpointSerde,
			metadata: metadataSerde,
			// parentConfig: row.parent_checkpoint_id
			parentConfig: rowSDB.parent_checkpoint_id
				? {
					configurable: {
						// thread_id: row.thread_id,
						thread_id: rowSDB.thread_id,
						checkpoint_ns,
						// checkpoint_id: row.parent_checkpoint_id,
						checkpoint_id: rowSDB.parent_checkpoint_id,
					},
				}
				: undefined,
			pendingWrites,
		};

		console.log("getTuple▶7:", { finalOutput });
		return finalOutput
	}

	async *list(
		config: RunnableConfig,
		options?: CheckpointListOptions
	): AsyncGenerator<CheckpointTuple> {
		const { limit, before } = options ?? {};
		debugLog(debug_checkpointer, `${serviceName} list▶`)

		const thread_id = config.configurable?.thread_id;
		// let sql = `SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ${before ? "AND checkpoint_id < ?" : ""
		//     } ORDER BY checkpoint_id DESC`;
		// if (limit) {
		//     sql += ` LIMIT ${limit}`;
		// }
		// const args = [thread_id, before?.configurable?.checkpoint_id].filter(
		//     Boolean
		// );

		// const rows: CheckpointRow[] = this.db
		//     .prepare(sql)
		//     .all(...args) as CheckpointRow[];


		const binding = {
			thread_id,
			beforeId: before?.configurable?.checkpoint_id || null
		}
		const query = /*surql*/ `
            SELECT * FROM checkpoints
            WHERE thread_id = $thread_id
            ${before ? `AND checkpoint_id < $beforeId` : ""}
            ORDER BY checkpoint_id DESC
            ${limit ? `LIMIT $limit` : ""}
        `
		const rows = await this.db.query(query, binding) as CheckpointRow[]


		// if (rows) {
		if (Array.isArray(rows) && rows.length > 0) {
			for (const row of rows) {
				yield {
					config: {
						configurable: {
							thread_id: row.thread_id,
							checkpoint_ns: row.checkpoint_ns,
							checkpoint_id: row.checkpoint_id,
						},
					},
					checkpoint: (await this.serde.loadsTyped(
						row.cpType ?? "json",
						row.checkpoint
					)) as Checkpoint,
					metadata: (await this.serde.loadsTyped(
						row.cpType ?? "json",
						row.metadata
					)) as CheckpointMetadata,
					parentConfig: row.parent_checkpoint_id
						? {
							configurable: {
								thread_id: row.thread_id,
								checkpoint_ns: row.checkpoint_ns,
								checkpoint_id: row.parent_checkpoint_id,
							},
						}
						: undefined,
				};
			}
		}
	}

	/**
	 * 
	 * @note PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
	 * 
	 * @param config 
	 * @param checkpoint 
	 * @param metadata 
	 * @returns 
	 */
	async put(
		config: RunnableConfig,
		checkpoint: Checkpoint,
		metadata: CheckpointMetadata
	): Promise<RunnableConfig> {
		debugLog(debug_checkpointer, `${serviceName} put▶`)

		const [type1, serializedCheckpoint] = this.serde.dumpsTyped(checkpoint);
		const [type2, serializedMetadata] = this.serde.dumpsTyped(metadata);
		if (type1 !== type2) {
			throw new Error(
				"Failed to serialized checkpoint and metadata to the same type."
			);
		}
		// const row = [
		// 	config.configurable?.thread_id?.toString(),
		// 	config.configurable?.checkpoint_ns,
		// 	checkpoint.id,
		// 	config.configurable?.checkpoint_id,
		// 	type1,
		// 	serializedCheckpoint,
		// 	serializedMetadata,
		// ];

		// this.db
		// 	.prepare(
		// 		`INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)`
		// 	)
		// 	.run(...row);

		const binding: CheckpointRow = {
			thread_id: config.configurable?.thread_id?.toString(),
			checkpoint_ns: config.configurable?.checkpoint_ns,
			checkpoint_id: checkpoint.id,
			parent_checkpoint_id: config.configurable?.checkpoint_id,
			cpType: type1,
			checkpoint: Buffer.from(serializedCheckpoint).toString("utf-8"),
			metadata: Buffer.from(serializedMetadata).toString("utf-8"),
			// checkpoint: serializedCheckpoint,
			// metadata: serializedMetadata,
		}

		// TODO: review if this is the same as Sqlite `INSERT OR REPLACE INTO`
		const query = /*surql*/ `
			INSERT INTO checkpoints {
				id: [$thread_id, $checkpoint_ns, $checkpoint_id],
				thread_id: $thread_id,
				checkpoint_ns: $checkpoint_ns,
				checkpoint_id: $checkpoint_id,
				parent_checkpoint_id: $parent_checkpoint_id,
				cpType: $cpType,
				checkpoint: $checkpoint,
				metadata: $metadata
			} 
			ON DUPLICATE KEY UPDATE
				parent_checkpoint_id = $Input.parent_checkpoint_id,
				cpType = $Input.cpType,
				checkpoint = $Input.checkpoint,
				metadata = $Input.metadata
			;
		`
		// const query = /*surql*/ `
		// 	INSERT INTO checkpoints {
		// 		thread_id: $thread_id
		// 		checkpoint_ns: $checkpoint_ns
		// 		checkpoint_id: $checkpoint_id
		// 		parent_checkpoint_id: $parent_checkpoint_id
		// 		cpType: $cpType
		// 		checkpoint: $checkpoint
		// 		metadata: $metadata
		// 	}
		// `
		// WHERE thread_id = $thread_id
		// AND checkpoint_ns = $checkpoint_ns
		// AND checkpoint_id = $checkpoint_id

		await this.db.query(query, binding as unknown as SurrealBindings)
		// const results = await SdbQuery.query(query, binding as unknown as SurrealBindings)

		const configurable = {
			thread_id: config.configurable?.thread_id,
			checkpoint_ns: config.configurable?.checkpoint_ns,
			checkpoint_id: checkpoint.id,
		}

		return {
			configurable: configurable,
		};
	}

	/**
	 * @note PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
	 * 
	 * @param config 
	 * @param writes 
	 * @param taskId 
	 */
	async putWrites(
		config: RunnableConfig,
		writes: PendingWrite[],
		taskId: string
	): Promise<void> {
		debugLog(debug_checkpointer, `${serviceName} putWrites▶`)

		// const stmt = this.db.prepare(`
		// 	INSERT OR REPLACE INTO writes 
		// 	(thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) 
		// 	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
		// `);

		// const transaction = this.db.transaction((rows) => {
		// 	for (const row of rows) {
		// 		stmt.run(...row);
		// 	}
		// });

		/** get the rows from the writes */
		// const rows: WritesRow[][] = writes.map((write, idx) => {
		const rows = writes.map((write, idx) => {
			const [cpType, serializedWrite] = this.serde.dumpsTyped(write[1]);
			return [
				config.configurable?.thread_id,
				config.configurable?.checkpoint_ns,
				config.configurable?.checkpoint_id,
				taskId,
				idx,
				write[0],
				cpType,
				serializedWrite,
			];
		});

		console.log("writes:", writes)
		console.log("rows:", rows)

		// transaction(rows);

		// // TODO loop over the `rows` and run the SdbQuery fn using `query` and spread the variables from each `row` in `rows` as a `binding` in the SdbQuery

		// // const binding: WritesRow = {
		// // 	thread_id,
		// // 	checkpoint_ns,
		// // 	checkpoint_id,
		// // 	task_id,
		// // 	idx,
		// // 	channel,
		// // 	type,
		// // 	value
		// // }
		// // const query = /*surql*/ `
		// // 	INSERT INTO writes {
		// // 		thread_id: $thread_id,
		// // 		checkpoint_ns: $checkpoint_ns,
		// // 		checkpoint_id: $checkpoint_id,
		// // 		task_id: $task_id,
		// // 		idx: $idx,
		// // 		channel: $channel,
		// // 		type: $type,
		// // 		value: $value				
		// // 	}
		// // `

		// SurrealDB QUERY
		const query = /*surql*/ `
			INSERT INTO writes {
				id: [$thread_id, $checkpoint_ns, $checkpoint_id, $task_id, $idx],
				thread_id: $thread_id,
				checkpoint_ns: $checkpoint_ns,
				checkpoint_id: $checkpoint_id,
				task_id: $task_id,
				idx: $idx,
				channel: $channel,
				cpType: $cpType,
				cpValue: $cpValue
			} 
			ON DUPLICATE KEY UPDATE
				channel = $channel,
				cpType = $cpType,
				cpValue = $cpValue				
			;
		`

		for (const row of rows) {
			const [thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, cpType, cpValue] = row
			// const binding: WritesRow = {
			const binding = {
				thread_id,
				checkpoint_ns,
				checkpoint_id,
				task_id,
				idx,
				channel,
				cpType,
				cpValue,
			}
			await this.db.query(query, binding as unknown as SurrealBindings)
		}

		// // const results = await SdbQuery.query(query, binding as unknown as SurrealBindings)
	}
}

@hgoona commented on GitHub (Oct 3, 2024): WORK-IN-PROGRESS for `SdbSaver`: I have kept the SqliteSaver code commented out as a direct reference to hopeful come up with an equivalent query in SurrealDB🤞🏾 - I've attempted to NOT deviate from the SqliteSaver query structure as much as possible (given my current lack of clarity on the above). ```ts // sdbSaver.ts // import Database, { Database as DatabaseType } from "better-sqlite3"; import { debugLog } from "$lib/debugLog"; import type { SdbQueryClass } from "$lib/server/sdbUtils"; // import { SdbQuery } from "$lib/server/sdbUtils"; import type { RunnableConfig } from "@langchain/core/runnables"; import { BaseCheckpointSaver, type Checkpoint, type CheckpointListOptions, type CheckpointTuple, type SerializerProtocol, type PendingWrite, type CheckpointMetadata, } from "@langchain/langgraph-checkpoint"; // import type Surreal from "surrealdb"; interface CheckpointRow { checkpoint: string; metadata: string; parent_checkpoint_id?: string; thread_id: string; checkpoint_id: string; checkpoint_ns?: string; cpType?: string; //fka: 'type' } interface WritesRow { thread_id: string; checkpoint_ns: string; checkpoint_id: string; task_id: string; idx: number; channel: string; cpType?: string; //fka: 'type' cpValue?: string; //fka: 'value } // DEBUGS const debug_checkpointer = true // const debug_checkpointer = false // const debug_output = true // // const debug_output = false const serviceName = "Memory5b_sdb" /** * Based on SqliteSaver */ export class SdbSaver extends BaseCheckpointSaver { // constructor(serde?: SerializerProtocol) { // super(serde); // } // public db: Surreal; // constructor(db: Surreal, serde?: SerializerProtocol) { public db: SdbQueryClass; constructor(db: SdbQueryClass, serde?: SerializerProtocol) { super(serde); this.db = db; } // // constructor(db: DatabaseType, serde?: SerializerProtocol) { // // super(serde); // // this.db = db; // // this.isSetup = false; // // } // static fromConnString(connStringOrLocalPath: string): SqliteSaver { // return new SqliteSaver(new Database(connStringOrLocalPath)); // } // protected setup(): void { // if (this.isSetup) { // return; // } // this.db.pragma("journal_mode=WAL"); // this.db.exec(` // CREATE TABLE IF NOT EXISTS checkpoints ( // thread_id TEXT NOT NULL, // checkpoint_ns TEXT NOT NULL DEFAULT '', // checkpoint_id TEXT NOT NULL, // parent_checkpoint_id TEXT, // type TEXT, // checkpoint BLOB, // metadata BLOB, // PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id) // );`); // this.db.exec(` // CREATE TABLE IF NOT EXISTS writes ( // thread_id TEXT NOT NULL, // checkpoint_ns TEXT NOT NULL DEFAULT '', // checkpoint_id TEXT NOT NULL, // task_id TEXT NOT NULL, // idx INTEGER NOT NULL, // channel TEXT NOT NULL, // type TEXT, // value BLOB, // PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) // );`); // this.isSetup = true; // } // NOTE // PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) // PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id) async getTuple(config: RunnableConfig): Promise<CheckpointTuple | undefined> { debugLog(debug_checkpointer, `${serviceName} getTuple▶`) const { thread_id, // checkpoint_ns = "", // why is this here?? checkpoint_ns, // why is this here?? checkpoint_id, } = config.configurable ?? {}; console.log("getTuple:", { thread_id, checkpoint_ns, checkpoint_id }); // let row: CheckpointRow; let rowSDB: CheckpointRow; if (checkpoint_id) { console.log("getTuple▶1"); // row = this.db // .prepare( // `SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?` // ) // .get(thread_id, checkpoint_ns, checkpoint_id) as CheckpointRow; const query = /*surql*/ ` SELECT * FROM checkpoints WHERE thread_id = $thread_id AND checkpoint_ns = $checkpoint_ns AND checkpoint_id = $checkpoint_id ` // rowSDB = await SdbQuery.query(query, { thread_id, checkpoint_ns, checkpoint_id }) as CheckpointRow rowSDB = await this.db.query(query, { thread_id, checkpoint_ns, checkpoint_id }) as CheckpointRow } else { // no checkpoint_id console.log("getTuple▶2"); // row = this.db // .prepare( // `SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1` // ) // .get(thread_id, checkpoint_ns) as CheckpointRow; const query = /*surql*/ ` SELECT * FROM checkpoints WHERE thread_id = $thread_id AND checkpoint_ns = $checkpoint_ns ORDER BY checkpoint_id DESC LIMIT 1 ` rowSDB = await this.db.query(query, { thread_id, checkpoint_ns }) as CheckpointRow } console.log("getTuple:", { rowSDB }); // if (row === undefined) { if (rowSDB === undefined) { return undefined; } let finalConfig = config; if (!checkpoint_id) { console.log("getTuple▶3"); finalConfig = { configurable: { // thread_id: row.thread_id, thread_id: rowSDB.thread_id, checkpoint_ns: checkpoint_ns ? checkpoint_ns : "", //prevent returning 'undefined' // checkpoint_id: row.checkpoint_id, checkpoint_id: rowSDB.checkpoint_id, }, }; } if ( finalConfig.configurable?.thread_id === undefined || finalConfig.configurable?.checkpoint_id === undefined ) { console.log("getTuple▶4"); throw new Error("Missing thread_id or checkpoint_id"); } console.log("getTuple▶5"); // find any pending writes // const pendingWritesRows = this.db // .prepare( // `SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?` // ) // .all( // finalConfig.configurable.thread_id.toString(), // checkpoint_ns, // finalConfig.configurable.checkpoint_id.toString() // ) as WritesRow[]; const query = /*surql*/ ` SELECT * FROM writes WHERE thread_id = $thread_id AND checkpoint_ns = $checkpoint_ns AND checkpoint_id = $checkpoint_id ` const bindings = { thread_id: finalConfig.configurable.thread_id.toString(), checkpoint_ns, checkpoint_id: finalConfig.configurable.checkpoint_id.toString() } const pendingWritesRowsSDB = await this.db.query(query, { bindings }, true) as WritesRow[] const pendingWrites = await Promise.all( // pendingWritesRows.map(async (row) => { pendingWritesRowsSDB.map(async (rowSDB) => { return [ // row.task_id, // row.channel, // await this.serde.loadsTyped(row.type ?? "json", row.value ?? ""), rowSDB.task_id, rowSDB.channel, await this.serde.loadsTyped(rowSDB.cpType ?? "json", rowSDB.cpValue ?? ""), ] as [string, string, unknown]; }) ); console.log("getTuple▶6"); // const checkpointSerde = (await this.serde.loadsTyped( // // row.type ?? "json", // // row.checkpoint // rowSDB.cpType ?? "json", // rowSDB.checkpoint // )) as Checkpoint // const metadataSerde = (await this.serde.loadsTyped( // // row.type ?? "json", // // row.metadata // rowSDB.cpType ?? "json", // rowSDB.metadata // )) as CheckpointMetadata let checkpointSerde: Checkpoint = {} as Checkpoint let metadataSerde try { checkpointSerde = await this.serde.loadsTyped( rowSDB.cpType ?? "json", rowSDB.checkpoint ?? "{}" ) as Checkpoint metadataSerde = (await this.serde.loadsTyped( rowSDB.cpType ?? "json", rowSDB.metadata ?? "{}" )) as CheckpointMetadata } catch (error) { console.error(error) // handle the error as needed } const finalOutput: CheckpointTuple = { config: finalConfig, checkpoint: checkpointSerde, metadata: metadataSerde, // parentConfig: row.parent_checkpoint_id parentConfig: rowSDB.parent_checkpoint_id ? { configurable: { // thread_id: row.thread_id, thread_id: rowSDB.thread_id, checkpoint_ns, // checkpoint_id: row.parent_checkpoint_id, checkpoint_id: rowSDB.parent_checkpoint_id, }, } : undefined, pendingWrites, }; console.log("getTuple▶7:", { finalOutput }); return finalOutput } async *list( config: RunnableConfig, options?: CheckpointListOptions ): AsyncGenerator<CheckpointTuple> { const { limit, before } = options ?? {}; debugLog(debug_checkpointer, `${serviceName} list▶`) const thread_id = config.configurable?.thread_id; // let sql = `SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ${before ? "AND checkpoint_id < ?" : "" // } ORDER BY checkpoint_id DESC`; // if (limit) { // sql += ` LIMIT ${limit}`; // } // const args = [thread_id, before?.configurable?.checkpoint_id].filter( // Boolean // ); // const rows: CheckpointRow[] = this.db // .prepare(sql) // .all(...args) as CheckpointRow[]; const binding = { thread_id, beforeId: before?.configurable?.checkpoint_id || null } const query = /*surql*/ ` SELECT * FROM checkpoints WHERE thread_id = $thread_id ${before ? `AND checkpoint_id < $beforeId` : ""} ORDER BY checkpoint_id DESC ${limit ? `LIMIT $limit` : ""} ` const rows = await this.db.query(query, binding) as CheckpointRow[] // if (rows) { if (Array.isArray(rows) && rows.length > 0) { for (const row of rows) { yield { config: { configurable: { thread_id: row.thread_id, checkpoint_ns: row.checkpoint_ns, checkpoint_id: row.checkpoint_id, }, }, checkpoint: (await this.serde.loadsTyped( row.cpType ?? "json", row.checkpoint )) as Checkpoint, metadata: (await this.serde.loadsTyped( row.cpType ?? "json", row.metadata )) as CheckpointMetadata, parentConfig: row.parent_checkpoint_id ? { configurable: { thread_id: row.thread_id, checkpoint_ns: row.checkpoint_ns, checkpoint_id: row.parent_checkpoint_id, }, } : undefined, }; } } } /** * * @note PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id) * * @param config * @param checkpoint * @param metadata * @returns */ async put( config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata ): Promise<RunnableConfig> { debugLog(debug_checkpointer, `${serviceName} put▶`) const [type1, serializedCheckpoint] = this.serde.dumpsTyped(checkpoint); const [type2, serializedMetadata] = this.serde.dumpsTyped(metadata); if (type1 !== type2) { throw new Error( "Failed to serialized checkpoint and metadata to the same type." ); } // const row = [ // config.configurable?.thread_id?.toString(), // config.configurable?.checkpoint_ns, // checkpoint.id, // config.configurable?.checkpoint_id, // type1, // serializedCheckpoint, // serializedMetadata, // ]; // this.db // .prepare( // `INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)` // ) // .run(...row); const binding: CheckpointRow = { thread_id: config.configurable?.thread_id?.toString(), checkpoint_ns: config.configurable?.checkpoint_ns, checkpoint_id: checkpoint.id, parent_checkpoint_id: config.configurable?.checkpoint_id, cpType: type1, checkpoint: Buffer.from(serializedCheckpoint).toString("utf-8"), metadata: Buffer.from(serializedMetadata).toString("utf-8"), // checkpoint: serializedCheckpoint, // metadata: serializedMetadata, } // TODO: review if this is the same as Sqlite `INSERT OR REPLACE INTO` const query = /*surql*/ ` INSERT INTO checkpoints { id: [$thread_id, $checkpoint_ns, $checkpoint_id], thread_id: $thread_id, checkpoint_ns: $checkpoint_ns, checkpoint_id: $checkpoint_id, parent_checkpoint_id: $parent_checkpoint_id, cpType: $cpType, checkpoint: $checkpoint, metadata: $metadata } ON DUPLICATE KEY UPDATE parent_checkpoint_id = $Input.parent_checkpoint_id, cpType = $Input.cpType, checkpoint = $Input.checkpoint, metadata = $Input.metadata ; ` // const query = /*surql*/ ` // INSERT INTO checkpoints { // thread_id: $thread_id // checkpoint_ns: $checkpoint_ns // checkpoint_id: $checkpoint_id // parent_checkpoint_id: $parent_checkpoint_id // cpType: $cpType // checkpoint: $checkpoint // metadata: $metadata // } // ` // WHERE thread_id = $thread_id // AND checkpoint_ns = $checkpoint_ns // AND checkpoint_id = $checkpoint_id await this.db.query(query, binding as unknown as SurrealBindings) // const results = await SdbQuery.query(query, binding as unknown as SurrealBindings) const configurable = { thread_id: config.configurable?.thread_id, checkpoint_ns: config.configurable?.checkpoint_ns, checkpoint_id: checkpoint.id, } return { configurable: configurable, }; } /** * @note PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) * * @param config * @param writes * @param taskId */ async putWrites( config: RunnableConfig, writes: PendingWrite[], taskId: string ): Promise<void> { debugLog(debug_checkpointer, `${serviceName} putWrites▶`) // const stmt = this.db.prepare(` // INSERT OR REPLACE INTO writes // (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) // VALUES (?, ?, ?, ?, ?, ?, ?, ?) // `); // const transaction = this.db.transaction((rows) => { // for (const row of rows) { // stmt.run(...row); // } // }); /** get the rows from the writes */ // const rows: WritesRow[][] = writes.map((write, idx) => { const rows = writes.map((write, idx) => { const [cpType, serializedWrite] = this.serde.dumpsTyped(write[1]); return [ config.configurable?.thread_id, config.configurable?.checkpoint_ns, config.configurable?.checkpoint_id, taskId, idx, write[0], cpType, serializedWrite, ]; }); console.log("writes:", writes) console.log("rows:", rows) // transaction(rows); // // TODO loop over the `rows` and run the SdbQuery fn using `query` and spread the variables from each `row` in `rows` as a `binding` in the SdbQuery // // const binding: WritesRow = { // // thread_id, // // checkpoint_ns, // // checkpoint_id, // // task_id, // // idx, // // channel, // // type, // // value // // } // // const query = /*surql*/ ` // // INSERT INTO writes { // // thread_id: $thread_id, // // checkpoint_ns: $checkpoint_ns, // // checkpoint_id: $checkpoint_id, // // task_id: $task_id, // // idx: $idx, // // channel: $channel, // // type: $type, // // value: $value // // } // // ` // SurrealDB QUERY const query = /*surql*/ ` INSERT INTO writes { id: [$thread_id, $checkpoint_ns, $checkpoint_id, $task_id, $idx], thread_id: $thread_id, checkpoint_ns: $checkpoint_ns, checkpoint_id: $checkpoint_id, task_id: $task_id, idx: $idx, channel: $channel, cpType: $cpType, cpValue: $cpValue } ON DUPLICATE KEY UPDATE channel = $channel, cpType = $cpType, cpValue = $cpValue ; ` for (const row of rows) { const [thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, cpType, cpValue] = row // const binding: WritesRow = { const binding = { thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, cpType, cpValue, } await this.db.query(query, binding as unknown as SurrealBindings) } // // const results = await SdbQuery.query(query, binding as unknown as SurrealBindings) } } ```

yindo commented

@hgoona commented on GitHub (Oct 3, 2024):

Test file, tested with Vitest:

// sdbSaver.test.ts

import { describe, it, expect, beforeAll, type TestOptions } from "vitest";
// import { describe, it, expect } from "@jest/globals";
import {
	type Checkpoint,
	type CheckpointTuple,
	uuid6,
} from "@langchain/langgraph-checkpoint";
import { TestSdb1 } from "$lib/server/sdbTestDb";
import { SdbQueryClass } from "$lib/server/sdbUtils";
import { SdbSaver } from "./sdbSaver";
// import { SqliteSaver } from "../index.js";

const checkpoint1: Checkpoint = {
	v: 1,
	id: uuid6(-1),
	ts: "2024-04-19T17:19:07.952Z",
	channel_values: {
		someKey1: "someValue1",
	},
	channel_versions: {
		someKey2: 1,
	},
	versions_seen: {
		someKey3: {
			someKey4: 1,
		},
	},
	pending_sends: [],
};
console.log({ checkpoint1 });

const checkpoint2: Checkpoint = {
	v: 1,
	id: uuid6(1),
	ts: "2024-04-20T17:19:07.952Z",
	channel_values: {
		someKey1: "someValue2",
	},
	channel_versions: {
		someKey2: 2,
	},
	versions_seen: {
		someKey3: {
			someKey4: 2,
		},
	},
	pending_sends: [],
};

describe("SdbSaver - should save and retrieve checkpoints correctly", () => {
	/** 
	 * New instance of SdbSaver with (test) DB attached 
	 */
	let sdbSaver: SdbSaver
	
	/**
	 * SDBTester is a SdbQueryClass 
	 * that connects to the TestSdb1 database (in-Memory DB).
	 */
	let SDBTester: SdbQueryClass
	
	beforeAll(() => {
		SDBTester = new SdbQueryClass(TestSdb1)

		// initialize SdbSaver
		sdbSaver = new SdbSaver(SDBTester); // must be an SdbQueryClass
	})

	it("confirm TestSdb - is connected", async () => {
		const info = await SDBTester.query("INFO FOR ROOT;")
		// console.log({ info })

		expect(info).not.toBeNull()
	})

	// CHECKPOINT SAVER TESTS
	// it("should save and retrieve checkpoints correctly", async () => {
	// const sqliteSaver = SqliteSaver.fromConnString(":memory:");
	// const sdbSaver = new SdbSaver(TestSdb1);

	it("get undefined checkpoint", async () => {
		// get undefined checkpoint
		const undefinedCheckpoint = await sdbSaver.getTuple({
			configurable: { thread_id: "1" },
		});
		expect(undefinedCheckpoint).toBeUndefined();
	});


	it("save first checkpoint", async () => {
		// save first checkpoint
		const runnableConfig = await sdbSaver.put(
			{ configurable: { thread_id: "1" } },
			checkpoint1,
			{ source: "update", step: -1, writes: null, parents: {} }
		);
		expect(runnableConfig).toEqual({
			configurable: {
				thread_id: "1",
				checkpoint_id: checkpoint1.id,
			},
		});
	});

	it("add some writes", async () => {
		// add some writes
		await sdbSaver.putWrites(
			{
				configurable: {
					checkpoint_id: checkpoint1.id,
					checkpoint_ns: "",
					thread_id: "1",
				},
			},
			[["bar", "baz"]],
			"foo"
		);
	});


	it("get first checkpoint tuple", async () => {
		// get first checkpoint tuple
		const firstCheckpointTuple = await sdbSaver.getTuple({
			configurable: { thread_id: "1" },
		});
		console.log("firstCheckpointTuple:", JSON.stringify(firstCheckpointTuple, null, 2));
		console.log("checkpoint1.id:", checkpoint1.id);

		expect(firstCheckpointTuple?.config).toEqual({
			configurable: {
				thread_id: "1",
				checkpoint_ns: "",
				checkpoint_id: checkpoint1.id,
			},
		});
		expect(firstCheckpointTuple?.checkpoint).toEqual(checkpoint1);
		expect(firstCheckpointTuple?.parentConfig).toBeUndefined();
		expect(firstCheckpointTuple?.pendingWrites).toEqual([
			["foo", "bar", "baz"],
		]);
	});

	it("save second checkpoint", async () => {
		// save second checkpoint
		await sdbSaver.put(
			{
				configurable: {
					thread_id: "1",
					checkpoint_id: "2024-04-18T17:19:07.952Z",
				},
			},
			checkpoint2,
			{ source: "update", step: -1, writes: null, parents: {} }
		);

		// verify that parentTs is set and retrieved correctly for second checkpoint
		const secondCheckpointTuple = await sdbSaver.getTuple({
			configurable: { thread_id: "1" },
		});
		expect(secondCheckpointTuple?.parentConfig).toEqual({
			configurable: {
				thread_id: "1",
				checkpoint_ns: "",
				checkpoint_id: "2024-04-18T17:19:07.952Z",
			},
		});
	});

	it("list checkpoints", async () => {
		// list checkpoints
		const checkpointTupleGenerator = await sdbSaver.list({
			configurable: { thread_id: "1" },
		});
		const checkpointTuples: CheckpointTuple[] = [];
		for await (const checkpoint of checkpointTupleGenerator) {
			checkpointTuples.push(checkpoint);
		}
		expect(checkpointTuples.length).toBe(2);

		const checkpointTuple1 = checkpointTuples[0];
		const checkpointTuple2 = checkpointTuples[1];
		expect(checkpointTuple1.checkpoint.ts).toBe("2024-04-20T17:19:07.952Z");
		expect(checkpointTuple2.checkpoint.ts).toBe("2024-04-19T17:19:07.952Z");
	});


	//keep the Vitest running to inspect the in-memory DB
	const options: TestOptions = {
		timeout: 2147483647, //max❗❗❗ 24.8 days
	}
	it("KEEP_TEST_DB_ALIVE",
		options,
		async () => {
			await new Promise(resolve => { })
		}
	);

});

@hgoona commented on GitHub (Oct 3, 2024): Test file, tested with Vitest: ```ts // sdbSaver.test.ts import { describe, it, expect, beforeAll, type TestOptions } from "vitest"; // import { describe, it, expect } from "@jest/globals"; import { type Checkpoint, type CheckpointTuple, uuid6, } from "@langchain/langgraph-checkpoint"; import { TestSdb1 } from "$lib/server/sdbTestDb"; import { SdbQueryClass } from "$lib/server/sdbUtils"; import { SdbSaver } from "./sdbSaver"; // import { SqliteSaver } from "../index.js"; const checkpoint1: Checkpoint = { v: 1, id: uuid6(-1), ts: "2024-04-19T17:19:07.952Z", channel_values: { someKey1: "someValue1", }, channel_versions: { someKey2: 1, }, versions_seen: { someKey3: { someKey4: 1, }, }, pending_sends: [], }; console.log({ checkpoint1 }); const checkpoint2: Checkpoint = { v: 1, id: uuid6(1), ts: "2024-04-20T17:19:07.952Z", channel_values: { someKey1: "someValue2", }, channel_versions: { someKey2: 2, }, versions_seen: { someKey3: { someKey4: 2, }, }, pending_sends: [], }; describe("SdbSaver - should save and retrieve checkpoints correctly", () => { /** * New instance of SdbSaver with (test) DB attached */ let sdbSaver: SdbSaver /** * SDBTester is a SdbQueryClass * that connects to the TestSdb1 database (in-Memory DB). */ let SDBTester: SdbQueryClass beforeAll(() => { SDBTester = new SdbQueryClass(TestSdb1) // initialize SdbSaver sdbSaver = new SdbSaver(SDBTester); // must be an SdbQueryClass }) it("confirm TestSdb - is connected", async () => { const info = await SDBTester.query("INFO FOR ROOT;") // console.log({ info }) expect(info).not.toBeNull() }) // CHECKPOINT SAVER TESTS // it("should save and retrieve checkpoints correctly", async () => { // const sqliteSaver = SqliteSaver.fromConnString(":memory:"); // const sdbSaver = new SdbSaver(TestSdb1); it("get undefined checkpoint", async () => { // get undefined checkpoint const undefinedCheckpoint = await sdbSaver.getTuple({ configurable: { thread_id: "1" }, }); expect(undefinedCheckpoint).toBeUndefined(); }); it("save first checkpoint", async () => { // save first checkpoint const runnableConfig = await sdbSaver.put( { configurable: { thread_id: "1" } }, checkpoint1, { source: "update", step: -1, writes: null, parents: {} } ); expect(runnableConfig).toEqual({ configurable: { thread_id: "1", checkpoint_id: checkpoint1.id, }, }); }); it("add some writes", async () => { // add some writes await sdbSaver.putWrites( { configurable: { checkpoint_id: checkpoint1.id, checkpoint_ns: "", thread_id: "1", }, }, [["bar", "baz"]], "foo" ); }); it("get first checkpoint tuple", async () => { // get first checkpoint tuple const firstCheckpointTuple = await sdbSaver.getTuple({ configurable: { thread_id: "1" }, }); console.log("firstCheckpointTuple:", JSON.stringify(firstCheckpointTuple, null, 2)); console.log("checkpoint1.id:", checkpoint1.id); expect(firstCheckpointTuple?.config).toEqual({ configurable: { thread_id: "1", checkpoint_ns: "", checkpoint_id: checkpoint1.id, }, }); expect(firstCheckpointTuple?.checkpoint).toEqual(checkpoint1); expect(firstCheckpointTuple?.parentConfig).toBeUndefined(); expect(firstCheckpointTuple?.pendingWrites).toEqual([ ["foo", "bar", "baz"], ]); }); it("save second checkpoint", async () => { // save second checkpoint await sdbSaver.put( { configurable: { thread_id: "1", checkpoint_id: "2024-04-18T17:19:07.952Z", }, }, checkpoint2, { source: "update", step: -1, writes: null, parents: {} } ); // verify that parentTs is set and retrieved correctly for second checkpoint const secondCheckpointTuple = await sdbSaver.getTuple({ configurable: { thread_id: "1" }, }); expect(secondCheckpointTuple?.parentConfig).toEqual({ configurable: { thread_id: "1", checkpoint_ns: "", checkpoint_id: "2024-04-18T17:19:07.952Z", }, }); }); it("list checkpoints", async () => { // list checkpoints const checkpointTupleGenerator = await sdbSaver.list({ configurable: { thread_id: "1" }, }); const checkpointTuples: CheckpointTuple[] = []; for await (const checkpoint of checkpointTupleGenerator) { checkpointTuples.push(checkpoint); } expect(checkpointTuples.length).toBe(2); const checkpointTuple1 = checkpointTuples[0]; const checkpointTuple2 = checkpointTuples[1]; expect(checkpointTuple1.checkpoint.ts).toBe("2024-04-20T17:19:07.952Z"); expect(checkpointTuple2.checkpoint.ts).toBe("2024-04-19T17:19:07.952Z"); }); //keep the Vitest running to inspect the in-memory DB const options: TestOptions = { timeout: 2147483647, //max❗❗❗ 24.8 days } it("KEEP_TEST_DB_ALIVE", options, async () => { await new Promise(resolve => { }) } ); }); ```

yindo commented

@hgoona commented on GitHub (Oct 7, 2024):

Hi @jacoblee93 @benjamincburns ! I've noted the work being done on these threads (https://github.com/langchain-ai/langgraphjs/pull/555#issuecomment-2397834075)

I was wondering if either of you would be able to review this custom Saver I've made for SurrealDB?
I've finally got it Passing my test file (99% the same as the original Sqlite Test file), however, when used in a production LangGraph it fails!

I'd be extremely grateful for any assistance on this!! I think I may have only a slight issue somewhere...

Here's the code 👇🏾

@hgoona commented on GitHub (Oct 7, 2024): Hi @jacoblee93 @benjamincburns ! I've noted the work being done on these threads (https://github.com/langchain-ai/langgraphjs/pull/555#issuecomment-2397834075) I was wondering if either of you would be able to review this custom Saver I've made for SurrealDB? I've finally got it Passing my test file (99% the same as the original Sqlite Test file), however, when used in a production LangGraph it fails! I'd be extremely grateful for any assistance on this!! I think I may have only a slight issue somewhere... Here's the code 👇🏾

yindo commented

@hgoona commented on GitHub (Oct 7, 2024):

SdbSaver.ts

import { debugLog } from "$lib/debugLog";
import type { SdbQueryClass } from "$lib/server/sdbUtils";
import type { RunnableConfig } from "@langchain/core/runnables";
import {
	BaseCheckpointSaver,
	type Checkpoint,
	type CheckpointListOptions,
	type CheckpointTuple,
	type SerializerProtocol,
	type PendingWrite,
	type CheckpointMetadata,
} from "@langchain/langgraph-checkpoint";

interface CheckpointRow {
	checkpoint: string;
	metadata: string;
	parent_checkpoint_id?: string;
	thread_id: string;
	checkpoint_id: string;
	checkpoint_ns?: string;
	cpType?: string; //fka: 'type'
}

interface WritesRow {
	thread_id: string;
	checkpoint_ns: string;
	checkpoint_id: string;
	task_id: string;
	idx: number;
	channel: string;
	cpType?: string; //fka: 'type'
	cpValue?: string //| ArrayLike<string>; //fka: 'value
}


// DEBUGS
const debug_checkpointer = true
// const debug_checkpointer = false

// const debug_output = true
// // const debug_output = false

const serviceName = "SdbSaver"

/** Option to Serialize or not */
const SERIALIZE = true;
// const SERIALIZE = false;



/**
 * Based on SqliteSaver
 */
export class SdbSaver extends BaseCheckpointSaver {
	public db: SdbQueryClass;

	constructor(db: SdbQueryClass, serde?: SerializerProtocol) {
		super(serde);
		this.db = db;
	}
	// //   constructor(db: DatabaseType, serde?: SerializerProtocol) {
	// //     super(serde);
	// //     this.db = db;
	// //     this.isSetup = false;
	// //   }

	//   static fromConnString(connStringOrLocalPath: string): SqliteSaver {
	//     return new SqliteSaver(new Database(connStringOrLocalPath));
	//   }

	//   protected setup(): void {
	//     if (this.isSetup) {
	//       return;
	//     }

	//     this.db.pragma("journal_mode=WAL");
	//     this.db.exec(`
	// CREATE TABLE IF NOT EXISTS checkpoints (
	//   thread_id TEXT NOT NULL,
	//   checkpoint_ns TEXT NOT NULL DEFAULT '',
	//   checkpoint_id TEXT NOT NULL,
	//   parent_checkpoint_id TEXT,
	//   type TEXT,
	//   checkpoint BLOB,
	//   metadata BLOB,
	//   PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
	// );`);
	//     this.db.exec(`
	// CREATE TABLE IF NOT EXISTS writes (
	//   thread_id TEXT NOT NULL,
	//   checkpoint_ns TEXT NOT NULL DEFAULT '',
	//   checkpoint_id TEXT NOT NULL,
	//   task_id TEXT NOT NULL,
	//   idx INTEGER NOT NULL,
	//   channel TEXT NOT NULL,
	//   type TEXT,
	//   value BLOB,
	//   PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
	// );`);

	//     this.isSetup = true;
	//   }


	// NOTE
	//  checkpoints table
	//  PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
	//  writes table
	//  PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)

	async getTuple(config: RunnableConfig): Promise<CheckpointTuple | undefined> {
		debugLog(debug_checkpointer, `${serviceName} getTuple▶`)
		// const debug_getTuple = true;
		const debug_getTuple = false;

		const {
			thread_id,
			checkpoint_ns,
			checkpoint_id,
		} = config.configurable ?? {};

		debugLog(debug_getTuple, "getTuple:", {
			thread_id,
			checkpoint_ns,
			checkpoint_id
		});

		let rowSDB: CheckpointRow;

		if (checkpoint_id) {
			debugLog(debug_getTuple, "getTuple▶1");

			const bindings = {
				thread_id,
				checkpoint_ns: checkpoint_ns ?? "",
				checkpoint_id
			}

			const query = /*surql*/ `
					SELECT * FROM checkpoints
					WHERE thread_id = $thread_id
					AND checkpoint_ns = $checkpoint_ns
					AND checkpoint_id = $checkpoint_id
				`

			rowSDB = await this.db.query(query, bindings) as CheckpointRow
		} else { // no checkpoint_id
			debugLog(debug_getTuple, "getTuple▶2");

			const bindings = {
				thread_id,
				checkpoint_ns: checkpoint_ns ?? "",
			}

			const query = /*surql*/ `
				SELECT * FROM checkpoints
				WHERE thread_id = $thread_id
				AND checkpoint_ns = $checkpoint_ns
				ORDER BY checkpoint_id DESC 
				LIMIT 1
			`

			rowSDB = await this.db.query(query, bindings) as CheckpointRow
		}

		debugLog(debug_getTuple, "getTuple:", { rowSDB });

		if (!rowSDB) {
			return undefined;
		}

		let finalConfig = config;
		if (!checkpoint_id) {
			debugLog(debug_getTuple, "getTuple▶3");

			finalConfig = {
				configurable: {
					thread_id: rowSDB.thread_id,
					checkpoint_ns: checkpoint_ns ? checkpoint_ns : "", //prevent returning 'undefined'
					checkpoint_id: rowSDB.checkpoint_id,
				},
			};
		}

		if (!finalConfig.configurable?.thread_id || !finalConfig.configurable?.checkpoint_id) {
			debugLog(debug_getTuple, "getTuple▶4");

			throw new Error("Missing thread_id or checkpoint_id");
		}
		debugLog(debug_getTuple, "getTuple▶5");

		// find any pending writes
		const query = /*surql*/ `
			SELECT * FROM writes:[
				$thread_id,
				$checkpoint_ns,
				$checkpoint_id,
				NONE,
				NONE
			]..
		`
		const bindings = {
			thread_id: finalConfig.configurable.thread_id.toString(),
			checkpoint_ns: checkpoint_ns ?? "",
			checkpoint_id: finalConfig.configurable.checkpoint_id.toString()
		}

		const pendingWritesRowsSDB = await this.db.query(query, {
			bindings
		}, true) as WritesRow[]

		debugLog(debug_getTuple, "getTuple▶6a", { pendingWritesRowsSDB });

		const pendingWrites = await Promise.all(
			pendingWritesRowsSDB.map(async (rowSDB: WritesRow) => {
				try {
					const cpValueType = typeof rowSDB.cpValue
					console.log("typeof:", cpValueType)
					console.log("typeof val:", rowSDB.cpValue)

					const cpValueBuffer = (cpValueType == "object")
						? Buffer.from(Object.values(rowSDB.cpValue!).map(Number))
						: rowSDB.cpValue!;

					const cpValue = SERIALIZE
						? await this.serde.loadsTyped(rowSDB.cpType ?? "json", cpValueBuffer)
						: cpValueBuffer

					return [
						rowSDB.task_id,
						rowSDB.channel,
						cpValue
					] as [string, string, unknown];
				} catch (error) {
					console.error("Error in Promise.all callback:", error);
					throw error;
				}
			})
		);

		// SERIALIZE STEP
		const checkpointBuffer = Buffer.from(Object.values(rowSDB.checkpoint as unknown as number[])); // ✅/2
		const metadataBuffer = Buffer.from(Object.values(rowSDB.metadata as unknown as number[])); // ✅/2

		const checkpointData = SERIALIZE
			? (await this.serde.loadsTyped(
				rowSDB.cpType ?? "json",
				checkpointBuffer
			)) as Checkpoint
			: (JSON.parse(rowSDB.checkpoint) ?? {})

		const checkpointMetadata = SERIALIZE
			? (await this.serde.loadsTyped(
				rowSDB.cpType ?? "json",
				metadataBuffer
			)) as CheckpointMetadata
			: (JSON.parse(rowSDB.metadata) ?? {})

		const finalOutput: CheckpointTuple = {
			config: finalConfig,
			checkpoint: checkpointData,
			metadata: checkpointMetadata,
			parentConfig: rowSDB.parent_checkpoint_id
				? {
					configurable: {
						thread_id: rowSDB.thread_id,
						checkpoint_ns: checkpoint_ns ?? "",
						checkpoint_id: rowSDB.parent_checkpoint_id,
					},
				}
				: undefined,
			pendingWrites,
		};

		debugLog(debug_getTuple, "getTuple▶7:", JSON.stringify(finalOutput, null, 2));
		return finalOutput
	};


	async *list(
		config: RunnableConfig,
		options?: CheckpointListOptions
	): AsyncGenerator<CheckpointTuple> {
		const { limit, before } = options ?? {};
		debugLog(debug_checkpointer, `${serviceName} list▶`)

		const thread_id = config.configurable?.thread_id;

		const binding = {
			thread_id,
			beforeId: before?.configurable?.checkpoint_id || null
		}
		const query = /*surql*/ `
            SELECT * FROM checkpoints
            WHERE thread_id = $thread_id
            ${before ? `AND checkpoint_id < $beforeId` : ""}
            ORDER BY checkpoint_id DESC
            ${limit ? `LIMIT $limit` : ""}
        `
		const rows = await this.db.query(query, binding) as CheckpointRow[]


		if (Array.isArray(rows) && rows.length > 0) {
			for (const row of rows) {

				// SERIALIZE STEP
				const checkpointBuffer = Buffer.from(Object.values(row.checkpoint as unknown as number[])); // ✅/2
				const metadataBuffer = Buffer.from(Object.values(row.metadata as unknown as number[])); // ✅/2

				const checkpointData = SERIALIZE
					? (await this.serde.loadsTyped(
						row.cpType ?? "json",
						checkpointBuffer
					)) as Checkpoint
					: (JSON.parse(row.checkpoint) ?? {})

				const checkpointMetadata = SERIALIZE
					? (await this.serde.loadsTyped(
						row.cpType ?? "json",
						metadataBuffer
					)) as CheckpointMetadata
					: (JSON.parse(row.metadata) ?? {})

				yield {
					config: {
						configurable: {
							thread_id: row.thread_id,
							checkpoint_ns: row.checkpoint_ns,
							checkpoint_id: row.checkpoint_id,
						},
					},
					checkpoint: checkpointData,
					metadata: checkpointMetadata,
					parentConfig: row.parent_checkpoint_id
						? {
							configurable: {
								thread_id: row.thread_id,
								checkpoint_ns: row.checkpoint_ns,
								checkpoint_id: row.parent_checkpoint_id,
							},
						}
						: undefined,
				};
			}
		}
	}


	/**
	 * @note PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
	 * 
	 * @param config 
	 * @param checkpoint 
	 * @param metadata 
	 * @returns 
	 */
	async put(
		config: RunnableConfig,
		checkpoint: Checkpoint,
		metadata: CheckpointMetadata
	): Promise<RunnableConfig> {
		debugLog(debug_checkpointer, `${serviceName} put▶`)

		// const debug_put = true;
		// const debug_put = false;

		const [type1, serializedCheckpoint] = this.serde.dumpsTyped(checkpoint);
		const [type2, serializedMetadata] = this.serde.dumpsTyped(metadata);
		if (type1 !== type2) {
			throw new Error(
				"Failed to serialized checkpoint and metadata to the same type."
			);
		}

		const checkpointData = SERIALIZE
			? serializedCheckpoint
			: Buffer.from(serializedCheckpoint).toString("utf-8")
		const checkpointMetadata = SERIALIZE
			? serializedMetadata
			: Buffer.from(serializedMetadata).toString("utf-8")

		// const binding: CheckpointRow = {
		const binding = {
			thread_id: config.configurable?.thread_id?.toString(),
			checkpoint_ns: config.configurable?.checkpoint_ns ?? "",
			checkpoint_id: checkpoint.id,
			parent_checkpoint_id: config.configurable?.checkpoint_id,
			cpType: type1,
			checkpoint: checkpointData,
			metadata: checkpointMetadata,
		}

		const query = /*surql*/ `
			UPSERT checkpoints:[$thread_id, $checkpoint_ns, $checkpoint_id] CONTENT {
				thread_id: $thread_id,
				checkpoint_ns: $checkpoint_ns,
				checkpoint_id: $checkpoint_id,
				parent_checkpoint_id: $parent_checkpoint_id,
				cpType: $cpType,
				checkpoint: $checkpoint,
				metadata: $metadata
			};
		`

		await this.db.query(query, binding as unknown as SurrealBindings)

		const configurable = {
			thread_id: config.configurable?.thread_id,
			checkpoint_ns: config.configurable?.checkpoint_ns ?? "",
			checkpoint_id: checkpoint.id,
		}

		return {
			configurable: configurable,
		};
	}


	/**
	 * @note PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
	 * 
	 * @param config 
	 * @param writes 
	 * @param taskId 
	 */
	async putWrites(
		config: RunnableConfig,
		writes: PendingWrite[],
		taskId: string
	): Promise<void> {
		debugLog(debug_checkpointer, `${serviceName} putWrites▶`)

		/** get the rows from the writes */
		const rows = writes.map((write, idx) => {
			const [cpType, serializedWrite] = this.serde.dumpsTyped(write[1]);

			console.log("putWrites1:", serializedWrite)
			console.log("putWrites2:", write[1])

			const cpValue = SERIALIZE ? serializedWrite : write[1]

			return {
				thread_id: config.configurable?.thread_id,
				checkpoint_ns: config.configurable?.checkpoint_ns ?? "",
				checkpoint_id: config.configurable?.checkpoint_id,
				task_id: taskId,
				idx: idx,
				channel: write[0],
				cpType: cpType,
				cpValue: cpValue,
			};
		});

		console.log("writes:", writes)
		console.log("rows:", rows)

		// SurrealDB QUERY
		const query = /*surql*/ `
			CREATE writes:[$thread_id, $checkpoint_ns, $checkpoint_id, $task_id, $idx] SET
				thread_id = $thread_id,
				checkpoint_ns = $checkpoint_ns,
				checkpoint_id = $checkpoint_id,
				task_id = $task_id,
				idx = $idx,
				channel = $channel,
				cpType = $cpType,
				cpValue = $cpValue
		`

		for (const row of rows) {
			const { thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, cpType, cpValue } = row
			const binding = {
				thread_id,
				checkpoint_ns: checkpoint_ns ?? "",
				checkpoint_id,
				task_id,
				idx,
				channel,
				cpType,
				cpValue,
			}
			await this.db.query(query, binding as unknown as SurrealBindings)
		}

	}
}

@hgoona commented on GitHub (Oct 7, 2024): SdbSaver.ts ```ts import { debugLog } from "$lib/debugLog"; import type { SdbQueryClass } from "$lib/server/sdbUtils"; import type { RunnableConfig } from "@langchain/core/runnables"; import { BaseCheckpointSaver, type Checkpoint, type CheckpointListOptions, type CheckpointTuple, type SerializerProtocol, type PendingWrite, type CheckpointMetadata, } from "@langchain/langgraph-checkpoint"; interface CheckpointRow { checkpoint: string; metadata: string; parent_checkpoint_id?: string; thread_id: string; checkpoint_id: string; checkpoint_ns?: string; cpType?: string; //fka: 'type' } interface WritesRow { thread_id: string; checkpoint_ns: string; checkpoint_id: string; task_id: string; idx: number; channel: string; cpType?: string; //fka: 'type' cpValue?: string //| ArrayLike<string>; //fka: 'value } // DEBUGS const debug_checkpointer = true // const debug_checkpointer = false // const debug_output = true // // const debug_output = false const serviceName = "SdbSaver" /** Option to Serialize or not */ const SERIALIZE = true; // const SERIALIZE = false; /** * Based on SqliteSaver */ export class SdbSaver extends BaseCheckpointSaver { public db: SdbQueryClass; constructor(db: SdbQueryClass, serde?: SerializerProtocol) { super(serde); this.db = db; } // // constructor(db: DatabaseType, serde?: SerializerProtocol) { // // super(serde); // // this.db = db; // // this.isSetup = false; // // } // static fromConnString(connStringOrLocalPath: string): SqliteSaver { // return new SqliteSaver(new Database(connStringOrLocalPath)); // } // protected setup(): void { // if (this.isSetup) { // return; // } // this.db.pragma("journal_mode=WAL"); // this.db.exec(` // CREATE TABLE IF NOT EXISTS checkpoints ( // thread_id TEXT NOT NULL, // checkpoint_ns TEXT NOT NULL DEFAULT '', // checkpoint_id TEXT NOT NULL, // parent_checkpoint_id TEXT, // type TEXT, // checkpoint BLOB, // metadata BLOB, // PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id) // );`); // this.db.exec(` // CREATE TABLE IF NOT EXISTS writes ( // thread_id TEXT NOT NULL, // checkpoint_ns TEXT NOT NULL DEFAULT '', // checkpoint_id TEXT NOT NULL, // task_id TEXT NOT NULL, // idx INTEGER NOT NULL, // channel TEXT NOT NULL, // type TEXT, // value BLOB, // PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) // );`); // this.isSetup = true; // } // NOTE // checkpoints table // PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id) // writes table // PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) async getTuple(config: RunnableConfig): Promise<CheckpointTuple | undefined> { debugLog(debug_checkpointer, `${serviceName} getTuple▶`) // const debug_getTuple = true; const debug_getTuple = false; const { thread_id, checkpoint_ns, checkpoint_id, } = config.configurable ?? {}; debugLog(debug_getTuple, "getTuple:", { thread_id, checkpoint_ns, checkpoint_id }); let rowSDB: CheckpointRow; if (checkpoint_id) { debugLog(debug_getTuple, "getTuple▶1"); const bindings = { thread_id, checkpoint_ns: checkpoint_ns ?? "", checkpoint_id } const query = /*surql*/ ` SELECT * FROM checkpoints WHERE thread_id = $thread_id AND checkpoint_ns = $checkpoint_ns AND checkpoint_id = $checkpoint_id ` rowSDB = await this.db.query(query, bindings) as CheckpointRow } else { // no checkpoint_id debugLog(debug_getTuple, "getTuple▶2"); const bindings = { thread_id, checkpoint_ns: checkpoint_ns ?? "", } const query = /*surql*/ ` SELECT * FROM checkpoints WHERE thread_id = $thread_id AND checkpoint_ns = $checkpoint_ns ORDER BY checkpoint_id DESC LIMIT 1 ` rowSDB = await this.db.query(query, bindings) as CheckpointRow } debugLog(debug_getTuple, "getTuple:", { rowSDB }); if (!rowSDB) { return undefined; } let finalConfig = config; if (!checkpoint_id) { debugLog(debug_getTuple, "getTuple▶3"); finalConfig = { configurable: { thread_id: rowSDB.thread_id, checkpoint_ns: checkpoint_ns ? checkpoint_ns : "", //prevent returning 'undefined' checkpoint_id: rowSDB.checkpoint_id, }, }; } if (!finalConfig.configurable?.thread_id || !finalConfig.configurable?.checkpoint_id) { debugLog(debug_getTuple, "getTuple▶4"); throw new Error("Missing thread_id or checkpoint_id"); } debugLog(debug_getTuple, "getTuple▶5"); // find any pending writes const query = /*surql*/ ` SELECT * FROM writes:[ $thread_id, $checkpoint_ns, $checkpoint_id, NONE, NONE ].. ` const bindings = { thread_id: finalConfig.configurable.thread_id.toString(), checkpoint_ns: checkpoint_ns ?? "", checkpoint_id: finalConfig.configurable.checkpoint_id.toString() } const pendingWritesRowsSDB = await this.db.query(query, { bindings }, true) as WritesRow[] debugLog(debug_getTuple, "getTuple▶6a", { pendingWritesRowsSDB }); const pendingWrites = await Promise.all( pendingWritesRowsSDB.map(async (rowSDB: WritesRow) => { try { const cpValueType = typeof rowSDB.cpValue console.log("typeof:", cpValueType) console.log("typeof val:", rowSDB.cpValue) const cpValueBuffer = (cpValueType == "object") ? Buffer.from(Object.values(rowSDB.cpValue!).map(Number)) : rowSDB.cpValue!; const cpValue = SERIALIZE ? await this.serde.loadsTyped(rowSDB.cpType ?? "json", cpValueBuffer) : cpValueBuffer return [ rowSDB.task_id, rowSDB.channel, cpValue ] as [string, string, unknown]; } catch (error) { console.error("Error in Promise.all callback:", error); throw error; } }) ); // SERIALIZE STEP const checkpointBuffer = Buffer.from(Object.values(rowSDB.checkpoint as unknown as number[])); // ✅/2 const metadataBuffer = Buffer.from(Object.values(rowSDB.metadata as unknown as number[])); // ✅/2 const checkpointData = SERIALIZE ? (await this.serde.loadsTyped( rowSDB.cpType ?? "json", checkpointBuffer )) as Checkpoint : (JSON.parse(rowSDB.checkpoint) ?? {}) const checkpointMetadata = SERIALIZE ? (await this.serde.loadsTyped( rowSDB.cpType ?? "json", metadataBuffer )) as CheckpointMetadata : (JSON.parse(rowSDB.metadata) ?? {}) const finalOutput: CheckpointTuple = { config: finalConfig, checkpoint: checkpointData, metadata: checkpointMetadata, parentConfig: rowSDB.parent_checkpoint_id ? { configurable: { thread_id: rowSDB.thread_id, checkpoint_ns: checkpoint_ns ?? "", checkpoint_id: rowSDB.parent_checkpoint_id, }, } : undefined, pendingWrites, }; debugLog(debug_getTuple, "getTuple▶7:", JSON.stringify(finalOutput, null, 2)); return finalOutput }; async *list( config: RunnableConfig, options?: CheckpointListOptions ): AsyncGenerator<CheckpointTuple> { const { limit, before } = options ?? {}; debugLog(debug_checkpointer, `${serviceName} list▶`) const thread_id = config.configurable?.thread_id; const binding = { thread_id, beforeId: before?.configurable?.checkpoint_id || null } const query = /*surql*/ ` SELECT * FROM checkpoints WHERE thread_id = $thread_id ${before ? `AND checkpoint_id < $beforeId` : ""} ORDER BY checkpoint_id DESC ${limit ? `LIMIT $limit` : ""} ` const rows = await this.db.query(query, binding) as CheckpointRow[] if (Array.isArray(rows) && rows.length > 0) { for (const row of rows) { // SERIALIZE STEP const checkpointBuffer = Buffer.from(Object.values(row.checkpoint as unknown as number[])); // ✅/2 const metadataBuffer = Buffer.from(Object.values(row.metadata as unknown as number[])); // ✅/2 const checkpointData = SERIALIZE ? (await this.serde.loadsTyped( row.cpType ?? "json", checkpointBuffer )) as Checkpoint : (JSON.parse(row.checkpoint) ?? {}) const checkpointMetadata = SERIALIZE ? (await this.serde.loadsTyped( row.cpType ?? "json", metadataBuffer )) as CheckpointMetadata : (JSON.parse(row.metadata) ?? {}) yield { config: { configurable: { thread_id: row.thread_id, checkpoint_ns: row.checkpoint_ns, checkpoint_id: row.checkpoint_id, }, }, checkpoint: checkpointData, metadata: checkpointMetadata, parentConfig: row.parent_checkpoint_id ? { configurable: { thread_id: row.thread_id, checkpoint_ns: row.checkpoint_ns, checkpoint_id: row.parent_checkpoint_id, }, } : undefined, }; } } } /** * @note PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id) * * @param config * @param checkpoint * @param metadata * @returns */ async put( config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata ): Promise<RunnableConfig> { debugLog(debug_checkpointer, `${serviceName} put▶`) // const debug_put = true; // const debug_put = false; const [type1, serializedCheckpoint] = this.serde.dumpsTyped(checkpoint); const [type2, serializedMetadata] = this.serde.dumpsTyped(metadata); if (type1 !== type2) { throw new Error( "Failed to serialized checkpoint and metadata to the same type." ); } const checkpointData = SERIALIZE ? serializedCheckpoint : Buffer.from(serializedCheckpoint).toString("utf-8") const checkpointMetadata = SERIALIZE ? serializedMetadata : Buffer.from(serializedMetadata).toString("utf-8") // const binding: CheckpointRow = { const binding = { thread_id: config.configurable?.thread_id?.toString(), checkpoint_ns: config.configurable?.checkpoint_ns ?? "", checkpoint_id: checkpoint.id, parent_checkpoint_id: config.configurable?.checkpoint_id, cpType: type1, checkpoint: checkpointData, metadata: checkpointMetadata, } const query = /*surql*/ ` UPSERT checkpoints:[$thread_id, $checkpoint_ns, $checkpoint_id] CONTENT { thread_id: $thread_id, checkpoint_ns: $checkpoint_ns, checkpoint_id: $checkpoint_id, parent_checkpoint_id: $parent_checkpoint_id, cpType: $cpType, checkpoint: $checkpoint, metadata: $metadata }; ` await this.db.query(query, binding as unknown as SurrealBindings) const configurable = { thread_id: config.configurable?.thread_id, checkpoint_ns: config.configurable?.checkpoint_ns ?? "", checkpoint_id: checkpoint.id, } return { configurable: configurable, }; } /** * @note PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) * * @param config * @param writes * @param taskId */ async putWrites( config: RunnableConfig, writes: PendingWrite[], taskId: string ): Promise<void> { debugLog(debug_checkpointer, `${serviceName} putWrites▶`) /** get the rows from the writes */ const rows = writes.map((write, idx) => { const [cpType, serializedWrite] = this.serde.dumpsTyped(write[1]); console.log("putWrites1:", serializedWrite) console.log("putWrites2:", write[1]) const cpValue = SERIALIZE ? serializedWrite : write[1] return { thread_id: config.configurable?.thread_id, checkpoint_ns: config.configurable?.checkpoint_ns ?? "", checkpoint_id: config.configurable?.checkpoint_id, task_id: taskId, idx: idx, channel: write[0], cpType: cpType, cpValue: cpValue, }; }); console.log("writes:", writes) console.log("rows:", rows) // SurrealDB QUERY const query = /*surql*/ ` CREATE writes:[$thread_id, $checkpoint_ns, $checkpoint_id, $task_id, $idx] SET thread_id = $thread_id, checkpoint_ns = $checkpoint_ns, checkpoint_id = $checkpoint_id, task_id = $task_id, idx = $idx, channel = $channel, cpType = $cpType, cpValue = $cpValue ` for (const row of rows) { const { thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, cpType, cpValue } = row const binding = { thread_id, checkpoint_ns: checkpoint_ns ?? "", checkpoint_id, task_id, idx, channel, cpType, cpValue, } await this.db.query(query, binding as unknown as SurrealBindings) } } } ```

yindo commented

@hgoona commented on GitHub (Oct 7, 2024):

SdbSaver Test:

import {
	describe,
	it,
	expect,
	beforeAll,
	type TestOptions
} from "vitest";

import {
	type Checkpoint,
	type CheckpointTuple,
	uuid6,
} from "@langchain/langgraph-checkpoint";
import { TestSdb1 } from "$lib/server/sdbTestDb";
import { SdbQueryClass } from "$lib/server/sdbUtils";
import { SdbSaver } from "./sdbSaver";
// import { SqliteSaver } from "../index.js";

const checkpoint1: Checkpoint = {
	v: 1,
	id: uuid6(-1),
	ts: "2024-04-19T17:19:07.952Z",
	channel_values: {
		someKey1: "someValue1",
	},
	channel_versions: {
		someKey2: 1,
	},
	versions_seen: {
		someKey3: {
			someKey4: 1,
		},
	},
	pending_sends: [],
};
console.log({ checkpoint1 });

const checkpoint2: Checkpoint = {
	v: 1,
	id: uuid6(1),
	ts: "2024-04-20T17:19:07.952Z",
	channel_values: {
		someKey1: "someValue2",
	},
	channel_versions: {
		someKey2: 2,
	},
	versions_seen: {
		someKey3: {
			someKey4: 2,
		},
	},
	pending_sends: [],
};

describe("SdbSaver - should save and retrieve checkpoints correctly", () => {
	/** 
	 * New instance of SdbSaver with (test) DB attached 
	 */
	let sdbSaver: SdbSaver

	/**
	 * SDBTester is a SdbQueryClass 
	 * that connects to the TestSdb1 database (in-Memory DB).
	 */
	let SDBTester: SdbQueryClass

	beforeAll(async () => {
		SDBTester = new SdbQueryClass(TestSdb1)

		// initialize SdbSaver ❗❗❗❗❗
		sdbSaver = new SdbSaver(SDBTester); // must be an SdbQueryClass
		// sdbSaver = new SdbSaver2(SDBTester); // must be an SdbQueryClass


		const query = /*surql*/ `
			-- init TABLES
			DEFINE TABLE IF NOT EXISTS checkpoints;
			DEFINE TABLE IF NOT EXISTS writes;

			-- define checkpoint fields
			DEFINE FIELD thread_id ON TABLE checkpoints TYPE string; // TEXT NOT NULL, 
			DEFINE FIELD checkpoint_ns ON TABLE checkpoints TYPE string DEFAULT ""; // TEXT NOT NULL DEFAULT '', 
			DEFINE FIELD checkpoint_id ON TABLE checkpoints TYPE string; // TEXT NOT NULL, 
			DEFINE FIELD parent_checkpoint_id ON TABLE checkpoints TYPE option<string>; // TEXT, 
			DEFINE FIELD cpType ON TABLE checkpoints TYPE option<string>; // TEXT, 
			-- DEFINE FIELD checkpoint ON TABLE checkpoints TYPE option<string>; // BLOB, 
			-- DEFINE FIELD metadata ON TABLE checkpoints TYPE option<string>; // BLOB, 

			-- define writes fields
			DEFINE FIELD thread_id ON TABLE writes TYPE string; // TEXT NOT NULL,
			DEFINE FIELD checkpoint_ns ON TABLE writes TYPE string DEFAULT ""; // TEXT NOT NULL DEFAULT '',
			DEFINE FIELD checkpoint_id ON TABLE writes TYPE string; // TEXT NOT NULL,
			DEFINE FIELD task_id ON TABLE writes TYPE string; // TEXT NOT NULL,
			DEFINE FIELD idx ON TABLE writes TYPE number; // INTEGER NOT NULL,
			DEFINE FIELD channel ON TABLE writes TYPE string; // TEXT NOT NULL,
			-- DEFINE FIELD cpType ON TABLE writes TYPE option<string>; // TEXT,
			-- DEFINE FIELD cpValue ON TABLE writes TYPE option<string>; // TEXT or could be a BLOB

			-- review
			INFO FOR TABLE checkpoints;
			INFO FOR TABLE writes;
		`

		const tableInfo = await SDBTester.query(query, {}, true) as Array<unknown>;
		console.log({ tableInfo })


	})

	it("confirm TestSdb - is connected", async () => {
		const info = await SDBTester.query("INFO FOR ROOT;")
		// console.log({ info })

		expect(info).not.toBeNull()
	})

	// CHECKPOINT SAVER TESTS
	// it("should save and retrieve checkpoints correctly", async () => {
	// const sqliteSaver = SqliteSaver.fromConnString(":memory:");
	// const sdbSaver = new SdbSaver(TestSdb1);

	it("get undefined checkpoint", async () => {
		// get undefined checkpoint
		const undefinedCheckpoint = await sdbSaver.getTuple({
			configurable: { thread_id: "1" },
		});
		expect(undefinedCheckpoint).toBeUndefined();
	});


	it("save first checkpoint", async () => {
		// save first checkpoint
		const runnableConfig = await sdbSaver.put(
			{ configurable: { thread_id: "1" } },
			checkpoint1,
			{ source: "update", step: -1, writes: null, parents: {} }
		);
		expect(runnableConfig).toEqual({
			configurable: {
				thread_id: "1",
				checkpoint_ns: "", //NONE stored as `""`
				checkpoint_id: checkpoint1.id,
			},
		});
	});

	it("add some writes", async () => {
		// add some writes
		await sdbSaver.putWrites(
			{
				configurable: {
					checkpoint_id: checkpoint1.id,
					checkpoint_ns: "",
					thread_id: "1",
				},
			},
			[["bar", "baz"]],
			"foo"
		);
	});


	it("get first checkpoint tuple", async () => { //❓❓❓
		// get first checkpoint tuple
		const firstCheckpointTuple = await sdbSaver.getTuple({
			configurable: { thread_id: "1" },
		});
		console.log("firstCheckpointTuple:", JSON.stringify(firstCheckpointTuple, null, 2));
		console.log("checkpoint1.id:", checkpoint1.id);

		expect(firstCheckpointTuple?.config).toEqual({
			configurable: {
				thread_id: "1",
				checkpoint_ns: "",
				checkpoint_id: checkpoint1.id,
			},
		});
		expect(firstCheckpointTuple?.checkpoint).toEqual(checkpoint1);
		expect(firstCheckpointTuple?.parentConfig).toBeUndefined();
		expect(firstCheckpointTuple?.pendingWrites).toEqual([
			["foo", "bar", "baz"],
		]);
	});

	it("save second checkpoint", async () => {
		// save second checkpoint
		await sdbSaver.put(
			{
				configurable: {
					thread_id: "1",
					checkpoint_id: "2024-04-18T17:19:07.952Z",
				},
			},
			checkpoint2,
			{ source: "update", step: -1, writes: null, parents: {} }
		);

		// verify that parentTs is set and retrieved correctly for second checkpoint
		const secondCheckpointTuple = await sdbSaver.getTuple({
			configurable: { thread_id: "1" },
		});

		// SCENARIO when checkpoint_ns = "undefined" in Database
		// expect(secondCheckpointTuple?.parentConfig?.configurable?.thread_id).toBe("1");
		// expect(secondCheckpointTuple?.parentConfig?.configurable?.checkpoint_ns ?? "").toBe("");
		// console.log("chk_ns:",secondCheckpointTuple?.parentConfig?.configurable?.checkpoint_ns)
		// expect(secondCheckpointTuple?.parentConfig?.configurable?.checkpoint_id).toBe("2024-04-18T17:19:07.952Z");
		expect(secondCheckpointTuple?.parentConfig).toEqual({
			configurable: {
				thread_id: "1",
				checkpoint_ns: "",
				checkpoint_id: "2024-04-18T17:19:07.952Z",
			},
		});
	});

	it("list checkpoints", async () => {
		// list checkpoints
		const checkpointTupleGenerator = await sdbSaver.list({
			configurable: { thread_id: "1" },
		});
		const checkpointTuples: CheckpointTuple[] = [];
		for await (const checkpoint of checkpointTupleGenerator) {
			checkpointTuples.push(checkpoint);
		}
		expect(checkpointTuples.length).toBe(2);

		const checkpointTuple1 = checkpointTuples[0];
		const checkpointTuple2 = checkpointTuples[1];
		expect(checkpointTuple1.checkpoint.ts).toBe("2024-04-20T17:19:07.952Z");
		expect(checkpointTuple2.checkpoint.ts).toBe("2024-04-19T17:19:07.952Z");
	});


	// // TESTING ================================================================================================
	// //keep the Vitest running to inspect the in-memory DB
	// const options: TestOptions = {
	// 	timeout: 2147483647, //max❗❗❗ 24.8 days
	// }
	// it("KEEP_TEST_DB_ALIVE",
	// 	options,
	// 	async () => {
	// 		await new Promise(resolve => { })
	// 	}
	// );
	// // TESTING ================================================================================================


});

/**
 * HOW TO RUN VITEST TEST:
 * bun run test -- --run --reporter default ✅
 * bun run test -- --run --reporter verbose
 * bun run test -- --run --reporter verbose --silent=false
 * 
 */

@hgoona commented on GitHub (Oct 7, 2024): SdbSaver Test: ```ts import { describe, it, expect, beforeAll, type TestOptions } from "vitest"; import { type Checkpoint, type CheckpointTuple, uuid6, } from "@langchain/langgraph-checkpoint"; import { TestSdb1 } from "$lib/server/sdbTestDb"; import { SdbQueryClass } from "$lib/server/sdbUtils"; import { SdbSaver } from "./sdbSaver"; // import { SqliteSaver } from "../index.js"; const checkpoint1: Checkpoint = { v: 1, id: uuid6(-1), ts: "2024-04-19T17:19:07.952Z", channel_values: { someKey1: "someValue1", }, channel_versions: { someKey2: 1, }, versions_seen: { someKey3: { someKey4: 1, }, }, pending_sends: [], }; console.log({ checkpoint1 }); const checkpoint2: Checkpoint = { v: 1, id: uuid6(1), ts: "2024-04-20T17:19:07.952Z", channel_values: { someKey1: "someValue2", }, channel_versions: { someKey2: 2, }, versions_seen: { someKey3: { someKey4: 2, }, }, pending_sends: [], }; describe("SdbSaver - should save and retrieve checkpoints correctly", () => { /** * New instance of SdbSaver with (test) DB attached */ let sdbSaver: SdbSaver /** * SDBTester is a SdbQueryClass * that connects to the TestSdb1 database (in-Memory DB). */ let SDBTester: SdbQueryClass beforeAll(async () => { SDBTester = new SdbQueryClass(TestSdb1) // initialize SdbSaver ❗❗❗❗❗ sdbSaver = new SdbSaver(SDBTester); // must be an SdbQueryClass // sdbSaver = new SdbSaver2(SDBTester); // must be an SdbQueryClass const query = /*surql*/ ` -- init TABLES DEFINE TABLE IF NOT EXISTS checkpoints; DEFINE TABLE IF NOT EXISTS writes; -- define checkpoint fields DEFINE FIELD thread_id ON TABLE checkpoints TYPE string; // TEXT NOT NULL, DEFINE FIELD checkpoint_ns ON TABLE checkpoints TYPE string DEFAULT ""; // TEXT NOT NULL DEFAULT '', DEFINE FIELD checkpoint_id ON TABLE checkpoints TYPE string; // TEXT NOT NULL, DEFINE FIELD parent_checkpoint_id ON TABLE checkpoints TYPE option<string>; // TEXT, DEFINE FIELD cpType ON TABLE checkpoints TYPE option<string>; // TEXT, -- DEFINE FIELD checkpoint ON TABLE checkpoints TYPE option<string>; // BLOB, -- DEFINE FIELD metadata ON TABLE checkpoints TYPE option<string>; // BLOB, -- define writes fields DEFINE FIELD thread_id ON TABLE writes TYPE string; // TEXT NOT NULL, DEFINE FIELD checkpoint_ns ON TABLE writes TYPE string DEFAULT ""; // TEXT NOT NULL DEFAULT '', DEFINE FIELD checkpoint_id ON TABLE writes TYPE string; // TEXT NOT NULL, DEFINE FIELD task_id ON TABLE writes TYPE string; // TEXT NOT NULL, DEFINE FIELD idx ON TABLE writes TYPE number; // INTEGER NOT NULL, DEFINE FIELD channel ON TABLE writes TYPE string; // TEXT NOT NULL, -- DEFINE FIELD cpType ON TABLE writes TYPE option<string>; // TEXT, -- DEFINE FIELD cpValue ON TABLE writes TYPE option<string>; // TEXT or could be a BLOB -- review INFO FOR TABLE checkpoints; INFO FOR TABLE writes; ` const tableInfo = await SDBTester.query(query, {}, true) as Array<unknown>; console.log({ tableInfo }) }) it("confirm TestSdb - is connected", async () => { const info = await SDBTester.query("INFO FOR ROOT;") // console.log({ info }) expect(info).not.toBeNull() }) // CHECKPOINT SAVER TESTS // it("should save and retrieve checkpoints correctly", async () => { // const sqliteSaver = SqliteSaver.fromConnString(":memory:"); // const sdbSaver = new SdbSaver(TestSdb1); it("get undefined checkpoint", async () => { // get undefined checkpoint const undefinedCheckpoint = await sdbSaver.getTuple({ configurable: { thread_id: "1" }, }); expect(undefinedCheckpoint).toBeUndefined(); }); it("save first checkpoint", async () => { // save first checkpoint const runnableConfig = await sdbSaver.put( { configurable: { thread_id: "1" } }, checkpoint1, { source: "update", step: -1, writes: null, parents: {} } ); expect(runnableConfig).toEqual({ configurable: { thread_id: "1", checkpoint_ns: "", //NONE stored as `""` checkpoint_id: checkpoint1.id, }, }); }); it("add some writes", async () => { // add some writes await sdbSaver.putWrites( { configurable: { checkpoint_id: checkpoint1.id, checkpoint_ns: "", thread_id: "1", }, }, [["bar", "baz"]], "foo" ); }); it("get first checkpoint tuple", async () => { //❓❓❓ // get first checkpoint tuple const firstCheckpointTuple = await sdbSaver.getTuple({ configurable: { thread_id: "1" }, }); console.log("firstCheckpointTuple:", JSON.stringify(firstCheckpointTuple, null, 2)); console.log("checkpoint1.id:", checkpoint1.id); expect(firstCheckpointTuple?.config).toEqual({ configurable: { thread_id: "1", checkpoint_ns: "", checkpoint_id: checkpoint1.id, }, }); expect(firstCheckpointTuple?.checkpoint).toEqual(checkpoint1); expect(firstCheckpointTuple?.parentConfig).toBeUndefined(); expect(firstCheckpointTuple?.pendingWrites).toEqual([ ["foo", "bar", "baz"], ]); }); it("save second checkpoint", async () => { // save second checkpoint await sdbSaver.put( { configurable: { thread_id: "1", checkpoint_id: "2024-04-18T17:19:07.952Z", }, }, checkpoint2, { source: "update", step: -1, writes: null, parents: {} } ); // verify that parentTs is set and retrieved correctly for second checkpoint const secondCheckpointTuple = await sdbSaver.getTuple({ configurable: { thread_id: "1" }, }); // SCENARIO when checkpoint_ns = "undefined" in Database // expect(secondCheckpointTuple?.parentConfig?.configurable?.thread_id).toBe("1"); // expect(secondCheckpointTuple?.parentConfig?.configurable?.checkpoint_ns ?? "").toBe(""); // console.log("chk_ns:",secondCheckpointTuple?.parentConfig?.configurable?.checkpoint_ns) // expect(secondCheckpointTuple?.parentConfig?.configurable?.checkpoint_id).toBe("2024-04-18T17:19:07.952Z"); expect(secondCheckpointTuple?.parentConfig).toEqual({ configurable: { thread_id: "1", checkpoint_ns: "", checkpoint_id: "2024-04-18T17:19:07.952Z", }, }); }); it("list checkpoints", async () => { // list checkpoints const checkpointTupleGenerator = await sdbSaver.list({ configurable: { thread_id: "1" }, }); const checkpointTuples: CheckpointTuple[] = []; for await (const checkpoint of checkpointTupleGenerator) { checkpointTuples.push(checkpoint); } expect(checkpointTuples.length).toBe(2); const checkpointTuple1 = checkpointTuples[0]; const checkpointTuple2 = checkpointTuples[1]; expect(checkpointTuple1.checkpoint.ts).toBe("2024-04-20T17:19:07.952Z"); expect(checkpointTuple2.checkpoint.ts).toBe("2024-04-19T17:19:07.952Z"); }); // // TESTING ================================================================================================ // //keep the Vitest running to inspect the in-memory DB // const options: TestOptions = { // timeout: 2147483647, //max❗❗❗ 24.8 days // } // it("KEEP_TEST_DB_ALIVE", // options, // async () => { // await new Promise(resolve => { }) // } // ); // // TESTING ================================================================================================ }); /** * HOW TO RUN VITEST TEST: * bun run test -- --run --reporter default ✅ * bun run test -- --run --reporter verbose * bun run test -- --run --reporter verbose --silent=false * */ ```

yindo commented

https://github.com/langchain-ai/langgraphjs/blob/main/libs/langgraph/src/tests/pregel.postgres_saver.int.test.ts

@jacoblee93 commented on GitHub (Oct 8, 2024):

Hey @hgoona, have just merged #471. I've opted to just add an integration test that imports and uses the pregel.test.ts suite:

I think a more granular test suite + docs would be nice too though, so will keep this open.

@jacoblee93 commented on GitHub (Oct 8, 2024): Hey @hgoona, have just merged #471. I've opted to just add an integration test that imports and uses the `pregel.test.ts` suite: https://github.com/langchain-ai/langgraphjs/blob/main/libs/langgraph/src/tests/pregel.postgres_saver.int.test.ts I think a more granular test suite + docs would be nice too though, so will keep this open.

yindo commented

@hgoona commented on GitHub (Oct 8, 2024):

Hi @jacoblee93 I'm not sure I understand/follow. Are you saying the Postgres Saver + Postgres Saver test is the best example I should follow in order to setup my SurrealDB Saver?
Or something else? Can I use one of these tests to test my SdbSaver?? If so which one? (sorry I'm new to, and a bit unclear of the details happening)

@hgoona commented on GitHub (Oct 8, 2024): Hi @jacoblee93 I'm not sure I understand/follow. Are you saying the Postgres Saver + Postgres Saver test is the best example I should follow in order to setup my SurrealDB Saver? Or something else? Can I use one of these tests to test my SdbSaver?? If so which one? (sorry I'm new to, and a bit unclear of the details happening)

yindo commented

@benjamincburns commented on GitHub (Oct 8, 2024):

Are you saying the Postgres Saver + Postgres Saver test is the best example I should follow in order to setup my SurrealDB Saver?
Or something else? Can I use one of these tests to test my SdbSaver?? If so which one? (sorry I'm new to, and a bit unclear of the details happening)

Not answering for Jacob here, but hopefully by the end of the week or early next you'll be able to use the tooling that I'm building to validate SdbSaver (or any other arbitrary checkpoint saver). You could give it a try now, but I'm not a member of the LangGraph team (nor do I work for LangChain), so it's quite likely that my understanding of some of the finer details of the checkpoint saver contract are flawed. Writing this test suite is how I'm developing that understanding, and I'm finding holes in my own knowledge/assumptions as I go.

With that in mind, if you still want to try it out, you can see how I wired it up to the existing mongodb checkpointer here. Note that this is a link to a specific commit, as I may break things on that branch as I go, so here's a link to the checkpoint-validation package under my checkpoint-acceptance-tests branch.

I should also warn you that I'm not in a position right now to provide any additional assistance with how to wire that up, how to interpret the test results, etc. If you wait until after I've had a chance to submit it as a PR (or more ideally, until after it has been reviewed by the LangGraph team), you'll be able to have more confidence in the results, and there will be a bit more documentation available as to how to go about making it all work.

Also bear in mind that it's currently an incomplete set of tests of each of the abstract methods on types that extend BaseCheckpointSaver, except for the list method, which I haven't written tests for yet.

I was wondering if either of you would be able to review this custom Saver I've made for SurrealDB?

I'm not able to do this at the moment, sorry. Also instead of pasting code into issues (especially lengthy or complex code), I'd recommend sharing a link to a PR (even just a PR to your own fork or other repo). PRs offer a better UI for reading code and providing feedback than issues do, and it'll help keep the comment history on topic and more readable.

@benjamincburns commented on GitHub (Oct 8, 2024): > Are you saying the Postgres Saver + Postgres Saver test is the best example I should follow in order to setup my SurrealDB Saver? > Or something else? Can I use one of these tests to test my SdbSaver?? If so which one? (sorry I'm new to, and a bit unclear of the details happening) Not answering for Jacob here, but hopefully by the end of the week or early next you'll be able to use the tooling that I'm building to validate `SdbSaver` (or any other arbitrary checkpoint saver). You could give it a try now, but I'm not a member of the LangGraph team (nor do I work for LangChain), so it's quite likely that my understanding of some of the finer details of the checkpoint saver contract are flawed. Writing this test suite is how I'm developing that understanding, and I'm finding holes in my own knowledge/assumptions as I go. With that in mind, if you still want to try it out, you can see how I wired it up to the existing mongodb checkpointer [here](https://github.com/benjamincburns/langgraphjs/blob/8d861c6f5011ef74821027103858a70bb3678cb0/libs/checkpoint-validation/src/tests/mongodb.spec.ts). Note that this is a link to a specific commit, as I may break things on that branch as I go, so [here's a link to the `checkpoint-validation` package under my `checkpoint-acceptance-tests` branch](https://github.com/benjamincburns/langgraphjs/tree/checkpoint-acceptance-tests/libs/checkpoint-validation). I should also warn you that I'm not in a position right now to provide any additional assistance with how to wire that up, how to interpret the test results, etc. If you wait until after I've had a chance to submit it as a PR (or more ideally, until after it has been reviewed by the LangGraph team), you'll be able to have more confidence in the results, and there will be a bit more documentation available as to how to go about making it all work. Also bear in mind that it's currently an incomplete set of tests of each of the abstract methods on types that extend `BaseCheckpointSaver`, except for the `list` method, which I haven't written tests for yet. > I was wondering if either of you would be able to review this custom Saver I've made for SurrealDB? I'm not able to do this at the moment, sorry. Also instead of pasting code into issues (especially lengthy or complex code), I'd recommend sharing a link to a PR (even just a PR to your own fork or other repo). PRs offer a better UI for reading code and providing feedback than issues do, and it'll help keep the comment history on topic and more readable.

yindo commented

@hgoona commented on GitHub (Oct 8, 2024):

@benjamincburns thank you for the feedback! I think I follow what you're saying: I've done my first fork of the LangGraph library, created a branch for SurrealdbSaver here:
https://github.com/hgoona/langgraphjs/blob/SurrealDb_checkpointer/libs/checkpoint-surrealdb/src/index.ts

Note: it was based off the Sqlite version. I haven't edited the test file or anything else copied from that Sqlite save, except to update the index.ts and the package.json (which is to say - this probably has errors!)

Please take a look! This current version of the SurrealdbSaver works with my LangGraph in a very very basic test, but it fails the Vitest test! I haven't had a chance to try it against your setup @jacoblee93 @benjamincburns << if you get a chance, let me know if it works for you!

Would appreciate more feedback on the code! Hopefully, I've done it right (though the test is broken for me)

@hgoona commented on GitHub (Oct 8, 2024): @benjamincburns thank you for the feedback! I think I follow what you're saying: I've done my first fork of the LangGraph library, created a branch for SurrealdbSaver here: https://github.com/hgoona/langgraphjs/blob/SurrealDb_checkpointer/libs/checkpoint-surrealdb/src/index.ts Note: it was based off the Sqlite version. I haven't edited the test file or anything else copied from that Sqlite save, except to update the index.ts and the package.json (which is to say - this probably has errors!) Please take a look! This current version of the SurrealdbSaver works with my LangGraph in a very very basic test, but it fails the Vitest test! I haven't had a chance to try it against your setup @jacoblee93 @benjamincburns << if you get a chance, let me know if it works for you! Would appreciate more feedback on the code! Hopefully, I've done it right (though the test is broken for me)

yindo commented