Keep Chat History

Why Chat History Must Be Stored

The chatbot core from Lesson 61 keeps the conversation in memory (bot.history). That is lost when the program stops, when a web server restarts, or when a request is handled by a different server process. To let users come back to a conversation, the history must be persisted: saved to durable storage and loaded again later.

Remember that Claude has no memory between requests (Lesson 21). The stored history is the conversation. If it is lost or corrupted, the conversation is lost or corrupted.

What to Store

Store Full Content, Not Just Text

User messages are usually strings. Assistant messages should be stored as their complete content blocks, as produced by final.to_dict()["content"], the format the core uses. Storing only the text works for simple chats, but it loses thinking blocks, tool calls, and citations, which some features require (Lesson 24).

The content is stored as JSON text in the database and decoded when loaded:

Choosing Storage

This lesson uses SQLite, because it requires no installation and uses the same SQL ideas as production databases. The design transfers directly to PostgreSQL or the Django ORM.

The Database Schema

Two tables: one row per conversation, and one row per message.

CREATE TABLE IF NOT EXISTS conversations (
    id         TEXT PRIMARY KEY,
    user_id    TEXT NOT NULL,
    title      TEXT NOT NULL DEFAULT 'New chat',
    created_at REAL NOT NULL,
    updated_at REAL NOT NULL
);

CREATE TABLE IF NOT EXISTS messages (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
    role            TEXT NOT NULL,
    content         TEXT NOT NULL,
    created_at      REAL NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_messages_conversation ON messages(conversation_id, id);

Design decisions:

• messages.id is an auto-incrementing integer. It increases with every insert, so ORDER BY id returns messages in exactly the order they were stored, even when two messages share the same timestamp.
• ON DELETE CASCADE deletes a conversation's messages automatically when the conversation is deleted. In SQLite, foreign-key enforcement must be switched on per connection with PRAGMA foreign_keys = ON.
• The index on (conversation_id, id) makes loading one conversation fast, even with millions of messages in the table.
• user_id and title on conversations prepare for sessions (Lesson 63).
• Timestamps use time.time() (seconds since 1970 as a floating-point number), which is simple to compare and sort.

Saving Turns, Not Whole Histories

There are two ways to save history:

The chatbot core returns exactly the new messages of each successful turn in Reply.new_messages (Lesson 61), so appending is straightforward.

Both messages of a turn are inserted in one transaction, so either both are stored or neither is. A half-stored turn, a user message without its answer, would corrupt the conversation.

Practical Example: A SQLite History Store

history_store.py

"""history_store.py: persist chat conversations and messages in SQLite."""

import json
import sqlite3
import time

SCHEMA = """
CREATE TABLE IF NOT EXISTS conversations (
    id         TEXT PRIMARY KEY,
    user_id    TEXT NOT NULL,
    title      TEXT NOT NULL DEFAULT 'New chat',
    created_at REAL NOT NULL,
    updated_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS messages (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
    role            TEXT NOT NULL,
    content         TEXT NOT NULL,
    created_at      REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messages_conversation ON messages(conversation_id, id);
"""


class HistoryStore:
    def __init__(self, path: str = "chat.db"):
        self.db = sqlite3.connect(path)
        self.db.execute("PRAGMA foreign_keys = ON")
        self.db.executescript(SCHEMA)

    def ensure_conversation(self, conversation_id: str, user_id: str) -> None:
        now = time.time()
        with self.db:
            self.db.execute(
                "INSERT OR IGNORE INTO conversations (id, user_id, created_at, updated_at) "
                "VALUES (?, ?, ?, ?)",
                (conversation_id, user_id, now, now),
            )

    def append(self, conversation_id: str, messages: list[dict]) -> None:
        now = time.time()
        rows = [
            (conversation_id, m["role"], json.dumps(m["content"], ensure_ascii=False), now)
            for m in messages
        ]
        with self.db:
            self.db.executemany(
                "INSERT INTO messages (conversation_id, role, content, created_at) VALUES (?, ?, ?, ?)",
                rows,
            )
            self.db.execute("UPDATE conversations SET updated_at = ? WHERE id = ?", (now, conversation_id))

    def load(self, conversation_id: str) -> list[dict]:
        rows = self.db.execute(
            "SELECT role, content FROM messages WHERE conversation_id = ? ORDER BY id",
            (conversation_id,),
        ).fetchall()
        return [{"role": role, "content": json.loads(content)} for role, content in rows]

Using It with the Chatbot Core

"""demo_history.py: a conversation that survives program restarts."""

from anthropic import Anthropic

from chatbot_core import BotConfig, ChatBot
from history_store import HistoryStore

CONVERSATION_ID = "demo-conversation"
USER_ID = "student-1"

store = HistoryStore("chat.db")
store.ensure_conversation(CONVERSATION_ID, USER_ID)

history = store.load(CONVERSATION_ID)
print(f"Loaded {len(history)} stored messages.")

bot = ChatBot(Anthropic(), BotConfig(system="You are a concise Python tutor."), history=history)

question = input("You: ")
reply = bot.send(question)

if reply.ok:
    store.append(CONVERSATION_ID, reply.new_messages)
    print(f"Claude: {reply.text}")
else:
    print(f"[Not saved: {reply.error}]")

Example: two separate runs of the program

$ python demo_history.py
Loaded 0 stored messages.
You: My name is Arjun and I'm learning Flask.
Claude: Nice to meet you, Arjun! Flask is a great lightweight framework to start with.

$ python demo_history.py
Loaded 2 stored messages.
You: What am I learning?
Claude: You're learning Flask, Arjun.

How It Works

1. HistoryStore.__init__ opens (or creates) the database file, turns on foreign-key enforcement, and creates the tables if they do not exist. CREATE ... IF NOT EXISTS makes this safe to run every time the program starts.
2. ensure_conversation() uses INSERT OR IGNORE, so calling it for an existing conversation does nothing. That avoids "already exists" errors.
3. with self.db: turns the block into a transaction: it commits if the block succeeds and rolls back if an exception occurs. In append(), both message rows and the updated_at change are saved together or not at all.
4. ? placeholders pass values to SQLite safely. Never build SQL by inserting values into the string with f-strings, which would open the door to SQL injection.
5. json.dumps(..., ensure_ascii=False) encodes content of any shape, strings or lists of blocks, and keeps non-English text readable in the database.
6. load() returns messages in insertion order, decoded back into exactly the format the API expects.
7. The demo loads the stored history into a new ChatBot, and after a successful turn appends reply.new_messages. The second run knows the user's name and topic only because the first run's turn was stored. Failed or refused turns are not stored.

Common Mistakes

Storing Only the Text of Assistant Replies

What happens: thinking blocks and tool calls are lost, and features that require them fail when the history is sent back.

Fix: store the complete content blocks as JSON.

Saving Half a Turn

What happens: a crash between saving the user message and the assistant message leaves an unanswered user message, followed later by another user message.

Fix: save both messages of a turn in one transaction.

Ordering by Timestamp Only

What happens: messages saved in the same instant come back in an unpredictable order.

Fix: order by an auto-incrementing ID.

Building SQL with f-Strings

What happens: user input can change the SQL statement (SQL injection).

Fix: always use parameter placeholders.

Storing Failed or Refused Turns

What happens: the stored conversation contains broken or unwanted exchanges.

Fix: store only turns whose Reply.ok is true.

Best Practices

• Persist every successful turn immediately, as one transaction.
• Store complete content as JSON, and decode it when loading.
• Order messages by an increasing ID.
• Index messages by conversation for fast loading.
• Use parameterized queries for all database access.
• Store owner IDs and timestamps from the start, for sessions and cleanup.
• Treat stored history as sensitive data: protect access, and define how long it is kept.

Exercise

1. Run demo_history.py three times with related questions, then open chat.db with the sqlite3 command-line tool and run SELECT role, substr(content, 1, 60) FROM messages;.
2. Add a count(conversation_id) method that returns the number of stored messages.
3. Add a delete_conversation(conversation_id) method, and verify that its messages are deleted too.
4. Write a Django model pair (Conversation and ChatMessage) that mirrors this schema, using a JSONField for content.

Step 3:

def delete_conversation(self, conversation_id: str) -> None:
    with self.db:
        self.db.execute("DELETE FROM conversations WHERE id = ?", (conversation_id,))

Because of ON DELETE CASCADE and PRAGMA foreign_keys = ON, the conversation's messages are deleted automatically. Check with SELECT COUNT(*) FROM messages WHERE conversation_id = ?.

Step 4:

from django.conf import settings
from django.db import models


class Conversation(models.Model):
    id = models.CharField(primary_key=True, max_length=64)
    owner = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    title = models.CharField(max_length=200, default="New chat")
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)


class ChatMessage(models.Model):
    conversation = models.ForeignKey(Conversation, on_delete=models.CASCADE, related_name="messages")
    role = models.CharField(max_length=20)
    content = models.JSONField()
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ["id"]

JSONField stores both strings and lists of blocks without manual encoding, and ordering = ["id"] returns messages in insertion order by default.