Build a Chatbot With Memory in an Afternoon

A chatbot is four things: somewhere to keep the conversation, a client that streams tokens out of a model, a server that joins the two, and a page. In Python’s standard library that is about two hundred lines, and the afternoon is spent almost entirely on the two parts a demo leaves out — keeping the history from growing until it costs more than the answers, and stopping when the spend hits a number.

What you are building

One process. A browser posts a message to /send; the server appends it to a SQLite conversation, calls the model with stream: true, and relays each token to the browser over server-sent events; when the stream finishes it appends the assistant turn and records the token usage against a daily budget.

Nothing here is framework-specific on purpose. The model call is the OpenAI-compatible POST /v1/chat/completions shape, which every gateway and most providers speak, so the same file works against a different model by changing one string. If you are calling a provider’s native API instead, the request and response fields differ and you should check their current reference — only this file changes.

File one: the store

Two tables and a counter. Messages are append-only — you never update a turn, because the whole conversation is the input to the next request and a mutated history is the hardest class of bug to see.

# store.py  (Python 3.11, stdlib only)
import sqlite3, time

DB = sqlite3.connect("chat.db", check_same_thread=False)
DB.executescript("""
CREATE TABLE IF NOT EXISTS conversation (
  id      INTEGER PRIMARY KEY,
  created REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS message (
  id      INTEGER PRIMARY KEY,
  conv_id INTEGER NOT NULL REFERENCES conversation(id),
  role    TEXT NOT NULL CHECK (role IN ('system','user','assistant')),
  content TEXT NOT NULL,
  created REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS message_conv ON message(conv_id, id);
CREATE TABLE IF NOT EXISTS spend (
  day           TEXT PRIMARY KEY,
  input_tokens  INTEGER NOT NULL DEFAULT 0,
  output_tokens INTEGER NOT NULL DEFAULT 0
);
""")

def new_conversation(system_prompt):
    cur = DB.execute("INSERT INTO conversation (created) VALUES (?)", (time.time(),))
    conv_id = cur.lastrowid
    add_message(conv_id, "system", system_prompt)
    DB.commit()
    return conv_id

def add_message(conv_id, role, content):
    DB.execute(
        "INSERT INTO message (conv_id, role, content, created) VALUES (?,?,?,?)",
        (conv_id, role, content, time.time()),
    )
    DB.commit()

def history(conv_id):
    rows = DB.execute(
        "SELECT role, content FROM message WHERE conv_id = ? ORDER BY id",
        (conv_id,),
    ).fetchall()
    return [{"role": r, "content": c} for r, c in rows]

check_same_thread=False is there because http.server’s threading handler serves each request on its own thread. SQLite tolerates that for one writer; if you outgrow one writer you have outgrown SQLite for this table, not this design.

File two: the streaming client

Server-sent events are lines. Each line that starts with data: carries one JSON object; the stream ends with the literal data: [DONE]. That is the entire protocol, and urllib.request gives you the lines for free because the response object is iterable.

# llm.py
import json, os, urllib.request

BASE  = os.environ["LLM_BASE_URL"]   # e.g. https://<your-gateway>/v1
KEY   = os.environ["LLM_API_KEY"]
MODEL = os.environ.get("LLM_MODEL", "gpt-4o-mini")

def stream_chat(messages, max_tokens=800, temperature=0.7):
    """Yield (kind, value) pairs: ('text', str) then ('usage', dict|None)."""
    body = json.dumps({
        "model": MODEL,
        "messages": messages,
        "max_tokens": max_tokens,
        "temperature": temperature,
        "stream": True,
    }).encode()
    req = urllib.request.Request(
        BASE + "/chat/completions",
        data=body,
        headers={
            "Authorization": "Bearer " + KEY,
            "Content-Type": "application/json",
        },
    )
    usage = None
    with urllib.request.urlopen(req, timeout=120) as resp:
        for raw in resp:
            line = raw.decode("utf-8").strip()
            if not line.startswith("data:"):
                continue
            payload = line[5:].strip()
            if payload == "[DONE]":
                break
            event = json.loads(payload)
            if event.get("usage"):
                usage = event["usage"]
            for choice in event.get("choices", []):
                piece = (choice.get("delta") or {}).get("content")
                if piece:
                    yield ("text", piece)
    yield ("usage", usage)

File three: the server

Two routes. POST /send stores the user turn and returns immediately with an id; GET /stream?conv=... is the EventSource the browser holds open. Splitting them means a dropped connection cannot lose the user’s message, which is the difference between an annoying refresh and a lost question.

# app.py  (sketch of the streaming handler)
class Handler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        url = urllib.parse.urlparse(self.path)
        if url.path != "/stream":
            return self.send_error(404)
        conv_id = int(urllib.parse.parse_qs(url.query)["conv"][0])

        if over_budget():
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.end_headers()
            self.wfile.write(b"event: blocked\ndata: daily cap reached\n\n")
            return

        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache")
        self.send_header("X-Accel-Buffering", "no")
        self.end_headers()

        answer = []
        for kind, value in stream_chat(compact(history(conv_id))):
            if kind == "text":
                answer.append(value)
                chunk = json.dumps({"t": value})
                self.wfile.write(("data: " + chunk + "\n\n").encode())
                self.wfile.flush()
            else:
                record_spend(history(conv_id), "".join(answer), value)
        add_message(conv_id, "assistant", "".join(answer))
        self.wfile.write(b"data: [DONE]\n\n")

The X-Accel-Buffering: no header matters the moment you put nginx in front. Without it, nginx buffers the response and the reader sees nothing for eight seconds and then the whole answer at once — the bug that makes people think streaming is not working when it is working perfectly two hops away. Every proxy in the path has to be told not to buffer, and that includes some CDNs.

History that does not grow without bound

A chat sends the entire conversation on every turn. Turn 40 pays for turns 1 to 39 again. Left alone, cost per message grows roughly linearly with conversation length and the thing eventually hits the context window and starts failing.

The fix is compaction: keep the system prompt, keep the last N turns verbatim, and replace everything before that with one summary message that the model itself wrote.

KEEP_RECENT = 12          # turns kept verbatim
COMPACT_AT  = 24          # compact once history exceeds this

def compact(messages):
    if len(messages) <= COMPACT_AT:
        return messages
    system, rest = messages[0], messages[1:]
    old, recent = rest[:-KEEP_RECENT], rest[-KEEP_RECENT:]
    transcript = "\n".join(m["role"] + ": " + m["content"] for m in old)
    summary = complete([
        {"role": "system",
         "content": "Summarise this conversation. Keep names, numbers, "
                    "decisions and anything the user asked to be remembered. "
                    "Drop pleasantries. Under 200 words."},
        {"role": "user", "content": transcript},
    ])
    return [system,
            {"role": "system", "content": "Earlier context: " + summary},
            *recent]

Two things people get wrong here. Compact before the request, not after, or the turn that triggers compaction is the expensive one. And write the summary back to the store as its own row rather than recomputing it every turn — otherwise you have added a second model call to every message, and summarising the same prefix repeatedly costs more than the tokens you saved.

The cost cap

A cap that only warns is not a cap. This one refuses. Record usage after every completion, check the day’s total before every request, and return a clear message rather than an error when it is reached.

# Prices per million tokens — YOUR model's posted prices.
IN_PER_M, OUT_PER_M = 0.15, 0.60
DAILY_CAP_USD = 2.00

def approx_tokens(text):
    "No tokenizer dependency: ~4 characters per token for English prose."
    return max(1, len(text) // 4)

def record_spend(sent, answer, usage):
    if usage:
        i, o = usage["prompt_tokens"], usage["completion_tokens"]
    else:
        i = sum(approx_tokens(m["content"]) for m in sent)
        o = approx_tokens(answer)
    day = time.strftime("%Y-%m-%d")
    DB.execute("INSERT INTO spend (day) VALUES (?) ON CONFLICT(day) DO NOTHING", (day,))
    DB.execute("UPDATE spend SET input_tokens = input_tokens + ?, "
               "output_tokens = output_tokens + ? WHERE day = ?", (i, o, day))
    DB.commit()

def over_budget():
    day = time.strftime("%Y-%m-%d")
    row = DB.execute("SELECT input_tokens, output_tokens FROM spend WHERE day = ?",
                     (day,)).fetchone()
    if not row:
        return False
    i, o = row
    return (i * IN_PER_M + o * OUT_PER_M) / 1_000_000 >= DAILY_CAP_USD

The four-characters-per-token approximation is wrong by roughly ten per cent on English prose and much worse on code, JSON or non-Latin scripts — a cap built on it should sit comfortably below the number you actually care about. The same sentence in Hindi or Thai can cost several times the tokens of its English original, so if your users are not writing English, reconcile against real usage numbers rather than trusting the estimate.

The two failures you will hit

• Nothing appears until the answer is finished. Almost always buffering, not the model. Test by curling your own endpoint — if tokens arrive one at a time there, the buffer is in a proxy, a CDN or the browser waiting for a first flush of about a kilobyte. Sending a comment line as soon as the stream opens defeats the last of those.
• The bot forgets something the user said four turns ago. Check whether compaction ran. A summary written by a model that was told to be brief will drop exactly the detail a user thinks is obviously important — their name, a date, a preference. The fix is in the summarisation prompt, not in the retrieval: name the categories that must survive, as the prompt above does.

When you want the bot to answer from documents rather than from the conversation, that is a different build — retrieval from scratch is the next file to write, and it plugs into exactly the messages array above.

What to log

The temptation is to log the answers, because the answers are what people ask about. In practice the answer is the least useful field: it is recoverable from the conversation, and it is the one thing a privacy review will object to storing. Log the numbers instead, one row per completed turn.

Report latency as percentiles rather than a mean. One cold start drags a mean past every request anyone experienced, and p50 with p95 tells you both what it usually feels like and how bad the tail is. Two numbers, one line of SQL, and it is the difference between a dashboard that reflects the product and one that reflects arithmetic.

The one text field worth keeping is stop_reason’s companion: the first hundred characters of any answer that was truncated. Truncated answers are the complaint people report as “it cut off”, and having the opening makes it obvious whether the model was mid-list, mid-code-block, or genuinely rambling — three different fixes, and only one of them is raising the limit.