fix(store): add value serialization for LangChain message objects (#128)

RedisStore was passing values directly to redisvl.SearchIndex.load() which
uses standard json.dumps() internally. This caused failures when storing
values containing LangChain message objects (HumanMessage, AIMessage, etc.).

Changes:
- Add _serialize_value() method to BaseRedisStore that uses JsonPlusRedisSerializer
- Add _deserialize_value() method for proper revival of complex objects
- Update _row_to_item() and _row_to_search_item() to accept deserialize function
- Update all call sites in RedisStore and AsyncRedisStore

The serialization is smart: simple JSON-serializable values are stored as-is
for backward compatibility, while complex objects get the serde wrapper.

Closes #128

Author: bsbodden

langgraph/store/redis/__init__.py

@@ -255,7 +255,9 @@ def _batch_get_ops(
             for idx, key in items:
                 if key in key_to_row:
                     data, doc_id = key_to_row[key]
-                    results[idx] = _row_to_item(namespace, data)
+                    results[idx] = _row_to_item(
+                        namespace, data, deserialize_fn=self._deserialize_value
+                    )
 
                     # Find the corresponding operation by looking it up in the operation list
                     # This is needed because idx is the index in the overall operation list
@@ -585,6 +587,7 @@ def _batch_search_ops(
                                 _decode_ns(store_doc["prefix"]),
                                 store_doc,
                                 score=score,
+                                deserialize_fn=self._deserialize_value,
                             )
                         )
 
@@ -651,7 +654,13 @@ def _batch_search_ops(
                         )
                         refresh_keys.append(vector_key)
 
-                    items.append(_row_to_search_item(_decode_ns(data["prefix"]), data))
+                    items.append(
+                        _row_to_search_item(
+                            _decode_ns(data["prefix"]),
+                            data,
+                            deserialize_fn=self._deserialize_value,
+                        )
+                    )
 
                 # Refresh TTL if requested
                 if op.refresh_ttl and refresh_keys and self.ttl_config:

langgraph/store/redis/aio.py

@@ -470,7 +470,9 @@ async def _batch_get_ops(
             for idx, key in items:
                 if key in key_to_row:
                     data, doc_id = key_to_row[key]
-                    results[idx] = _row_to_item(namespace, data)
+                    results[idx] = _row_to_item(
+                        namespace, data, deserialize_fn=self._deserialize_value
+                    )
 
                     # Find the corresponding operation by looking it up in the operation list
                     # This is needed because idx is the index in the overall operation list
@@ -870,6 +872,7 @@ async def _batch_search_ops(
                                 _decode_ns(store_doc["prefix"]),
                                 store_doc,
                                 score=score,
+                                deserialize_fn=self._deserialize_value,
                             )
                         )
 
@@ -937,7 +940,13 @@ async def _batch_search_ops(
                         )
                         refresh_keys.append(vector_key)
 
-                    items.append(_row_to_search_item(_decode_ns(data["prefix"]), data))
+                    items.append(
+                        _row_to_search_item(
+                            _decode_ns(data["prefix"]),
+                            data,
+                            deserialize_fn=self._deserialize_value,
+                        )
+                    )
 
                 # Refresh TTL if requested
                 if op.refresh_ttl and refresh_keys and self.ttl_config:

langgraph/store/redis/base.py

@@ -9,6 +9,7 @@
 from datetime import datetime, timedelta, timezone
 from typing import (
     Any,
+    Callable,
     Dict,
     Generic,
     Iterable,
@@ -40,6 +41,8 @@
 from redisvl.query.filter import Tag, Text
 from redisvl.utils.token_escaper import TokenEscaper
 
+from langgraph.checkpoint.redis.jsonplus_redis import JsonPlusRedisSerializer
+
 from .token_unescaper import TokenUnescaper
 from .types import IndexType, RedisClientType
 
@@ -124,6 +127,9 @@ class BaseRedisStore(Generic[RedisClientType, IndexType]):
     supports_ttl: bool = True
     ttl_config: Optional[TTLConfig] = None
 
+    # Serializer for handling complex objects like LangChain messages
+    _serde: JsonPlusRedisSerializer
+
     def _apply_ttl_to_keys(
         self,
         main_key: str,
@@ -223,6 +229,8 @@ def __init__(
         self._redis = conn
         # Store cluster_mode; None means auto-detect in RedisStore or AsyncRedisStore
         self.cluster_mode = cluster_mode
+        # Initialize the serializer for handling complex objects like LangChain messages
+        self._serde = JsonPlusRedisSerializer()
 
         # Store custom prefixes
         self.store_prefix = store_prefix
@@ -357,6 +365,84 @@ async def aset_client_info(self) -> None:
                 # Silently fail if even echo doesn't work
                 pass
 
+    def _serialize_value(self, value: Any) -> Any:
+        """Serialize a value for storage in Redis.
+
+        This method handles complex objects like LangChain messages by
+        serializing them to a JSON-compatible format.
+
+        The method is smart about serialization:
+        - If the value is a simple JSON-serializable dict/list, it's stored as-is
+        - If the value contains complex objects (HumanMessage, etc.), it uses
+          the serde wrapper format
+
+        Args:
+            value: The value to serialize (can contain HumanMessage, AIMessage, etc.)
+
+        Returns:
+            A JSON-serializable representation of the value
+        """
+        import json
+
+        if value is None:
+            return None
+
+        # First, try standard JSON serialization to check if it's needed
+        try:
+            json.dumps(value)
+            # Value is already JSON-serializable, return as-is for backward
+            # compatibility and to preserve filter functionality
+            return value
+        except (TypeError, ValueError):
+            # Value contains non-JSON-serializable objects, use serde wrapper
+            pass
+
+        # Use the serializer to handle complex objects
+        type_str, data_bytes = self._serde.dumps_typed(value)
+        # Store the serialized data with type info for proper deserialization
+        return {
+            "__serde_type__": type_str,
+            "__serde_data__": (
+                data_bytes.decode("utf-8") if type_str == "json" else data_bytes.hex()
+            ),
+        }
+
+    def _deserialize_value(self, value: Any) -> Any:
+        """Deserialize a value from Redis storage.
+
+        This method handles both new serialized format and legacy plain values
+        for backward compatibility.
+
+        Args:
+            value: The value from Redis (may be serialized or plain)
+
+        Returns:
+            The deserialized value with proper Python objects (HumanMessage, etc.)
+        """
+        if value is None:
+            return None
+
+        # Check if this is a serialized value (new format)
+        if (
+            isinstance(value, dict)
+            and "__serde_type__" in value
+            and "__serde_data__" in value
+        ):
+            type_str = value["__serde_type__"]
+            data_str = value["__serde_data__"]
+
+            # Convert back to bytes
+            if type_str == "json":
+                data_bytes = data_str.encode("utf-8")
+            else:
+                data_bytes = bytes.fromhex(data_str)
+
+            return self._serde.loads_typed((type_str, data_bytes))
+
+        # Legacy format: value is stored as-is (plain JSON-serializable data)
+        # Return as-is for backward compatibility
+        return value
+
     def _get_batch_GET_ops_queries(
         self,
         get_ops: Sequence[tuple[int, GetOp]],
@@ -433,7 +519,7 @@ def _prepare_batch_PUT_queries(
                 doc = RedisDocument(
                     prefix=_namespace_to_text(op.namespace),
                     key=op.key,
-                    value=op.value,
+                    value=self._serialize_value(op.value),
                     created_at=now,
                     updated_at=now,
                     ttl_minutes=ttl_minutes,
@@ -568,10 +654,27 @@ def _decode_ns(ns: str) -> tuple[str, ...]:
     return tuple(_token_unescaper.unescape(ns).split("."))
 
 
-def _row_to_item(namespace: tuple[str, ...], row: dict[str, Any]) -> Item:
-    """Convert a row from Redis to an Item."""
+def _row_to_item(
+    namespace: tuple[str, ...],
+    row: dict[str, Any],
+    deserialize_fn: Optional[Callable[[Any], Any]] = None,
+) -> Item:
+    """Convert a row from Redis to an Item.
+
+    Args:
+        namespace: The namespace tuple for this item
+        row: The raw row data from Redis
+        deserialize_fn: Optional function to deserialize the value (handles
+            LangChain messages, etc.)
+
+    Returns:
+        An Item with properly deserialized value
+    """
+    value = row["value"]
+    if deserialize_fn is not None:
+        value = deserialize_fn(value)
     return Item(
-        value=row["value"],
+        value=value,
         key=row["key"],
         namespace=namespace,
         created_at=datetime.fromtimestamp(row["created_at"] / 1_000_000, timezone.utc),
@@ -583,10 +686,25 @@ def _row_to_search_item(
     namespace: tuple[str, ...],
     row: dict[str, Any],
     score: Optional[float] = None,
+    deserialize_fn: Optional[Callable[[Any], Any]] = None,
 ) -> SearchItem:
-    """Convert a row from Redis to a SearchItem."""
+    """Convert a row from Redis to a SearchItem.
+
+    Args:
+        namespace: The namespace tuple for this item
+        row: The raw row data from Redis
+        score: Optional similarity score from vector search
+        deserialize_fn: Optional function to deserialize the value (handles
+            LangChain messages, etc.)
+
+    Returns:
+        A SearchItem with properly deserialized value
+    """
+    value = row["value"]
+    if deserialize_fn is not None:
+        value = deserialize_fn(value)
     return SearchItem(
-        value=row["value"],
+        value=value,
         key=row["key"],
         namespace=namespace,
         created_at=datetime.fromtimestamp(row["created_at"] / 1_000_000, timezone.utc),