DOCSEARCHbyAlgolia
Submit Docs

langgraph-checkpoint-mysql

repository·main·Indexed 41 days ago

https://github.com/tjni/langgraph-checkpoint-mysql

A MySQL implementation of the LangGraph CheckpointSaver for persisting state, conversation history, and state recovery in MySQL or MariaDB databases. Version 3.0.0 provides PyMySQLSaver for synchronous operations and AIOMySQLSaver for asynchronous operations, supporting MySQL >= 8.0.19 or MariaDB >= 10.7.1.

Tokens
1.4K
Snippets
3
Records
4
Agent score
50%

What's inside langgraph-checkpoint-mysql

  1. Install langgraph-checkpoint-mysql with specific drivers

    main

    Install the package with the extra dependency required for your preferred driver:

    • For synchronous usage with PyMySQLSaver: pip install langgraph-checkpoint-mysql[pymysql]
    • For asynchronous usage with AIOMySQLSaver: pip install langgraph-checkpoint-mysql[aiomysql]
    • For asynchronous usage with AsyncMySaver: pip install langgraph-checkpoint-mysql[asyncmy]

    Requirements:

    • MySQL >= 8.0.19 or MariaDB >= 10.7.1.
    • Note: MySQL >= 9.6.0 has discontinued the MD5 function in generated columns, which may affect compatibility.
    pip install langgraph-checkpoint-mysql[pymysql]
pip install langgraph-checkpoint-mysql[aiomysql]
pip install langgraph-checkpoint-mysql[asyncmy]

  2. Initialize MySQL checkpointer tables with .setup()

    main

    When using a MySQL checkpointer for the first time, you must call the .setup() method to create the required database tables.

    Important Connection Note: If you are manually creating MySQL connections and passing them to PyMySQLSaver or AIOMySQLSaver, you must include autocommit=True. Without this, the .setup() method may fail to persist the newly created tables to the database.

  3. Use AIOMySQLSaver for asynchronous checkpointing

    main

    Use AIOMySQLSaver for asynchronous operations. This implementation provides async with context management and awaitable methods for checkpoint operations.

    Key methods:

    • .aput(config, checkpoint, metadata, maybe_writes): Asynchronously stores a checkpoint.
    • .aget(config): Asynchronously loads a checkpoint.
    • .alist(config): Asynchronously lists checkpoints (returns an async iterator).
    from langgraph.checkpoint.mysql.aio import AIOMySQLSaver

DB_URI = "mysql://mysql:mysql@localhost:3306/mysql"
write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}

async with AIOMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {"my_key": "meow", "node": "node"},
        "channel_versions": {"__start__": 2, "my_key": 3, "start:node": 3, "node": 3},
        "versions_seen": {"__input__": {}, "__start__": {"__start__": 1}, "node": {"start:node": 2}},
    }

    # store checkpoint
    await checkpointer.aput(write_config, checkpoint, {}, {})

    # load checkpoint
    await checkpointer.aget(read_config)

    # list checkpoints
    [c async for c in checkpointer.alist(read_config)]

  4. Use PyMySQLSaver for synchronous checkpointing

    main

    Use PyMySQLSaver to store, load, and list checkpoints synchronously. Use from_conn_string(DB_URI) to initialize the checkpointer and ensure you call .setup() on the first run.

    Key methods:

    • .setup(): Creates required tables.
    • .put(config, checkpoint, metadata, maybe_writes): Stores a checkpoint.
    • .get(config): Loads a checkpoint.
    • .list(config): Lists checkpoints.
    from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver

write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}
DB_URI = "mysql://mysql:mysql@localhost:3306/mysql"

with PyMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    # call .setup() the first time you're using the checkpointer
    checkpointer.setup()
    
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {"my_key": "meow", "node": "node"},
        "channel_versions": {"__start__": 2, "my_key": 3, "start:node": 3, "node": 3},
        "versions_seen": {"__input__": {}, "__start__": {"__start__": 1}, "node": {"start:node": 2}},
    }

    # store checkpoint
    checkpointer.put(write_config, checkpoint, {}, {})

    # load checkpoint
    checkpointer.get(read_config)

    # list checkpoints
    list(checkpointer.list(read_config))