LangChain RunnableWithMessageHistory tutorial

This langchain runnablewithmessagehistory tutorial walks you through building a stateful conversational chain using LangChain Expression Language (LCEL) and the RunnableWithMessageHistory wrapper. You’ll start with a stateless prompt-model pipeline, then add per-session memory that survives across multiple turns without manual message threading.

Prerequisites

• Python 3.10 or newer
• langchain-core and langchain-openai installed
• An OpenAI API key, or credentials for any OpenAI-compatible endpoint

pip install "langchain-core==0.3.0" "langchain-openai==0.2.0"
export OPENAI_API_KEY="sk-your-key"

If you prefer a gateway that fronts many providers, keep your key for that instead. The code below stays identical.

Setting up the base chain

A conversational chain needs a prompt that reserves a slot for prior messages. Use MessagesPlaceholder for that slot.

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
base_chain = prompt | model

base_chain is stateless. Each invocation starts fresh; the model has no memory of previous calls.

Adding conversation history with RunnableWithMessageHistory

The wrapper needs three things: the runnable to wrap, a function that returns a history object per session, and the key names that map to your prompt. Implement a minimal in-memory history backend:

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables import RunnableWithMessageHistory

class InMemoryHistory(BaseChatMessageHistory):
    def __init__(self):
        self.messages = []
    def add_messages(self, messages):
        self.messages.extend(messages)
    def clear(self):
        self.messages = []

store: dict[str, BaseChatMessageHistory] = {}

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

with_history = RunnableWithMessageHistory(
    base_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

input_messages_key tells the wrapper which dict key holds the new human message. history_messages_key matches the MessagesPlaceholder name. Get these wrong and history silently disappears.

Running a multi-turn conversation

Pass a config dict with a session_id under configurable. The wrapper loads that session’s history, injects it, runs the chain, and persists the new human and AI messages.

config = {"configurable": {"session_id": "user-123"}}

resp1 = with_history.invoke({"input": "My name is Ada."}, config=config)
print(resp1.content)

Expected output (wording may vary):

Hello Ada, how can I help you today?

Now ask a follow-up that requires recall:

resp2 = with_history.invoke({"input": "What is my name?"}, config=config)
print(resp2.content)

Expected output:

Your name is Ada.

Without the wrapper, the second call would have no context and would guess or say it doesn’t know.

Inspecting the stored history

Debugging memory issues is easier when you look at what’s actually stored.

history = store["user-123"].messages
for m in history:
    print(f"{m.type}: {m.content}")

Output:

human: My name is Ada.
ai: Hello Ada, how can I help you today?
human: What is my name?
ai: Your name is Ada.

The wrapper automatically appended both sides of each exchange. You never manually call add_messages in the happy path.

How injection works under the hood

Before invocation, RunnableWithMessageHistory calls get_session_history(session_id), reads .messages, and passes them as the history prompt variable. After the runnable returns, it calls add_messages([human_message, ai_message]). If your backend is slow or remote, that round-trip happens on every call.

Customizing the history backend

In-memory storage dies with the process. For production, back it with Redis, Postgres, or a file. The contract is small:

from redis import Redis
from langchain_core.messages import BaseMessage

class RedisHistory(BaseChatMessageHistory):
    def __init__(self, session_id: str, redis_client: Redis):
        self.session_id = session_id
        self.redis = redis_client

    @property
    def messages(self) -> list[BaseMessage]:
        # deserialize from redis list
        ...

    def add_messages(self, messages: list[BaseMessage]) -> None:
        # serialize and rpush to self.session_id
        ...

    def clear(self) -> None:
        self.redis.delete(self.session_id)

Swap the store function to return RedisHistory and the rest of the langchain runnablewithmessagehistory tutorial code stays unchanged.

Swapping in an OpenAI-compatible gateway

If you route across multiple model providers, you can keep the exact same LangChain code by pointing ChatOpenAI at an OpenAI-compatible base URL. n4n.ai exposes one OpenAI-compatible endpoint that addresses 240+ models and forwards provider cache-control hints, so the RunnableWithMessageHistory layer is oblivious to the swap.

import os
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="anthropic/claude-3.5-sonnet",
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["N4N_API_KEY"],
)
base_chain = prompt | model
with_history = RunnableWithMessageHistory(
    base_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

The session history, prompt template, and invocation pattern are identical. You only changed the model construction.

Streaming with history

The wrapper supports the same interfaces as the inner runnable. Streaming works as expected:

for chunk in with_history.stream({"input": "Tell me a short joke."}, config=config):
    print(chunk.content, end="", flush=True)

History is still loaded and saved; you just consume tokens incrementally.

Common pitfalls

• Missing key mappings: If input_messages_key or history_messages_key don’t match your prompt, history is dropped without error.
• Sync backends in async code: Using invoke inside async def with a blocking Redis client will block the loop. Implement aget_messages/add_messages async variants or use the async-specific runnable.
• Session collisions: Never hardcode session_id. Derive it from the authenticated user or conversation UUID.
• Message ordering: MessagesPlaceholder must sit between system and human turns. Putting it after the human input breaks most chat models.

Why not the legacy ConversationChain

Older ConversationChain bundled memory and prompt into one object, making it hard to compose with LCEL, streaming, or custom routing. RunnableWithMessageHistory is a thin, composable wrapper that works with any Runnable, including those with multiple steps, tools, or fallbacks. For new code, it’s the correct primitive.

This langchain runnablewithmessagehistory tutorial gave you a runnable memory layer, a pluggable backend, and a drop-in path to multi-provider inference. Copy the snippets, replace the store with your own, and ship.