🧠 Conversational Memory#

Memory Chat shows how conversational AI applications create the illusion of persistent memory on top of completely stateless LLM APIs. Because the model forgets everything the moment an HTTP response finishes, the application must capture every turn and replay the prior transcript on each new request.

This demo uses a stateless client-held memory architecture: the browser UI (main.js) maintains the array of past {role, content} messages and sends it alongside each new question. On the server, LangChain's MessagesPlaceholder("history") slots those historical messages directly between the fixed system persona and the latest human question, then returns the updated history list with the new user and assistant turn appended.

1. The Illusion of Memory in Stateless LLMs

LLMs do not have per-user session state inside their neural weights. When an assistant "remembers" that you said your name is Alice three turns ago, it is only because your earlier message ("user", "My name is Alice") was re-transmitted inside the messages array of the current API call.

2. Dynamic Prompt Assembly with MessagesPlaceholder

A multi-turn prompt has both fixed and variable parts. ChatPromptTemplate.from_messages() combines them cleanly:

• Fixed Head: ("system", "You are a friendly assistant...") always sits at index 0 so the model never loses its core instructions.
• Variable Middle: MessagesPlaceholder("history") expands a Python list of N past (role, content) tuples into N distinct chat messages in chronological order.
• Current Tail: ("human", "{question}") appends the user's new input at the very end.

3. Client-Held vs. Server-Held Session State

There are two ways to store chat history in web apps:

• Client-Held (this demo): The browser stores the transcript in a JavaScript array (memoryHistory in main.js) and sends it with every POST request. The backend requires zero database infrastructure and scales horizontally without sticky sessions, and "Clear Memory" simply resets the browser array.
• Server-Held (see the DeskBuddy Redis demo): The browser sends only a session_id, and the server loads/saves messages in Redis or Postgres. This reduces payload size and prevents clients from tampering with past messages.

4. The Quadratic Cost of Growing Context

Because the entire history is re-sent and re-tokenized on every turn, token usage grows linearly with each message (and cumulative cost across an N-turn chat grows quadratically, O(N2)). Production systems cap history using a sliding window, token budget trimmer, or periodic conversation summarization.

Core AI Module (src/python/02_langchain_memory_chat.py)

src/python/02_langchain_memory_chat.py implements multi-turn conversational memory over a stateless LLM using LangChain's ChatPromptTemplate and MessagesPlaceholder. On every turn, prior messages are spliced between the fixed system persona and the new human question, and the latest exchange is appended to the history list for subsequent turns.

Schemas, Prompt Chain, and Memory Function

• class ChatMessage(BaseModel): Represents a single turn in the conversation transcript with role: str ("user" or "assistant") and content: str.
• class MemoryChatRequest(BaseModel) & class MemoryChatResponse(BaseModel): Pydantic schemas defining the input payload (message: str = "My favorite color is navy blue." and history: list[ChatMessage] = []) and output payload (message: str, response: str, and updated history: list[ChatMessage]).
• Chain Setup and chat_with_memory(question: str, history: list[ChatMessage]) -> tuple[str, list[ChatMessage]]:
  1. Step ① — Initialize the ChatOpenAI Client: Configures model = ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), api_key=os.getenv("OPENAI_API_KEY") or "missing-key", base_url=os.getenv("OPENAI_BASE_URL") or None, timeout=15.0).
  2. Step ② — Build the Multi-Turn Prompt Template: Uses ChatPromptTemplate.from_messages([...]) to combine three ordered sections:
     • ("system", "You are a friendly assistant. Use the conversation history to stay consistent.")
     • MessagesPlaceholder("history") — a dynamic slot that expands a list of past (role, content) tuples into chronological chat messages.
     • ("human", "{question}") — the current user question at the tail of the prompt.
  3. Step ③ — Assemble the LCEL Pipeline: Connects chain = prompt | model using the LCEL pipe operator.
  4. Step ④ — Format Prior History Tuples: Inside chat_with_memory(), converts the incoming ChatMessage objects into LangChain-compatible tuples via lc_history = [(msg.role, msg.content) for msg in history].
  5. Step ⑤ — Invoke the Chain with Context: Calls answer = chain.invoke({"history": lc_history, "question": question}).content so the LLM sees the full conversation history before answering.
  6. Step ⑥ — Append the New Turn to History: Returns answer alongside a new list [*history, ChatMessage(role="user", content=question), ChatMessage(role="assistant", content=answer)] so the next turn retains both the user's question and the assistant's reply.

How the AI Components Connect End-to-End

1. chat_with_memory(question, history) receives the user's new question along with the list of prior ChatMessage turns.
2. The prior messages are converted into (role, content) tuples and injected into MessagesPlaceholder("history") between the system instruction and ("human", "{question}").
3. chain.invoke() sends the complete, ordered message sequence to ChatOpenAI (gpt-4o-mini), allowing the stateless model to resolve references to earlier turns.
4. The assistant's reply is extracted from .content and appended with the user's question as new ChatMessage entries, producing the updated history for the next turn.

LangChain Conversational Memory Chat Service.

"""LangChain Conversational Memory Chat Service.

This service demonstrates conversational memory using LangChain LCEL (LangChain Expression Language):
1. Loads API credentials and model configuration from environment variables (.env).
2. Maintains conversation context across multiple turns using ChatPromptTemplate with MessagesPlaceholder.
3. Uses Pydantic models (ChatMessage, MemoryChatRequest, MemoryChatResponse) for structured serialization.
4. Dynamically appends user questions and assistant answers to preserve context across turns.
"""

import os

from dotenv import find_dotenv, load_dotenv
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

# ---------------------------------------------------------------------------
# Configuration & Environment Setup
# ---------------------------------------------------------------------------
# Load API keys and model names from .env
load_dotenv(find_dotenv())


# ---------------------------------------------------------------------------
# Request & Response Schemas (Pydantic)
# ---------------------------------------------------------------------------
class ChatMessage(BaseModel):
    """Single chat turn in conversation history."""

    role: str
    content: str


class MemoryChatRequest(BaseModel):
    """Input payload for memory chat queries."""

    message: str = "My favorite color is navy blue."
    history: list[ChatMessage] = []


class MemoryChatResponse(BaseModel):
    """Output payload containing assistant response and updated history."""

    message: str
    response: str
    history: list[ChatMessage]


# ---------------------------------------------------------------------------
# Model & Prompt Chain Setup
# ---------------------------------------------------------------------------
# ① Initialize the ChatOpenAI client
#   (placeholder key avoids an import-time crash when OPENAI_API_KEY is unset)
model = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
    api_key=os.getenv("OPENAI_API_KEY") or "missing-key",
    base_url=os.getenv("OPENAI_BASE_URL") or None,
    timeout=15.0,
)

# ② Build prompt template injecting system instructions, chat history, and current question
#   MessagesPlaceholder("history") dynamically receives the list of past (role, content) tuples
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a friendly assistant. Use the conversation history to stay consistent."),
    MessagesPlaceholder("history"),
    ("human", "{question}"),
])

# ③ Assemble the LCEL pipeline (Prompt | Model)
chain = prompt | model


# ---------------------------------------------------------------------------
# Core Service Logic
# ---------------------------------------------------------------------------
def chat_with_memory(question: str, history: list[ChatMessage]) -> tuple[str, list[ChatMessage]]:
    """Execute a conversational chat turn with memory retention.

    Args:
        question: User message or query.
        history: Previous chat messages to provide context.

    Returns:
        Tuple of (assistant_response, updated_chat_history).
    """
    # ④ Convert the history sent by the browser into (role, content) tuples
    lc_history = [(msg.role, msg.content) for msg in history]

    # ⑤ Invoke the chain with accumulated history and the current user question
    answer = chain.invoke({"history": lc_history, "question": question}).content

    # ⑥ Append the current question and answer to preserve context for the next turn
    return answer, [
        *history,
        ChatMessage(role="user", content=question),
        ChatMessage(role="assistant", content=answer),
    ]

Flask Blueprint server and route handler (memory_chat_service) in src/python/app.py.

"""Flask server for 02-langchain-agents."""

import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_summarize = _load_service("01_langchain_summarizer.py", "langchain_summarizer")
mod_memory = _load_service("02_langchain_memory_chat.py", "langchain_memory_chat")
mod_agent = _load_service("03_agent_one_tool.py", "agent_one_tool")
mod_lc_tool = _load_service("04_langchain_one_tool.py", "langchain_one_tool")
mod_lc_multi = _load_service("05_langchain_multi_tools.py", "langchain_multi_tools")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/summarize", methods=["POST"])
def summarizer_service():
    # ① Parse input text from JSON payload and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    text = (data.get("text") or data.get("message") or "").strip()
    if not text:
        return jsonify({"error": "Text cannot be empty"}), 400
    try:
        # ② Run the LangChain LCEL summarizer chain in 01_langchain_summarizer.py
        summary = mod_summarize.summarize(text)
        return jsonify({"result": summary, "summary": summary})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/memory-chat", methods=["POST"])
def memory_chat_service():
    # ① Parse user message and prior conversation history from JSON payload
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    history = data.get("history") or []
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Convert history dicts to ChatMessage models and run chat_with_memory()
        chat_history = [
            mod_memory.ChatMessage(**m) if isinstance(m, dict) else m
            for m in history
        ]
        reply, updated_history = mod_memory.chat_with_memory(message, chat_history)
        return jsonify({
            "result": reply,
            "response": reply,
            "history": [m.model_dump() if hasattr(m, "model_dump") else m for m in updated_history],
        })
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/agent", methods=["POST"])
def agent_service():
    # ① Parse user query from JSON payload and reject empty messages
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Run the raw OpenAI tool-calling loop in 03_agent_one_tool.py
        reply = mod_agent.run_agent_with_tool(message)
        return jsonify({"result": reply, "response": reply})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/langchain-one-tool", methods=["POST"])
def langchain_one_tool_service():
    # ① Parse user query from JSON payload and reject empty messages
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Run the LangChain single-tool agent in 04_langchain_one_tool.py
        reply, tool_called = mod_lc_tool.run_langchain_one_tool(message)
        return jsonify({"result": reply, "response": reply, "tool_called": tool_called})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/multi-tool-agent", methods=["POST"])
def multi_tool_agent_service():
    # ① Parse user query from JSON payload and reject empty messages
    data = request.get_json(force=True, silent=True) or {}
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "Message cannot be empty"}), 400
    try:
        # ② Run the LangGraph ReAct multi-tool agent in 05_langchain_multi_tools.py
        reply, tool_called = mod_lc_multi.run_multi_tool_agent(message)
        return jsonify({"result": reply, "response": reply, "tool_called": tool_called})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Use these numbered takeaways to quickly revise the core AI and LangChain concepts demonstrated in the Memory Chat feature:

1. LLMs are Stateless by Default: Every HTTP call to an LLM API is independent; the model has no built-in memory of earlier requests unless the application explicitly re-sends prior messages in the prompt payload.
2. Conversational Memory Pattern: Chat memory works by storing each turn's HumanMessage and AIMessage in a session store and splicing them into the prompt before the latest user input on subsequent turns.
3. MessagesPlaceholder in Prompt Templates: Using MessagesPlaceholder(variable_name="history") inside ChatPromptTemplate.from_messages() reserves a typed list slot where a variable number of historical message objects can be dynamically inserted between the system prompt and the new user message.
4. RunnableWithMessageHistory Wrapper: Instead of manually loading and appending messages on every turn, RunnableWithMessageHistory wraps an LCEL pipeline to automatically load history before execution and persist the new user/assistant exchange afterward.
5. Session Isolation via configurable.session_id: Passing config={"configurable": {"session_id": session_id}} routes each request to its own isolated InMemoryChatMessageHistory bucket so concurrent users never leak context into each other's chats.
6. LCEL Composition (prompt | model | StrOutputParser()): The underlying pipeline remains a clean 3-stage declarative chain, proving that state management can be layered around a stateless chain without modifying its internal logic.
7. Turn Counting & Session Reset: Because each completed turn adds exactly two messages (one HumanMessage and one AIMessage), dividing len(history.messages) by 2 gives the exact turn count, while popping the key from _store cleanly resets the conversation.
8. In-Memory vs. Production Persistence: InMemoryChatMessageHistory lives inside the Python process RAM (ideal for local demos and single-worker prototypes), whereas multi-worker production deployments swap the factory function to back onto Redis, Postgres, or DynamoDB without changing the chain code.