A mokey patch demo for use memory with history in LCEL

[954-Ivory]

I just submitted an issuse(#16525).
But I don't know why my description is missing.

It goes something like this:
This is why BaseChatMemory.chat memory doesn't pruning #14957 (comment)

So I made some mokey patches to fix the problem temporarily.
And the following demo implements to use history and memory in LCEL.
Because there are too many modules involved, I wanted core contributors to help me refine this idea.

import json
from typing import Union, Any, List, Optional
from langchain.memory.chat_memory import BaseChatMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.load import load
from langchain_core.messages import BaseMessage, message_to_dict
from langchain_core.runnables import Runnable, RunnableConfig
from langchain_core.runnables.history import (
    RunnableWithMessageHistory,
    MessagesOrDictWithMessages,
    GetSessionHistoryCallable
)
from langchain_core.tracers.schemas import Run


class MemoryList(list):
    def __init__(self, *args, history=None, **kwargs):
        self.__history: RedisChatMessageHistory = history
        super().__init__(*args, **kwargs)

    def pop(self, __index=-1):
        if __index == 0:
            self.__history.redis_client.rpop(self.__history.key)
        elif __index == -1:
            self.__history.redis_client.lpop(self.__history.key)
        else:
            raise IndexError("Redis doesn't support pop by index.")
        return super().pop(__index)

    def clear(self):
        self.__history.clear()
        super().clear()


class RedisChatMessageHistoryFixed(RedisChatMessageHistory):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        msgs = super().messages.copy()
        self._messages: MemoryList = MemoryList(msgs, history=self)

    @property
    def messages(self) -> List[BaseMessage]:  # type: ignore
        msgs = super().messages.copy()
        self._messages: MemoryList = MemoryList(msgs, history=self)
        return self._messages

    @messages.setter
    def messages(self, msgs):
        self._messages.clear()
        if msgs:
            self.redis_client.lpush(self.key, *[json.dumps(message_to_dict(msg)) for msg in msgs])


class RunnableWithMessageHistoryWithMemory(RunnableWithMessageHistory):
    memory: Optional[BaseChatMemory] = None

    def __init__(
            self,
            runnable: Runnable[
                MessagesOrDictWithMessages,
                Union[str, BaseMessage, MessagesOrDictWithMessages]
            ],
            get_session_history: GetSessionHistoryCallable,
            memory: BaseChatMemory = None,
            **kwargs: Any
    ):
        super().__init__(runnable, get_session_history, **kwargs)
        if memory:
            self.memory = memory
            self.memory.input_key = self.input_messages_key
            self.memory.output_key = self.output_messages_key

    def _enter_history(self, input: Any, config: RunnableConfig) -> List[BaseMessage]:
        hist = config["configurable"]["message_history"]
        if not isinstance(self.memory.chat_memory, RedisChatMessageHistoryFixed):
            self.memory.chat_memory = hist
        # return only historic messages
        if self.history_messages_key:
            # Some of the 'BaseChatMemory' pruning features are in `load_memory_variables()`,
            # such as `ConversationSummaryBufferMemory`.
            # So we should extract the `messages` from 'load_memory_variables()'.
            messages = self.memory.load_memory_variables({})[self.history_messages_key].copy()
            hist.messages = messages
            return messages
        # return all messages
        else:
            input_val = (
                input if not self.input_messages_key else input[self.input_messages_key]
            )
            return hist.messages.copy() + self._get_input_messages(input_val)

    def _exit_history(self, run: Run, config: RunnableConfig) -> None:
        hist = config["configurable"]["message_history"]
        if not isinstance(self.memory.chat_memory, RedisChatMessageHistoryFixed):
            self.memory.chat_memory = hist
        # Get the input messages
        inputs = load(run.inputs)
        input_val = inputs[self.input_messages_key or "input"]
        input_messages = self._get_input_messages(input_val)

        # If historic messages were prepended to the input messages, remove them to
        # avoid adding duplicate messages to history.
        if not self.history_messages_key:
            historic_messages = config["configurable"]["message_history"].messages
            input_messages = input_messages[len(historic_messages):]

        # Get the output messages
        output_val = load(run.outputs)
        output_messages = self._get_output_messages(output_val)

        messages = zip(input_messages, output_messages)
        # `BaseChatMemory.save_context()` will call `add_message()` and `prune()`.
        # `RunnableWithMessageHistory` just call the `add_message()`.
        for i, o in messages:
            self.memory.save_context(
                {self.input_messages_key or 'input': i.content},
                {self.output_messages_key or 'output': o.content}
            )



if __name__ == '__main__':
    REDIS_URL = ...
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", 'You are a helpful assistant.'),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{question}"),
        ]
    )

    model = ChatOpenAI(
        model="gpt-3.5-turbo",
    )
    chain = prompt | model
    chain_with_history = RunnableWithMessageHistoryPlus(
        chain,
        lambda session_id: RedisChatMessageHistoryFixed(session_id, url=REDIS_URL),
        memory=ConversationSummaryBufferMemory(
            llm=model,
            memory_key="history",
            return_messages=True,
            max_token_limit=2000
        ),
        input_messages_key="question",
        history_messages_key="history",
    )


    def chat(question):
        res = chain_with_history.stream(
            {"question": question},
            config={"configurable": {"session_id": 'test'}},
        )
        for message in res:
            print(message.content, end='')


    while _question := input('human：'):
        chat(_question)
        print()