RunnableWithMessageHistory Class — langchain Architecture

Source Code

libs/core/langchain_core/runnables/history.py lines 38–616

class RunnableWithMessageHistory(RunnableBindingBase):  # type: ignore[no-redef]
    """`Runnable` that manages chat message history for another `Runnable`.

    A chat message history is a sequence of messages that represent a conversation.

    `RunnableWithMessageHistory` wraps another `Runnable` and manages the chat message
    history for it; it is responsible for reading and updating the chat message
    history.

    The formats supported for the inputs and outputs of the wrapped `Runnable`
    are described below.

    `RunnableWithMessageHistory` must always be called with a config that contains
    the appropriate parameters for the chat message history factory.

    By default, the `Runnable` is expected to take a single configuration parameter
    called `session_id` which is a string. This parameter is used to create a new
    or look up an existing chat message history that matches the given `session_id`.

    In this case, the invocation would look like this:

    `with_history.invoke(..., config={"configurable": {"session_id": "bar"}})`
    ; e.g., `{"configurable": {"session_id": "<SESSION_ID>"}}`.

    The configuration can be customized by passing in a list of
    `ConfigurableFieldSpec` objects to the `history_factory_config` parameter (see
    example below).

    In the examples, we will use a chat message history with an in-memory
    implementation to make it easy to experiment and see the results.

    For production use cases, you will want to use a persistent implementation
    of chat message history, such as `RedisChatMessageHistory`.

    Example: Chat message history with an in-memory implementation for testing.

        ```python
        from operator import itemgetter

        from langchain_openai.chat_models import ChatOpenAI

        from langchain_core.chat_history import BaseChatMessageHistory
        from langchain_core.documents import Document
        from langchain_core.messages import BaseMessage, AIMessage
        from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
        from pydantic import BaseModel, Field
        from langchain_core.runnables import (
            RunnableLambda,
            ConfigurableFieldSpec,
            RunnablePassthrough,
        )
        from langchain_core.runnables.history import RunnableWithMessageHistory


        class InMemoryHistory(BaseChatMessageHistory, BaseModel):
            \"\"\"In memory implementation of chat message history.\"\"\"

            messages: list[BaseMessage] = Field(default_factory=list)

            def add_messages(self, messages: list[BaseMessage]) -> None:
                \"\"\"Add a list of messages to the store\"\"\"
                self.messages.extend(messages)

            def clear(self) -> None:
                self.messages = []

        # Here we use a global variable to store the chat message history.
        # This will make it easier to inspect it to see the underlying results.
        store = {}

        def get_by_session_id(session_id: str) -> BaseChatMessageHistory:
            if session_id not in store:
                store[session_id] = InMemoryHistory()
            return store[session_id]


        history = get_by_session_id("1")
        history.add_message(AIMessage(content="hello"))
        print(store)  # noqa: T201

        ```