stream_log endpoint fails on chain with object DynamoDBChatMessageHistory

[mesutgurlek]

Hi,

In my chain implementation, I have implemented a chain with:

• Custom Class that is passed in passthrough_dict until the end of the chain.
• DynamoDBChatMessageHistory which is used both in RunnableWithMessageHistory and manually in the chain.

At some point, in APIHandler class, there is a object serialization part, that raises error:

Specific code block that raises error (inside api_handler.py, in _stream_events function):

                    if (
                        self._names_in_stream_allow_list is None
                        or self._runnable.config.get("run_name")
                        in self._names_in_stream_allow_list
                    ):
                        # Strip internal metadata from the event
                        event["metadata"] = _strip_internal_keys(event["metadata"])
                        yield {
                            # EventSourceResponse expects a string for data
                            # so after serializing into bytes, we decode into utf-8
                            # to get a string.
                            "data": self._serializer.dumps(event).decode("utf-8"),
                            "event": "data",
                        }

Possible solutions:

• I have extended my Custom Class to pydantic BaseModel, which solved serialization. However, chain still raises error for DynamoDBChatMessageHistory class.
• Another option might be providing stream_log_name_allow_list inside add_routes(), however there is neither documentation nor example about this parameter, so I am not sure which log names to add.

@eyurtsev Do you have any suggestion?

Thanks in advance!

[eyurtsev]

What's the failure you're seeing?

[mesutgurlek]

Hi @eyurtsev, here is the mention from my previous question:

'''At some point, in APIHandler class, there is a object serialization part, that raises error:

Specific code block that raises error (inside api_handler.py, in _stream_events function):'''

Here is the related error:

+-+---------------- 1 ----------------
| RuntimeError: super(): __class__ cell not found
| 
| The above exception was the direct cause of the following exception:
| 
| Traceback (most recent call last):
|   File ".venv/lib/python3.11/site-packages/sse_starlette/sse.py", line 258, in wrap
|     await func()
|   File ".venv/lib/python3.11/site-packages/sse_starlette/sse.py", line 245, in stream_response
|     async for data in self.body_iterator:
|   File ".venv/lib/python3.11/site-packages/langserve/api_handler.py", line 1085, in _stream_log
|     "data": self._serializer.dumps(data).decode("utf-8"),
|             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|   File ".venv/lib/python3.11/site-packages/langserve/serialization.py", line 168, in dumps
|     return orjson.dumps(obj, default=default)
|            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
| TypeError: Type is not JSON serializable: ContextLogger
+------------------------------------
NotImplementedError in LogStreamCallbackHandler.on_chain_end callback: NotImplementedError("Trying to load an object that doesn't implement serialization: {'lc': 1, 'type': 'not_implemented', 'id': ['rag', 'ContextLogger'], 'repr': '<rag.ContextLogger object at 0x135c996d0>'}")
NotImplementedError in LogStreamCallbackHandler.on_chain_end callback: NotImplementedError("Trying to load an object that doesn't implement serialization: {'lc': 1, 'type': 'not_implemented', 'id': ['rag', 'ContextLogger'], 'repr': '<rag.ContextLogger object at 0x135c996d0>'}")
NotImplementedError in LogStreamCallbackHandler.on_chain_end callback: NotImplementedError("Trying to load an object that doesn't implement serialization: {'lc': 1, 'type': 'not_implemented', 'id': ['rag', 'ContextLogger'], 'repr': '<rag.ContextLogger object at 0x135c996d0>'}")
NotImplementedError in LogStreamCallbackHandler.on_chain_end callback: NotImplementedError("Trying to load an object that doesn't implement serialization: {'lc': 1, 'type': 'not_implemented', 'id': ['rag', 'ContextLogger'], 'repr': '<rag.ContextLogger object at 0x135c996d0>'}")

[mesutgurlek]

Hi again @eyurtsev,

I think I got the problem specifically, so I can provide the summary, but still dont know how to workaround it.

• I want to use DynamoDBChatMessageHistory class for session management.
• I also want to use langchain playground which uses astream_log endpoint.
• astream_log returns intermadiate steps as follows:

yield {
    # EventSourceResponse expects a string for data
    # so after serializing into bytes, we decode into utf-8
    # to get a string.
    "data": self._serializer.dumps(data).decode("utf-8"),
    "event": "data",
}

• Serializer is the WellKnownLCSerializer:

self._serializer = WellKnownLCSerializer()

• WellKnowLCSerializer is defined as follows:

class WellKnownLCSerializer(Serializer):
    def dumpd(self, obj: Any) -> Any:
        """Convert the given object to a JSON serializable object."""
        return orjson.loads(orjson.dumps(obj, default=default))

    def dumps(self, obj: Any) -> bytes:
        """Dump the given object as a JSON string."""
        return orjson.dumps(obj, default=default)

    def loadd(self, obj: Any) -> Any:
        """Load the given object."""
        return _decode_lc_objects(obj)

    def loads(self, s: bytes) -> Any:
        """Load the given JSON string."""
        return self.loadd(orjson.loads(s))

• orjson.dumps(obj, default=default) serializes pydantic BaseModel successfully but does not provide serialization for third party classes.

def default(obj) -> Any:
    """Default serialization for well known objects."""
    if isinstance(obj, BaseModel):
        return obj.dict()
    return super().default(obj)

At this point, if I want to use a custom class implementation or a LangChain class that does not extend Runnable or BaseModel wont be serialized in the astream_log endpoint response.
Do you know any workaround or best practices for my case?