08. RunnableWithMessageHistory

Add message history (memory)

RunnableWithMessageHistory Using to add message history to certain types of tasks (chains) is a very useful feature in programming.

This feature is especially important when you need to keep the context of previous messages when implementing interactive applications or complex data processing tasks.

By managing message history, developers can better control the flow of applications and respond appropriately to user's previous requests.

Actual utilization example

• Interactive chatbot development: You can adjust the chatbot's response based on the conversation history with the user.

• Complex data processing: During the data processing process, you can refer to the results of the previous step to determine the logic of the next step.

• Applications that require state management: You can remember the user's previous choices and provide the next screen or information accordingly.

RunnableWithMessageHistory Is a powerful tool that allows you to stay in the state of your application, improve your user experience, and implement more sophisticated response mechanisms.

tutorial

Copy

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

model = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are an assistant who is proficient in {ability}. Please respond in 20 characters or less.",
        ),
        # Use conversation history as a variable, history becomes the key of MessageHistory
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),  # Use user input as a variable
    ]
)
runnable = prompt | model  # Create a runnable object by connecting a prompt and a model

Managing message history is very important for interactive applications or complex data processing tasks. Effective management of message records requires two main elements:

1. Runnable : Mainly like Retriever, Chain BaseChatMessageHistory Interact with runnable Object.

2. BaseChatMessageHistory Callable object (callable) returning an instance of : An object to manage message history. This object is used to store, search, and update message history. Message recording is necessary to maintain the context of the conversation and generate a response based on the user's previous input.

There are many ways to implement message recording, memory integrations The page introduces various storage options and integration methods.

Here we will look at two main methods.

Inmemory ChatMessageHistory use

This method manages message recording within memory. Mainly used in development stages or simple applications. The inmemory approach provides a fast approach speed, but it has the disadvantage of losing message history when restarting the application.

RedisChatMessageHistory Use to utilize permanent storage

The method of using Redis allows you to permanently save message history. Redis is an open source in-memory data structure repository that provides high performance, allowing you to reliably manage message records even in a distributed environment. This method is suitable for complex applications or long-running services.

When choosing how to manage message records, you should consider the application's requirements, the amount of traffic expected, the importance of message data, and the retention period. The inmemory method is simple and fast to implement, but if data permanence is required, it may be more appropriate to use a permanent repository like Redis.

Volatile Conversation: In-Memory

Below is a simple example where chat history is stored in memory.

RunnableWithMessageHistory Setting parameters

• runnable

• BaseChatMessageHistory Objects that have been or have been inherited. ex) ChatMessageHistory

• input_messages_key : key to specify user query input when chain invoke()

• history_messages_key : key specified by conversation record

Copy

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}  # A dictionary to store session records.


# Function to retrieve session records based on session ID
def get_session_history(session_ids: str) -> BaseChatMessageHistory:
    print(session_ids)
    if session_ids not in store:  # If the session ID is not in the store
        # Create a new ChatMessageHistory object and save it to the store.
        store[session_ids] = ChatMessageHistory()
    return store[session_ids]  # Returns the session record for the given session ID.


with_message_history = (
    RunnableWithMessageHistory(  # RunnableWithMessageHistory Object creation
        runnable,  # Runnable object to execute
        get_session_history,  # Function to retrieve session records
        input_messages_key="input",  # Key in input message
        history_messages_key="history",  # 기록 메시지의 키
    )
)

input_messages_key Specifies the key to be processed with the latest input message, history_messages_key Specifies the key to add the previous message.

Looking at the following code RunnableWithMessageHistory At the initial value session_id You can see the key being inserted as Default, just by this code RunnableWithMessageHistory Manage conversation threads session_id You can indirectly know that it is Hanha.

In other words, conversation thread-specific management session_id You can see that it is implemented very much.

Reference code: RunnableWithMessageHistory With reference to implementation,

Copy

if history_factory_config:
    _config_specs = history_factory_config
else:
    # If not provided, then we'll use the default session_id field
    _config_specs = [
        ConfigurableFieldSpec(
            id="session_id",
            annotation=str,
            name="Session ID",
            description="Unique identifier for a session.",
            default="",
            is_shared=True,
        ),
    ]

therefore, invoke() city config={"configurable": {"session_id": "세션ID입력"}} You can see that you must specify the code.

Copy

with_message_history.invoke(
    # Pass the math related question "What is the meaning of cosine?" as input.
    {"ability": "math", "input": "What does cosine mean?"},
    # Pass the session ID "abc123" as configuration information.
    config={"configurable": {"session_id": "abc123"}},
)

Copy

 abc123

Copy

 AIMessage (content=' Cosine is a trigonometric function that represents the ratio of the adjacent side to the hypotenuse in a right triangle.', response_metadata={'token_usage': {'completion_tok 

same session_id If you enter, you can get the contents of the previous conversation thread, so you can talk!

Copy

# Calls including message history.
with_message_history.invoke(
    # Set the capabilities and inputs.
    {"ability": "math", "input": "Please reply to the previous question in Korean."},
    # Specifies configuration options.
    config={"configurable": {"session_id": "abc123"}},
)

Copy

 abc123

Copy

 AIMessage (content=' cosine is a delta that represents the ratio of adjacent and hypotenuse in a right triangle.', response_metadata={'token_usage': {'completion_tokens': 47,'prompt_tokens': 98,'total_tok 

But different session_id If you specify, the answer will not be performed properly because there is no dialog.

(The example below session_id : def234 Since it doesn't exist, you can confirm that you are doing the wrong answer)

Copy

# New session_id does not remember previous conversations.
with_message_history.invoke(
    # Conveys mathematical skills and input messages.
    {"ability": "math", "input": "Please reply to the previous question in Korean"},
    # Set a new session_id.
    config={"configurable": {"session_id": "def234"}},
)

Copy

 def234

Copy

 AIMessage (content=' is an assistant proficient in mathematics.', response_metadata={'token_usage': {'completion_tokens': 19,'prompt_tokens': 58,'total_tokens': 77< 

The configuration parameters used to track message history ConfigurableFieldSpec List of objects history_factory_config You can customize it by passing it as a parameter.

like this history_factory_config When you set up a new one session_id The setting is overwritten.

In the example below user_id Wow conversation_id LA uses two parameters.

Copy

from langchain_core.runnables import ConfigurableFieldSpec

store = {}  # Initializes an empty dictionary.


def get_session_history(user_id: str, conversation_id: str) -> BaseChatMessageHistory:
    # Returns the session record corresponding to the given user_id and conversation_id.
    if (user_id, conversation_id) not in store:
        # If the key is not in the store, create a new one.ChatMessageHistory Create and save.
        store[(user_id, conversation_id)] = ChatMessageHistory()
    return store[(user_id, conversation_id)]


with_message_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
    history_factory_config=[  # This will replace the existing "session_id" setting.
        ConfigurableFieldSpec(
            id="user_id",  # Used as the first argument to the get_session_history function.
            annotation=str,
            name="User ID",
            description="Your unique identifier.",
            default="",
            is_shared=True,
        ),
        ConfigurableFieldSpec(
            id="conversation_id",  # get_session_history Used as the second argument to a function.
            annotation=str,
            name="Conversation ID",
            description="A unique identifier for the conversation.",
            default="",
            is_shared=True,
        ),
    ],
)

Copy

with_message_history.invoke(
    # Pass a dictionary containing abilities and inputs.
    {"ability": "math", "input": "Hello"},
    # Pass a config dictionary.
    config={"configurable": {"user_id": "123", "conversation_id": "1"}},
)

Copy

 AIMessage (content='Hello. What can I do for you?', response_metadata={'token_usage': {' completion_tokens': 21,'prompt_tokens': 43,'total_tokens': 64},'model_name':'gpt-3. 

Example using Runnable with various Keys

Enter Messages object, output in the form of dict

This is when you receive the message as input and return the dictionary as output.

• [important]: input_messages_key Omit ="input". Then you will set the Message object to be inserted as input.

Copy

from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableParallel

# create chain
chain = RunnableParallel({"output_message": ChatOpenAI()})


def get_session_history(session_id: str) -> BaseChatMessageHistory:
    # If a conversation history corresponding to the session ID does not exist in the store, a new ChatMessageHistory is created..
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    # Returns the conversation history corresponding to the session ID.
    return store[session_id]


# Create a RunnableWithMessageHistory object that adds conversation history functionality to the chain.
with_message_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    # Set the key of the input message to "input" (if omitted, input as a Message object)
    # input_messages_key="input",
    # Set the key of the output message to "output_message". (If omitted, output as a Message object)
    output_messages_key="output_message",
)

# Runs a chain with the given message and settings.
with_message_history.invoke(
    # 혹은 "what is the definition of cosine?" 도 가능
    [HumanMessage(content="what is the definition of cosine?")],
    config={"configurable": {"session_id": "abc123"}},
)

Copy

 {'output_message': AIMessage (content='The cosine function is a trigonometric function that returns the cosmine value of a given angle. It represents the ratio of the length of the adjacent side to the length of the hypotenuse in a right triangle with that angle. The cosine function is commonly denoted as "cos" and the angle is typically measure in radians.', respons_metadata={'token_usage': 66,'prom_tokens': 12 

Copy

with_message_history.invoke(
    # I would like to request a reply in Korean regarding the previous reply.
    [HumanMessage(content="Please reply to the previous question in Korean!")],
    # Pass configuration options in dictionary form.
    config={"configurable": {"session_id": "abc123"}},
)

Copy

 {'output_message': AIMessage (content=' The cosine function is a trigonometry that returns the cosine value of a given angle, representing the ratio of the length of the adjacency to the length of the hypotenuse. Mathematically, the cosine value for the angle θ is defined as the length of the adjacent sides of a right triangle with that angle divided by the length of the hypotenuse. '4p','foken_usage': ={':'f','prompt_tokens': 1020,'total_tokens': 

Enter Messages object, output of Messages object

• [important]: output_messages_key Omit ="output_message". Then it returns the Message object as output.

Copy

with_message_history = RunnableWithMessageHistory(
    ChatOpenAI(),  # Chat uses the OpenAI language model.
    get_session_history,  # Specifies a function to retrieve the conversation session history.
    # Set the key of the input message to "input" (if omitted, input as a Message object)
    # input_messages_key="input",
    # Set the key of the output message to "output_message". (If omitted, output as a Message object)
    # output_messages_key="output_message",
)

Copy

with_message_history.invoke(
    # I would like to request a reply in Korean regarding the previous reply.
    [HumanMessage(content="What does cosine mean?")],
    # Pass configuration options in dictionary form.
    config={"configurable": {"session_id": "def123"}},
)

Copy

 Error in RootListenersTracer.on_llm_end callback: KeyError('input')

Copy

 AIMessage (content=' cosine is one of the trigonomets, meaning a ratio representing the cosine value for the hypotenuse and hypotenuse for the angle θ in a right triangle. Simply put, cosine is a trigonometric that represents the ratio of adjacent and hypotenuse to a specific angle in a right triangle.', response_metadata={'token_usage': {' completion_tokens': 138,'prompt_tokens': 24,'total_tokens':1 

Dict with single key for all message inputs and outputs

This is how you use a single key for all input and output messages.

• itemgetter("input_messages") Extract input messages using

Copy

from operator import itemgetter

with_message_history = RunnableWithMessageHistory(
    # "input_messages" Use the key to get the input message and pass it to ChatOpenAI().
    itemgetter("input_messages") | ChatOpenAI(),
    get_session_history,  # A function to retrieve session records.
    input_messages_key="input_messages",  # Specifies the key of the input message.
)

Copy

with_message_history.invoke(
    {"input_messages": "What does cosine mean?"},
    # Pass configuration options in dictionary form.
    config={"configurable": {"session_id": "xyz123"}},
)

Copy

 AIMessage (content=' cosine is one of the trigonomets representing the ratio of the length of the sides to each of the triangles. Cosine is defined as the ratio of the length of the side adjacent to the hypotenuse, and is mainly used in right triangle. cosine is a value that changes with the angle of the triangle, used in various mathematical applications and fields as a delta.', response_metadata={'token_usage': {' completion_tokens': 144,'prompt_tokens': 24,'total_tokens': 168 

Permanent storage

Persistent storage refers to a storage mechanism that maintains data even when the program ends or the system reboots. This can be implemented through databases, file systems, or other non-volatile storage devices.

Permanent storage stores the status of the application, maintains user settings, Essential for long-term data retention is. This allows the program to restart from the point where it was stopped in the previous run, Users continue to do this without data loss You can.

• RunnableWithMessageHistory has get_session_history Independent of how callable objects retrieve chat message history.

• Examples using local file systems here See.

• Below Redis Shows how to use. How to implement chat message history using other providers memory integrations Check the page.

Redis installation

If Redis is not installed, it must be installed first.

Copy

%pip install -qU redis

Redis server powered

If there is no Redis deployment to connect to the existing, launch a local Redis Stack server.

Here is the instruction that drives the Redis server with Docker.

Copy

docker run -d -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

REDIS_URL Assign the variable with the link URL of the Redis database.

• URL "redis://localhost:6379/0" It is set to.

Copy

# Specifies the URL of the Redis server.
REDIS_URL = "redis://localhost:6379/0"

LangSmith tracking settings

Set LangSmith for tracking. LangSmith is not essential, but it can help.

Copy

from dotenv import load_dotenv
import os

load_dotenv()

# Set the LANGCHAIN_TRACING_V2 environment variable to "true".
os.environ["LANGCHAIN_TRACING_V2"] = "true"
# Setting LANGCHAIN_PROJECT
os.environ["LANGCHAIN_PROJECT"] = "RunnableWithMessageHistory"

To update the message recording implementation, define a new callable object, this time just return an instance of the RedisChatMessageHistory.

Copy

from langchain_community.chat_message_histories import RedisChatMessageHistory


def get_message_history(session_id: str) -> RedisChatMessageHistory:
    # Returns a RedisChatMessageHistory object based on the session ID.
    return RedisChatMessageHistory(session_id, url=REDIS_URL)


with_message_history = RunnableWithMessageHistory(
    runnable,  # executable object
    get_message_history,  # Function to get message history
    input_messages_key="input",  # Key in input message
    history_messages_key="history",  # 기록 메시지의 키
)

You can call it the same way as before.

Copy

with_message_history.invoke(
    # Pass the math related question "What is the meaning of cosine?" as input.
    {"ability": "math", "input": "What does cosine mean?"},
    # Set the session ID to "redis123" as a configuration option.
    config={"configurable": {"session_id": "redis123"}},
)

Copy

 AIMessage (content=' Trigonometric function that represents the ratio of the adjacent side to the hypotenuse in a right triangle.', response_metadata={'token_usage': {'completion_tokens' 

same session_id Use to make a second call. This time, I'll ask you to answer the previous answer in Koreanglo.

Copy

with_message_history.invoke(
    # I would like to request a Korean translation of my previous answer.
    {"ability": "math", "input": "Please translate the previous answer into Korean."},
    # Set the session ID to "foobar" as the configuration value.
    config={"configurable": {"session_id": "redis123"}},
)

Copy

 AIMessage (content=' is a delta function that represents the ratio of the length of the adjacency to the length of the hypotenuse.', response_metadata={'token_usage': {'completion_tokens': 95,'tot 

Different this time session_id I will use to ask a question.

Copy

with_message_history.invoke(
    # I would like to request a Korean translation of my previous answer.
    {"ability": "math", "input": "Please translate the previous answer into Korean."},
    # Session ID as a setting value "redis456"로 지정합니다.
    config={"configurable": {"session_id": "redis456"}},
)

Copy

 AIMessage (content=' is an assistant proficient in mathematics.', response_metadata={'token_usage': {'completion_tokens': 19,'prompt_tokens': 60,'total_tokens': 79TA