Class PostgresSaver

LangGraph checkpointer that uses a Postgres instance as the backing store. Uses the node-postgres package internally to connect to a Postgres instance.

Example

import { ChatOpenAI } from "@langchain/openai";
import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";
import { createReactAgent } from "@langchain/langgraph/prebuilt";

const checkpointer = PostgresSaver.fromConnString(
  "postgresql://user:password@localhost:5432/db"
);

// NOTE: you need to call .setup() the first time you're using your checkpointer
await checkpointer.setup();

const graph = createReactAgent({
  tools: [getWeather],
  llm: new ChatOpenAI({
    model: "gpt-4o-mini",
  }),
  checkpointSaver: checkpointer,
});
const config = { configurable: { thread_id: "1" } };

await graph.invoke({
  messages: [{
    role: "user",
    content: "what's the weather in sf"
  }],
}, config);

Hierarchy (view full)

Constructors

constructor

Properties

isSetup pool serde

Methods

_dumpBlobs _dumpCheckpoint _dumpMetadata _dumpWrites _loadBlobs _loadCheckpoint _loadMetadata _loadWrites _searchWhere end get getNextVersion getTuple list put putWrites setup fromConnString

Constructors

constructor

Properties

Protected isSetup

isSetup: boolean

Private pool

pool: any

serde

serde: SerializerProtocol

Methods

Protected _dumpBlobs

  • _dumpBlobs(threadId, checkpointNs, values, versions): [string, string, string, string, string, undefined | Uint8Array][]

  • Parameters

    • threadId: string
    • checkpointNs: string
    • values: Record<string, unknown>
    • versions: ChannelVersions

    Returns [string, string, string, string, string, undefined | Uint8Array][]

Protected _dumpCheckpoint

  • _dumpCheckpoint(checkpoint): Record<string, unknown>

  • Parameters

    Returns Record<string, unknown>

Protected _dumpMetadata

  • _dumpMetadata(metadata): any

  • Parameters

    Returns any

Protected _dumpWrites

  • _dumpWrites(threadId, checkpointNs, checkpointId, taskId, writes): [string, string, string, string, number, string, string, Uint8Array][]

  • Parameters

    • threadId: string
    • checkpointNs: string
    • checkpointId: string
    • taskId: string
    • writes: [string, unknown][]

    Returns [string, string, string, string, number, string, string, Uint8Array][]

Protected _loadBlobs

  • _loadBlobs(blobValues): Promise<Record<string, unknown>>

  • Parameters

    • blobValues: [Uint8Array, Uint8Array, Uint8Array][]

    Returns Promise<Record<string, unknown>>

Protected _loadCheckpoint

  • _loadCheckpoint(checkpoint, channelValues, pendingSends): Promise<Checkpoint<string, string>>

  • Parameters

    • checkpoint: Omit<Checkpoint<string, string>, "channel_values" | "pending_sends">
    • channelValues: [Uint8Array, Uint8Array, Uint8Array][]
    • pendingSends: [Uint8Array, Uint8Array][]

    Returns Promise<Checkpoint<string, string>>

Protected _loadMetadata

  • _loadMetadata(metadata): Promise<any>

  • Parameters

    • metadata: Record<string, unknown>

    Returns Promise<any>

Protected _loadWrites

  • _loadWrites(writes): Promise<[string, string, unknown][]>

  • Parameters

    • writes: [Uint8Array, Uint8Array, Uint8Array, Uint8Array][]

    Returns Promise<[string, string, unknown][]>

Protected _searchWhere

  • _searchWhere(config?, filter?, before?): [string, unknown[]]

  • Return WHERE clause predicates for alist() given config, filter, cursor.

    This method returns a tuple of a string and a tuple of values. The string is the parametered WHERE clause predicate (including the WHERE keyword): "WHERE column1 = $1 AND column2 IS $2". The list of values contains the values for each of the corresponding parameters.

    Parameters

    • Optional config: RunnableConfig<Record<string, any>>
    • Optional filter: Record<string, unknown>
    • Optional before: RunnableConfig<Record<string, any>>

    Returns [string, unknown[]]

end

  • end(): Promise<void>

  • Returns Promise<void>

get

  • get(config): Promise<undefined | Checkpoint<string, string>>

  • Parameters

    • config: RunnableConfig<Record<string, any>>

    Returns Promise<undefined | Checkpoint<string, string>>

getNextVersion

  • getNextVersion(current, _channel): number

  • Generate the next version ID for a channel.

    Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions, as long as they are monotonically increasing.

    Parameters

    • current: undefined | number
    • _channel: ChannelProtocol<unknown, unknown, unknown>

    Returns number

getTuple

  • getTuple(config): Promise<undefined | CheckpointTuple>

  • Get a checkpoint tuple from the database. This method retrieves a checkpoint tuple from the Postgres database based on the provided config. If the config's configurable field contains a "checkpoint_id" key, the checkpoint with the matching thread_id and namespace is retrieved. Otherwise, the latest checkpoint for the given thread_id is retrieved.

    Parameters

    • config: RunnableConfig<Record<string, any>>

      The config to use for retrieving the checkpoint.

    Returns Promise<undefined | CheckpointTuple>

    The retrieved checkpoint tuple, or undefined.

list

  • list(config, options?): AsyncGenerator<CheckpointTuple, any, unknown>

  • List checkpoints from the database.

    This method retrieves a list of checkpoint tuples from the Postgres database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

    Parameters

    Returns AsyncGenerator<CheckpointTuple, any, unknown>

put

  • put(config, checkpoint, metadata, newVersions): Promise<RunnableConfig<Record<string, any>>>

  • Save a checkpoint to the database.

    This method saves a checkpoint to the Postgres database. The checkpoint is associated with the provided config and its parent config (if any).

    Parameters

    Returns Promise<RunnableConfig<Record<string, any>>>

putWrites

  • putWrites(config, writes, taskId): Promise<void>

  • Store intermediate writes linked to a checkpoint.

    This method saves intermediate writes associated with a checkpoint to the Postgres database.

    Parameters

    • config: RunnableConfig<Record<string, any>>

      Configuration of the related checkpoint.

    • writes: PendingWrite<string>[]

      List of writes to store.

    • taskId: string

      Identifier for the task creating the writes.

    Returns Promise<void>

setup

  • setup(): Promise<void>

  • Set up the checkpoint database asynchronously.

    This method creates the necessary tables in the Postgres database if they don't already exist and runs database migrations. It MUST be called directly by the user the first time checkpointer is used.

    Returns Promise<void>

Static fromConnString

  • fromConnString(connString): PostgresSaver

  • Parameters

    • connString: string

    Returns PostgresSaver

Settings

Member Visibility

Theme

On This Page

constructorisSetuppoolserde_dumpBlobs_dumpCheckpoint_dumpMetadata_dumpWrites_loadBlobs_loadCheckpoint_loadMetadata_loadWrites_searchWhereendgetgetNextVersiongetTuplelistputputWritessetupfromConnString