'TypeError: Object of type float32 is not JSON serializable' when implemeting MemorySaver checkpointer

[jcalvoch]

System Information

OS: Linux
OS Version: #212-Ubuntu SMP Fri Jul 5 09:47:39 UTC 2024
Python Version: 3.10.14 (main, Mar 21 2024, 16:24:04) [GCC 11.2.0]

Package Information

langchain_core: 0.2.37
langchain: 0.2.15
langchain_community: 0.2.15
langsmith: 0.1.108
langchain_ibm: 0.1.12
langchain_text_splitters: 0.2.2
langchainhub: 0.1.21
langgraph: 0.2.16

Optional packages not installed

langserve

Other Dependencies

aiohttp: 3.9.5
async-timeout: 4.0.2
dataclasses-json: 0.6.7
httpx: 0.27.2
ibm-watsonx-ai: 1.0.11
jsonpatch: 1.33
langgraph-checkpoint: 1.0.8
numpy: 1.23.5
orjson: 3.10.7
packaging: 24.1
pydantic: 2.8.2
PyYAML: 6.0
requests: 2.31.0
SQLAlchemy: 1.4.39
tenacity: 8.5.0
types-requests: 2.32.0.20240712
typing-extensions: 4.12.2

Currently working on a project based on https://langchain-ai.github.io/langgraph/tutorials/rag/langgraph_self_rag_local/. For my use case I would like to implement memory. However, I am getting the following error when implementing MemorySaver() to my workflow.

---CHECK HALLUCINATIONS---
---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---
---GRADE GENERATION vs QUESTION---
---DECISION: GENERATION ADDRESSES QUESTION---
"Node 'generate':"
'\n---\n'

TypeError Traceback (most recent call last)
Cell In[36], line 8
6 inputs = {"question": "--omitted--?", "generate_attempts":0, "session_id": "1"}
7 try:
----> 8 for output in app.stream(inputs, {"recursion_limit": 15, "configurable": {"thread_id": "42"}}, stream_mode="updates", ):
9 for key, value in output.items():
10 # Node
11 pprint(f"Node '{key}':")

File /opt/conda/envs/Python-RT23.1/lib/python3.10/site-packages/langgraph/pregel/init.py:1211, in Pregel.stream(self, input, config, stream_mode, output_keys, interrupt_before, interrupt_after, debug, subgraphs)
1194 # assign defaults
1195 (
1196 debug,
1197 stream_modes,
(...)
1208 debug=debug,
1209 )
-> 1211 with SyncPregelLoop(
1212 input,
1213 config=config,
1214 store=self.store,
1215 checkpointer=checkpointer,
1216 nodes=self.nodes,
1217 specs=self.channels,
1218 output_keys=output_keys,
1219 stream_keys=self.stream_channels_asis,
1220 ) as loop:
1221 if subgraphs:
1222 loop.config["configurable"][CONFIG_KEY_STREAM] = loop.stream

File /opt/conda/envs/Python-RT23.1/lib/python3.10/site-packages/langgraph/pregel/loop.py:589, in SyncPregelLoop.exit(self, exc_type, exc_value, traceback)
582 def exit(
583 self,
584 exc_type: Optional[Type[BaseException]],
(...)
587 ) -> Optional[bool]:
588 # unwind stack
--> 589 return self.stack.exit(exc_type, exc_value, traceback)

File /opt/conda/envs/Python-RT23.1/lib/python3.10/contextlib.py:576, in ExitStack.exit(self, *exc_details)
572 try:
573 # bare "raise exc_details[1]" replaces our carefully
574 # set-up context
575 fixed_ctx = exc_details[1].context
--> 576 raise exc_details[1]
577 except BaseException:
578 exc_details[1].context = fixed_ctx

File /opt/conda/envs/Python-RT23.1/lib/python3.10/contextlib.py:561, in ExitStack.exit(self, *exc_details)
559 assert is_sync
560 try:
--> 561 if cb(*exc_details):
562 suppressed_exc = True
563 pending_raise = False

File /opt/conda/envs/Python-RT23.1/lib/python3.10/site-packages/langgraph/pregel/executor.py:92, in BackgroundExecutor.exit(self, exc_type, exc_value, traceback)
90 for task in self.tasks:
91 try:
---> 92 task.result()
93 except concurrent.futures.CancelledError:
94 pass

File /opt/conda/envs/Python-RT23.1/lib/python3.10/concurrent/futures/_base.py:451, in Future.result(self, timeout)
449 raise CancelledError()
450 elif self._state == FINISHED:
--> 451 return self.__get_result()
453 self._condition.wait(timeout)
455 if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:

File /opt/conda/envs/Python-RT23.1/lib/python3.10/concurrent/futures/_base.py:403, in Future.__get_result(self)
401 if self._exception:
402 try:
--> 403 raise self._exception
404 finally:
405 # Break a reference cycle with the exception in self._exception
406 self = None

File /opt/conda/envs/Python-RT23.1/lib/python3.10/site-packages/langgraph/pregel/executor.py:59, in BackgroundExecutor.done(self, task)
57 def done(self, task: concurrent.futures.Future) -> None:
58 try:
---> 59 task.result()
60 except GraphInterrupt:
61 # This exception is an interruption signal, not an error
62 # so we don't want to re-raise it on exit
63 self.tasks.pop(task)

File /opt/conda/envs/Python-RT23.1/lib/python3.10/concurrent/futures/_base.py:451, in Future.result(self, timeout)
449 raise CancelledError()
450 elif self._state == FINISHED:
--> 451 return self.__get_result()
453 self._condition.wait(timeout)
455 if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:

File /opt/conda/envs/Python-RT23.1/lib/python3.10/concurrent/futures/_base.py:403, in Future.__get_result(self)
401 if self._exception:
402 try:
--> 403 raise self._exception
404 finally:
405 # Break a reference cycle with the exception in self._exception
406 self = None

File /opt/conda/envs/Python-RT23.1/lib/python3.10/concurrent/futures/thread.py:58, in _WorkItem.run(self)
55 return
57 try:
---> 58 result = self.fn(*self.args, **self.kwargs)
59 except BaseException as exc:
60 self.future.set_exception(exc)

File /opt/conda/envs/Python-RT23.1/lib/python3.10/site-packages/langgraph/checkpoint/memory/init.py:363, in MemorySaver.put_writes(self, config, writes, task_id)
361 for idx, (c, v) in enumerate(writes):
362 inner_key = (task_id, WRITES_IDX_MAP.get(c, idx))
--> 363 self.writes[outer_key][inner_key] = (task_id, c, self.serde.dumps_typed(v))

File /opt/conda/envs/Python-RT23.1/lib/python3.10/site-packages/langgraph/checkpoint/serde/jsonplus.py:178, in JsonPlusSerializer.dumps_typed(self, obj)
176 return "bytearray", obj
177 else:
--> 178 return "json", self.dumps(obj)

File /opt/conda/envs/Python-RT23.1/lib/python3.10/site-packages/langgraph/checkpoint/serde/jsonplus.py:168, in JsonPlusSerializer.dumps(self, obj)
167 def dumps(self, obj: Any) -> bytes:
--> 168 return json.dumps(obj, default=self._default, ensure_ascii=False).encode(
169 "utf-8", "ignore"
170 )

File /opt/conda/envs/Python-RT23.1/lib/python3.10/json/init.py:238, in dumps(obj, skipkeys, ensure_ascii, check_circular, allow_nan, cls, indent, separators, default, sort_keys, **kw)
232 if cls is None:
233 cls = JSONEncoder
234 return cls(
235 skipkeys=skipkeys, ensure_ascii=ensure_ascii,
236 check_circular=check_circular, allow_nan=allow_nan, indent=indent,
237 separators=separators, default=default, sort_keys=sort_keys,
--> 238 **kw).encode(obj)

File /opt/conda/envs/Python-RT23.1/lib/python3.10/json/encoder.py:199, in JSONEncoder.encode(self, o)
195 return encode_basestring(o)
196 # This doesn't pass the iterator directly to ''.join() because the
197 # exceptions aren't as detailed. The list call should be roughly
198 # equivalent to the PySequence_Fast that ''.join() would do.
--> 199 chunks = self.iterencode(o, _one_shot=True)
200 if not isinstance(chunks, (list, tuple)):
201 chunks = list(chunks)

File /opt/conda/envs/Python-RT23.1/lib/python3.10/json/encoder.py:257, in JSONEncoder.iterencode(self, o, _one_shot)
252 else:
253 _iterencode = _make_iterencode(
254 markers, self.default, _encoder, self.indent, floatstr,
255 self.key_separator, self.item_separator, self.sort_keys,
256 self.skipkeys, _one_shot)
--> 257 return _iterencode(o, 0)

File /opt/conda/envs/Python-RT23.1/lib/python3.10/site-packages/langgraph/checkpoint/serde/jsonplus.py:121, in JsonPlusSerializer._default(self, obj)
119 return repr(obj)
120 else:
--> 121 raise TypeError(
122 f"Object of type {obj.class.name} is not JSON serializable"
123 )

TypeError: Object of type float32 is not JSON serializable

This is the code that I am trying to implement:

from langgraph.graph import END, StateGraph, START
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
    
workflow = StateGraph(GraphState)

workflow.add_node("retrieve", retrieve)  # retrieve
workflow.add_node("grade_documents", grade_documents)  # grade documents
workflow.add_node("generate", generate)  # generate
workflow.add_node("transform_query", transform_query)  # transform_query


workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",
    },
)
workflow.add_edge("transform_query", "retrieve")
workflow.add_conditional_edges(
    "generate",
    grade_generation_v_documents_and_question,
    {
        "not supported": "generate",
        "useful": END,
        "not useful": "transform_query",
    },
)

# Compile
app = workflow.compile(checkpointer=checkpointer)