RunnableWithMessageHistory in LangChain's Expression Language (LCEL) for managing conversation history in chatbots, virtual assistants, and other conversational AI applications. It seamlessly integrates with existing LangChain components to automatically handle message history management and updates.
Key Features
Message History Management
Maintains conversation context across multiple interactions.
Automatically tracks and stores chat messages.
Enables contextual responses based on previous conversations.
Flexible Input/Output Support
Handles both message objects and Python dictionaries.
Supports various input formats, including:
Single messages
Message sequences
Dictionary inputs with custom keys
Provides consistent output handling regardless of input format.
Session Management
Manages conversations through unique identifiers, such as:
Simple session IDs
Combined user and conversation IDs
Maintains separate conversation threads for different users or contexts.
Ensures conversation continuity within the same session.
Storage Options
Offers in-memory storage for development and testing.
Supports persistent storage (e.g., Redis, files) for production environments.
Provides easy integration with various storage backends.
Advantages Over Legacy Approaches
More flexible than the older ConversationChain.
Offers better state management.
Provides improved integration with modern LangChain components.
Summary
RunnableWithMessageHistory is the recommended standard for conversation management in LangChain, offering:
Simplified conversation state management.
An enhanced user experience through context preservation.
Flexible configuration options for diverse use cases.
Table of Contents
References
Environment Setup
[Note]
The langchain-opentutorial is a bundle of easy-to-use environment setup guidance, useful functions and utilities for tutorials.
Alternatively, you can set and load OPENAI_API_KEY from a .env file.
[Note] This is only necessary if you haven't already set OPENAI_API_KEY in previous steps.
from dotenv import load_dotenv
load_dotenv(override=True)
True
Getting Started with RunnableWithMessageHistory
Managing conversation history is crucial for conversational applications and complex data processing tasks. RunnableWithMessageHistory simplifies the message history implementation. To use it effectively, you need these two key components:
Runnable objects,
Creating Runnable objects, such as retriever or chain, are the primary components that interacts with BaseChatMessageHistory.
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model_name="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are an assistant skilled in {ability}. Keep responses under 20 words.",
),
# Use the conversation history as a variable, with 'history' as the key for MessageHistory.
MessagesPlaceholder(variable_name="history"),
("human", "{input}"), # Use user input as a variable
]
)
runnable = (
prompt | model
) # Create a runnable object by connecting the prompt and the model
Message History Manager (callable)
This is a callable that returns an instance of BaseChatMessageHistory. It handles message storage, retrieval, updates, and maintains conversation context for contextual responses.
Implementation Options
This tutorial covers two primary approaches in implementation:
In-Memory ChatMessageHistory
Manages message history in memory, making it ideal for development and simple applications.
Provides fast access speeds.
Message history is lost on application restart.
Persistent Storage with RedisChatMessageHistory
Enables permanent message storage using Remote Dictionary Server (Redis), a high-performance, open-source in-memory data structure store.
Suitable for distributed environments.
Ideal for complex applications and long-running services.
Consider these factors when selecting a message history management approach:
Application requirements
Expected traffic volume
Message data importance
Retention period requirements
While in-memory implementation offers simplicity and speed, persistent storage solutions like Redis are more appropriate when data durability is a concern.
Understanding In-Memory Conversation History
In-memory conversation history provides a simple and fast way to manage chat message history during development and testing. This approach stores conversation data in memory, offering quick access but without persistence across application restarts.
Core Configuration Parameters
Required Components
runnable: The chain or model (e.g., ChatOpenAI) to execute.
get_session_history: A function returning a BaseChatMessageHistory instance.
input_messages_key: Specifies the key for user input in invoke() calls.
history_messages_key: Defines the key for accessing conversation history.
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
store = {} # Initialize empty store for message histories.
# function for getting session logs by session_id.
def get_session_history(session_id: str) -> BaseChatMessageHistory:
print(session_id)
if session_id not in store: # if session_id isn't in the store,
# create new ChatMessageHistory and put it in the store.
store[session_id] = ChatMessageHistory()
return store[session_id] # return the session logs for session_id.
with_message_history = RunnableWithMessageHistory(
runnable,
get_session_history,
input_messages_key="input",
history_messages_key="history",
)
Default Session Implementation
RunnableWithMessageHistory uses session_id as its default identifier for managing conversation threads, as shown in its core implementation:
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,
),
]
Using Session Management
To utilize session management, specify a session ID in your invoke call:
AIMessage(content='Cosine is a trigonometric function representing the adjacent side over hypotenuse in a right triangle.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 22, 'prompt_tokens': 31, 'total_tokens': 53, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_5f20662549', 'finish_reason': 'stop', 'logprobs': None}, id='run-4afc17da-4dd9-46ff-888c-fff45266820e-0', usage_metadata={'input_tokens': 31, 'output_tokens': 22, 'total_tokens': 53, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})
Using the same session_id continues the conversation by retrieving the previous thread's content (this continuous conversation is called a session):
# Call with message history
with_message_history.invoke(
# Set ability and input
{"ability": "math", "input": "Please translate the previous content to Korean"},
# Specify configuration options
config={"configurable": {"session_id": "abc123"}},
)
However, using a different session_id will result in an inaccurate response because there is no corresponding history.
For example, if session_id is def234 and no history exists for that ID, you'll see an irrelevant response (see the following code snippet).
# New session_id means no previous conversation memory
with_message_history.invoke(
# Pass math ability and input message
{"ability": "math", "input": "Please translate the previous content to Korean"},
# Set a new session_id
config={"configurable": {"session_id": "def234"}},
)
You can customize the configuration parameters for tracking message history by passing a list of ConfigurableFieldSpec objects through the history_factory_config parameter.
Setting a new history_factory_config overrides the existing session_id configuration.
The following example demonstrates using two parameters: user_id and conversation_id.
from langchain_core.runnables import ConfigurableFieldSpec
store = {} # Initialize empty store for message histories.
def get_session_history(user_id: str, conversation_id: str) -> BaseChatMessageHistory:
# Return session history for given user_id and conversation_id combination.
if (user_id, conversation_id) not in store:
store[(user_id, conversation_id)] = ChatMessageHistory()
return store[(user_id, conversation_id)]
# Configure RunnableWithMessageHistory with custom identifiers
with_message_history = RunnableWithMessageHistory(
runnable,
get_session_history,
input_messages_key="input",
history_messages_key="history",
history_factory_config=[ # replacing the "session_id"
ConfigurableFieldSpec(
id="user_id",
annotation=str,
name="User ID",
description="Unique identifier for user.",
default="",
is_shared=True,
),
ConfigurableFieldSpec(
id="conversation_id",
annotation=str,
name="Conversation ID",
description="Unique identifier for conversation.",
default="",
is_shared=True,
),
],
)
AIMessage(content='Cosine is a trigonometric function representing the adjacent side over the hypotenuse in a right triangle.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 23, 'prompt_tokens': 30, 'total_tokens': 53, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_703d4ff298', 'finish_reason': 'stop', 'logprobs': None}, id='run-ebd536b5-e599-4c68-b8b8-e011d43b1d0c-0', usage_metadata={'input_tokens': 30, 'output_tokens': 23, 'total_tokens': 53, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})
Example of Runnables Using Different Keys
This example demonstrates how to handle inputs and output messages with RunnableWithMessageHistory.
Messages Input with Dictionary Output
Direct Message Object Handling
Omitting input_messages_key="input" configures the system to accept Message objects as input.
from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableParallel
# Create a chain that outputs dictionary with message.
chain = RunnableParallel(
{"output_message": ChatOpenAI(model_name="gpt-4o", temperature=0)}
)
def get_session_history(session_id: str) -> BaseChatMessageHistory:
# Create new ChatMessageHistory if session doesn't exist.
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
# Create RunnableWithMessageHistory with message history capability.
with_message_history = RunnableWithMessageHistory(
chain,
get_session_history,
# input_messages_key="input" is omitted to accept Message objects
output_messages_key="output_message", # Specify output format as dictionary
)
# Execute chain with message input
with_message_history.invoke(
[HumanMessage(content="what is the definition of cosine?")],
config={"configurable": {"session_id": "abc123"}},
)
{'output_message': AIMessage(content='The cosine of an angle in a right-angled triangle is defined as the ratio of the length of the adjacent side to the length of the hypotenuse. In mathematical terms, for an angle \\( \\theta \\):\n\n\\[\n\\cos(\\theta) = \\frac{\\text{Adjacent side}}{\\text{Hypotenuse}}\n\\]\n\nIn the context of the unit circle, which is a circle with a radius of 1 centered at the origin of a coordinate plane, the cosine of an angle \\( \\theta \\) is the x-coordinate of the point where the terminal side of the angle intersects the circle. This definition extends the concept of cosine to all real numbers, not just angles in right triangles.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 144, 'prompt_tokens': 14, 'total_tokens': 158, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_b7d65f1a5b', 'finish_reason': 'stop', 'logprobs': None}, id='run-6e321281-a1c6-4ea9-a4d4-3b7752534c74-0', usage_metadata={'input_tokens': 14, 'output_tokens': 144, 'total_tokens': 158, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})}
# Continue conversation with same session ID
with_message_history.invoke(
[HumanMessage(content="Please translate the previous content to Korean!")],
config={"configurable": {"session_id": "abc123"}},
)
Continuing conversations seamlessly using session IDs.
Message Objects for both Input and Output
Continuing from the previous example, you can also configure RunnableWithMessageHistory to handle Message objects directly for both input and output.
Direct Message Object Handling
Omitting output_messages_key="output_message" configures the system to return Message objects as output.
with_message_history = RunnableWithMessageHistory(
ChatOpenAI(model_name="gpt-4o", temperature=0), # Use ChatOpenAI language model
get_session_history, # Function to retrieve conversation history
# input_messages_key="input", # Omit to accept Message objects as input
# output_messages_key="output_message" # Omit to return Message objects as output
)
# Invoke with Message object input
with_message_history.invoke(
[HumanMessage(content="What is the meaning of cosine?")],
config={"configurable": {"session_id": "def123"}},
)
AIMessage(content='The term "cosine" can refer to a few different concepts depending on the context, but it is most commonly associated with trigonometry in mathematics. Here are the primary meanings:\n\n1. **Trigonometric Function**: In trigonometry, the cosine of an angle in a right triangle is defined as the ratio of the length of the adjacent side to the length of the hypotenuse. If \\(\\theta\\) is an angle in a right triangle, then:\n \\[\n \\cos(\\theta) = \\frac{\\text{Adjacent side}}{\\text{Hypotenuse}}\n \\]\n The cosine function is one of the fundamental trigonometric functions, along with sine and tangent.\n\n2. **Unit Circle Definition**: In the context of the unit circle, which is a circle with a radius of 1 centered at the origin of a coordinate plane, the cosine of an angle \\(\\theta\\) is the x-coordinate of the point on the unit circle that is reached by moving counterclockwise from the positive x-axis by an angle \\(\\theta\\).\n\n3. **Cosine Function in Calculus**: The cosine function, often written as \\(\\cos(x)\\), is a periodic function with a period of \\(2\\pi\\). It is an even function, meaning \\(\\cos(-x) = \\cos(x)\\), and it is used extensively in calculus and analysis.\n\n4. **Cosine Similarity**: In the context of data analysis and machine learning, cosine similarity is a measure of similarity between two non-zero vectors. It is calculated as the cosine of the angle between the two vectors, which is the dot product of the vectors divided by the product of their magnitudes. It is often used in text analysis and information retrieval.\n\nThese are the primary contexts in which the term "cosine" is used, each with its own specific meaning and application.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 389, 'prompt_tokens': 14, 'total_tokens': 403, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_d28bcae782', 'finish_reason': 'stop', 'logprobs': None}, id='run-0ba9374f-a492-4770-b1df-a0bf740b57e0-0', usage_metadata={'input_tokens': 14, 'output_tokens': 389, 'total_tokens': 403, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})
Dictionary with Single Key for All Messages
Using a Single Key for Input/Output
This approach uses one key for both input and output messages.
It utilizes itemgetter("input_messages") to extract input messages from the dictionary.
from operator import itemgetter
with_message_history = RunnableWithMessageHistory(
itemgetter("input_messages")
| ChatOpenAI(model_name="gpt-4o", temperature=0), # Extract and process messages.
get_session_history, # Session history management
input_messages_key="input_messages", # Specify input message key
)
# Invoke with dictionary input
with_message_history.invoke(
{"input_messages": "What is the meaning of cosine?"},
config={"configurable": {"session_id": "xyz123"}},
)
AIMessage(content='The term "cosine" refers to a trigonometric function that is fundamental in mathematics, particularly in the study of triangles and periodic phenomena. In the context of a right-angled triangle, the cosine of an angle is defined as the ratio of the length of the adjacent side to the length of the hypotenuse. Mathematically, for an angle \\( \\theta \\), it is expressed as:\n\n\\[\n\\cos(\\theta) = \\frac{\\text{Adjacent side}}{\\text{Hypotenuse}}\n\\]\n\nIn the unit circle, which is a circle with a radius of one centered at the origin of a coordinate plane, the cosine of an angle \\( \\theta \\) is the x-coordinate of the point where the terminal side of the angle intersects the circle.\n\nCosine is also an even function, meaning that \\( \\cos(-\\theta) = \\cos(\\theta) \\), and it has a range of values from -1 to 1. It is periodic with a period of \\( 2\\pi \\), meaning that \\( \\cos(\\theta + 2\\pi) = \\cos(\\theta) \\).\n\nCosine is widely used in various fields such as physics, engineering, and computer science, particularly in wave analysis, signal processing, and the study of oscillations.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 266, 'prompt_tokens': 14, 'total_tokens': 280, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_d28bcae782', 'finish_reason': 'stop', 'logprobs': None}, id='run-7616dbf4-706f-4f2b-ba66-a32f863db461-0', usage_metadata={'input_tokens': 14, 'output_tokens': 266, 'total_tokens': 280, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})
This configuration enables:
Direct handling of Message objects.
Simplified input/output processing.
Flexible conversion between different message formats.
Consistent session management.
Understanding Persistent Storage
Persistent storage ensures that data is retained even after a program terminates or the system restarts . This is typically achieved using databases, file systems, or other non-volatile storage devices.
Persistent storage is essential for long-term data preservation in applications. It enables.:
State preservation across sessions.
User preference retention.
Continuous operation without data loss .
Recovery from previous execution points.
Implementation Options
RunnableWithMessageHistory offers flexible storage options that are independent of how get_session_history retrieves the chat message history.
Using Redis for Persistence
This section demonstrates how to use Redis for persistent message history storage.
Installation
%pip install -qU redis
Note: you may need to restart the kernel to use updated packages.
Redis Server Setup
Launch a local Redis Stack server using Docker:
docker run -d -p 6379:6379 -p 8001:8001 redis/redis-stack:latest
Check port availability (terminate any processes using the port or use different ports).
Redis Connection
Set up the Redis connection URL: "redis://localhost:{port}/0"
REDIS_URL = "redis://localhost:6379/0"
Implementing Redis Message History
To use Redis for message history, define a new callable that returns an instance of RedisChatMessageHistory :
from langchain_community.chat_message_histories.redis import RedisChatMessageHistory
def get_message_history(session_id: str) -> RedisChatMessageHistory:
# Return RedisChatMessageHistory instance based on session ID.
return RedisChatMessageHistory(session_id, url=REDIS_URL)
with_message_history = RunnableWithMessageHistory(
runnable, # Runnable object
get_message_history, # Message history retrieval
input_messages_key="input", # Key for input messages
history_messages_key="history", # Key for history messages
)
Testing Conversation Continuity
First Interaction
You can call the function / chain as before.
# Initial query with new session ID
with_message_history.invoke(
{"ability": "math", "input": "What does cosine mean?"},
config={"configurable": {"session_id": "redis123"}},
)
AIMessage(content='Cosine is the ratio of the length of the adjacent side to the hypotenuse in a right triangle.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 23, 'prompt_tokens': 214, 'total_tokens': 237, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_5f20662549', 'finish_reason': 'stop', 'logprobs': None}, id='run-08fb9a1f-b16d-4e19-9c2d-404052d3b111-0', usage_metadata={'input_tokens': 214, 'output_tokens': 23, 'total_tokens': 237, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})
Continuing the Conversation
Make the second call using the same session_id .
# Second query using same session ID
with_message_history.invoke(
{"ability": "math", "input": "Please translate the previous response to Korean"},
config={"configurable": {"session_id": "redis123"}},
)
We will ask the question using a different session_id for this time.
# Query with different session ID
with_message_history.invoke(
{"ability": "math", "input": "Please translate the previous response to Korean"},
config={"configurable": {"session_id": "redis456"}},
)
[Note] The last response will be inaccurate because there's no conversation history associated with that session ID redis456.
Setting up your environment is the first step. See the guide for more details.
Check out the for more details.
LangChain offers several implementations for managing message history. You can explore various memory integrations for persistent storage, as documented in the page.
It supports the local file system (see an example )
It integrates with various storage providers (see )
