langchain-ai/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-07-15 12:35:43 -04:00
Code Issues 314 Packages Projects Releases 17 Wiki Activity

Issue with from langchain_core.runnables import RunnableConfig #790

New Issue
Closed
opened 2026-02-20 17:41:47 -05:00 by yindo · 4 comments
yindo commented 2026-02-20 17:41:47 -05:00
Owner
Copy Link
Copy Source

Originally created by @shubhamgajbhiye1994 on GitHub (Jul 11, 2025).

Checked other resources

  • This is a bug, not a usage question. For questions, please use GitHub Discussions.
  • I added a clear and detailed title that summarizes the issue.
  • I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
  • I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

source path in langgraph checkpoint:langgraph/checkpoint/memory/__init__.py


import logging
import os
import pickle
import random
import shutil
from collections import defaultdict
from collections.abc import AsyncIterator, Iterator, Sequence
from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack
from types import TracebackType
from typing import Any, Optional, Union

from langchain_core.runnables import RunnableConfig

from langgraph.checkpoint.base import (
    WRITES_IDX_MAP,
    BaseCheckpointSaver,
    ChannelVersions,
    Checkpoint,
    CheckpointMetadata,
    CheckpointTuple,
    SerializerProtocol,
    get_checkpoint_id,
    get_checkpoint_metadata,
)
from langgraph.checkpoint.serde.types import TASKS, ChannelProtocol

logger = logging.getLogger(__name__)


class InMemorySaver(
    BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager
):
    """An in-memory checkpoint saver.

    This checkpoint saver stores checkpoints in memory using a defaultdict.

    Note:
        Only use `InMemorySaver` for debugging or testing purposes.
        For production use cases we recommend installing [langgraph-checkpoint-postgres](https://pypi.org/project/langgraph-checkpoint-postgres/) and using `PostgresSaver` / `AsyncPostgresSaver`.

        If you are using the LangGraph Platform, no checkpointer needs to be specified. The correct managed checkpointer will be used automatically.

    Args:
        serde: The serializer to use for serializing and deserializing checkpoints. Defaults to None.

    Examples:

            import asyncio

            from langgraph.checkpoint.memory import InMemorySaver
            from langgraph.graph import StateGraph

            builder = StateGraph(int)
            builder.add_node("add_one", lambda x: x + 1)
            builder.set_entry_point("add_one")
            builder.set_finish_point("add_one")

            memory = InMemorySaver()
            graph = builder.compile(checkpointer=memory)
            coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
            asyncio.run(coro)  # Output: 2
    """

    # thread ID ->  checkpoint NS -> checkpoint ID -> checkpoint mapping
    storage: defaultdict[
        str,
        dict[
            str, dict[str, tuple[tuple[str, bytes], tuple[str, bytes], Optional[str]]]
        ],
    ]
    # (thread ID, checkpoint NS, checkpoint ID) -> (task ID, write idx)
    writes: defaultdict[
        tuple[str, str, str],
        dict[tuple[str, int], tuple[str, str, tuple[str, bytes], str]],
    ]
    blobs: dict[
        tuple[
            str, str, str, Union[str, int, float]
        ],  # thread id, checkpoint ns, channel, version
        tuple[str, bytes],
    ]

    def __init__(
        self,
        *,
        serde: Optional[SerializerProtocol] = None,
        factory: type[defaultdict] = defaultdict,
    ) -> None:
        super().__init__(serde=serde)
        self.storage = factory(lambda: defaultdict(dict))
        self.writes = factory(dict)
        self.blobs = factory()
        self.stack = ExitStack()
        if factory is not defaultdict:
            self.stack.enter_context(self.storage)  # type: ignore[arg-type]
            self.stack.enter_context(self.writes)  # type: ignore[arg-type]
            self.stack.enter_context(self.blobs)  # type: ignore[arg-type]

    def __enter__(self) -> "InMemorySaver":
        return self.stack.__enter__()

    def __exit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        return self.stack.__exit__(exc_type, exc_value, traceback)

    async def __aenter__(self) -> "InMemorySaver":
        return self.stack.__enter__()

    async def __aexit__(
        self,
        __exc_type: Optional[type[BaseException]],
        __exc_value: Optional[BaseException],
        __traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        return self.stack.__exit__(__exc_type, __exc_value, __traceback)

    def _load_blobs(
        self, thread_id: str, checkpoint_ns: str, versions: ChannelVersions
    ) -> dict[str, Any]:
        channel_values: dict[str, Any] = {}
        for k, v in versions.items():
            kk = (thread_id, checkpoint_ns, k, v)
            if kk in self.blobs:
                vv = self.blobs[kk]
                if vv[0] != "empty":
                    channel_values[k] = self.serde.loads_typed(vv)
        return channel_values

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the in-memory storage.

        This method retrieves a checkpoint tuple from the in-memory storage based on the
        provided config. If the config contains a "checkpoint_id" key, the checkpoint with
        the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config: The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        thread_id: str = config["configurable"]["thread_id"]
        checkpoint_ns: str = config["configurable"].get("checkpoint_ns", "")
        if checkpoint_id := get_checkpoint_id(config):
            if saved := self.storage[thread_id][checkpoint_ns].get(checkpoint_id):
                checkpoint, metadata, parent_checkpoint_id = saved
                writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                if parent_checkpoint_id:
                    sends = sorted(
                        (
                            (*w, k[1])
                            for k, w in self.writes[
                                (thread_id, checkpoint_ns, parent_checkpoint_id)
                            ].items()
                            if w[1] == TASKS
                        ),
                        key=lambda w: (w[3], w[0], w[4]),
                    )
                else:
                    sends = []
                checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
                return CheckpointTuple(
                    config=config,
                    checkpoint={
                        **checkpoint_,
                        "channel_values": self._load_blobs(
                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
                        ),
                        "pending_sends": [self.serde.loads_typed(s[2]) for s in sends],
                    },
                    metadata=self.serde.loads_typed(metadata),
                    pending_writes=[
                        (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes
                    ],
                    parent_config=(
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                )
        else:
            if checkpoints := self.storage[thread_id][checkpoint_ns]:
                checkpoint_id = max(checkpoints.keys())
                checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
                writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                if parent_checkpoint_id:
                    sends = sorted(
                        (
                            (*w, k[1])
                            for k, w in self.writes[
                                (thread_id, checkpoint_ns, parent_checkpoint_id)
                            ].items()
                            if w[1] == TASKS
                        ),
                        key=lambda w: (w[3], w[0], w[4]),
                    )
                else:
                    sends = []

                checkpoint_ = self.serde.loads_typed(checkpoint)

                return CheckpointTuple(
                    config={
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
                            "checkpoint_id": checkpoint_id,
                        }
                    },
                    checkpoint={
                        **checkpoint_,
                        "channel_values": self._load_blobs(
                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
                        ),
                        "pending_sends": [self.serde.loads_typed(s[2]) for s in sends],
                    },
                    metadata=self.serde.loads_typed(metadata),
                    pending_writes=[
                        (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes
                    ],
                    parent_config=(
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": parent_checkpoint_id,
                            }
                        }
                        if parent_checkpoint_id
                        else None
                    ),
                )

    def list(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the in-memory storage.

        This method retrieves a list of checkpoint tuples from the in-memory storage based
        on the provided criteria.

        Args:
            config: Base configuration for filtering checkpoints.
            filter: Additional filtering criteria for metadata.
            before: List checkpoints created before this configuration.
            limit: Maximum number of checkpoints to return.

        Yields:
            Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
        """
        thread_ids = (config["configurable"]["thread_id"],) if config else self.storage
        config_checkpoint_ns = (
            config["configurable"].get("checkpoint_ns") if config else None
        )
        config_checkpoint_id = get_checkpoint_id(config) if config else None
        for thread_id in thread_ids:
            for checkpoint_ns in self.storage[thread_id].keys():
                if (
                    config_checkpoint_ns is not None
                    and checkpoint_ns != config_checkpoint_ns
                ):
                    continue

                for checkpoint_id, (
                    checkpoint,
                    metadata_b,
                    parent_checkpoint_id,
                ) in sorted(
                    self.storage[thread_id][checkpoint_ns].items(),
                    key=lambda x: x[0],
                    reverse=True,
                ):
                    # filter by checkpoint ID from config
                    if config_checkpoint_id and checkpoint_id != config_checkpoint_id:
                        continue

                    # filter by checkpoint ID from `before` config
                    if (
                        before
                        and (before_checkpoint_id := get_checkpoint_id(before))
                        and checkpoint_id >= before_checkpoint_id
                    ):
                        continue

                    # filter by metadata
                    metadata = self.serde.loads_typed(metadata_b)
                    if filter and not all(
                        query_value == metadata.get(query_key)
                        for query_key, query_value in filter.items()
                    ):
                        continue

                    # limit search results
                    if limit is not None and limit <= 0:
                        break
                    elif limit is not None:
                        limit -= 1

                    writes = self.writes[
                        (thread_id, checkpoint_ns, checkpoint_id)
                    ].values()

                    if parent_checkpoint_id:
                        sends = sorted(
                            (
                                (*w, k[1])
                                for k, w in self.writes[
                                    (thread_id, checkpoint_ns, parent_checkpoint_id)
                                ].items()
                                if w[1] == TASKS
                            ),
                            key=lambda w: (w[3], w[0], w[4]),
                        )
                    else:
                        sends = []

                    checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)

                    yield CheckpointTuple(
                        config={
                            "configurable": {
                                "thread_id": thread_id,
                                "checkpoint_ns": checkpoint_ns,
                                "checkpoint_id": checkpoint_id,
                            }
                        },
                        checkpoint={
                            **checkpoint_,
                            "channel_values": self._load_blobs(
                                thread_id,
                                checkpoint_ns,
                                checkpoint_["channel_versions"],
                            ),
                            "pending_sends": [
                                self.serde.loads_typed(s[2]) for s in sends
                            ],
                        },
                        metadata=metadata,
                        parent_config=(
                            {
                                "configurable": {
                                    "thread_id": thread_id,
                                    "checkpoint_ns": checkpoint_ns,
                                    "checkpoint_id": parent_checkpoint_id,
                                }
                            }
                            if parent_checkpoint_id
                            else None
                        ),
                        pending_writes=[
                            (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes
                        ],
                    )

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Save a checkpoint to the in-memory storage.

        This method saves a checkpoint to the in-memory storage. The checkpoint is associated
        with the provided config.

        Args:
            config: The config to associate with the checkpoint.
            checkpoint: The checkpoint to save.
            metadata: Additional metadata to save with the checkpoint.
            new_versions: New versions as of this write

        Returns:
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.
        """
        c = checkpoint.copy()
        c.pop("pending_sends")  # type: ignore[misc]
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        values: dict[str, Any] = c.pop("channel_values")  # type: ignore[misc]
        for k, v in new_versions.items():
            self.blobs[(thread_id, checkpoint_ns, k, v)] = (
                self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
            )
        self.storage[thread_id][checkpoint_ns].update(
            {
                checkpoint["id"]: (
                    self.serde.dumps_typed(c),
                    self.serde.dumps_typed(get_checkpoint_metadata(config, metadata)),
                    config["configurable"].get("checkpoint_id"),  # parent
                )
            }
        )
        return {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": checkpoint["id"],
            }
        }

    def put_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[tuple[str, Any]],
        task_id: str,
        task_path: str = "",
    ) -> None:
        """Save a list of writes to the in-memory storage.

        This method saves a list of writes to the in-memory storage. The writes are associated
        with the provided config.

        Args:
            config: The config to associate with the writes.
            writes: The writes to save.
            task_id: Identifier for the task creating the writes.
            task_path: Path of the task creating the writes.

        Returns:
            RunnableConfig: The updated config containing the saved writes' timestamp.
        """
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
        checkpoint_id = config["configurable"]["checkpoint_id"]
        outer_key = (thread_id, checkpoint_ns, checkpoint_id)
        outer_writes_ = self.writes.get(outer_key)
        for idx, (c, v) in enumerate(writes):
            inner_key = (task_id, WRITES_IDX_MAP.get(c, idx))
            if inner_key[1] >= 0 and outer_writes_ and inner_key in outer_writes_:
                continue

            self.writes[outer_key][inner_key] = (
                task_id,
                c,
                self.serde.dumps_typed(v),
                task_path,
            )

    def delete_thread(self, thread_id: str) -> None:
        """Delete all checkpoints and writes associated with a thread ID.

        Args:
            thread_id: The thread ID to delete.

        Returns:
            None
        """
        if thread_id in self.storage:
            del self.storage[thread_id]
        for k in list(self.writes.keys()):
            if k[0] == thread_id:
                del self.writes[k]
        for k in list(self.blobs.keys()):
            if k[0] == thread_id:
                del self.blobs[k]

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Asynchronous version of get_tuple.

        This method is an asynchronous wrapper around get_tuple that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config: The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        return self.get_tuple(config)

    async def alist(
        self,
        config: Optional[RunnableConfig],
        *,
        filter: Optional[dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """Asynchronous version of list.

        This method is an asynchronous wrapper around list that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config: The config to use for listing the checkpoints.

        Yields:
            AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.
        """
        for item in self.list(config, filter=filter, before=before, limit=limit):
            yield item

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        """Asynchronous version of put.

        Args:
            config: The config to associate with the checkpoint.
            checkpoint: The checkpoint to save.
            metadata: Additional metadata to save with the checkpoint.
            new_versions: New versions as of this write

        Returns:
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.
        """
        return self.put(config, checkpoint, metadata, new_versions)

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[tuple[str, Any]],
        task_id: str,
        task_path: str = "",
    ) -> None:
        """Asynchronous version of put_writes.

        This method is an asynchronous wrapper around put_writes that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config: The config to associate with the writes.
            writes: The writes to save, each as a (channel, value) pair.
            task_id: Identifier for the task creating the writes.
            task_path: Path of the task creating the writes.

        Returns:
            None
        """
        return self.put_writes(config, writes, task_id, task_path)

    async def adelete_thread(self, thread_id: str) -> None:
        """Delete all checkpoints and writes associated with a thread ID.

        Args:
            thread_id: The thread ID to delete.

        Returns:
            None
        """
        return self.delete_thread(thread_id)

    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
        if current is None:
            current_v = 0
        elif isinstance(current, int):
            current_v = current
        else:
            current_v = int(current.split(".")[0])
        next_v = current_v + 1
        next_h = random.random()
        return f"{next_v:032}.{next_h:016}"


MemorySaver = InMemorySaver  # Kept for backwards compatibility


class PersistentDict(defaultdict):
    """Persistent dictionary with an API compatible with shelve and anydbm.

    The dict is kept in memory, so the dictionary operations run as fast as
    a regular dictionary.

    Write to disk is delayed until close or sync (similar to gdbm's fast mode).

    Input file format is automatically discovered.
    Output file format is selectable between pickle, json, and csv.
    All three serialization formats are backed by fast C implementations.

    Adapted from https://code.activestate.com/recipes/576642-persistent-dict-with-multiple-standard-file-format/

    """

    def __init__(self, *args: Any, filename: str, **kwds: Any) -> None:
        self.flag = "c"  # r=readonly, c=create, or n=new
        self.mode = None  # None or an octal triple like 0644
        self.format = "pickle"  # 'csv', 'json', or 'pickle'
        self.filename = filename
        super().__init__(*args, **kwds)

    def sync(self) -> None:
        "Write dict to disk"
        if self.flag == "r":
            return
        tempname = self.filename + ".tmp"
        fileobj = open(tempname, "wb" if self.format == "pickle" else "w")
        try:
            self.dump(fileobj)
        except Exception:
            os.remove(tempname)
            raise
        finally:
            fileobj.close()
        shutil.move(tempname, self.filename)  # atomic commit
        if self.mode is not None:
            os.chmod(self.filename, self.mode)

    def close(self) -> None:
        self.sync()
        self.clear()

    def __enter__(self) -> "PersistentDict":
        return self

    def __exit__(self, *exc_info: Any) -> None:
        self.close()

    def dump(self, fileobj: Any) -> None:
        if self.format == "pickle":
            pickle.dump(dict(self), fileobj, 2)
        else:
            raise NotImplementedError("Unknown format: " + repr(self.format))

    def load(self) -> None:
        # try formats from most restrictive to least restrictive
        if self.flag == "n":
            return
        with open(self.filename, "rb" if self.format == "pickle" else "r") as fileobj:
            for loader in (pickle.load,):
                fileobj.seek(0)
                try:
                    return self.update(loader(fileobj))
                except EOFError:
                    return
                except Exception:
                    logging.error(f"Failed to load file: {fileobj.name}")
                    raise
            raise ValueError("File not in a supported format")

Error Message and Stack Trace (if applicable)

This "from langchain_core.runnables import RunnableConfig" does not support latest version of langgraph which give import error along with even installed langgraph-checkpoint lib with or without same error.

Description

This "from langchain_core.runnables import RunnableConfig" does not support latest version of langgraph which give import error along with even installed langgraph-checkpoint lib with or without same error.

System Info

This "from langchain_core.runnables import RunnableConfig" does not support latest version of langgraph which give import error along with even installed langgraph-checkpoint lib with or without same error.

Originally created by @shubhamgajbhiye1994 on GitHub (Jul 11, 2025). ### Checked other resources - [x] This is a bug, not a usage question. For questions, please use GitHub Discussions. - [x] I added a clear and detailed title that summarizes the issue. - [x] I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example). - [x] I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue. ### Example Code ```python source path in langgraph checkpoint:langgraph/checkpoint/memory/__init__.py import logging import os import pickle import random import shutil from collections import defaultdict from collections.abc import AsyncIterator, Iterator, Sequence from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack from types import TracebackType from typing import Any, Optional, Union from langchain_core.runnables import RunnableConfig from langgraph.checkpoint.base import ( WRITES_IDX_MAP, BaseCheckpointSaver, ChannelVersions, Checkpoint, CheckpointMetadata, CheckpointTuple, SerializerProtocol, get_checkpoint_id, get_checkpoint_metadata, ) from langgraph.checkpoint.serde.types import TASKS, ChannelProtocol logger = logging.getLogger(__name__) class InMemorySaver( BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager ): """An in-memory checkpoint saver. This checkpoint saver stores checkpoints in memory using a defaultdict. Note: Only use `InMemorySaver` for debugging or testing purposes. For production use cases we recommend installing [langgraph-checkpoint-postgres](https://pypi.org/project/langgraph-checkpoint-postgres/) and using `PostgresSaver` / `AsyncPostgresSaver`. If you are using the LangGraph Platform, no checkpointer needs to be specified. The correct managed checkpointer will be used automatically. Args: serde: The serializer to use for serializing and deserializing checkpoints. Defaults to None. Examples: import asyncio from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph builder = StateGraph(int) builder.add_node("add_one", lambda x: x + 1) builder.set_entry_point("add_one") builder.set_finish_point("add_one") memory = InMemorySaver() graph = builder.compile(checkpointer=memory) coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}}) asyncio.run(coro) # Output: 2 """ # thread ID -> checkpoint NS -> checkpoint ID -> checkpoint mapping storage: defaultdict[ str, dict[ str, dict[str, tuple[tuple[str, bytes], tuple[str, bytes], Optional[str]]] ], ] # (thread ID, checkpoint NS, checkpoint ID) -> (task ID, write idx) writes: defaultdict[ tuple[str, str, str], dict[tuple[str, int], tuple[str, str, tuple[str, bytes], str]], ] blobs: dict[ tuple[ str, str, str, Union[str, int, float] ], # thread id, checkpoint ns, channel, version tuple[str, bytes], ] def __init__( self, *, serde: Optional[SerializerProtocol] = None, factory: type[defaultdict] = defaultdict, ) -> None: super().__init__(serde=serde) self.storage = factory(lambda: defaultdict(dict)) self.writes = factory(dict) self.blobs = factory() self.stack = ExitStack() if factory is not defaultdict: self.stack.enter_context(self.storage) # type: ignore[arg-type] self.stack.enter_context(self.writes) # type: ignore[arg-type] self.stack.enter_context(self.blobs) # type: ignore[arg-type] def __enter__(self) -> "InMemorySaver": return self.stack.__enter__() def __exit__( self, exc_type: Optional[type[BaseException]], exc_value: Optional[BaseException], traceback: Optional[TracebackType], ) -> Optional[bool]: return self.stack.__exit__(exc_type, exc_value, traceback) async def __aenter__(self) -> "InMemorySaver": return self.stack.__enter__() async def __aexit__( self, __exc_type: Optional[type[BaseException]], __exc_value: Optional[BaseException], __traceback: Optional[TracebackType], ) -> Optional[bool]: return self.stack.__exit__(__exc_type, __exc_value, __traceback) def _load_blobs( self, thread_id: str, checkpoint_ns: str, versions: ChannelVersions ) -> dict[str, Any]: channel_values: dict[str, Any] = {} for k, v in versions.items(): kk = (thread_id, checkpoint_ns, k, v) if kk in self.blobs: vv = self.blobs[kk] if vv[0] != "empty": channel_values[k] = self.serde.loads_typed(vv) return channel_values def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]: """Get a checkpoint tuple from the in-memory storage. This method retrieves a checkpoint tuple from the in-memory storage based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved. Args: config: The config to use for retrieving the checkpoint. Returns: Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found. """ thread_id: str = config["configurable"]["thread_id"] checkpoint_ns: str = config["configurable"].get("checkpoint_ns", "") if checkpoint_id := get_checkpoint_id(config): if saved := self.storage[thread_id][checkpoint_ns].get(checkpoint_id): checkpoint, metadata, parent_checkpoint_id = saved writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values() if parent_checkpoint_id: sends = sorted( ( (*w, k[1]) for k, w in self.writes[ (thread_id, checkpoint_ns, parent_checkpoint_id) ].items() if w[1] == TASKS ), key=lambda w: (w[3], w[0], w[4]), ) else: sends = [] checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint) return CheckpointTuple( config=config, checkpoint={ **checkpoint_, "channel_values": self._load_blobs( thread_id, checkpoint_ns, checkpoint_["channel_versions"] ), "pending_sends": [self.serde.loads_typed(s[2]) for s in sends], }, metadata=self.serde.loads_typed(metadata), pending_writes=[ (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes ], parent_config=( { "configurable": { "thread_id": thread_id, "checkpoint_ns": checkpoint_ns, "checkpoint_id": parent_checkpoint_id, } } if parent_checkpoint_id else None ), ) else: if checkpoints := self.storage[thread_id][checkpoint_ns]: checkpoint_id = max(checkpoints.keys()) checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id] writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values() if parent_checkpoint_id: sends = sorted( ( (*w, k[1]) for k, w in self.writes[ (thread_id, checkpoint_ns, parent_checkpoint_id) ].items() if w[1] == TASKS ), key=lambda w: (w[3], w[0], w[4]), ) else: sends = [] checkpoint_ = self.serde.loads_typed(checkpoint) return CheckpointTuple( config={ "configurable": { "thread_id": thread_id, "checkpoint_ns": checkpoint_ns, "checkpoint_id": checkpoint_id, } }, checkpoint={ **checkpoint_, "channel_values": self._load_blobs( thread_id, checkpoint_ns, checkpoint_["channel_versions"] ), "pending_sends": [self.serde.loads_typed(s[2]) for s in sends], }, metadata=self.serde.loads_typed(metadata), pending_writes=[ (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes ], parent_config=( { "configurable": { "thread_id": thread_id, "checkpoint_ns": checkpoint_ns, "checkpoint_id": parent_checkpoint_id, } } if parent_checkpoint_id else None ), ) def list( self, config: Optional[RunnableConfig], *, filter: Optional[dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None, ) -> Iterator[CheckpointTuple]: """List checkpoints from the in-memory storage. This method retrieves a list of checkpoint tuples from the in-memory storage based on the provided criteria. Args: config: Base configuration for filtering checkpoints. filter: Additional filtering criteria for metadata. before: List checkpoints created before this configuration. limit: Maximum number of checkpoints to return. Yields: Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples. """ thread_ids = (config["configurable"]["thread_id"],) if config else self.storage config_checkpoint_ns = ( config["configurable"].get("checkpoint_ns") if config else None ) config_checkpoint_id = get_checkpoint_id(config) if config else None for thread_id in thread_ids: for checkpoint_ns in self.storage[thread_id].keys(): if ( config_checkpoint_ns is not None and checkpoint_ns != config_checkpoint_ns ): continue for checkpoint_id, ( checkpoint, metadata_b, parent_checkpoint_id, ) in sorted( self.storage[thread_id][checkpoint_ns].items(), key=lambda x: x[0], reverse=True, ): # filter by checkpoint ID from config if config_checkpoint_id and checkpoint_id != config_checkpoint_id: continue # filter by checkpoint ID from `before` config if ( before and (before_checkpoint_id := get_checkpoint_id(before)) and checkpoint_id >= before_checkpoint_id ): continue # filter by metadata metadata = self.serde.loads_typed(metadata_b) if filter and not all( query_value == metadata.get(query_key) for query_key, query_value in filter.items() ): continue # limit search results if limit is not None and limit <= 0: break elif limit is not None: limit -= 1 writes = self.writes[ (thread_id, checkpoint_ns, checkpoint_id) ].values() if parent_checkpoint_id: sends = sorted( ( (*w, k[1]) for k, w in self.writes[ (thread_id, checkpoint_ns, parent_checkpoint_id) ].items() if w[1] == TASKS ), key=lambda w: (w[3], w[0], w[4]), ) else: sends = [] checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint) yield CheckpointTuple( config={ "configurable": { "thread_id": thread_id, "checkpoint_ns": checkpoint_ns, "checkpoint_id": checkpoint_id, } }, checkpoint={ **checkpoint_, "channel_values": self._load_blobs( thread_id, checkpoint_ns, checkpoint_["channel_versions"], ), "pending_sends": [ self.serde.loads_typed(s[2]) for s in sends ], }, metadata=metadata, parent_config=( { "configurable": { "thread_id": thread_id, "checkpoint_ns": checkpoint_ns, "checkpoint_id": parent_checkpoint_id, } } if parent_checkpoint_id else None ), pending_writes=[ (id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes ], ) def put( self, config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions, ) -> RunnableConfig: """Save a checkpoint to the in-memory storage. This method saves a checkpoint to the in-memory storage. The checkpoint is associated with the provided config. Args: config: The config to associate with the checkpoint. checkpoint: The checkpoint to save. metadata: Additional metadata to save with the checkpoint. new_versions: New versions as of this write Returns: RunnableConfig: The updated config containing the saved checkpoint's timestamp. """ c = checkpoint.copy() c.pop("pending_sends") # type: ignore[misc] thread_id = config["configurable"]["thread_id"] checkpoint_ns = config["configurable"]["checkpoint_ns"] values: dict[str, Any] = c.pop("channel_values") # type: ignore[misc] for k, v in new_versions.items(): self.blobs[(thread_id, checkpoint_ns, k, v)] = ( self.serde.dumps_typed(values[k]) if k in values else ("empty", b"") ) self.storage[thread_id][checkpoint_ns].update( { checkpoint["id"]: ( self.serde.dumps_typed(c), self.serde.dumps_typed(get_checkpoint_metadata(config, metadata)), config["configurable"].get("checkpoint_id"), # parent ) } ) return { "configurable": { "thread_id": thread_id, "checkpoint_ns": checkpoint_ns, "checkpoint_id": checkpoint["id"], } } def put_writes( self, config: RunnableConfig, writes: Sequence[tuple[str, Any]], task_id: str, task_path: str = "", ) -> None: """Save a list of writes to the in-memory storage. This method saves a list of writes to the in-memory storage. The writes are associated with the provided config. Args: config: The config to associate with the writes. writes: The writes to save. task_id: Identifier for the task creating the writes. task_path: Path of the task creating the writes. Returns: RunnableConfig: The updated config containing the saved writes' timestamp. """ thread_id = config["configurable"]["thread_id"] checkpoint_ns = config["configurable"].get("checkpoint_ns", "") checkpoint_id = config["configurable"]["checkpoint_id"] outer_key = (thread_id, checkpoint_ns, checkpoint_id) outer_writes_ = self.writes.get(outer_key) for idx, (c, v) in enumerate(writes): inner_key = (task_id, WRITES_IDX_MAP.get(c, idx)) if inner_key[1] >= 0 and outer_writes_ and inner_key in outer_writes_: continue self.writes[outer_key][inner_key] = ( task_id, c, self.serde.dumps_typed(v), task_path, ) def delete_thread(self, thread_id: str) -> None: """Delete all checkpoints and writes associated with a thread ID. Args: thread_id: The thread ID to delete. Returns: None """ if thread_id in self.storage: del self.storage[thread_id] for k in list(self.writes.keys()): if k[0] == thread_id: del self.writes[k] for k in list(self.blobs.keys()): if k[0] == thread_id: del self.blobs[k] async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]: """Asynchronous version of get_tuple. This method is an asynchronous wrapper around get_tuple that runs the synchronous method in a separate thread using asyncio. Args: config: The config to use for retrieving the checkpoint. Returns: Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found. """ return self.get_tuple(config) async def alist( self, config: Optional[RunnableConfig], *, filter: Optional[dict[str, Any]] = None, before: Optional[RunnableConfig] = None, limit: Optional[int] = None, ) -> AsyncIterator[CheckpointTuple]: """Asynchronous version of list. This method is an asynchronous wrapper around list that runs the synchronous method in a separate thread using asyncio. Args: config: The config to use for listing the checkpoints. Yields: AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples. """ for item in self.list(config, filter=filter, before=before, limit=limit): yield item async def aput( self, config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: ChannelVersions, ) -> RunnableConfig: """Asynchronous version of put. Args: config: The config to associate with the checkpoint. checkpoint: The checkpoint to save. metadata: Additional metadata to save with the checkpoint. new_versions: New versions as of this write Returns: RunnableConfig: The updated config containing the saved checkpoint's timestamp. """ return self.put(config, checkpoint, metadata, new_versions) async def aput_writes( self, config: RunnableConfig, writes: Sequence[tuple[str, Any]], task_id: str, task_path: str = "", ) -> None: """Asynchronous version of put_writes. This method is an asynchronous wrapper around put_writes that runs the synchronous method in a separate thread using asyncio. Args: config: The config to associate with the writes. writes: The writes to save, each as a (channel, value) pair. task_id: Identifier for the task creating the writes. task_path: Path of the task creating the writes. Returns: None """ return self.put_writes(config, writes, task_id, task_path) async def adelete_thread(self, thread_id: str) -> None: """Delete all checkpoints and writes associated with a thread ID. Args: thread_id: The thread ID to delete. Returns: None """ return self.delete_thread(thread_id) def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str: if current is None: current_v = 0 elif isinstance(current, int): current_v = current else: current_v = int(current.split(".")[0]) next_v = current_v + 1 next_h = random.random() return f"{next_v:032}.{next_h:016}" MemorySaver = InMemorySaver # Kept for backwards compatibility class PersistentDict(defaultdict): """Persistent dictionary with an API compatible with shelve and anydbm. The dict is kept in memory, so the dictionary operations run as fast as a regular dictionary. Write to disk is delayed until close or sync (similar to gdbm's fast mode). Input file format is automatically discovered. Output file format is selectable between pickle, json, and csv. All three serialization formats are backed by fast C implementations. Adapted from https://code.activestate.com/recipes/576642-persistent-dict-with-multiple-standard-file-format/ """ def __init__(self, *args: Any, filename: str, **kwds: Any) -> None: self.flag = "c" # r=readonly, c=create, or n=new self.mode = None # None or an octal triple like 0644 self.format = "pickle" # 'csv', 'json', or 'pickle' self.filename = filename super().__init__(*args, **kwds) def sync(self) -> None: "Write dict to disk" if self.flag == "r": return tempname = self.filename + ".tmp" fileobj = open(tempname, "wb" if self.format == "pickle" else "w") try: self.dump(fileobj) except Exception: os.remove(tempname) raise finally: fileobj.close() shutil.move(tempname, self.filename) # atomic commit if self.mode is not None: os.chmod(self.filename, self.mode) def close(self) -> None: self.sync() self.clear() def __enter__(self) -> "PersistentDict": return self def __exit__(self, *exc_info: Any) -> None: self.close() def dump(self, fileobj: Any) -> None: if self.format == "pickle": pickle.dump(dict(self), fileobj, 2) else: raise NotImplementedError("Unknown format: " + repr(self.format)) def load(self) -> None: # try formats from most restrictive to least restrictive if self.flag == "n": return with open(self.filename, "rb" if self.format == "pickle" else "r") as fileobj: for loader in (pickle.load,): fileobj.seek(0) try: return self.update(loader(fileobj)) except EOFError: return except Exception: logging.error(f"Failed to load file: {fileobj.name}") raise raise ValueError("File not in a supported format") ``` ### Error Message and Stack Trace (if applicable) ```shell This "from langchain_core.runnables import RunnableConfig" does not support latest version of langgraph which give import error along with even installed langgraph-checkpoint lib with or without same error. ``` ### Description This "from langchain_core.runnables import RunnableConfig" does not support latest version of langgraph which give import error along with even installed langgraph-checkpoint lib with or without same error. ### System Info This "from langchain_core.runnables import RunnableConfig" does not support latest version of langgraph which give import error along with even installed langgraph-checkpoint lib with or without same error.
yindo added the invalid label 2026-02-20 17:41:47 -05:00
yindo closed this issue 2026-02-20 17:41:47 -05:00
yindo commented 2026-02-20 17:41:47 -05:00
Author
Owner
Copy Link
Copy Source

@eyurtsev commented on GitHub (Jul 14, 2025):

@shubhamgajbhiye1994 please provide a full stack trace. The issue description that you have makes it impossible to provide help.

Before doing any of the below. First:

  1. Include a full stack trace
  2. Provide output from either pip freeze or python -m langchain_core.sys_info

Here's an output from our helpful LLM that's trying to understand:

Issue Explanation

The issue is about an import error when trying to use from langchain_core.runnables import RunnableConfig in the latest version of LangGraph. The user reports that this import is causing compatibility issues with the latest LangGraph version, even when the langgraph-checkpoint library is installed.

Looking at the documentation, I can see that RunnableConfig is indeed a valid import from langchain_core.runnables. According to the LangChain documentation, RunnableConfig is a fundamental part of the Runnable interface and should be imported from langchain_core.runnables.config.

Minimal Reproducible Example

"""
Minimal Reproducible Example for LangGraph RunnableConfig Import Issue

This script demonstrates the import error that occurs when trying to use
RunnableConfig from langchain_core.runnables in the latest version of LangGraph.

Prerequisites:
- Python 3.9+
- Latest versions of langchain-core, langgraph, and langgraph-checkpoint
"""

import sys
import pkg_resources

# Step 1: Print environment information
print("=== Environment Information ===")
print(f"Python Version: {sys.version}")

# Check installed package versions
packages = ['langchain-core', 'langgraph', 'langgraph-checkpoint']
for package in packages:
    try:
        version = pkg_resources.get_distribution(package).version
        print(f"{package}: {version}")
    except pkg_resources.DistributionNotFound:
        print(f"{package}: NOT INSTALLED")

print("\n=== Attempting Import ===")

# Step 2: Try the problematic import
try:
    # This is the import that's causing issues according to the issue
    from langchain_core.runnables import RunnableConfig
    print("✓ Successfully imported RunnableConfig from langchain_core.runnables")
    print(f"  RunnableConfig type: {type(RunnableConfig)}")
except ImportError as e:
    print(f"✗ ImportError: {e}")
    
    # Try alternative import paths
    print("\n=== Trying Alternative Imports ===")
    
    # Alternative 1: Direct import from config module
    try:
        from langchain_core.runnables.config import RunnableConfig
        print("✓ Successfully imported from langchain_core.runnables.config")
    except ImportError as e:
        print(f"✗ Failed to import from langchain_core.runnables.config: {e}")
    
    # Alternative 2: Import the entire module
    try:
        import langchain_core.runnables
        print("✓ Successfully imported langchain_core.runnables module")
        # Check if RunnableConfig is available
        if hasattr(langchain_core.runnables, 'RunnableConfig'):
            print("  RunnableConfig is available as an attribute")
        else:
            print("  RunnableConfig is NOT available as an attribute")
    except ImportError as e:
        print(f"✗ Failed to import langchain_core.runnables module: {e}")

# Step 3: Test usage in LangGraph context (as shown in the issue)
print("\n=== Testing in LangGraph Context ===")

try:
    # This mimics the usage in the langgraph/checkpoint/memory/__init__.py file
    from langchain_core.runnables import RunnableConfig
    from typing import Optional, Any
    
    # Simple test function that uses RunnableConfig
    def test_runnable_config(config: RunnableConfig) -> Optional[Any]:
        """Test function that accepts RunnableConfig as parameter"""
        thread_id = config.get("configurable", {}).get("thread_id")
        return thread_id
    
    # Create a test config
    test_config = {
        "configurable": {
            "thread_id": "test-thread-123",
            "checkpoint_ns": "test-namespace"
        }
    }
    
    # Test the function
    result = test_runnable_config(test_config)
    print(f"✓ Function executed successfully with thread_id: {result}")
    
except Exception as e:
    print(f"✗ Error in LangGraph context: {type(e).__name__}: {e}")

# Step 4: Test InMemorySaver import (from the issue's code)
print("\n=== Testing InMemorySaver Import ===")

try:
    # This is what would fail based on the issue description
    from langgraph.checkpoint.memory import InMemorySaver
    print("✓ Successfully imported InMemorySaver")
    
    # Try to create an instance
    saver = InMemorySaver()
    print("✓ Successfully created InMemorySaver instance")
    
except ImportError as e:
    print(f"✗ ImportError when importing InMemorySaver: {e}")
except Exception as e:
    print(f"✗ Error creating InMemorySaver: {type(e).__name__}: {e}")

print("\n=== Diagnosis Complete ===")

Expected vs Actual Behavior

Expected Behavior:

  • The import from langchain_core.runnables import RunnableConfig should work without errors
  • The InMemorySaver class from langgraph.checkpoint.memory should be importable and usable
  • All components should work together seamlessly

Actual Behavior (based on the issue):

  • Import error occurs when trying to import RunnableConfig from langchain_core.runnables
  • This prevents the use of LangGraph checkpoint functionality
  • The error persists even with langgraph-checkpoint installed

Additional Notes and Potential Solutions

  1. Version Compatibility: The issue might be related to version mismatches between langchain-core, langgraph, and langgraph-checkpoint. Ensure all packages are updated to compatible versions.

  2. Import Path Changes: It's possible that in newer versions, the import path has changed. The correct import might be:

    from langchain_core.runnables.config import RunnableConfig

  3. Installation Order: Try reinstalling the packages in this order:

    pip uninstall langchain-core langgraph langgraph-checkpoint -y
pip install langchain-core
pip install langgraph
pip install langgraph-checkpoint

  4. TypedDict Issue: Based on the documentation, RunnableConfig might be a TypedDict, which could cause import issues in certain Python versions or configurations.

This MRE should help identify exactly where the import issue occurs and provide diagnostic information to help resolve the problem.

@eyurtsev commented on GitHub (Jul 14, 2025): @shubhamgajbhiye1994 please provide a full stack trace. The issue description that you have makes it impossible to provide help. Before doing any of the below. First: 1. Include a full stack trace 2. Provide output from either pip freeze or `python -m langchain_core.sys_info` ---- Here's an output from our helpful LLM that's trying to understand: ## Issue Explanation The issue is about an import error when trying to use `from langchain_core.runnables import RunnableConfig` in the latest version of LangGraph. The user reports that this import is causing compatibility issues with the latest LangGraph version, even when the `langgraph-checkpoint` library is installed. Looking at the documentation, I can see that `RunnableConfig` is indeed a valid import from `langchain_core.runnables`. According to the LangChain documentation, `RunnableConfig` is a fundamental part of the Runnable interface and should be imported from `langchain_core.runnables.config`. ## Minimal Reproducible Example ```python """ Minimal Reproducible Example for LangGraph RunnableConfig Import Issue This script demonstrates the import error that occurs when trying to use RunnableConfig from langchain_core.runnables in the latest version of LangGraph. Prerequisites: - Python 3.9+ - Latest versions of langchain-core, langgraph, and langgraph-checkpoint """ import sys import pkg_resources # Step 1: Print environment information print("=== Environment Information ===") print(f"Python Version: {sys.version}") # Check installed package versions packages = ['langchain-core', 'langgraph', 'langgraph-checkpoint'] for package in packages: try: version = pkg_resources.get_distribution(package).version print(f"{package}: {version}") except pkg_resources.DistributionNotFound: print(f"{package}: NOT INSTALLED") print("\n=== Attempting Import ===") # Step 2: Try the problematic import try: # This is the import that's causing issues according to the issue from langchain_core.runnables import RunnableConfig print("✓ Successfully imported RunnableConfig from langchain_core.runnables") print(f" RunnableConfig type: {type(RunnableConfig)}") except ImportError as e: print(f"✗ ImportError: {e}") # Try alternative import paths print("\n=== Trying Alternative Imports ===") # Alternative 1: Direct import from config module try: from langchain_core.runnables.config import RunnableConfig print("✓ Successfully imported from langchain_core.runnables.config") except ImportError as e: print(f"✗ Failed to import from langchain_core.runnables.config: {e}") # Alternative 2: Import the entire module try: import langchain_core.runnables print("✓ Successfully imported langchain_core.runnables module") # Check if RunnableConfig is available if hasattr(langchain_core.runnables, 'RunnableConfig'): print(" RunnableConfig is available as an attribute") else: print(" RunnableConfig is NOT available as an attribute") except ImportError as e: print(f"✗ Failed to import langchain_core.runnables module: {e}") # Step 3: Test usage in LangGraph context (as shown in the issue) print("\n=== Testing in LangGraph Context ===") try: # This mimics the usage in the langgraph/checkpoint/memory/__init__.py file from langchain_core.runnables import RunnableConfig from typing import Optional, Any # Simple test function that uses RunnableConfig def test_runnable_config(config: RunnableConfig) -> Optional[Any]: """Test function that accepts RunnableConfig as parameter""" thread_id = config.get("configurable", {}).get("thread_id") return thread_id # Create a test config test_config = { "configurable": { "thread_id": "test-thread-123", "checkpoint_ns": "test-namespace" } } # Test the function result = test_runnable_config(test_config) print(f"✓ Function executed successfully with thread_id: {result}") except Exception as e: print(f"✗ Error in LangGraph context: {type(e).__name__}: {e}") # Step 4: Test InMemorySaver import (from the issue's code) print("\n=== Testing InMemorySaver Import ===") try: # This is what would fail based on the issue description from langgraph.checkpoint.memory import InMemorySaver print("✓ Successfully imported InMemorySaver") # Try to create an instance saver = InMemorySaver() print("✓ Successfully created InMemorySaver instance") except ImportError as e: print(f"✗ ImportError when importing InMemorySaver: {e}") except Exception as e: print(f"✗ Error creating InMemorySaver: {type(e).__name__}: {e}") print("\n=== Diagnosis Complete ===") ``` ## Expected vs Actual Behavior ### Expected Behavior: - The import `from langchain_core.runnables import RunnableConfig` should work without errors - The `InMemorySaver` class from `langgraph.checkpoint.memory` should be importable and usable - All components should work together seamlessly ### Actual Behavior (based on the issue): - Import error occurs when trying to import `RunnableConfig` from `langchain_core.runnables` - This prevents the use of LangGraph checkpoint functionality - The error persists even with `langgraph-checkpoint` installed ## Additional Notes and Potential Solutions 1. **Version Compatibility**: The issue might be related to version mismatches between `langchain-core`, `langgraph`, and `langgraph-checkpoint`. Ensure all packages are updated to compatible versions. 2. **Import Path Changes**: It's possible that in newer versions, the import path has changed. The correct import might be: ```python from langchain_core.runnables.config import RunnableConfig ``` 3. **Installation Order**: Try reinstalling the packages in this order: ```bash pip uninstall langchain-core langgraph langgraph-checkpoint -y pip install langchain-core pip install langgraph pip install langgraph-checkpoint ``` 4. **TypedDict Issue**: Based on the documentation, `RunnableConfig` might be a TypedDict, which could cause import issues in certain Python versions or configurations. This MRE should help identify exactly where the import issue occurs and provide diagnostic information to help resolve the problem.
yindo commented 2026-02-20 17:41:47 -05:00
Author
Owner
Copy Link
Copy Source

@shubhamgajbhiye1994 commented on GitHub (Jul 15, 2025):

Actaully I am facing issue with latest version of langgraph , But when I freeze my langgraph version and other packages e.g. langgraph version 0.4.8.3 , after freezing and deployement its working properly.

@shubhamgajbhiye1994 commented on GitHub (Jul 15, 2025): Actaully I am facing issue with latest version of langgraph , But when I freeze my langgraph version and other packages e.g. langgraph version 0.4.8.3 , after freezing and deployement its working properly.
yindo commented 2026-02-20 17:41:47 -05:00
Author
Owner
Copy Link
Copy Source

@eyurtsev commented on GitHub (Jul 16, 2025):

@shubhamgajbhiye1994 please provide a full stack trace. The issue description that you have makes it impossible to provide help.

Before doing any of the below. First:

Include a full stack trace
Provide output from either pip freeze or python -m langchain_core.sys_info

@eyurtsev commented on GitHub (Jul 16, 2025): @shubhamgajbhiye1994 please provide a full stack trace. The issue description that you have makes it impossible to provide help. Before doing any of the below. First: Include a full stack trace Provide output from either pip freeze or python -m langchain_core.sys_info
yindo commented 2026-02-20 17:41:47 -05:00
Author
Owner
Copy Link
Copy Source

@sydney-runkle commented on GitHub (Jul 20, 2025):

Going to close this given we don't have an MRE. If you're still experiencing an issue, please do as Eugene has suggested above :)

@sydney-runkle commented on GitHub (Jul 20, 2025): Going to close this given we don't have an MRE. If you're still experiencing an issue, please do as Eugene has suggested above :)
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
agents
awaiting author revision
bug
change
chat models
checkpointer
ci
code quality
code quality
dependencies
documentation
drawing
duplicate
enhancement
good first issue
help wanted
invalid
maintainer
meta
missing information
not maintained by langchain
open-swe-auto
open-swe-max
open-swe-max-auto
pending
performance
pull-request
Mirrored from GitHub Pull Request
 question
requires design
sdk
server
stale
studio
tests
type checking
usage question
v0.5
v1.1
No Label invalid
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#790
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?