Build a local server for the trained model

Problem

When you need this. Another program should be able to call the trained generator without importing its Python modules. Add a local HTTP interface that validates inputs and returns the generated text and stop reason.

The series has trained a model, saved its best checkpoint and made generation fast. Until now every result came from calling Python functions directly. A model that other programs use sits behind an interface, and that interface has its own ways to fail: malformed requests, oversized bodies, wrong types, settings out of range. This build serves the model over HTTP and evaluates it through that interface.

The contract has two routes on 127.0.0.1. GET /health reports that the server is up and which checkpoint it loaded. POST /generate takes a JSON object with a prompt and optional max_new_tokens, temperature, top_k and seed, and returns the text, the chosen tokens and the stop reason. A body that is not a JSON object, has unknown fields, wrong types or values out of range gets status 400; a body over the size limit gets 413; a missing length gets 411; any other route gets 404.

Success has three checks, all recorded. The served text equals direct generation for every one of the 60 test openings. Every malformed request in the error list gets its documented status code. And the evaluation reports the served model against the series’ baselines on the test stories and on a fresh set of stories generated after training ended.

Concepts

An HTTP client sends a request to a server at a URL. The server reads the request and returns a response with a status code and a body. Here the body is JSON, a text format for objects, arrays and scalar values. The client sends generation settings in a JSON object; the server validates them before invoking the cached generator.

Three ideas are specific to this implementation:

Request contract: The fields a request may contain, their types and allowed values, checked before any model code runs. A request that breaks the contract gets an error that names the problem; it never reaches the model with a guessed default.
Equivalence check: A test that the served reply equals what calling the generator directly returns. It shows the server adds no change of its own, such as a lost seed or a different default.
Fresh evaluation set: Stories generated for this build with a new seed, 2027, after the model was trained and its checkpoint chosen. The test stories were scored in several earlier builds; a set made after those decisions gives an additional check on the same story task. Stories that already appear word for word in the corpus are removed first.

Architecture

Serving one request · validate, generate, reply

A local HTTP client sends a generation request. The server checks the route, body length and request fields before generating. At startup it loads the model checkpoint with its tokenizer and corpus identity. Successful generation returns JSON; invalid routes, lengths or fields return errors. The server handles one request at a time.

Start at evaluate_service.py, the HTTP client script, which sends requests with urllib. Everything inside the dashed boundary runs in one Python process. The handler first checks the route and the request length. It returns 404 for an unknown route, 411 for a missing Content-Length and 413 for a body over the size limit. parse_request() then checks the JSON body and its fields. A bad body, field or setting gets 400, and so does a ValueError raised by the generator. generate_cached() runs with the parameters from best.json. The server loaded and verified that checkpoint once, at start, on the dashed hand-off. The reply returns to the client on the path labeled 200 · JSON. The interface and generation results are inspected after running the server.

server.py holds parse_request(), which enforces the contract, a Service that loads the checkpoint once, and the request handler. It uses Python’s http.server, handles one request at a time and listens on 127.0.0.1 only. evaluate_service.py starts the server on an unused port, sends every evaluation request through HTTP with urllib, and writes eval-report.json and service-time.json.

Working code

Runtime: Python 3.10 or newer. Dependencies: Python standard library only.

Download and extract the complete lab using Download the complete lab above. Open build-a-server-for-your-trained-llm/; keep its Python modules, tests and fixture files together. The bundled README includes setup commands for your platform.

server.pySource · pythonDownload

"""Serve the trained one-block model over HTTP with the standard library.

  GET  /health    -> {"status": "ok", "checkpoint": {"step": ..., "dev_loss": ...}}
  POST /generate  -> {"text": ..., "tokens": [...], "stopped": ..., "settings": {...}}

Run: python3 server.py --port 8000 [--checkpoint best.json]   (listens on 127.0.0.1 only)
One request is handled at a time. There is no authentication, TLS or rate limit:
this is a local teaching server, not a public service.
"""
import argparse
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path

import checkpoint as ck
import kv_cache as kv
import train as t
from bpe import Tokenizer

HERE = Path(__file__).parent
MAX_BODY_BYTES = 4096
MAX_NEW_TOKENS = 64
FIELDS = {"prompt": str, "max_new_tokens": int, "temperature": (int, float), "top_k": int, "seed": int}


class BadRequest(ValueError):
    pass


def parse_request(body):
    """Validate a /generate body and return keyword arguments for generate_cached()."""
    try:
        data = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError):
        raise BadRequest("body must be a JSON object") from None
    if not isinstance(data, dict):
        raise BadRequest("body must be a JSON object")
    unknown = sorted(set(data) - set(FIELDS))
    if unknown:
        raise BadRequest(f"unknown fields: {unknown}")
    if "prompt" not in data:
        raise BadRequest("missing field: prompt")
    for name, kind in FIELDS.items():
        value = data.get(name)
        if name in data and not (name == "top_k" and value is None) and (isinstance(value, bool) or not isinstance(value, kind)):
            raise BadRequest(f"field {name} has the wrong type")
    max_new = data.get("max_new_tokens", MAX_NEW_TOKENS)
    if not 1 <= max_new <= MAX_NEW_TOKENS:
        raise BadRequest(f"max_new_tokens must be from 1 to {MAX_NEW_TOKENS}")
    return {"prompt": data["prompt"], "max_new_tokens": max_new, "temperature": data.get("temperature", 0.0),
            "top_k": data.get("top_k"), "seed": data.get("seed", 0)}


class Service:
    """Loads the verified checkpoint once and answers requests from it."""

    def __init__(self, checkpoint_path=HERE / "best.json"):
        self.tok = Tokenizer.load(HERE / "tokenizer.json")
        best = ck.load(checkpoint_path, t.CONFIG, ck.input_signature(self.tok, (HERE / "corpus.txt").read_text()))
        self.params = best["params"]
        self.checkpoint = {"step": best["step"], "dev_loss": best["dev_loss"]}

    def generate(self, body):
        settings = parse_request(body)
        result = kv.generate_cached(self.params, self.tok, **settings)
        return {"text": result["text"], "tokens": result["steps"], "stopped": result["stopped"],
                "settings": settings, "checkpoint": self.checkpoint}


def make_handler(service):
    class Handler(BaseHTTPRequestHandler):
        def send_json(self, status, payload):
            data = json.dumps(payload).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)

        def do_GET(self):
            if self.path == "/health":
                self.send_json(200, {"status": "ok", "checkpoint": service.checkpoint})
            else:
                self.send_json(404, {"error": f"no route {self.path}"})

        def do_POST(self):
            if self.path != "/generate":
                return self.send_json(404, {"error": f"no route {self.path}"})
            try:
                length = int(self.headers.get("Content-Length", ""))
            except ValueError:
                return self.send_json(411, {"error": "Content-Length is required"})
            if length < 0:
                return self.send_json(400, {"error": "Content-Length must not be negative"})
            if length > MAX_BODY_BYTES:
                return self.send_json(413, {"error": f"body larger than {MAX_BODY_BYTES} bytes"})
            body = self.rfile.read(length)
            try:
                self.send_json(200, service.generate(body))
            except ValueError as error:  # BadRequest and the generator's own checks
                self.send_json(400, {"error": str(error)})

        def log_message(self, *args):  # keep test output quiet; a real service would log requests
            pass

    return Handler


def make_server(port=0, service=None):
    """Create the server bound to 127.0.0.1 only; both start() and the command line use it."""
    return HTTPServer(("127.0.0.1", port), make_handler(service or Service()))


def start(port=0, service=None):
    """Start the server on 127.0.0.1 in a background thread and return it."""
    server = make_server(port, service)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--port", type=int, default=8000)
    parser.add_argument("--checkpoint", default=str(HERE / "best.json"))
    args = parser.parse_args()
    server = make_server(args.port, Service(Path(args.checkpoint)))
    host, port = server.server_address
    print(f"serving on http://{host}:{port}", flush=True)
    server.serve_forever()

parse_request() rejects booleans for number fields explicitly, because in Python True is also an integer and would otherwise pass as max_new_tokens=1. Explicit JSON null is accepted only for top_k, where it means no filter. A null prompt, token limit, temperature or seed gets 400. test_null_fields_obey_the_request_contract checks these cases through the HTTP server. Errors raised by the generator itself, such as an empty prompt, become 400 replies too. The handler reads the body only after checking its declared length.

Start the server, then send a request from a second terminal:

Terminal commandsTerminal · bash

python3 server.py --port 8000
curl -s -X POST http://127.0.0.1:8000/generate -d '{"prompt": "mia has a red ball."}'

Read evaluate_service.py

evaluate_service.pySource · pythonDownload

"""Evaluate the served model through its HTTP interface and record the results.

Run: python3 evaluate_service.py          (print the report; under a minute on a laptop CPU)
     python3 evaluate_service.py --write  (rewrite eval-report.json and service-time.json)
     python3 evaluate_service.py --check  (exit 1 if eval-report.json differs)
service-time.json holds request latencies from the --write run and is not compared.
"""
import json
import platform
import statistics
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

import generate as g
import make_corpus
import kv_cache as kv
import server
import train as t

HERE = Path(__file__).parent
FRESH_SEED, FRESH_STORIES = 2027, 60  # a second set, generated for this build after training ended


def story_checks(base, openings):
    checks = {"well_formed": 0, "keeps_color_and_object": 0}
    for opening in openings:
        code, reply = call(base, "POST", "/generate", {"prompt": opening})
        if code != 200:
            raise RuntimeError(f"service returned {code} for {opening!r}: {reply}")
        result = g.follows_story(opening, reply["text"])
        checks = {k: checks[k] + result[k] for k in checks}
    return checks


def call(base, method, path, body=None, raw=None):
    data = raw if raw is not None else (None if body is None else json.dumps(body).encode())
    request = urllib.request.Request(base + path, data=data, method=method,
                                     headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return response.status, json.loads(response.read())
    except urllib.error.HTTPError as error:
        return error.code, json.loads(error.read())


ERROR_CASES = [
    ("not JSON", "POST", "/generate", None, b"mia has"),
    ("missing prompt", "POST", "/generate", {"temperature": 1.0}, None),
    ("unknown field", "POST", "/generate", {"prompt": "mia", "top_p": 0.9}, None),
    ("temperature as text", "POST", "/generate", {"prompt": "mia", "temperature": "hot"}, None),
    ("negative temperature", "POST", "/generate", {"prompt": "mia", "temperature": -1}, None),
    ("max_new_tokens too large", "POST", "/generate", {"prompt": "mia", "max_new_tokens": 500}, None),
    ("empty prompt", "POST", "/generate", {"prompt": ""}, None),
    ("body over the size limit", "POST", "/generate", None, json.dumps({"prompt": "mia " * 2000}).encode()),
    ("unknown route", "GET", "/models", None, None),
]


def run():
    service = server.Service()
    srv = server.start(service=service)
    base = f"http://127.0.0.1:{srv.server_address[1]}"
    try:
        _, _, test = t.split_stories((HERE / "corpus.txt").read_text())
        openings = [s.split(". ")[0] + "." for s in test]
        status, health = call(base, "GET", "/health")
        matches, checks, latencies = 0, {"well_formed": 0, "keeps_color_and_object": 0}, []
        for opening in openings:
            start = time.perf_counter()
            code, reply = call(base, "POST", "/generate", {"prompt": opening})
            latencies.append(time.perf_counter() - start)
            if code != 200:
                raise RuntimeError(f"service returned {code} for {opening!r}: {reply}")
            direct = kv.generate_cached(service.params, service.tok, opening)
            matches += reply["text"] == direct["text"] and reply["stopped"] == direct["stopped"]
            result = g.follows_story(opening, reply["text"])
            checks = {k: checks[k] + result[k] for k in checks}
        errors = []
        for name, method, path, body, raw in ERROR_CASES:
            code, reply = call(base, method, path, body, raw)
            errors.append({"case": name, "status": code, "error": reply.get("error")})
        corpus_text = (HERE / "corpus.txt").read_text()
        known = set(corpus_text.split("\n"))
        fresh_all = [s + "\n" for s in make_corpus.corpus(FRESH_SEED, FRESH_STORIES).split("\n") if s]
        fresh = [s for s in fresh_all if s.rstrip("\n") not in known]
        fresh_openings = [s.split(". ")[0] + "." for s in fresh]
        fresh_encoded = t.encode_all(service.tok, fresh, t.CONFIG["max_len"])
        te = t.encode_all(service.tok, test, t.CONFIG["max_len"])
        report = {
            "health": {"status": status, **health},
            "test_stories": len(openings),
            "served_equals_direct": matches,
            "greedy_story_checks": checks,
            "test_loss": round(t.mean_loss(service.params, te), 4),
            "fresh_set": {"seed": FRESH_SEED, "generated": len(fresh_all), "duplicates_removed": len(fresh_all) - len(fresh),
                          "stories": len(fresh), "test_loss": round(t.mean_loss(service.params, fresh_encoded), 4),
                          "greedy_story_checks": story_checks(base, fresh_openings),
                          "color_recall": t.color_recall(service.params, service.tok, fresh)},
            "color_recall_test": t.color_recall(service.params, service.tok, test),
            "error_cases": errors,
        }
        timing = {"requests": len(latencies), "median_ms": round(1000 * statistics.median(latencies), 1),
                  "max_ms": round(1000 * max(latencies), 1), "python": platform.python_version(),
                  "machine": platform.machine(), "system": platform.system()}
        return report, timing
    finally:
        srv.shutdown()
        srv.server_close()


if __name__ == "__main__":
    report, timing = run()
    text = json.dumps(report, indent=1) + "\n"
    if "--write" in sys.argv:
        (HERE / "eval-report.json").write_text(text)
        (HERE / "service-time.json").write_text(json.dumps(timing, indent=1) + "\n")
    elif "--check" in sys.argv:
        sys.exit(0 if (HERE / "eval-report.json").read_text() == text else "eval-report.json is out of date")
    else:
        print(text, end="")

Run the evaluation, which starts and stops its own server:

Terminal commandsTerminal · bash

python3 evaluate_service.py

python3 evaluate_service.py --check exits with an error if your report differs from the recorded eval-report.json.

Check the interface

The health route reports the checkpoint from step 450. For all 60 test openings, the served text and stop reason equal direct generation in 60 of 60. Each error case gets its documented status:

Request	Status	Error message
not JSON	400	`body must be a JSON object`
missing prompt	400	`missing field: prompt`
unknown field	400	`unknown fields: ['top_p']`
temperature as text	400	`field temperature has the wrong type`
negative temperature	400	`temperature must be 0 or a finite number >= 0.01, got -1`
max_new_tokens too large	400	`max_new_tokens must be from 1 to 64`
empty prompt	400	`empty prompt: the model needs at least one token to continue`
body over the size limit	413	`body larger than 4096 bytes`
unknown route	404	`no route /models`

Measured on one machine; your numbers will differ. The 60 generation requests took a median of 8.2 ms and at most 12.8 ms each, including HTTP overhead. That is one run, not a load test: the server handles one request at a time.

Evaluate against the baselines

Every row below uses the same tokenizer and the same 60 test stories. The loss is mean next-token cross-entropy; the color column counts stories where the model’s top next token after “now ⟨name⟩ has the” is the first sentence’s color.

Model	Test loss	Color recalled
Uniform over 96 tokens	4.5643	not scored
Bigram	1.1702	0 of 60
Always the most common training color	not a language model	18 of 60
One-block transformer, final step	0.2694	39 of 60
One-block transformer, best checkpoint (served)	0.2628	36 of 60

Through the server, greedy generation from the test openings gives 57 of 60 well-formed stories and keeps the color and object in 35.

Check your understanding

The fresh set started as 60 stories generated with seed 2027; 8 were exact copies of corpus stories and were removed, leaving 52. Before you open the comparison, predict whether the served model scores better, the same or worse on them than on the test stories.

Compare your prediction

Measure	Test stories (60)	Fresh stories (52)
Loss	0.2628	0.3172
Color recalled	36 of 60	25 of 52
Well-formed greedy stories	57 of 60	43 of 52
Color and object kept	35 of 60	25 of 52

Every measure is worse on the fresh stories, counted as a share of the stories in each set. Both sets come from the same generator. The original test split contains exact training-text duplicates and was scored in earlier lessons; its tokenizer also saw the test text. The fresh set excludes every exact corpus match, so it addresses that overlap, while still testing the same narrow synthetic pattern.

Read the explanation

Small held-out sets give noisy estimates. Report every set you evaluated on, say how and when each was made and how often it was scored, and do not keep only the better number.

Inspect the recorded report

eval-report.jsonRecorded output · jsonDownload

{
 "health": {
  "status": "ok",
  "checkpoint": {
   "step": 450,
   "dev_loss": 0.2941
  }
 },
 "test_stories": 60,
 "served_equals_direct": 60,
 "greedy_story_checks": {
  "well_formed": 57,
  "keeps_color_and_object": 35
 },
 "test_loss": 0.2628,
 "fresh_set": {
  "seed": 2027,
  "generated": 60,
  "duplicates_removed": 8,
  "stories": 52,
  "test_loss": 0.3172,
  "greedy_story_checks": {
   "well_formed": 43,
   "keeps_color_and_object": 25
  },
  "color_recall": {
   "correct": 25,
   "total": 52
  }
 },
 "color_recall_test": {
  "correct": 36,
  "total": 60
 },
 "error_cases": [
  {
   "case": "not JSON",
   "status": 400,
   "error": "body must be a JSON object"
  },
  {
   "case": "missing prompt",
   "status": 400,
   "error": "missing field: prompt"
  },
  {
   "case": "unknown field",
   "status": 400,
   "error": "unknown fields: ['top_p']"
  },
  {
   "case": "temperature as text",
   "status": 400,
   "error": "field temperature has the wrong type"
  },
  {
   "case": "negative temperature",
   "status": 400,
   "error": "temperature must be 0 or a finite number >= 0.01, got -1"
  },
  {
   "case": "max_new_tokens too large",
   "status": 400,
   "error": "max_new_tokens must be from 1 to 64"
  },
  {
   "case": "empty prompt",
   "status": 400,
   "error": "empty prompt: the model needs at least one token to continue"
  },
  {
   "case": "body over the size limit",
   "status": 413,
   "error": "body larger than 4096 bytes"
  },
  {
   "case": "unknown route",
   "status": 404,
   "error": "no route /models"
  }
 ]
}

Failure cases

Each case has a named test in test_server.py.

Omitting a field can select its default. Supplying null is a different request: the server rejects it for non-nullable fields before inference. The optional top_k field accepts null to disable that filter, as its request contract allows.

Case	Result with this implementation	What it tells you
A malformed or out-of-range request	Status 400 with a message naming the problem	The contract is checked before the model runs
A boolean where a number is expected	Status 400	`True` would otherwise pass as the integer 1
Null in a non-nullable request field	Status 400	Null does not select a default
A body over the size limit	Status 413	Large bodies are refused before they are read
A request without a length	Status 411	The server never reads an unbounded body
A negative Content-Length	Status 400	`rfile.read(-1)` would wait for the client to close and block every later request
An unknown route	Status 404	Only the two documented routes exist
Binding to every network interface	The tests fail unless both `start()` and the command line listen on 127.0.0.1	This server has no authentication and must stay local

The complete lab includes these tests and their fixtures. Run the 11 standard-library tests. Each starts a real server on an unused local port:

Terminal commandsTerminal · bash

python3 -m unittest -v test_server.py

Read the tests

test_server.pySource · pythonDownload

"""Tests for server.py. Run: python3 -m unittest -v test_server.py

Each test talks to a real server on an unused port of 127.0.0.1.
"""
import http.client
import json
import subprocess
import sys
import unittest
from pathlib import Path

import evaluate_service as ev
import kv_cache as kv
import server

HERE = Path(__file__).parent


class ServerTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.service = server.Service()
        cls.srv = server.start(service=cls.service)
        cls.port = cls.srv.server_address[1]
        cls.base = f"http://127.0.0.1:{cls.port}"

    @classmethod
    def tearDownClass(cls):
        cls.srv.shutdown()
        cls.srv.server_close()

    def post(self, body):
        return ev.call(self.base, "POST", "/generate", body)

    def test_listens_on_loopback_only(self):
        self.assertEqual(self.srv.server_address[0], "127.0.0.1")

    def test_health_names_the_checkpoint(self):
        status, reply = ev.call(self.base, "GET", "/health")
        self.assertEqual((status, reply["status"]), (200, "ok"))
        self.assertEqual(reply["checkpoint"], self.service.checkpoint)

    def test_served_text_equals_direct_generation(self):
        for body in ({"prompt": "mia has a red ball."},
                     {"prompt": "zoe has a blue kite.", "temperature": 1.0, "top_k": 3, "seed": 4}):
            with self.subTest(body=body):
                status, reply = self.post(body)
                direct = kv.generate_cached(self.service.params, self.service.tok, **server.parse_request(json.dumps(body)))
                self.assertEqual(status, 200)
                self.assertEqual((reply["text"], reply["stopped"]), (direct["text"], direct["stopped"]))

    def test_bad_requests_get_400(self):
        cases = [(None, b"mia has"), ({"temperature": 1.0}, None), ({"prompt": "mia", "top_p": 0.9}, None),
                 ({"prompt": "mia", "temperature": True}, None), ({"prompt": "mia", "max_new_tokens": True}, None),
                 ({"prompt": "mia", "max_new_tokens": 0}, None),
                 ({"prompt": "mia", "temperature": -1}, None), ({"prompt": ""}, None), ([], None)]
        for body, raw in cases:
            with self.subTest(body=body, raw=raw):
                status, reply = ev.call(self.base, "POST", "/generate", body, raw)
                self.assertEqual(status, 400)
                self.assertTrue(reply["error"])

    def test_null_fields_obey_the_request_contract(self):
        for field in ("prompt", "max_new_tokens", "temperature", "seed"):
            body = {"prompt": "mia has a red ball.", field: None}
            with self.subTest(field=field):
                status, reply = self.post(body)
                self.assertEqual(status, 400)
                self.assertIn("wrong type", reply["error"])
        status, _ = self.post({"prompt": "mia has a red ball.", "top_k": None})
        self.assertEqual(status, 200)  # null means no top-k filter, as omission does

    def test_body_over_the_limit_gets_413(self):
        status, _ = ev.call(self.base, "POST", "/generate", raw=b"x" * (server.MAX_BODY_BYTES + 1))
        self.assertEqual(status, 413)

    def test_missing_length_gets_411(self):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
        conn.putrequest("POST", "/generate")
        conn.endheaders()
        self.assertEqual(conn.getresponse().status, 411)
        conn.close()

    def test_negative_length_gets_400(self):
        # rfile.read(-1) would wait for the client to close, blocking every later request.
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
        conn.putrequest("POST", "/generate")
        conn.putheader("Content-Length", "-1")
        conn.endheaders()
        self.assertEqual(conn.getresponse().status, 400)
        conn.close()
        status, _ = ev.call(self.base, "GET", "/health")
        self.assertEqual(status, 200)

    def test_command_line_server_listens_on_loopback_only(self):
        proc = subprocess.Popen([sys.executable, "server.py", "--port", "0"], cwd=HERE,
                                stdout=subprocess.PIPE, text=True)
        try:
            line = proc.stdout.readline().strip()
            self.assertRegex(line, r"^serving on http://127\.0\.0\.1:\d+$")
            status, reply = ev.call(line.split(" on ")[1], "GET", "/health")
            self.assertEqual((status, reply["status"]), (200, "ok"))
        finally:
            proc.terminate()
            proc.wait(timeout=10)
            proc.stdout.close()

    def test_unknown_routes_get_404(self):
        for method, path in (("GET", "/models"), ("POST", "/train")):
            with self.subTest(path=path):
                status, _ = ev.call(self.base, method, path, {"prompt": "mia"} if method == "POST" else None)
                self.assertEqual(status, 404)


class Recorded(unittest.TestCase):
    def test_recorded_report(self):
        result = subprocess.run([sys.executable, "evaluate_service.py", "--check"], cwd=HERE)
        self.assertEqual(result.returncode, 0)


if __name__ == "__main__":
    unittest.main()

test_served_text_equals_direct_generation sends a greedy and a sampled request and compares each reply with generate_cached() called directly with the same parsed settings.

Deployment

The deployable unit is the folder: server.py, the copied modules, tokenizer.json and best.json. Start it with python3 server.py --port 8000, check GET /health reports the expected checkpoint step, then send one known prompt and compare the reply with the recorded report. To roll back, restart it with --checkpoint pointing at the previous checkpoint file; load() refuses a file whose digest or configuration does not match. Nothing here is a production deployment.

Scaling

The latencies above are one local run. The server handles one request at a time, so concurrent clients wait in line; Python’s ThreadingHTTPServer would handle requests in threads, but each request would still run the model in plain Python. Throughput, concurrency and memory for many caches at once were not measured.

Key takeaways

Serving adds a contract: validate every field before the model runs, and give each kind of bad request its own status code.
Test that the served reply equals direct generation, so the interface cannot quietly change results.
Evaluate through the interface against simple baselines, on the original test set and on a fresh set made after training, and report both.

Best practices

Reject booleans and unknown fields explicitly in JSON requests.
Keep a local teaching server on 127.0.0.1, and do not add a public endpoint without authentication, TLS and limits.
Record how each evaluation set was made and how often it has been scored before.

References

Sources checked October 20, 2026.

Python 3.15: http.server: not recommended for production and implements only basic security checks; HTTPServer and the threaded ThreadingHTTPServer; BaseHTTPRequestHandler dispatching to do_GET() and do_POST(), with send_response(), send_header(), end_headers(), rfile and wfile; log_message() is typically overridden.
RFC 9110, HTTP Semantics, sections 15.5.1, 15.5.5, 15.5.12 and 15.5.14: 400 Bad Request, 404 Not Found, 411 Length Required and 413 Content Too Large.
Python 3.15: urllib.error: HTTPError is raised for error responses and can be read like a response, which is how the evaluation reads error bodies.

Build a local server for the trained model

Prerequisites

By the end, you can

Problem

Concepts

Architecture

Working code

Check the interface

Evaluate against the baselines

Check your understanding

Failure cases

Deployment

Scaling

Key takeaways

Best practices

References

Training, checkpoints and inference