TypeError: Type is not JSON serializable: numpy.float64

[maximeperrindev]

Checked other resources

• I added a very descriptive title to this issue.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.
• I am sure that this is a bug in LangChain rather than my code.

Example Code

from fastapi import FastAPI
from langchain_openai import ChatOpenAI
from langchain.schema.runnable import (
    ConfigurableField,
    Runnable,
    RunnableBranch,
    RunnableLambda,
    RunnableMap,
)
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_core import __version__
from langchain_community.vectorstores import Milvus
from langserve import add_routes as langserve_add_routes
from dotenv import load_dotenv
from operator import itemgetter
from file_manager import add_routes as file_manager_add_routes
from fastapi import FastAPI
from langchain.schema.output_parser import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.messages import get_buffer_string
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from prompts import CONDENSE_QUESTION_PROMPT, ANSWER_PROMPT, DEFAULT_DOCUMENT_PROMPT, RESPONSE_TEMPLATE
from langchain_community.retrievers.llama_index import LlamaIndexRetriever
from llama_index.core.indices.vector_store.base import VectorStoreIndex
from llama_index.vector_stores.milvus import MilvusVectorStore
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.retrievers.document_compressors import (
    DocumentCompressorPipeline,
    EmbeddingsFilter,
)
from datetime import datetime
from typing import Sequence
from langchain.schema.document import Document
from llama_index.core.vector_stores.types import VectorStore
from langchain.retrievers import ContextualCompressionRetriever
from langchain_openai import OpenAIEmbeddings
load_dotenv()

def get_chat_history(session_id: str) -> str:
    """Get the chat history for the session."""
    history = SQLChatMessageHistory(
        session_id=session_id,
        connection_string="postgresql://rag:rag_pass@127.0.0.1:5432/rag"
    )
    return history


def format_docs(docs: Sequence[Document]) -> str:
    formatted_docs = []
    for i, doc in enumerate(docs):
        doc_string = f"<doc id='{i}'>{doc.page_content}</doc>"
        formatted_docs.append(doc_string)
    return "\n".join(formatted_docs)

def _get_retriever(vector_store: VectorStore):
    embeddings = OpenAIEmbeddings()
    splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=20)
    relevance_filter = EmbeddingsFilter(embeddings=embeddings, similarity_threshold=0.8)
    pipeline_compressor = DocumentCompressorPipeline(
        transformers=[splitter, relevance_filter]
    )
    base_retriever = LlamaIndexRetriever(index=VectorStoreIndex.from_vector_store(vector_store).as_query_engine())
    retriever = ContextualCompressionRetriever(
        base_retriever=base_retriever, base_compressor=pipeline_compressor
    )

    return retriever.with_config(run_name="SourceRetriever")

def create_retriever_chain(
    llm, retriever
) -> Runnable:
    condense_question_chain = (
        CONDENSE_QUESTION_PROMPT | llm | StrOutputParser()
    ).with_config(
        run_name="CondenseQuestion",
    )
    conversation_chain = condense_question_chain | retriever
    return RunnableBranch(
        (
            RunnableLambda(lambda x: get_buffer_string(x["chat_history"])).with_config(
                run_name="HasChatHistoryCheck"
            ),
            conversation_chain.with_config(run_name="RetrievalChainWithHistory"),
        ),
        (
            RunnableLambda(itemgetter("question")).with_config(
                run_name="Itemgetter:question"
            )
            | retriever
        ).with_config(run_name="RetrievalChainWithNoHistory"),
    ).with_config(run_name="RouteDependingOnChatHistory")

def _build_chain() -> Runnable:
    vector_store = MilvusVectorStore(dim=1536)
    retriever = _get_retriever(vector_store)

    llm = ChatOpenAI(model="gpt-3.5-turbo").configurable_alternatives(
        ConfigurableField(id="llm"),
        default_key="gpt-3.5-turbo",
        gpt_4_turbo_preview=ChatOpenAI(model="gpt-4-turbo-preview")
    )
    retriever_chain = create_retriever_chain(llm, retriever) | RunnableLambda(
        format_docs
    ).with_config(run_name="FormatDocumentChunks")

    _context = RunnableMap(
        {
            "context": retriever_chain.with_config(run_name="RetrievalChain"),
            "question": RunnableLambda(itemgetter("question")).with_config(
                run_name="Itemgetter:question"
            ),
            "chat_history": RunnableLambda(itemgetter("chat_history")).with_config(
                run_name="Itemgetter:chat_history"
            ),
        }
    )

    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", RESPONSE_TEMPLATE),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{question}"),
        ]
    ).partial(current_date=datetime.now().isoformat())

    response_synthesizer = (prompt | llm | StrOutputParser()).with_config(
        run_name="GenerateResponse",
    )

    return (
        {
            "question": RunnableLambda(itemgetter("question")).with_config(
                run_name="Itemgetter:question"
            ),
            "chat_history": RunnableLambda(itemgetter("chat_history")).with_config(
                run_name="SerializeHistory"
            ),
        }
        | _context
        | response_synthesizer
    )
chain = _build_chain()
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_chat_history,
    input_messages_key="question",
    history_messages_key="chat_history",
)

def add_routes(app: FastAPI) -> None:
    """Add routes to the FastAPI app."""
    langserve_add_routes(
        app,
        chain_with_history,
    # disabled_endpoints=["playground", "batch"],
    )
    file_manager_add_routes(app)

Error Message and Stack Trace (if applicable)

RuntimeError: super(): class cell not found

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/uvicorn/protocols/http/httptools_impl.py", line 426, in run_asgi
result = await app( # type: ignore[func-returns-value]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/uvicorn/middleware/proxy_headers.py", line 84, in call
return await self.app(scope, receive, send)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/fastapi/applications.py", line 1054, in call
await super().call(scope, receive, send)
File "/usr/local/lib/python3.11/site-packages/starlette/applications.py", line 123, in call
await self.middleware_stack(scope, receive, send)
File "/usr/local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 186, in call
raise exc
File "/usr/local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 164, in call
await self.app(scope, receive, _send)
File "/usr/local/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 62, in call
await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
File "/usr/local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 64, in wrapped_app
raise exc
File "/usr/local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
await app(scope, receive, sender)
File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 758, in call
await self.middleware_stack(scope, receive, send)
File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 778, in app
await route.handle(scope, receive, send)
File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 299, in handle
await self.app(scope, receive, send)
File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 79, in app
await wrap_app_handling_exceptions(app, request)(scope, receive, send)
File "/usr/local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 64, in wrapped_app
raise exc
File "/usr/local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
await app(scope, receive, sender)
File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 77, in app
await response(scope, receive, send)
File "/usr/local/lib/python3.11/site-packages/sse_starlette/sse.py", line 255, in call
async with anyio.create_task_group() as task_group:
File "/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 597, in aexit
raise exceptions[0]
File "/usr/local/lib/python3.11/site-packages/sse_starlette/sse.py", line 258, in wrap
await func()
File "/usr/local/lib/python3.11/site-packages/sse_starlette/sse.py", line 245, in stream_response
async for data in self.body_iterator:
File "/usr/local/lib/python3.11/site-packages/langserve/api_handler.py", line 1085, in _stream_log
"data": self._serializer.dumps(data).decode("utf-8"),
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/langserve/serialization.py", line 168, in dumps
return orjson.dumps(obj, default=default)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: Type is not JSON serializable: numpy.float64

Description

I'm trying to run langserve api with RAG and Runnable with message history but i'm facing this bug

System Info

System Information

OS: Darwin
OS Version: Darwin Kernel Version 23.1.0: Mon Oct 9 21:27:27 PDT 2023; root:xnu-10002.41.9~6/RELEASE_X86_64
Python Version: 3.11.6 (main, Oct 2 2023, 13:45:54) [Clang 15.0.0 (clang-1500.0.40.1)]

Package Information

langchain_core: 0.1.24
langchain: 0.1.8
langchain_community: 0.0.21
langsmith: 0.1.3
langchain_cli: 0.0.21
langchain_experimental: 0.0.50
langchain_openai: 0.0.5
langchainhub: 0.1.14
langserve: 0.0.41

Packages not installed (Not Necessarily a Problem)

The following packages were not found:

langgraph

[dosubot[bot]]

🤖

Hey @maximeperrindev, great to see you back! Running into some quirky bugs, I see? Let's get to the bottom of this.

The error you're encountering, "TypeError: Type is not JSON serializable: numpy.float64", typically occurs when you're trying to serialize a numpy.float64 object into JSON format. JSON serialization is a process of converting an object state to a JSON formatted string. However, not all Python objects can be serialized into JSON format. In Python's json module, only Python's primitive datatypes and collections are serializable by default.

In your case, there is no direct evidence of numpy.float64 being used or serialized into JSON. However, the error might be occurring because the data being processed in the LangChain application contains numpy.float64 objects. These objects might be part of the input data or might be generated during the processing of the data.

The error might be occurring in the following lines of code:

data = {"chunk": log_entry["streamed_output"][0]}

data = {"chunk": state["streamed_output"][0]}

output: Output = await asyncio.create_task(coro, context=context)  # type: ignore

output = await coro

await run_manager.on_chain_end(dumpd(output))

In these lines, if the "streamed_output" or the output of the coroutine contains numpy.float64 objects, it might cause the error when these objects are being serialized into JSON format.

To fix this issue, you can convert numpy.float64 objects into Python's native float type before serializing them into JSON. Here is an example of how you can do this:

import numpy as np

# Assuming `data` is a numpy.float64 object
data = np.float64(10.5)

# Convert numpy.float64 to Python's native float type
data = float(data)

# Now, `data` can be serialized into JSON format

Please note that this is a general solution. The actual solution might vary depending on the structure of your data and where exactly the error is occurring.

Regarding the structure of the data being processed in the LangChain application, it is structured as a query. This query is represented by the StructuredQuery class, which is parsed from a JSON string. The JSON string is expected to have the keys "query" and "filter", and optionally "limit". The "query" key represents the main query to be executed, the "filter" key represents any filtering conditions to be applied to the results of the query, and the "limit" key represents a limit on the number of results to be returned.

I hope this helps! If you have any more questions or need further clarification, feel free to ask.

Sources

• libs/core/langchain_core/runnables/base.py
• libs/langchain/langchain/chains/query_constructor/base.py

About Dosu

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

[maximeperrindev]

@dosu-bot I don't even know the object is... just try to find what's wrong with my own code

[dosubot[bot]]

🤖

The error you're encountering is due to the fact that numpy.float64 is not JSON serializable by default. The orjson.dumps method, which is used in the _stream_log function, is trying to serialize a numpy.float64 object and failing.

The _stream_log function is part of the LangChain codebase and is responsible for running a given runnable in streaming mode and logging the run. The self._serializer.dumps(data) line is where the data is being serialized into a JSON format. The data variable here is the output from the runnable that is being logged.

To resolve this issue, you could convert the numpy.float64 objects to Python native data types before they are serialized. This can be done by creating a custom JSON encoder that handles numpy.float64 objects. Here's an example of how you could do this:

import numpy as np
import orjson

class CustomEncoder:
    def dumps(self, obj):
        if isinstance(obj, np.float64):
            return orjson.dumps(obj.item())
        return orjson.dumps(obj)

# Replace the original serializer with the custom one
self._serializer = CustomEncoder()

This custom encoder checks if the object to be serialized is a numpy.float64 instance. If it is, it converts the numpy.float64 object to a Python float using the item method before serializing it. If the object is not a numpy.float64 instance, it is serialized as usual.

Please note that this is a general solution and might need to be adapted depending on the specifics of your code. If this doesn't solve your issue, could you please provide more information about the data that is being processed when the error occurs? This would help in identifying a more specific solution.

Sources

• libs/core/langchain_core/tracers/log_stream.py

About Dosu

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

[maximeperrindev]

🤖

The error you're encountering is due to the fact that numpy.float64 is not JSON serializable by default. The orjson.dumps method, which is used in the _stream_log function, is trying to serialize a numpy.float64 object and failing.

The _stream_log function is part of the LangChain codebase and is responsible for running a given runnable in streaming mode and logging the run. The self._serializer.dumps(data) line is where the data is being serialized into a JSON format. The data variable here is the output from the runnable that is being logged.

To resolve this issue, you could convert the numpy.float64 objects to Python native data types before they are serialized. This can be done by creating a custom JSON encoder that handles numpy.float64 objects. Here's an example of how you could do this:

import numpy as np
import orjson

class CustomEncoder:
    def dumps(self, obj):
        if isinstance(obj, np.float64):
            return orjson.dumps(obj.item())
        return orjson.dumps(obj)

# Replace the original serializer with the custom one
self._serializer = CustomEncoder()

This custom encoder checks if the object to be serialized is a numpy.float64 instance. If it is, it converts the numpy.float64 object to a Python float using the item method before serializing it. If the object is not a numpy.float64 instance, it is serialized as usual.

Please note that this is a general solution and might need to be adapted depending on the specifics of your code. If this doesn't solve your issue, could you please provide more information about the data that is being processed when the error occurs? This would help in identifying a more specific solution.

Sources

• libs/core/langchain_core/tracers/log_stream.py

About Dosu

@dosu-bot
I don't want to modify langchain code. This is my current code :

def _build_chain() -> Runnable:
    vector_store = MilvusVectorStore(dim=1536)
    retriever = _get_retriever(vector_store)

    llm = ChatOpenAI(model="gpt-3.5-turbo").configurable_alternatives(
        ConfigurableField(id="llm"),
        default_key="gpt-3.5-turbo",
        gpt_4_turbo_preview=ChatOpenAI(model="gpt-4-turbo-preview")
    )
    retriever_chain = create_retriever_chain(llm, retriever) | RunnableLambda(
        format_docs
    ).with_config(run_name="FormatDocumentChunks")

    _context = RunnableMap(
        {
            "context": retriever_chain.with_config(run_name="RetrievalChain"),
            "question": RunnableLambda(itemgetter("question")).with_config(
                run_name="Itemgetter:question"
            ),
            "chat_history": RunnableLambda(itemgetter("chat_history")).with_config(
                run_name="Itemgetter:chat_history"
            ),
        }
    )

    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", RESPONSE_TEMPLATE),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{question}"),
        ]
    ).partial(current_date=datetime.now().isoformat())

    response_synthesizer = (prompt | llm | StrOutputParser()).with_config(
        run_name="GenerateResponse",
    )

    return (
        {
            "question": RunnableLambda(itemgetter("question")).with_config(
                run_name="Itemgetter:question"
            ),
            "chat_history": RunnableLambda(itemgetter("chat_history")).with_config(
                run_name="SerializeHistory"
            ),
        }
        | _context
        | response_synthesizer
    )
chain = _build_chain()
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_chat_history,
    input_messages_key="question",
    history_messages_key="chat_history",
)

def add_routes(app: FastAPI) -> None:
    """Add routes to the FastAPI app."""
    langserve_add_routes(
        app,
        chain_with_history,
    # disabled_endpoints=["playground", "batch"],
    )

How can i fix it ?

[eyurtsev]

@maximeperrindev it looks like either the input or output (probably output) of one of the chains is a numpy array. If you want to output it and are sending the data over a web-server, you need to provide a way to encode the data as json.

The easiest thing to do is add another runnable lambda that takes the numpy and outputs a string representation of the numpy that can be sent over the wire (i.e., search online for ways to encode JSON)

[maximeperrindev]

@eyurtsev Thanks for the reply.

I don't really know where's the numpy actually.. I tried to print the object just before the exception and i got something like this :

 'ops': [{'op': 'add', 'path': '/logs/SourceRetriever/final_output', 'value': {'documents': [_DocumentWithState(page_content='Dans la table tarif_periode, on aura :1 heure Tarif horaire heure creuse 120 1\n2 heure Tarif horaire heure pleine 100 1id type libelle montant public\n1 1 8 12\n1 1 14 18\n2 1 12 14tarif jour_semaine heure_debut heure_fin', metadata={'page_label': '30', 'file_name': 'ILOO-170124-124023-1.pdf', 'file_path': '../docs/ILOO-170124-124023-1.pdf', 'file_type': 'application/pdf', 'file_size': 1793252, 'creation_date': '2024-02-20', 'last_modified_date': '2024-02-20', 'last_accessed_date': '2024-02-20', 'document_title': 'Gestion des tarifs et des périodes de restriction', 'section_summary': 'La section traite de la gestion des tarifs et des périodes de restriction. Elle aborde les différents types de tarifs, leur validité dans le temps, les promotions, et les restrictions liées aux utilisateurs. Elle mentionne également les restrictions sur les jours de la semaine et les tranches horaires, avec des exemples concrets de mise en place de tarifs en fonction de ces restrictions.'}, state={'embedded_doc': [0.01386505144403404, 0.01544047015256333, 0.0004034453785579994, -0.023719945328791205, -0.0232834642513163, -0.010114053942431296, -0.014362910402930133, -0.017350069744242, -0.016299789363459074,], 'query_similarity_score': 0.8110104310364158}

Maybe the issue is here because it looks like a numpy array with embeddings

[xiaokuili]

EmbeddingsFiltert return Sequence[_DocumentWithState]
_DocumentWithState can‘t be _serializer .

I'm thinking how to resolve it .

[xiaokuili]

I fix it . I rewrite EmbeddingsFilter,return Document object.

class ToDocEmbeddingsFilter(EmbeddingsFilter):
    def compress_documents(
        self,
        documents: Sequence[Document],
        query: str,
        callbacks: Optional[Callbacks] = None,
    ) -> Sequence[Document]:
        docs = []
        for doc in super().compress_documents( documents, query, callbacks):
            docs.append(doc.to_document())
        return docs
        

def get_retriever():
    # embeddings = OpenAIEmbeddings()
    embeddings = get_embeddings_model()
    splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=20)
    relevance_filter = ToDocEmbeddingsFilter(embeddings=embeddings, similarity_threshold=0.1)
    

[maximeperrindev]

Thank you !! I've created a PR with a potential fix i found: #17946
Let me know if you have something to add to this pr