unfazed-checkpointer 0.0.1

Langgraph checkpointer Plugin for Mysql

Unfazed Checkpointer

unfazed-checkoutpointer is a unfazed plugin for langgraph.

important:

• only support async methods now.
• only support mysql/tidb protocol now.

reference docs:

• what is checkpoint
• checkpoint for pg

Usage

Install unfazed-checkpointer

make sure you already installed unfazed

pip install unfazed-checkpointer

Migrate the DB

currently we only support mysql or tidb

# for the first migration
unfazed-cli init-db


# for the upgrade
unfazed-cli migrate
unfazed-cli upgrade

Use it in your code

import uuid

from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command as LangGraphCommand
from langgraph.types import interrupt

from unfazed_checkpointer.services import TortoiseCheckpointSaver

# ==================== State & Nodes ====================


class SimpleState(t.TypedDict):
    foo: str
    bar: t.Annotated[list[str], operator.add]



class InterruptState(t.TypedDict):
    foo: str
    bar: t.Annotated[list[str], operator.add]


def interrupt_node(state: InterruptState) -> dict[str, t.Any]:
    """Runs second, interrupts and modifies foo on resume."""
    approved = interrupt("Do you want to proceed?")
    return {"foo": f"a-approved:{approved}", "bar": ["a"]}

async def main():
    workflow = StateGraph(SimpleState)
    workflow.add_node("node", interrupt_node)
    workflow.add_edge(START, "node")
    workflow.add_edge("node", END)
    checkpointer = TortoiseCheckpointSaver()
    graph = workflow.compile(checkpointer=checkpointer)
    thread_id = str(uuid.uuid4())
    config: RunnableConfig = {"configurable": {"thread_id": thread_id}}
    await graph.ainvoke(
            t.cast(SimpleState, {"foo": "init", "bar": ["init"]}), config
        )