Skip to content

No examples for using async checkpointer with Functional API #1869

@markdascher

Description

@markdascher

Every Functional API example either omits the checkpointer definition or uses InMemorySaver. However, as soon as you switch to an async checkpointer, you run into a problem, because langchain-ai/langgraph#1580 (which is used as a template by many third-party checkpointers as well) made it impossible to construct an async checkpointer outside of an async function. (Because get_running_loop can only be called from a coroutine or a callback.)

So it seems like you have to do something goofy like this:

def make_workflow(checkpointer: AsyncRedisSaver):
    checkpointer = AsyncRedisSaver(redis_client=redis_client)
    await checkpointer.setup()

    @entrypoint(checkpointer=checkpointer)
    async def workflow() -> None:
        …

    return workflow

But surely there's a better way?

Metadata

Metadata

Assignees

No one assigned

    Labels

    externalUser is not a member of langchain-ai

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions