Conversation memory and history

Message history, trimming and summarising long threads, per-user retrieval, and deciding what belongs outside the model's context.

Managing message history

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a support assistant. Use the conversation history when relevant."),
    MessagesPlaceholder("history"),
    ("human", "{input}"),
])

chain = prompt | model
store: dict[str, InMemoryChatMessageHistory] = {}

def get_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

chat = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
)

config = {"configurable": {"session_id": "user-42"}}
print(chat.invoke({"input": "My order has not arrived."}, config=config).content)
print(chat.invoke({"input": "It was meant to arrive on Tuesday."}, config=config).content)
print(len(store["user-42"].messages))      # 4: two turns

• InMemoryChatMessageHistory loses everything when the process restarts. Use a persistent store backed by Redis, Postgres or a table before you ship anything.
• The session id is a security boundary. Derive it from the authenticated user, never from a client-supplied string, or one user can read another's conversation.
• A conversation store grows without bound. Decide a retention policy at design time, not after the table is a hundred gigabytes.
• History is part of the prompt, so it costs tokens on every request. Trimming is a cost control mechanism as well as a context-window one.

Trimming and summarising

from langchain_core.messages import (
    SystemMessage, HumanMessage, AIMessage, trim_messages, RemoveMessage
)

def windowed(history, max_tokens=1500):
    return trim_messages(
        history,
        max_tokens=max_tokens,
        strategy="last",                 # keep the most recent messages
        token_counter=model,
        include_system=True,             # never drop the system instruction
        start_on="human",                # never begin with an assistant message
        allow_partial=False,
    )

def summarise_old(history, keep=6):
    """Replace the old middle of a thread with a summary, keeping the ends."""
    if len(history) <= keep + 2:
        return history, None
    old = history[1:-keep]
    text = "\n".join(f"{m.type}: {m.content}" for m in old)
    summary = model.invoke(
        "Summarise this support conversation in at most 5 bullet points, "
        "keeping customer identifiers and any decisions made:\n" + text
    )
    new = [history[0],
           SystemMessage(f"Summary of the earlier conversation:\n{summary.content}")] + history[-keep:]
    return new, [RemoveMessage(id=m.id) for m in old if m.id]

# a rolling summary is worth it once threads are long
trimmed = windowed(store["user-42"].messages)
print([m.type for m in trimmed])

• Token-aware trimming beats message-count trimming: a message count is not a token count, and one long pasted document can consume the whole window.
• Always keep the system message. Dropping it mid-conversation changes the assistant's behaviour and is a very confusing bug to diagnose.
• start_on="human" prevents a history that begins with an assistant turn, which some providers reject.
• The rolling summary must preserve identifiers, decisions and constraints. A summary that loses the order number turns a helpful assistant into a useless one.

What belongs outside the context

# per-user retrieval: never retrieve across tenants
def user_retriever(user_id: str, question: str, k: int = 6):
    return store.similarity_search(
        question,
        k=k,
        filter={"tenant_id": user_id},     # enforced in the query, not after
    )

def build_messages(user_id, question, history):
    profile = load_profile(user_id)         # a database call, not a memory
    system = (
        "You are a support assistant.\n"
        f"Plan: {profile['plan']}. Region: {profile['region']}.\n"
        "Only answer from the provided context. If the answer is absent, say so."
    )
    docs = user_retriever(user_id, question)
    context = "\n\n".join(d.page_content for d in docs)
    return [
        SystemMessage(system),
        *windowed(history),
        HumanMessage(f"Context:\n{context}\n\nQuestion: {question}"),
    ]

The test of a good memory design: restart the service, and the next request should behave identically. Anything held only in process memory, or only in a message list, fails that test.

FAQ

Related

Models, messages and providers Retrieval-augmented generation

Last refreshed 2026-09-18.