Memory

Build a Chatbot: From API Call to Deployed Bot

Chapter 4 · Conversation State and Memory

In chapter 1 the bot's memory was a Python dictionary, which meant it forgot everything when the program stopped, grew without limit, and could not tell one person from another. A real chatbot needs memory with four properties: it survives a restart, it stays within what the model can read, it is kept per conversation and per person, and the user controls it. This chapter builds exactly that, and tests every behaviour with the fake provider.

What was run, and what was not

Everything here ran on Python 3.14.0 with SQLite from the standard library, driven by the FakeProvider; the project now has 65 passing tests. No real model was called, so this chapter shows what the bot sends, not how a real model uses it, and the "summaries" are scripted strings rather than model-written ones. The token comparison uses OpenAI's open-source tiktoken 0.14.0 as a real tokeniser to check my estimate; it is OpenAI's tokeniser, not Claude's or Grok's, so treat it as an illustration of how counts vary, not as the count for your model.

Four Kinds of Memory

Kind	What it holds	Lives	Belongs to
History	Every message in this conversation	SQLite table `messages`	a session
Window	The part of the history actually sent to the model this time	built per call, never stored	a session
Summary	A short account of the older history that no longer fits	SQLite table `summaries`	a session
Facts	Things the person asked the bot to remember about them	SQLite table `facts`	a user

The two owners matter. A session id names one conversation in one place, such as web:3f9a, telegram:12345 or terminal; history and summaries belong to it. A user id names a person, and facts follow the person across channels. If the caller does not give a user id, the session id is used, which is right for a one-to-one chat. (Group chats, where one session has many users, need more care; later chapters return to that.)

The Store: SQLite

SQLite ships with Python, needs no server, keeps everything in one file, and gives you transactions, which is the property we need most. Here is the whole store: the schema, and one method per thing the bot does with it.

import json
import sqlite3
import time

from bot.providers import Message

SCHEMA = """
CREATE TABLE IF NOT EXISTS messages (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id TEXT NOT NULL,
    role       TEXT NOT NULL CHECK (role IN ('user', 'assistant')),
    content    TEXT NOT NULL,
    created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS messages_by_session ON messages (session_id, id);

CREATE TABLE IF NOT EXISTS summaries (
    session_id TEXT PRIMARY KEY,
    text       TEXT NOT NULL,
    upto_id    INTEGER NOT NULL            -- the summary covers messages with id <= upto_id
);

CREATE TABLE IF NOT EXISTS facts (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id    TEXT NOT NULL,
    text       TEXT NOT NULL,
    created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS facts_by_user ON facts (user_id, id);
"""


class Store:
    """Everything the bot remembers, in one SQLite database (a file path, or ':memory:')."""

    def __init__(self, path: str = ":memory:") -> None:
        self.db = sqlite3.connect(path)
        self.db.executescript(SCHEMA)

    def close(self) -> None:
        self.db.close()

    # --- conversation history -------------------------------------------------------------
    def add_exchange(self, session_id: str, user_text: str, assistant_text: str) -> None:
        """Save a question and its answer together, so a failed call never leaves half an exchange."""
        now = time.time()
        with self.db:  # one transaction: both rows or neither
            self.db.execute("INSERT INTO messages (session_id, role, content, created_at) VALUES (?, 'user', ?, ?)",
                            (session_id, user_text, now))
            self.db.execute("INSERT INTO messages (session_id, role, content, created_at) VALUES (?, 'assistant', ?, ?)",
                            (session_id, assistant_text, now))

    def messages(self, session_id: str, after_id: int = 0) -> list[tuple[int, Message]]:
        rows = self.db.execute("SELECT id, role, content FROM messages WHERE session_id = ? AND id > ? ORDER BY id",
                               (session_id, after_id)).fetchall()
        return [(i, {"role": role, "content": content}) for i, role, content in rows]

    def clear_session(self, session_id: str) -> int:
        with self.db:
            n = self.db.execute("DELETE FROM messages WHERE session_id = ?", (session_id,)).rowcount
            self.db.execute("DELETE FROM summaries WHERE session_id = ?", (session_id,))
        return n

    # --- summaries ------------------------------------------------------------------------
    def get_summary(self, session_id: str) -> tuple[str, int] | None:
        row = self.db.execute("SELECT text, upto_id FROM summaries WHERE session_id = ?", (session_id,)).fetchone()
        return (row[0], row[1]) if row else None

    def set_summary(self, session_id: str, text: str, upto_id: int) -> None:
        with self.db:
            self.db.execute("INSERT INTO summaries (session_id, text, upto_id) VALUES (?, ?, ?) "
                            "ON CONFLICT(session_id) DO UPDATE SET text = excluded.text, upto_id = excluded.upto_id",
                            (session_id, text, upto_id))

    # --- facts the user asked us to remember ------------------------------------------------
    def add_fact(self, user_id: str, text: str) -> int:
        with self.db:
            return self.db.execute("INSERT INTO facts (user_id, text, created_at) VALUES (?, ?, ?)",
                                   (user_id, text, time.time())).lastrowid

    def facts(self, user_id: str) -> list[tuple[int, str]]:
        return self.db.execute("SELECT id, text FROM facts WHERE user_id = ? ORDER BY id", (user_id,)).fetchall()

    def forget(self, user_id: str, fact_id: int | None = None) -> int:
        with self.db:
            if fact_id is None:
                return self.db.execute("DELETE FROM facts WHERE user_id = ?", (user_id,)).rowcount
            return self.db.execute("DELETE FROM facts WHERE user_id = ? AND id = ?", (user_id, fact_id)).rowcount

    # --- export ---------------------------------------------------------------------------
    def export(self, session_id: str, user_id: str) -> str:
        summary = self.get_summary(session_id)
        return json.dumps({
            "session": session_id,
            "summary": summary[0] if summary else None,
            "messages": [m for _, m in self.messages(session_id)],
            "facts": [text for _, text in self.facts(user_id)],
        }, indent=2, ensure_ascii=False)

Two details worth noticing:

add_exchange saves the question and its answer together, in one transaction. The bot calls it only after the provider has answered. If the call fails, nothing is stored, so a rate-limited request cannot leave a dangling user message that would make the next request start with two user turns in a row. A test checks exactly this.
The queries are short and run on the event loop. At this scale (a bot talking to a handful of people) that is fine, and I did not measure it under heavy load; a busy bot would move the calls to a thread or use an async SQLite library.

The Window: What Fits in the Budget

A model can read only so much at once, and every token costs money and time, so the bot sends a window: the newest messages that fit in a budget. Counting tokens exactly needs the model's own tokeniser, which we do not have in a vendor-neutral bot, so the code estimates with the common rule of thumb of about four characters per token:

from typing import Callable

from bot.providers import Message

TokenCounter = Callable[[str], int]

PER_MESSAGE_OVERHEAD = 4  # a guess at the framing tokens each message costs; real counts differ by model


def estimate_tokens(text: str) -> int:
    """A rough token count: about four characters per token, a common rule of thumb for English.

    It is an estimate. Real tokenisers differ by model and by language, so keep a safety margin, and
    prefer the usage numbers a provider reports after the fact (Reply.usage) when you have them.
    """
    return max(1, (len(text) + 3) // 4)


def message_tokens(message: Message, count: TokenCounter = estimate_tokens) -> int:
    return count(message["content"]) + PER_MESSAGE_OVERHEAD


def fit_window(messages: list[Message], budget: int, count: TokenCounter = estimate_tokens) -> list[Message]:
    """The newest messages that fit in `budget` tokens, oldest dropped first.

    The result never starts with an assistant message (a window that begins mid-exchange would confuse
    the model), and the final message is always kept, because it is the one being answered.
    """
    kept: list[Message] = []
    used = 0
    for message in reversed(messages):
        cost = message_tokens(message, count)
        if kept and used + cost > budget:
            break
        kept.append(message)
        used += cost
    kept.reverse()
    while len(kept) > 1 and kept[0]["role"] != "user":
        kept.pop(0)
    return kept


def build_system(base: str | None, facts: list[str], summary: str | None) -> str | None:
    parts = [base] if base else []
    if facts:
        parts.append("Things the user asked you to remember:\n" + "\n".join(f"- {f}" for f in facts))
    if summary:
        parts.append("Summary of the earlier part of this conversation:\n" + summary)
    return "\n\n".join(parts) if parts else None

The window never starts with an assistant message and always keeps the final message, because that is the one being answered. A demonstration with a deliberately small budget, on a six-exchange conversation (each stored message estimated at 16 or 28 tokens because the fake provider's replies are long):

stored messages: 12 | tokens per stored message: [16, 28] budget 10000: sent 13 messages, first role user, first starts 'message number 0', last starts 'message number 6' budget 100: sent 3 messages, first role user, first starts 'message number 6', last starts 'message number 6' budget 60: sent 3 messages, first role user, first starts 'message number 6', last starts 'message number 6' budget 30: sent 1 messages, first role user, first starts 'message number 6', last starts 'message number 6'

As the budget shrinks, the oldest messages fall out first, and at the smallest budget only the question being asked is sent. Nothing is deleted from the store; the window is a view, built fresh each call.

How Good Is the Four-Characters Rule?

I checked it against a real tokeniser. These are real counts from tiktoken's o200k_base encoding next to the estimate:

text chars estimate o200k_base English prose 105 27 21 Python code 132 33 33 Hungarian 87 22 35 Japanese 31 8 20 Numbers/IDs 67 17 28

For plain English the estimate over-counts a little (27 against 21), which is safe. For other kinds of text it under-counts, sometimes badly: Hungarian came out at 35 tokens against an estimate of 22, Japanese at 20 against 8, and identifiers and numbers at 28 against 17. This matters for a learner of Hungarian and Japanese, and for any bot whose users write in other languages: a window that "fits" by the estimate may not fit in reality. Three defences, in increasing order of effort: leave a generous safety margin (budget well below the model's real limit); use the token counts the provider reports after each call (Reply.usage, from chapter 2) to learn how big your real conversations are; or ask the provider to count before sending (Anthropic's API has a token-counting endpoint, POST /v1/messages/count_tokens, listed in its API overview; for OpenAI-family models, tiktoken does it locally). Also remember that the budget must leave room for the system prompt and for the reply, not only the history. Exercise 3 turns the safety margin into code.

Summaries: Keeping the Gist of What Fell Out

Dropping old messages loses information the person may still expect the bot to know ("as I said at the start..."). The alternative is to summarise the oldest part and keep the summary in the system prompt. The algorithm in the bot is deliberately simple:

If the unsummarised history plus the new message is over the budget, pick the oldest whole exchanges until what remains is under half the budget (so we do not summarise on every single message).
Ask the provider to summarise them, giving it the earlier summary too, so the summary accumulates.
Store the summary with the id of the last message it covers; the next call sends only the messages after that id.
If summarising fails, carry on with plain truncation: the user's message must never fail because a housekeeping call did.

A run where the fake provider returns a scripted summary (a real model's summary would be different, and sometimes wrong):

1. the summarising call used system prompt: 'You condense conversations. Write a short, factu...' and was sent this text: | Conversation to summarise: | user: Hi, I am Ada and I prefer short answers. | assistant: a1 | user: I am planning a trip to Edinburgh. | assistant: a2 | user: Which month is best? | assistant: a3 2. stored summary: Ada is planning a trip to Edinburgh and prefers short answers. 3. the final call's system prompt: | Be brief. | | Summary of the earlier part of this conversation: | Ada is planning a trip to Edinburgh and prefers short answers. messages sent with it: [('user', 'Thanks.'), ('assistant', 'a4'), ('user', 'Remind me what we decided.')] 4. messages still in the store: 10

Read the output in order: the summariser saw the first six messages as plain text; its answer was stored; the final call carried the persona, the summary and only the three most recent messages; and the store still has all ten messages. Summaries cost an extra model call, are written by a model and can drop or distort details, so the full history stays in the database as the source of truth. Whether to summarise at all is a setting (summarise=False by default).

Facts: Memory the User Controls

Remembering things about a person across conversations is useful ("I prefer short answers") and also the most sensitive thing a bot does. The design here is the cautious one: the bot remembers only what the person explicitly asks it to, with a command, never by silently extracting facts from what they say. Everything it holds is visible and deletable:

Command	What it does	Reaches the model?
/remember <text>	Stores a fact for this user and answers "Remembered as #n."	no
/facts	Lists the facts with their numbers	no
/forget <n> or /forget all	Deletes one fact, or all of them	no
/reset	Clears this conversation's history and summary (facts stay)	no
/export	Returns everything stored for this session and user as JSON	no
/help	Lists the commands	no

Commands are handled by the bot itself before any model call; a test checks that none of them reaches the provider, and that an unknown slash word such as /shrug is treated as ordinary text. Stored facts are added to the system prompt under a heading, in order, before the summary. Here is a two-process run against one database file: the second process is a completely new start of the program, and it still has the conversation and the fact.

--- run 1 bot> (fake) I was sent 1 message(s); the last was: 'My name is Ada' bot> Remembered as #1. bot> (fake) I was sent 3 message(s); the last was: 'What is my name?' --- run 2 (a new process, same database file) bot> #1 I prefer short answers bot> (fake) I was sent 5 message(s); the last was: 'Do you remember me?' bot> { "session": "terminal", "summary": null, "messages": [ { "role": "user", "content": "My name is Ada" }, { "role": "assistant", "content": "(fake) I was sent 1 message(s); the last was: 'My name is Ada'" }, { "role": "user", "content": "What is my name?" }, { "role": "assistant", "content": "(fake) I was sent 3 message(s); the last was: 'What is my name?'" }, { "role": "user", "content": "Do you remember me?" }, { "role": "assistant", "content": "(fake) I was sent 5 message(s); the last was: 'Do you remember me?'" } ], "facts": [ "I prefer short answers" ] }

Notice that in the second run the bot was sent five messages for a one-line question: the whole earlier conversation was loaded back from the file. That is persistence working, and also a reminder that history has a cost. Chapter 6 returns to what you may keep, for how long, and what you must tell people.

Added after writing chapter 5: a flaw in how facts are used

Stored facts are the user's own text, and this chapter puts them into the system prompt, where the model gives them more authority than a normal message. A user could store a fact that reads like an instruction. Chapter 5 fixes this (facts are flattened to one line, limited to 200 characters and fenced as data) and explains why that lowers the risk without removing it. The code listings in this chapter show the original version, as it was when this chapter was written.

One Message at a Time, Per Session

Chat platforms can deliver two messages from the same person almost together. If both are handled at once, each reads the history before the other has written to it. I ran the bot both ways with two concurrent messages and a provider that takes 50 ms to answer:

with lock messages sent to the model per call: [1, 3] stored roles: ['user', 'assistant', 'user', 'assistant'] without lock messages sent to the model per call: [1, 1] stored roles: ['user', 'assistant', 'user', 'assistant']

With the per-session lock, the second call was sent three messages: it saw the first exchange. Without it, both calls were sent one message, so the second answer was written without knowing about the first. The stored order stays valid either way, since the question and answer are saved together; the damage is lost context. The lock is per session (a dictionary of asyncio.Locks), so different people never wait for each other.

The Bot, Rewritten

All of this lives in the rewritten Bot. It replaces chapter 1's in-memory dictionary version; the constructor now takes a store and the memory settings, and chapter 1's tests still pass unchanged.

import asyncio

from bot.context import build_system, fit_window, message_tokens
from bot.providers import Message, Provider, ProviderError
from bot.store import Store

SUMMARISE_SYSTEM = (
    "You condense conversations. Write a short, factual summary of the conversation you are given, "
    "keeping names, decisions and open questions. Reply with the summary only."
)

HELP = """Commands:
/remember <text>   ask me to remember something about you
/facts             list what I remember
/forget <number>   forget one thing (or /forget all)
/reset             clear this conversation
/export            show everything I have stored for you as JSON"""


async def one_shot(provider: Provider, text: str) -> str:
    """A bare model call: no memory, every question stands alone."""
    return (await provider.complete([{"role": "user", "content": text}])).text


class Bot:
    """A provider plus a memory: stored history per session, a token budget, summaries and user facts."""

    def __init__(self, provider: Provider, system: str | None = None, *, store: Store | None = None,
                 history_budget: int = 2000, summarise: bool = False, max_tokens: int = 1024) -> None:
        self.provider = provider
        self.system = system
        self.store = store or Store()
        self.history_budget = history_budget
        self.summarise = summarise
        self.max_tokens = max_tokens
        self._locks: dict[str, asyncio.Lock] = {}

    async def handle(self, session_id: str, text: str, *, user_id: str | None = None) -> str:
        user_id = user_id or session_id
        command = await self._command(session_id, user_id, text)
        if command is not None:
            return command
        lock = self._locks.setdefault(session_id, asyncio.Lock())
        async with lock:  # one message at a time per session, or two quick messages would see the same history
            return await self._answer(session_id, user_id, text)

    async def _answer(self, session_id: str, user_id: str, text: str) -> str:
        rows = self.store.messages(session_id)
        summary = self.store.get_summary(session_id)
        if summary:
            rows = [r for r in rows if r[0] > summary[1]]
        new: Message = {"role": "user", "content": text}
        if self.summarise:
            rows, summary = await self._maybe_summarise(session_id, rows, summary, new)
        window = fit_window([m for _, m in rows] + [new], self.history_budget)
        system = build_system(self.system, [f for _, f in self.store.facts(user_id)], summary[0] if summary else None)
        reply = await self.provider.complete(window, system=system, max_tokens=self.max_tokens)
        self.store.add_exchange(session_id, text, reply.text)  # only now: a failed call stores nothing
        return reply.text

    async def _maybe_summarise(self, session_id, rows, summary, new):
        """If the unsummarised history is over budget, fold its oldest exchanges into the summary."""
        total = sum(message_tokens(m) for _, m in rows) + message_tokens(new)
        if total <= self.history_budget:
            return rows, summary
        split, remaining = 0, total
        while remaining > self.history_budget // 2 and split + 2 <= len(rows):
            remaining -= message_tokens(rows[split][1]) + message_tokens(rows[split + 1][1])
            split += 2
        if split == 0:
            return rows, summary
        transcript = "\n".join(f"{m['role']}: {m['content']}" for _, m in rows[:split])
        prompt = (f"Earlier summary:\n{summary[0]}\n\n" if summary else "") + f"Conversation to summarise:\n{transcript}"
        try:
            reply = await self.provider.complete([{"role": "user", "content": prompt}],
                                                 system=SUMMARISE_SYSTEM, max_tokens=300)
        except ProviderError:
            return rows, summary  # could not summarise: fall back to plain truncation, never fail the user's message
        upto = rows[split - 1][0]
        self.store.set_summary(session_id, reply.text, upto)
        return rows[split:], (reply.text, upto)

    async def _command(self, session_id: str, user_id: str, text: str) -> str | None:
        word, _, rest = text.strip().partition(" ")
        rest = rest.strip()
        match word.lower():
            case "/help":
                return HELP
            case "/reset":
                return f"Conversation cleared ({self.store.clear_session(session_id)} messages)."
            case "/remember":
                if not rest:
                    return "Usage: /remember <something to remember>"
                return f"Remembered as #{self.store.add_fact(user_id, rest)}."
            case "/facts":
                facts = self.store.facts(user_id)
                return "\n".join(f"#{i} {t}" for i, t in facts) if facts else "I am not remembering anything about you."
            case "/forget":
                if rest.lower() == "all":
                    return f"Forgot {self.store.forget(user_id)} thing(s)."
                if rest.isdigit():
                    n = self.store.forget(user_id, int(rest))
                    return "Forgotten." if n else f"I have no fact #{rest}."
                return "Usage: /forget <number> or /forget all"
            case "/export":
                return self.store.export(session_id, user_id)
        return None

A mistake I made while building this

My first versions of the two summary tests failed, one with a RateLimited error I had planted for a later step. The cause was the order of the scripted replies: I built the long history with summarising switched on, so the bot started summarising during the set-up, and the summary calls used up the scripted replies that the real steps needed. The fix was to arrange the state with a generous budget, then lower the budget just before the step being tested (the _bot_with_history helper). The takeaway: when a test fails because a script ran out of replies, the code under test is probably making a call you did not expect, and a test that mixes set-up with the thing being tested is hard to read. Keep the two apart.

What This Chapter Leaves Out

Encryption and retention. The database is a plain file. Who can read it, how long you keep it, and what you tell users are chapter 6 topics.
Search over old conversations. Looking up relevant past messages by meaning (embeddings, retrieval) is covered in Advanced AI Workflows & the Claude API 8; this chapter keeps recent history and a summary.
Group chats. One session with many users needs rules about whose facts are used. Later channel chapters return to it.
Real summaries. The tests script them. With a real model, check that summaries keep names and decisions and do not invent anything.

Hands-On Exercises

Exercise 1

Prove persistence without the terminal: create a Store on a temporary file, send two messages through one Bot, close it, build a brand-new Store and Bot on the same file with a new provider, send a third message, and print what the new provider was sent. How many messages, and which?

📄 View solution

Exercise 2

Build a ten-exchange history, then for budgets of 10000, 100, 80, 60, 40, 20 and 5 print how many messages were sent and how many were dropped from the window. What is the smallest budget at which the bot still sends at least one earlier exchange along with the new question, and how would you choose a budget for a model with a 200,000-token limit?

📄 View solution

Exercise 3

Compare estimate_tokens with a real tokeniser (pip install tiktoken) on English, Hungarian, Japanese and number-heavy text of your own. Compute real/estimate for each, and write an estimate_tokens_safe that multiplies by a safety factor. Does one factor cover all your samples? What would you do instead?

📄 View solution

Chapter 4 Quick Reference

History and summaries belong to a session (channel:chat_id); facts belong to a user (user_id, defaulting to the session id)
Store = SQLite (file or :memory:): add_exchange (one transaction, after a successful reply), messages, get_summary/set_summary, add_fact/facts/forget, clear_session, export
A failed provider call stores nothing; a failed summary falls back to truncation
Window = newest messages that fit a token budget, never starting with an assistant message, always keeping the last; the store is never trimmed
estimate_tokens (about 4 characters per token) over-counts English a little and under-counts Hungarian, Japanese and IDs (measured with tiktoken's o200k_base: 22 vs 35, 8 vs 20, 17 vs 28); leave a margin, use Reply.usage, or count with the provider's tool
Summaries: when over budget, fold the oldest whole exchanges into a stored summary (up to half the budget remains); the summary goes in the system prompt
Facts only by explicit /remember; /facts, /forget n|all, /reset, /export, /help never reach the model
A per-session asyncio.Lock keeps two quick messages from reading the same history (without it the second call lost the first exchange: 1 message sent instead of 3)

Coming next

Build a Chatbot: From API Call to Deployed Bot 5 gives the bot a character: a persona that holds up over a long conversation, scope limits ("only talks about..."), what to do about prompt-injection and jailbreak attempts from users, graceful refusals, output checks, and scripted conversations that test for persona drift. System prompts as such are covered in Advanced AI Workflows & the Claude API 2; the chapter covers only the bot-specific side.