Build a CPU-Only AI Inference Service

Build a local text inference API and measure caching, load, and backpressure.

Introduction

30 Second Summary

Support requests rarely arrive one at a time. A service that feels instant for one message can slow down when several messages land together.

In this project, you will build a CPU-only support-routing service with Python's standard library. Its multinomial Naive Bayes model sits behind a local HTTP endpoint that you test under concurrent load.

What You'll Build

You will watch a local service route support messages while its benchmark reveals how the system behaves under load.

By the end of this project, you'll have:

  • A CPU-only classifier that turns support messages into routing labels with relative scores.
  • A synchronous load test that shows eight simultaneous requests queue behind one-at-a-time processing.
  • A threaded service with bounded caching that compares cold and warm batches while reporting CPU time, peak memory, request counts, cache activity, and context switches.
  • Secret Mission: Add explicit backpressure that rejects excess inference requests with an HTTP 503 overload response.

Are there any prerequisites?

You need Linux, Visual Studio Code, and Python 3.3 or later. The project uses only Python's standard library, so you need no GPU, CUDA setup, cloud account, API key, or external package.

Before We Start

This opening checkpoint turns the project goal into a commitment before any hands-on work begins. You are building a CPU-only local service that routes support messages across four categories while revealing its behavior under concurrent load.

Verify Your CPU-Only Workspace

Your support-routing service uses the Python standard library on your CPU. It needs no CUDA installation or external package.

The only setup risk is an unverified python3 command or a runtime older than 3.3. You will confirm the runtime in Visual Studio Code before creating the three empty project files.

In this step, get ready to:
  • Prepare the cpu-inference-lab workspace in Visual Studio Code.
  • Verify that your Python runtime meets the project requirement.
  • Create the three empty files used by the project.
Open the project workspace

A Visual Studio Code workspace gives the editor one folder context. Its integrated terminal starts from that folder.

  • Open your Linux application launcher.
  • Type Visual Studio Code into the application search field.

You will see Visual Studio Code in the search results.

  • Select Visual Studio Code from the search results.

The editor opens with its welcome screen visible.

  • Select File in the menu bar.
  • Select the Open Folder... menu item.

The folder picker opens so you can choose the workspace location.

  • Select your Desktop folder.
  • Select New Folder.
  • Enter cpu-inference-lab as the folder name.

The folder picker now shows the new cpu-inference-lab folder.

  • Select the cpu-inference-lab folder.
  • Select Select Folder.
  • Confirm that you trust cpu-inference-lab if the Workspace Trust dialog appears.

The Explorer now shows cpu-inference-lab as the open workspace.

  • Select View in the menu bar.
  • Select Terminal.

The integrated terminal opens at the bottom of the editor. Its working location is your cpu-inference-lab folder.

Why this workspace stays local

Your CPU handles every prediction in this project. The GPU stays outside the service.

You do not connect a cloud account. You do not create an API key.

Can't see the workspace or terminal?

  • Confirm that cpu-inference-lab appears at the top of the Explorer.
  • Select View from the menu bar.
  • Select Terminal to restore the terminal panel.

If the folder opened in another window, continue in the window that shows cpu-inference-lab. Help me check my Visual Studio Code workspace.

Verify the Python runtime

The benchmark later measures short durations with time.perf_counter(). This timing API requires Python 3.3 or later.

Before you run the check, do you expect the installed runtime to meet that minimum?

  • Check the installed Python version by running this command:
python3 --version

What does this command check?

  • The python3 command selects the installed Python 3 runtime.
  • The --version option prints the runtime version.

✔️ I see a supported version

That clears the only runtime risk. A result of Python 3.3 or later supports the timing API used by the benchmark.

ⓧ I see an older version

This runtime predates the timing API required by the benchmark. Update Python before creating the project files.

  • Use your Linux distribution's software manager to update Python 3.
  • Close the current integrated terminal.
  • Select Terminal in the menu bar.
  • Select New Terminal.
  • Repeat the version check by running this command:
python3 --version

What does this command check?

The new terminal loads the updated command path. The command prints the Python runtime that Visual Studio Code can now reach.

You should now see Python 3.3 or later.

Still seeing the older version?

Your new terminal may still be reaching an older Python installation. Help me update the Python runtime used by Visual Studio Code on Linux.

ⓧ Command not found

Your terminal cannot reach a Python 3 runtime through the required command. Install Python before continuing.

  • Use your Linux distribution's software manager to install Python 3.
  • Close the current integrated terminal.
  • Select Terminal in the menu bar.
  • Select New Terminal.
  • Verify the installation by running this command:
python3 --version

What does this command check?

The command confirms that the new terminal can reach Python 3. It also prints the installed runtime version.

You should now see Python 3.3 or later.

Still seeing command not found?

Your Linux installation may expose Python through a different command path. Help me make python3 available in the Visual Studio Code terminal.

Create the project files

The classifier gets its own file. The service code and load test each get a separate file.

  • Select the Explorer view in the left Activity Bar.
  • Select the New File... button in the Explorer toolbar.
  • Enter classifier.py as the file name.

The empty classifier.py file opens in the editor.

  • Select the New File... button in the Explorer toolbar.
  • Enter server.py as the file name.

The empty server.py file appears beside classifier.py.

  • Select the New File... button in the Explorer toolbar.
  • Enter load_test.py as the file name.

The Explorer now shows all three Python files inside cpu-inference-lab.

✔️ Awesome, I've got everything!

Great. Your three project files exist in the correct workspace. Keep each file empty for this setup checkpoint.

ⓧ I'd like to double check the full code

All three target artifacts contain zero lines at this checkpoint.

  • Select classifier.py in the Explorer.
  • Confirm that its editor contains no code.
  • Select load_test.py in the Explorer.
  • Confirm that its editor contains no code.
  • Select server.py in the Explorer.
  • Confirm that its editor contains no code.

Before the final check, do you expect the same terminal to report a supported Python version from this workspace?

  • Run the workspace verification command again:
python3 --version

What does this final check prove?

The command confirms that the integrated terminal can reach the required runtime from cpu-inference-lab.

You will see Python 3.3 or later in the terminal. The Explorer will show classifier.py, load_test.py, plus server.py.

Missing a project file?

  • Correct any filename that differs from the spelling shown above.
  • Create any missing file with the New File... button in the Explorer toolbar.

Linux filenames are case-sensitive. Help me find a missing project file.

Your local runtime and project structure are ready. Next, you will build the support-message classifier and evaluate its first predictions.

Train the Support Classifier

Your workspace is ready for its first real inference task. A visible prediction proves that the project has a working model before you build the service around it.

You will implement a multinomial Naive Bayes classifier that learns from local support messages. It uses tokenization, class priors, logarithmic scores, and Laplace smoothing to route each message.

In this step, get ready to:
  • Add labeled training examples and held-out test messages.
  • Build the classifier's training and prediction logic.
  • Evaluate four predictions in the integrated terminal.
Add the labeled support messages

The model learns associations between words and routing labels from labeled examples. A separate test set checks how those associations transfer to messages that were excluded from training.

  • Switch back to the empty classifier.py tab from earlier.
  • Add the standard-library imports by pasting this code:
import math
import re
from collections import Counter, defaultdict

What do these imports provide?

  • The math module provides logarithmic scoring and score scaling.
  • The re module extracts lowercase word tokens from each message.
  • Counter and defaultdict store document counts and per-label word counts.
  • Save classifier.py.
  • Confirm the unsaved-change indicator disappears from the classifier.py tab.

Seeing an import warning?

Check that each import matches the snippet exactly. These modules are included with Python, so no package installation is needed.

Confirm that Visual Studio Code is using the verified Python runtime. Help me diagnose these Python import warnings.

  • Add the labeled training examples below the imports by pasting this code:
TRAINING_DATA = [
    ("I cannot sign in to my account", "account"),
    ("My password reset link does not work", "account"),
    ("The login page says my account is locked", "account"),
    ("I forgot my password and need access", "account"),
    ("My verification code never arrived", "account"),
    ("Please help me change my account email", "account"),
    ("I was charged twice for one order", "billing"),
    ("Where can I download my invoice", "billing"),
    ("I need a refund for my subscription", "billing"),
    ("My card payment was declined", "billing"),
    ("The price on my bill is incorrect", "billing"),
    ("Please cancel the recurring payment", "billing"),
    ("The mobile app crashes when it starts", "technical"),
    ("The screen freezes after I click save", "technical"),
    ("I see an error when uploading a file", "technical"),
    ("The dashboard is loading very slowly", "technical"),
    ("Notifications stopped working on my phone", "technical"),
    ("The website shows a blank page", "technical"),
    ("I feel anxious and cannot calm down", "wellbeing"),
    ("I have been overwhelmed by stress", "wellbeing"),
    ("I feel sad and need someone to talk to", "wellbeing"),
    ("My panic symptoms are getting worse", "wellbeing"),
    ("I am struggling with sleep and worry", "wellbeing"),
    ("I feel emotionally exhausted today", "wellbeing"),
]

How is the training data organized?

  • Each tuple pairs one support message with its expected routing label.
  • The four labels are account, billing, technical, and wellbeing.
  • Each label receives six examples so the class priors begin evenly balanced.
  • Save classifier.py.
  • Confirm TRAINING_DATA contains examples for all four routing labels.

Seeing red underlines in the data?

Check that every example has an opening parenthesis and a closing parenthesis. Confirm that the list ends with a closing square bracket.

Make sure each message and label remains inside double quotes. Help me find the syntax problem in TRAINING_DATA.

  • Add the held-out evaluation messages below TRAINING_DATA by pasting this code:
TEST_DATA = [
    ("I am locked out after changing my password", "account"),
    ("My subscription charged my card two times", "billing"),
    ("The app freezes and shows an error", "technical"),
    ("I feel overwhelmed and anxious", "wellbeing"),
]

Why keep separate test messages?

The classifier never learns directly from TEST_DATA. Each message tests whether the learned word patterns transfer to a new example.

  • Save classifier.py.
  • Confirm TEST_DATA contains one held-out message for each routing label.

Test data showing a syntax problem?

Confirm that TRAINING_DATA closes before TEST_DATA begins. Check the commas between the four test tuples.

Compare the bracket placement with the snippet above. Help me repair the TEST_DATA list.

Build the training and prediction logic

The classifier stores document counts for its class priors. It also stores token counts for the likelihood of each word under each label.

  • Add the classifier class below TEST_DATA by pasting this first section:
class NaiveBayesTextClassifier:
    def __init__(self, alpha=1.0):
        self.alpha = alpha
        self.document_counts = Counter()
        self.token_counts = defaultdict(Counter)
        self.total_tokens = Counter()
        self.vocabulary = set()
        self.total_documents = 0
        self.labels = []

What state does the classifier keep?

  • alpha controls the amount of Laplace smoothing.
  • document_counts tracks the number of training messages in each label.
  • token_counts and total_tokens track the words learned for each label.
  • vocabulary stores every unique training token.
  • Save classifier.py.
  • Confirm the class contains all seven model-state attributes.

Class indentation looking uneven?

Keep def __init__ indented once beneath the class. Keep every self assignment indented once more beneath the method.

Use spaces consistently throughout the class. Help me correct the class indentation.

  • Add the tokenizer below __init__ by pasting this method:
    @staticmethod
    def tokenize(text):
        return re.findall(r"[a-z0-9']+", text.lower())

What does tokenization do?

  • text.lower() normalizes uppercase and lowercase forms of the same word.
  • re.findall() returns the word-like matches from the normalized message.
  • The pattern keeps letters, digits, and apostrophes inside tokens.
  • Save classifier.py.
  • Confirm tokenize is aligned with __init__ inside the class.

Tokenizer outside the class?

Check that @staticmethod begins with four spaces. The method definition must use the same indentation.

Keep the regular expression exactly as shown. Help me place tokenize inside the class.

  • Add the training method below tokenize by pasting this code:
    def fit(self, rows):
        for text, label in rows:
            tokens = self.tokenize(text)
            self.document_counts[label] += 1
            self.token_counts[label].update(tokens)
            self.total_tokens[label] += len(tokens)
            self.vocabulary.update(tokens)

        self.total_documents = len(rows)
        self.labels = sorted(self.document_counts)
        return self

How does training work?

  • The loop tokenizes each message under its supplied label.
  • The counters record how often each label and token appears.
  • The vocabulary collects every unique token used by the smoothing calculation.
  • Returning self lets the model be created and trained in one expression.
  • Save classifier.py.
  • Confirm fit ends with return self at the method indentation level.

Training method showing an indentation error?

Keep the five counter updates inside the for loop. Keep the final three statements outside that loop.

Confirm the entire method remains inside the class. Help me repair the fit method indentation.

  • Start the prediction method below fit by pasting this code:
    def predict(self, text):
        tokens = self.tokenize(text)
        if not tokens:
            return {"label": "needs_review", "relative_score": 0.0}

        scores = {}
        vocabulary_size = len(self.vocabulary)

How does prediction begin?

  • The message is tokenized with the same rules used during training.
  • An empty message receives the fallback label needs_review.
  • scores holds one logarithmic score for each trained label.
  • vocabulary_size supplies the smoothing denominator with the number of known words.
  • Save classifier.py.
  • Confirm the empty-token branch returns both label and relative_score.

Prediction method nested incorrectly?

Align predict with the other class methods. Keep the empty-input return indented inside its if block.

Check that both dictionary keys remain inside double quotes. Help me fix the start of predict.

  • Complete predict directly below vocabulary_size by pasting this code:
        for label in self.labels:
            prior = self.document_counts[label] / float(self.total_documents)
            log_score = math.log(prior)
            denominator = (
                self.total_tokens[label] + self.alpha * vocabulary_size
            )

            for token in tokens:
                numerator = self.token_counts[label][token] + self.alpha
                log_score += math.log(numerator / float(denominator))

            scores[label] = log_score

        best_label = max(scores, key=scores.get)
        largest_score = max(scores.values())
        scaled_scores = {
            label: math.exp(score - largest_score)
            for label, score in scores.items()
        }
        score_total = sum(scaled_scores.values())
        relative_score = scaled_scores[best_label] / score_total

        return {
            "label": best_label,
            "relative_score": round(relative_score, 4),
        }

How does the model choose a label?

  • The prior represents the training share belonging to one label.
  • Laplace smoothing adds alpha to every token count.
  • Logarithms turn repeated probability multiplication into addition.
  • The scaled scores produce a relative score for the winning label.
  • Save classifier.py.
  • Confirm predict ends with a dictionary containing the winning label and rounded relative score.

How should you read the score?

The relative_score compares the winning label with the other labels for the current message. It is a useful ranking signal for this model.

Naive Bayes scores can look decisive even when the training set is small. Treat the value as relative model preference.

Scoring block showing a syntax error?

Check the closing parenthesis for denominator. Check both closing braces in scaled_scores and the returned dictionary.

Keep the token loop inside the label loop. Help me debug the predict scoring block.

Evaluate the held-out predictions

Evaluation compares each predicted label with the expected label. The printed rows also expose the relative score so you can inspect the model's preference.

  • Add the evaluation function below the classifier class by pasting this code:
def evaluate(model, rows):
    correct = 0
    for text, expected in rows:
        result = model.predict(text)
        predicted = result["label"]
        if predicted == expected:
            correct += 1
        print(
            "expected={0:10s} predicted={1:10s} score={2:.4f} text={3}".format(
                expected,
                predicted,
                result["relative_score"],
                text,
            )
        )

    print("evaluation: {0}/{1} correct".format(correct, len(rows)))

What does evaluation report?

  • The loop asks the trained model to classify every held-out message.
  • correct increases whenever the predicted label matches the expected label.
  • Each row prints the expected label, predicted label, relative score, and original message.
  • The final line summarizes the number of correct predictions.
  • Save classifier.py.
  • Confirm evaluate sits outside NaiveBayesTextClassifier with no leading indentation.

Evaluation function still inside the class?

Move def evaluate fully to the left edge of the file. Keep its loop and print statements indented beneath the function.

Check that the formatting string remains on one continuous quoted line. Help me fix the evaluate function.

  • Add the script entry point below evaluate by pasting this code:
if __name__ == "__main__":
    classifier = NaiveBayesTextClassifier(alpha=1.0).fit(TRAINING_DATA)
    evaluate(classifier, TEST_DATA)

What does the entry point do?

  • The first indented line creates the classifier with Laplace smoothing set to 1.0.
  • fit(TRAINING_DATA) trains the model from the local examples.
  • evaluate(classifier, TEST_DATA) runs the four held-out checks.
  • Save classifier.py.
  • Confirm the entry point appears at the left edge below evaluate.

Entry point showing a name error?

Confirm that NaiveBayesTextClassifier, TRAINING_DATA, evaluate, and TEST_DATA match their earlier definitions exactly.

Check the double underscores around __name__ and __main__. Help me repair the classifier entry point.

✔️ Awesome, I've got everything!

Great. Double-check that you saved classifier.py before running it.

ⓧ I'd like to double check the full code

import math
import re
from collections import Counter, defaultdict


TRAINING_DATA = [
    ("I cannot sign in to my account", "account"),
    ("My password reset link does not work", "account"),
    ("The login page says my account is locked", "account"),
    ("I forgot my password and need access", "account"),
    ("My verification code never arrived", "account"),
    ("Please help me change my account email", "account"),
    ("I was charged twice for one order", "billing"),
    ("Where can I download my invoice", "billing"),
    ("I need a refund for my subscription", "billing"),
    ("My card payment was declined", "billing"),
    ("The price on my bill is incorrect", "billing"),
    ("Please cancel the recurring payment", "billing"),
    ("The mobile app crashes when it starts", "technical"),
    ("The screen freezes after I click save", "technical"),
    ("I see an error when uploading a file", "technical"),
    ("The dashboard is loading very slowly", "technical"),
    ("Notifications stopped working on my phone", "technical"),
    ("The website shows a blank page", "technical"),
    ("I feel anxious and cannot calm down", "wellbeing"),
    ("I have been overwhelmed by stress", "wellbeing"),
    ("I feel sad and need someone to talk to", "wellbeing"),
    ("My panic symptoms are getting worse", "wellbeing"),
    ("I am struggling with sleep and worry", "wellbeing"),
    ("I feel emotionally exhausted today", "wellbeing"),
]

TEST_DATA = [
    ("I am locked out after changing my password", "account"),
    ("My subscription charged my card two times", "billing"),
    ("The app freezes and shows an error", "technical"),
    ("I feel overwhelmed and anxious", "wellbeing"),
]


class NaiveBayesTextClassifier:
    def __init__(self, alpha=1.0):
        self.alpha = alpha
        self.document_counts = Counter()
        self.token_counts = defaultdict(Counter)
        self.total_tokens = Counter()
        self.vocabulary = set()
        self.total_documents = 0
        self.labels = []

    @staticmethod
    def tokenize(text):
        return re.findall(r"[a-z0-9']+", text.lower())

    def fit(self, rows):
        for text, label in rows:
            tokens = self.tokenize(text)
            self.document_counts[label] += 1
            self.token_counts[label].update(tokens)
            self.total_tokens[label] += len(tokens)
            self.vocabulary.update(tokens)

        self.total_documents = len(rows)
        self.labels = sorted(self.document_counts)
        return self

    def predict(self, text):
        tokens = self.tokenize(text)
        if not tokens:
            return {"label": "needs_review", "relative_score": 0.0}

        scores = {}
        vocabulary_size = len(self.vocabulary)

        for label in self.labels:
            prior = self.document_counts[label] / float(self.total_documents)
            log_score = math.log(prior)
            denominator = (
                self.total_tokens[label] + self.alpha * vocabulary_size
            )

            for token in tokens:
                numerator = self.token_counts[label][token] + self.alpha
                log_score += math.log(numerator / float(denominator))

            scores[label] = log_score

        best_label = max(scores, key=scores.get)
        largest_score = max(scores.values())
        scaled_scores = {
            label: math.exp(score - largest_score)
            for label, score in scores.items()
        }
        score_total = sum(scaled_scores.values())
        relative_score = scaled_scores[best_label] / score_total

        return {
            "label": best_label,
            "relative_score": round(relative_score, 4),
        }


def evaluate(model, rows):
    correct = 0
    for text, expected in rows:
        result = model.predict(text)
        predicted = result["label"]
        if predicted == expected:
            correct += 1
        print(
            "expected={0:10s} predicted={1:10s} score={2:.4f} text={3}".format(
                expected,
                predicted,
                result["relative_score"],
                text,
            )
        )

    print("evaluation: {0}/{1} correct".format(correct, len(rows)))


if __name__ == "__main__":
    classifier = NaiveBayesTextClassifier(alpha=1.0).fit(TRAINING_DATA)
    evaluate(classifier, TEST_DATA)

Before you run the classifier, do you expect its predictions to match all four held-out labels?

  • Run the completed classifier in the integrated terminal from earlier by using this command:
python3 classifier.py

What should you see?

You will see four rows containing expected labels, predicted labels, relative scores, and held-out messages. The predicted labels match the expected account, billing, technical, and wellbeing labels.

The final evaluation summary reports four correct predictions out of four. Your local model is now producing visible inference results.

Classifier not printing four predictions?

Confirm the terminal is still inside the cpu-inference-lab folder. Check that classifier.py is saved before rerunning the command.

Use the first traceback line that points into classifier.py to locate the failing block. Help me debug this classifier traceback.

That is the first visible win complete: your CPU-only classifier can learn from local examples and route new support messages. Next, you will place this prediction behind a local JSON API.

Serve Synchronous Predictions

Your classifier can now route support messages from a local script. Another application still has no stable way to send it work or receive a prediction.

A local HTTP API gives the model a JSON boundary. This first server handles one request at a time so you can measure synchronous queueing in the next step.

In this step, get ready to:
  • Configure the local inference service.
  • Validate prediction requests through a synchronous handler.
  • Send one message to the endpoint and inspect its response.
Configure the local service

The service needs the trained model plus a loopback address that keeps requests on your computer. Its simulated delay makes later queueing behavior repeatable without claiming to measure your processor.

  • Select server.py in the Explorer sidebar in Visual Studio Code.
  • Add the imports and service configuration to server.py by pasting this code:
import json
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

from classifier import NaiveBayesTextClassifier, TRAINING_DATA


HOST = "127.0.0.1"
PORT = 8000
MAX_BODY_BYTES = 4096
SIMULATED_INFERENCE_SECONDS = 0.15
MODEL = NaiveBayesTextClassifier(alpha=1.0).fit(TRAINING_DATA)

What Does This Configuration Do?

  • The imports provide JSON encoding and timing support.
  • The HTTP classes provide the local server boundary.
  • The HOST value binds the service to your computer's loopback address.
  • The MODEL value trains the existing classifier from TRAINING_DATA when the service starts.
  • Save server.py.
  • Check that the configuration loads by running this command in the integrated terminal:
python3 server.py

What Does This Check Prove?

Python imports the server modules and trains MODEL. A clean return to the prompt confirms that this foundation loads without a traceback.

Good. Your classifier now loads as reusable server state instead of running the evaluation entry point.

Seeing an Import Error?

Confirm that server.py sits beside classifier.py inside cpu-inference-lab.

Check that the imported names match NaiveBayesTextClassifier and TRAINING_DATA exactly. Help me diagnose why server.py cannot import my classifier.

Build the synchronous request handler

A request handler translates between HTTP responses and Python dictionaries. The basic server processes each request completely before it starts the next one.

  • Place the cursor below the MODEL line in server.py.
  • Add the response helper and synchronous server entry point by pasting this code:
class InferenceHandler(BaseHTTPRequestHandler):
    def send_json(self, status, payload):
        body = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format_string, *args):
        return


if __name__ == "__main__":
    server = HTTPServer((HOST, PORT), InferenceHandler)
    print("inference service listening on http://{0}:{1}".format(HOST, PORT))
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nstopping service")
    finally:
        server.server_close()

What Does This Server Code Do?

  • The send_json() helper encodes a dictionary as UTF-8 JSON.
  • The helper sends the response status plus the body headers.
  • The HTTPServer instance listens only on 127.0.0.1 at port 8000.
  • The cleanup block closes the server socket when the process stops.
  • Save server.py.
  • Start the local service by running this command:
python3 server.py

What Should You See?

The command creates the synchronous server and keeps it running. The terminal prints the loopback URL where it is listening.

You will see inference service listening on http://127.0.0.1:8000. This confirms that the service owns the local port.

  • Stop the service by pressing Ctrl+C in the terminal.

Server Not Starting?

Check that the InferenceHandler class sits above the __main__ entry point.

Another local process may already be using port 8000. Stop any earlier server process before retrying. Help me diagnose why my local HTTPServer is not starting.

Adding a method inside an existing class is a fiddly edit. Its indentation must align with send_json() so Python treats both methods as part of InferenceHandler.

  • Place the cursor above def log_message(self, format_string, *args): inside InferenceHandler.
  • Add the request validation method by pasting this code:
    def do_POST(self):
        if self.path != "/predict":
            self.send_json(404, {"error": "not found"})
            return

        try:
            content_length = int(self.headers.get("Content-Length", "0"))
        except ValueError:
            content_length = 0

        if content_length <= 0 or content_length > MAX_BODY_BYTES:
            self.send_json(400, {"error": "invalid body size"})
            return

        try:
            payload = json.loads(self.rfile.read(content_length).decode("utf-8"))
        except (UnicodeDecodeError, ValueError):
            self.send_json(400, {"error": "invalid JSON"})
            return

        text = payload.get("text", "") if isinstance(payload, dict) else ""
        text = text.strip()
        if not text:
            self.send_json(400, {"error": "text is required"})
            return

How Does Validation Protect the Endpoint?

  • The path check accepts prediction work only at /predict.
  • The body-size check rejects empty requests and bodies above MAX_BODY_BYTES.
  • The JSON parser converts the UTF-8 body into a Python value.
  • The final check requires a non-empty text value.
  • Save server.py.
  • Confirm that the expanded handler loads by running this command:
python3 server.py

What Does This Run Confirm?

The listening message confirms that Python accepted the handler's structure. It also confirms that the service can still bind to its loopback port.

The service prints its listening URL again. Your request validator is now active inside the handler.

  • Stop the service by pressing Ctrl+C in the terminal.

Seeing an Indentation Error?

Align def do_POST(self): with def send_json(self, status, payload): inside the class.

Keep the validation statements indented beneath do_POST(). Help me fix the indentation in my InferenceHandler class.

A valid request now needs timed inference before the server can answer it. The fixed delay represents repeatable model work for this local experiment.

  • Find the final return inside the empty-text check in do_POST().
  • Place the cursor directly below that validation block.
  • Add the timed prediction response by pasting this code:
        started = time.perf_counter()
        time.sleep(SIMULATED_INFERENCE_SECONDS)
        result = dict(MODEL.predict(text.lower()))
        result["latency_ms"] = round((time.perf_counter() - started) * 1000.0, 2)
        self.send_json(200, result)

What Does the Inference Path Measure?

  • The first time.perf_counter() call records the start of the inference path.
  • The time.sleep() call adds the project-defined 150-millisecond delay.
  • The classifier predicts from a normalized lowercase message.
  • The response includes the measured latency_ms value alongside the model result.
  • Save server.py.
  • Start the completed synchronous service by running this command:
python3 server.py

What Is Running Now?

The process is waiting for one request at a time. It stays active in this terminal while the client runs from another terminal.

You will see the listening URL once more. Leave this server process running for the endpoint check.

Service Stops Immediately?

Confirm that server.serve_forever() remains inside the try block at the bottom of server.py.

Check the terminal for a traceback that names a missing parenthesis or indentation problem. Help me diagnose why server.py exits instead of continuing to serve requests.

✔️ Awesome, I've got everything!

Great. Double-check that you saved server.py while the service continues running.

ⓧ I'd like to double check the full code

import json
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

from classifier import NaiveBayesTextClassifier, TRAINING_DATA


HOST = "127.0.0.1"
PORT = 8000
MAX_BODY_BYTES = 4096
SIMULATED_INFERENCE_SECONDS = 0.15
MODEL = NaiveBayesTextClassifier(alpha=1.0).fit(TRAINING_DATA)


class InferenceHandler(BaseHTTPRequestHandler):
    def send_json(self, status, payload):
        body = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        if self.path != "/predict":
            self.send_json(404, {"error": "not found"})
            return

        try:
            content_length = int(self.headers.get("Content-Length", "0"))
        except ValueError:
            content_length = 0

        if content_length <= 0 or content_length > MAX_BODY_BYTES:
            self.send_json(400, {"error": "invalid body size"})
            return

        try:
            payload = json.loads(self.rfile.read(content_length).decode("utf-8"))
        except (UnicodeDecodeError, ValueError):
            self.send_json(400, {"error": "invalid JSON"})
            return

        text = payload.get("text", "") if isinstance(payload, dict) else ""
        text = text.strip()
        if not text:
            self.send_json(400, {"error": "text is required"})
            return

        started = time.perf_counter()
        time.sleep(SIMULATED_INFERENCE_SECONDS)
        result = dict(MODEL.predict(text.lower()))
        result["latency_ms"] = round((time.perf_counter() - started) * 1000.0, 2)
        self.send_json(200, result)

    def log_message(self, format_string, *args):
        return


if __name__ == "__main__":
    server = HTTPServer((HOST, PORT), InferenceHandler)
    print("inference service listening on http://{0}:{1}".format(HOST, PORT))
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nstopping service")
    finally:
        server.server_close()
Send one prediction request

The running service needs a separate client process to prove that another application can use its boundary. This temporary client sends one billing message and prints the decoded response.

  • Select load_test.py in the Explorer sidebar.
  • Add the one-request client to load_test.py by pasting this code:
import json
from urllib.request import Request, urlopen


PREDICT_URL = "http://127.0.0.1:8000/predict"


body = json.dumps({"text": "I was charged twice"}).encode("utf-8")
request = Request(
    PREDICT_URL,
    data=body,
    headers={"Content-Type": "application/json"},
)
with urlopen(request, timeout=5) as response:
    print(json.loads(response.read().decode("utf-8")))

How Does the Client Reach the Model?

  • The client serializes the support message into a JSON request body.
  • The Request object targets PREDICT_URL with the JSON content type.
  • The urlopen() call waits up to five seconds for the local response.
  • The final line decodes the response and prints the resulting dictionary.
  • Save load_test.py.

Client File Showing a Syntax Error?

Confirm that the closing parenthesis for Request() appears before the with urlopen line.

Check that the URL ends with /predict. Help me check the syntax in my one-request load_test.py client.

✔️ Awesome, I've got everything!

Great. Double-check that you saved load_test.py.

ⓧ I'd like to double check the full code

import json
from urllib.request import Request, urlopen


PREDICT_URL = "http://127.0.0.1:8000/predict"


body = json.dumps({"text": "I was charged twice"}).encode("utf-8")
request = Request(
    PREDICT_URL,
    data=body,
    headers={"Content-Type": "application/json"},
)
with urlopen(request, timeout=5) as response:
    print(json.loads(response.read().decode("utf-8")))
  • Create a second terminal in the Visual Studio Code terminal panel.

Before you run the client, which routing label do you expect for a message about being charged twice?

  • Send the test message to the running service by running this command in the second terminal:
python3 load_test.py

What Does the Response Prove?

  • The label value comes from the classifier behind the HTTP boundary.
  • The relative_score value preserves the model's normalized comparison score.
  • The latency_ms value includes the project-defined inference delay.

You will see a dictionary containing the billing label plus relative_score and latency_ms values. The measured latency should reflect at least the simulated delay while the exact total varies by system activity.

Request Not Reaching the Service?

Confirm that the first terminal still shows the running service at http://127.0.0.1:8000.

Check that PREDICT_URL includes the /predict path. Help me diagnose why load_test.py cannot reach my local prediction service.

That is the model-to-service path working end to end. Next, you will launch several requests together and expose the queue hidden inside this synchronous design.

Expose Synchronous Queueing

Your one-request check proved that the local inference endpoint can classify a support message. Now you need to see how that design behaves when several clients arrive together.

The synchronous server completes each delayed prediction before it accepts the next one. A concurrent client exposes this synchronous queueing by sending eight requests through a thread pool.

In this step, get ready to:
  • Define eight distinct support messages for the load test.
  • Launch the requests together with ThreadPoolExecutor.
  • Compare each request's latency with the total batch time.
Prepare eight concurrent requests

Distinct messages keep every request focused on the server's one-at-a-time behavior. The timing module measures each client's complete wait from request start to response.

  • In load_test.py, select the block from import json through the PREDICT_URL line.
  • Replace the selected block with this code:
import json
import time
from concurrent.futures import ThreadPoolExecutor
from urllib.request import Request, urlopen


PREDICT_URL = "http://127.0.0.1:8000/predict"
MESSAGES = [
    "I cannot access my account after a password reset",
    "My card was charged twice for the subscription",
    "The mobile app crashes on the loading screen",
    "I feel anxious and overwhelmed today",
    "The verification code never reached my email",
    "Please send a refund and a corrected invoice",
    "The website freezes whenever I upload a file",
    "I have panic symptoms and cannot sleep",
]

What does this code prepare?

  • The time module provides the high-resolution counter used for client timing.
  • ThreadPoolExecutor provides the workers that launch requests concurrently.
  • MESSAGES holds eight distinct inputs across the classifier's support categories.
  • Save load_test.py.
  • Run the existing one-request check in the second integrated terminal from earlier with this command:
python3 load_test.py

You'll see one prediction dictionary from the running endpoint. Good work. The expanded setup has preserved the request that already worked.

Did the one-request check stop working?

Check that every import appears exactly once at the top of load_test.py. Confirm that the server terminal from earlier is still running.

Make sure each message has matching quotation marks. Help me troubleshoot the updated load test setup.

Launch the batch and measure queueing

The current client performs one request at the top level of the file. You will turn that logic into a reusable function before the batch runner calls it eight times.

  • In load_test.py, find this current one-request block:
body = json.dumps({"text": "I was charged twice"}).encode("utf-8")
request = Request(
    PREDICT_URL,
    data=body,
    headers={"Content-Type": "application/json"},
)
with urlopen(request, timeout=5) as response:
    print(json.loads(response.read().decode("utf-8")))

This block sends one fixed message directly. Its response is printed as soon as that single request finishes.

  • Replace the selected one-request block with this function:
def request_prediction(text):
    body = json.dumps({"text": text}).encode("utf-8")
    request = Request(
        PREDICT_URL,
        data=body,
        headers={"Content-Type": "application/json"},
    )
    started = time.perf_counter()
    with urlopen(request, timeout=5) as response:
        result = json.loads(response.read().decode("utf-8"))
    result["client_ms"] = round(
        (time.perf_counter() - started) * 1000.0,
        2,
    )
    return result

What does this function measure?

  • request_prediction() builds a JSON request from the message supplied by a worker.
  • started records the moment before the client contacts the server.
  • client_ms includes connection time plus server queueing plus inference time.
  • latency_ms remains the server's measurement of the prediction work after its handler begins.
  • Place your cursor on the second blank line below return result.
  • Add the concurrent batch runner by pasting this code:
def run_batch():
    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=8) as executor:
        results = list(executor.map(request_prediction, MESSAGES))
    elapsed_ms = round((time.perf_counter() - started) * 1000.0, 2)

    for result in results:
        print(result)
    print("total_ms={0:.2f}".format(elapsed_ms))


if __name__ == "__main__":
    run_batch()

How does the batch expose queueing?

  • ThreadPoolExecutor(max_workers=8) gives all eight messages a worker.
  • executor.map() submits each message to request_prediction().
  • elapsed_ms measures the complete batch from submission to the final response.
  • run_batch() prints every result before it prints total_ms.
  • Save load_test.py.

✔️ Awesome, I've got everything!

Great. Double-check that you saved load_test.py before running the batch.

ⓧ I'd like to double check the full code

import json
import time
from concurrent.futures import ThreadPoolExecutor
from urllib.request import Request, urlopen


PREDICT_URL = "http://127.0.0.1:8000/predict"
MESSAGES = [
    "I cannot access my account after a password reset",
    "My card was charged twice for the subscription",
    "The mobile app crashes on the loading screen",
    "I feel anxious and overwhelmed today",
    "The verification code never reached my email",
    "Please send a refund and a corrected invoice",
    "The website freezes whenever I upload a file",
    "I have panic symptoms and cannot sleep",
]


def request_prediction(text):
    body = json.dumps({"text": text}).encode("utf-8")
    request = Request(
        PREDICT_URL,
        data=body,
        headers={"Content-Type": "application/json"},
    )
    started = time.perf_counter()
    with urlopen(request, timeout=5) as response:
        result = json.loads(response.read().decode("utf-8"))
    result["client_ms"] = round(
        (time.perf_counter() - started) * 1000.0,
        2,
    )
    return result


def run_batch():
    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=8) as executor:
        results = list(executor.map(request_prediction, MESSAGES))
    elapsed_ms = round((time.perf_counter() - started) * 1000.0, 2)

    for result in results:
        print(result)
    print("total_ms={0:.2f}".format(elapsed_ms))


if __name__ == "__main__":
    run_batch()

Before you run the batch, do you expect eight concurrent clients to finish near one 150 ms delay or to wait behind one another?

  • Run the concurrent load test in the second integrated terminal from earlier with this command:
python3 load_test.py

You'll see eight prediction dictionaries containing latency_ms plus client_ms. The final total_ms value is substantially greater than one 150 ms inference delay.

The later clients spend more time waiting even though the server work for each prediction stays close to the same delay. You have reproduced serial queueing under concurrent demand.

Did the batch fail or finish too quickly?

Confirm that the server from earlier still uses the synchronous HTTPServer. Check that MESSAGES contains eight distinct strings.

Make sure the batch uses ThreadPoolExecutor(max_workers=8). Help me diagnose the synchronous load-test result.

That slowdown gives you a measured bottleneck to fix. Next, you will let the service handle requests concurrently before adding caching and process metrics.

Add Threads Caching and Metrics

Your concurrent load test exposed the synchronous server's queue. Each request waited while HTTPServer finished the request ahead of it.

In this step, threads let requests overlap. A bounded cache avoids repeated model work. Linux process metrics reveal the runtime cost.

In this step, get ready to:
  • Cache repeated predictions with a fixed capacity.
  • Handle requests in separate server threads.
  • Compare cold latency with warm latency through process metrics.
Cache repeated predictions

Repeated messages produce the same prediction. A cache stores recent results so the model can skip work for a normalized message it has already seen.

  • Switch back to server.py in Visual Studio Code.
  • Select everything from import json through the existing MODEL assignment.
  • Replace the selected section by pasting this code:
import json
import resource
import time
from functools import lru_cache
from http.server import BaseHTTPRequestHandler, HTTPServer
from socketserver import ThreadingMixIn
from threading import Lock, active_count

from classifier import NaiveBayesTextClassifier, TRAINING_DATA


HOST = "127.0.0.1"
PORT = 8000
MAX_BODY_BYTES = 4096
SIMULATED_INFERENCE_SECONDS = 0.15

MODEL = NaiveBayesTextClassifier(alpha=1.0).fit(TRAINING_DATA)
METRICS_LOCK = Lock()
REQUESTS_TOTAL = 0


@lru_cache(maxsize=128)
def cached_predict(normalized_text):
    time.sleep(SIMULATED_INFERENCE_SECONDS)
    return MODEL.predict(normalized_text)

What does this code do?

  • The lru_cache decorator retains up to 128 recent prediction results.
  • The normalized message becomes the cache key.
  • The simulated delay now belongs to uncached model work.
  • The lock prepares the request counter for safe updates from multiple threads.
  • Save server.py.
  • Return to the server terminal from earlier.
  • Stop the running synchronous service by pressing Ctrl+C.
  • Check that the revised imports load by running this command:
python3 server.py

What does this command prove?

Python loads the revised module before starting the service. The listening message confirms that the new imports and cache definition have valid syntax.

You should see inference service listening on http://127.0.0.1:8000. The service is ready for the request-path edit.

  • Stop the service by pressing Ctrl+C.
  • In server.py, find the do_POST method.
  • Add this line directly below the method definition:
        global REQUESTS_TOTAL

Why declare the counter as global?

The request handler updates the module-level REQUESTS_TOTAL value. The declaration makes that shared update explicit inside do_POST.

  • Find the inference section that begins with started = time.perf_counter().
  • Select through the existing self.send_json(200, result) line.
  • Replace the selected section by pasting this code:
        with METRICS_LOCK:
            REQUESTS_TOTAL += 1

        started = time.perf_counter()
        result = dict(cached_predict(text.lower()))
        result["latency_ms"] = round(
            (time.perf_counter() - started) * 1000.0,
            2,
        )
        self.send_json(200, result)

What happens on each request?

  • The lock protects the shared request counter during its update.
  • Lowercasing the message gives equivalent text a consistent cache key.
  • The latency measurement includes either a cache lookup or the delayed model prediction.
  • Copying the cached result into a new dictionary keeps the stored value separate from the response-specific latency.
  • Save server.py.
  • Restart the service in the server terminal by running this command:
python3 server.py

What does this restart reset?

A fresh process starts with an empty cache. This makes the next eight distinct messages cold requests.

Before you run the client, do you expect its first batch to remain serial or become concurrent?

  • Switch back to the client terminal from earlier.
  • Populate the empty cache by running the existing load test:
python3 load_test.py

What does the first run show?

The eight distinct messages miss the empty cache. The synchronous server still handles them one at a time.

You should still see a total batch time far above one 150 ms delay. The completed predictions now occupy eight cache entries.

Before you repeat the test, do you expect the same messages to wait for the simulated inference delay again?

  • Repeat the same client batch by running this command:
python3 load_test.py

Why is the repeat faster?

Each normalized message now matches a cached result. The server returns those predictions without repeating the delayed model call.

You should see much lower response latencies and a much lower total time. The cache has removed repeated inference work.

Warm requests still taking 150 ms?

Keep the same server process running between both client commands. Restarting the service clears its in-memory cache.

Check that cached_predict(text.lower()) replaced the direct model call. Help me diagnose why repeated predictions are missing the cache.

Add threaded serving and metrics

Caching helps repeated work. Distinct cold requests still need concurrent handling so one slow prediction does not block every request behind it.

  • Stop the service in the server terminal by pressing Ctrl+C.
  • In server.py, find the end of do_POST.
  • Add this method directly before log_message:
    def do_GET(self):
        if self.path != "/metrics":
            self.send_json(404, {"error": "not found"})
            return

        usage = resource.getrusage(resource.RUSAGE_SELF)
        cache = cached_predict.cache_info()
        with METRICS_LOCK:
            request_count = REQUESTS_TOTAL

        self.send_json(
            200,
            {
                "requests_total": request_count,
                "cache_hits": cache.hits,
                "cache_misses": cache.misses,
                "cache_entries": cache.currsize,
                "active_threads": active_count(),
                "user_cpu_seconds": round(usage.ru_utime, 4),
                "system_cpu_seconds": round(usage.ru_stime, 4),
                "max_rss_raw": usage.ru_maxrss,
                "voluntary_context_switches": usage.ru_nvcsw,
                "involuntary_context_switches": usage.ru_nivcsw,
            },
        )

What does the metrics endpoint report?

The cache reports hits, misses, and current entries. The handler also reports the shared request count and active thread count.

The resource fields expose process CPU time, peak resident memory, and context switches. max_rss_raw stays unitless because its interpretation depends on the operating system.

  • Save server.py.
  • Start the service by running this command in the server terminal:
python3 server.py

What does this run confirm?

The running process now exposes prediction results through POST /predict. It also exposes runtime observations through GET /metrics.

  • Enter http://127.0.0.1:8000/metrics in your browser address bar.
  • Load the address to confirm the metrics response.

You should see a JSON object containing the request, cache, CPU, memory, thread, and context-switch fields. This proves the metrics boundary works before threading changes request handling.

  • Stop the service in the server terminal by pressing Ctrl+C.
  • In server.py, add this class directly below InferenceHandler:
class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
    daemon_threads = True

How does the threaded server work?

Placing ThreadingMixIn first lets it override request processing from HTTPServer. Each incoming request can run in its own thread.

Setting daemon_threads to True gives shutdown explicit behavior. The process does not wait for request threads before exiting.

  • Find the server assignment inside the final if __name__ == "__main__" block.
  • Replace that assignment with this line:
    server = ThreadedHTTPServer((HOST, PORT), InferenceHandler)

What changes at startup?

The service now creates ThreadedHTTPServer on the same loopback address. The API boundary remains unchanged while request handling becomes concurrent.

  • Save server.py.
  • Start the threaded service by running this command in the server terminal:
python3 server.py

What should remain stable?

The service keeps the same local address and JSON contract. Only the request-handling strategy has changed.

Before you send eight cold requests, do you expect their total time to resemble one delayed inference or eight delayed inferences?

  • Switch to the client terminal.
  • Test the empty threaded cache by running the current load test:
python3 load_test.py

What changed under load?

The eight distinct requests still perform delayed inference. Separate request threads let those delays overlap.

You should see the total batch time drop sharply from the synchronous result. Scheduler activity can change the exact total.

Batch still looks serial?

Confirm the startup block creates ThreadedHTTPServer. Starting HTTPServer keeps the old one-request-at-a-time behavior.

Confirm ThreadingMixIn appears before HTTPServer in the class definition. Help me diagnose why my threaded inference server still processes requests serially.

✔️ Awesome, I've got everything!

Great. Double-check that you saved server.py before continuing.

ⓧ I'd like to double check the full code

import json
import resource
import time
from functools import lru_cache
from http.server import BaseHTTPRequestHandler, HTTPServer
from socketserver import ThreadingMixIn
from threading import Lock, active_count

from classifier import NaiveBayesTextClassifier, TRAINING_DATA


HOST = "127.0.0.1"
PORT = 8000
MAX_BODY_BYTES = 4096
SIMULATED_INFERENCE_SECONDS = 0.15

MODEL = NaiveBayesTextClassifier(alpha=1.0).fit(TRAINING_DATA)
METRICS_LOCK = Lock()
REQUESTS_TOTAL = 0


@lru_cache(maxsize=128)
def cached_predict(normalized_text):
    time.sleep(SIMULATED_INFERENCE_SECONDS)
    return MODEL.predict(normalized_text)


class InferenceHandler(BaseHTTPRequestHandler):
    def send_json(self, status, payload):
        body = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        global REQUESTS_TOTAL

        if self.path != "/predict":
            self.send_json(404, {"error": "not found"})
            return

        try:
            content_length = int(self.headers.get("Content-Length", "0"))
        except ValueError:
            content_length = 0

        if content_length <= 0 or content_length > MAX_BODY_BYTES:
            self.send_json(400, {"error": "invalid body size"})
            return

        try:
            payload = json.loads(
                self.rfile.read(content_length).decode("utf-8")
            )
        except (UnicodeDecodeError, ValueError):
            self.send_json(400, {"error": "invalid JSON"})
            return

        text = payload.get("text", "") if isinstance(payload, dict) else ""
        text = text.strip()
        if not text:
            self.send_json(400, {"error": "text is required"})
            return

        with METRICS_LOCK:
            REQUESTS_TOTAL += 1

        started = time.perf_counter()
        result = dict(cached_predict(text.lower()))
        result["latency_ms"] = round(
            (time.perf_counter() - started) * 1000.0,
            2,
        )
        self.send_json(200, result)

    def do_GET(self):
        if self.path != "/metrics":
            self.send_json(404, {"error": "not found"})
            return

        usage = resource.getrusage(resource.RUSAGE_SELF)
        cache = cached_predict.cache_info()
        with METRICS_LOCK:
            request_count = REQUESTS_TOTAL

        self.send_json(
            200,
            {
                "requests_total": request_count,
                "cache_hits": cache.hits,
                "cache_misses": cache.misses,
                "cache_entries": cache.currsize,
                "active_threads": active_count(),
                "user_cpu_seconds": round(usage.ru_utime, 4),
                "system_cpu_seconds": round(usage.ru_stime, 4),
                "max_rss_raw": usage.ru_maxrss,
                "voluntary_context_switches": usage.ru_nvcsw,
                "involuntary_context_switches": usage.ru_nivcsw,
            },
        )

    def log_message(self, format_string, *args):
        return


class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
    daemon_threads = True


if __name__ == "__main__":
    server = ThreadedHTTPServer((HOST, PORT), InferenceHandler)
    print("inference service listening on http://{0}:{1}".format(HOST, PORT))
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nstopping service")
    finally:
        server.server_close()

How should I use this file?

Compare this reference with your saved server.py. Check imports, indentation, method placement, and the final server class.

Compare cold and warm batches

The client needs named batches so the timing difference is easy to read. It also needs the metrics endpoint so the final run connects latency with cache and process behavior.

  • Switch back to load_test.py.
  • Select the existing run_batch function.
  • Replace that function by pasting this code:
def run_batch(name):
    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=8) as executor:
        results = list(executor.map(request_prediction, MESSAGES))
    elapsed_ms = round((time.perf_counter() - started) * 1000.0, 2)

    print("\n{0} batch".format(name))
    for result in results:
        print(
            "label={0:10s} score={1:.4f} server_ms={2:7.2f} client_ms={3:7.2f}".format(
                result["label"],
                result["relative_score"],
                result["latency_ms"],
                result["client_ms"],
            )
        )
    print("{0} total_ms={1:.2f}".format(name, elapsed_ms))

What does the named batch show?

The function keeps eight concurrent client workers. It prints each prediction with server-side latency and client-side latency.

The supplied name labels the total as cold or warm. This makes the comparison visible in one terminal session.

  • Find the final if __name__ == "__main__" block.
  • Replace that block with this temporary runnable version:
if __name__ == "__main__":
    run_batch("cold")

Why run one named batch first?

This small check proves the revised function accepts its new name argument. It also confirms the formatted result still contains all eight predictions.

  • Save load_test.py.
  • Check the named output by running this command in the client terminal:
python3 load_test.py

What should the client print?

The output begins with the cold batch label. Each row includes a routing label, relative score, server latency, and client latency.

You should see eight formatted rows followed by a cold total. The updated batch function is working.

  • Add this metrics URL directly below PREDICT_URL:
METRICS_URL = "http://127.0.0.1:8000/metrics"

What is this URL for?

The client uses METRICS_URL to read the running server's metrics response. It keeps prediction traffic separate from process observation.

  • Replace the temporary main block with this final section:
def read_metrics():
    with urlopen(METRICS_URL, timeout=5) as response:
        return json.loads(response.read().decode("utf-8"))


if __name__ == "__main__":
    run_batch("cold")
    run_batch("warm")
    print("\nmetrics")
    print(json.dumps(read_metrics(), indent=2, sort_keys=True))

How does the final benchmark flow?

The first batch fills the fresh cache with eight distinct results. The second batch immediately requests the same messages.

The final request reads the server metrics as formatted JSON. One run now connects timing behavior with cache and process state.

  • Save load_test.py.
  • Return to the server terminal.
  • Stop the service by pressing Ctrl+C.
  • Clear the in-memory cache by starting a fresh server process:
python3 server.py

Why restart before the comparison?

The new process begins with zero cache entries. This gives the cold batch a clean baseline before the warm batch reuses its results.

Before you run the final benchmark, which batch do you expect to show the lower server latency?

  • Switch to the client terminal.
  • Run the complete benchmark with this command:
python3 load_test.py

What should the final benchmark show?

The cold requests overlap because the server handles them in separate threads. Their exact times can vary with scheduler activity.

The warm batch is faster because all eight messages can use cached results. The metrics JSON shows request totals, cache behavior, CPU time, peak resident memory, active threads, and context switches.

You should see a cold batch followed by a faster warm batch. The final JSON should contain cache hits from the repeated messages and cache misses from the first batch.

Final benchmark not showing a warm speedup?

Restart the server before the final client run so the cold label matches an empty cache. Keep that same server process running while both batches complete.

Confirm both calls use the unchanged MESSAGES list. Help me diagnose my cold and warm inference benchmark.

✔️ Awesome, I've got everything!

Great. Double-check that you saved load_test.py after adding both batches and the metrics request.

ⓧ I'd like to double check the full code

import json
import time
from concurrent.futures import ThreadPoolExecutor
from urllib.request import Request, urlopen


PREDICT_URL = "http://127.0.0.1:8000/predict"
METRICS_URL = "http://127.0.0.1:8000/metrics"
MESSAGES = [
    "I cannot access my account after a password reset",
    "My card was charged twice for the subscription",
    "The mobile app crashes on the loading screen",
    "I feel anxious and overwhelmed today",
    "The verification code never reached my email",
    "Please send a refund and a corrected invoice",
    "The website freezes whenever I upload a file",
    "I have panic symptoms and cannot sleep",
]


def request_prediction(text):
    body = json.dumps({"text": text}).encode("utf-8")
    request = Request(
        PREDICT_URL,
        data=body,
        headers={"Content-Type": "application/json"},
    )
    started = time.perf_counter()
    with urlopen(request, timeout=5) as response:
        result = json.loads(response.read().decode("utf-8"))
    result["client_ms"] = round(
        (time.perf_counter() - started) * 1000.0,
        2,
    )
    return result


def run_batch(name):
    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=8) as executor:
        results = list(executor.map(request_prediction, MESSAGES))
    elapsed_ms = round((time.perf_counter() - started) * 1000.0, 2)

    print("\n{0} batch".format(name))
    for result in results:
        print(
            "label={0:10s} score={1:.4f} server_ms={2:7.2f} client_ms={3:7.2f}".format(
                result["label"],
                result["relative_score"],
                result["latency_ms"],
                result["client_ms"],
            )
        )
    print("{0} total_ms={1:.2f}".format(name, elapsed_ms))


def read_metrics():
    with urlopen(METRICS_URL, timeout=5) as response:
        return json.loads(response.read().decode("utf-8"))


if __name__ == "__main__":
    run_batch("cold")
    run_batch("warm")
    print("\nmetrics")
    print(json.dumps(read_metrics(), indent=2, sort_keys=True))

How should I use this file?

Compare this reference with your saved load_test.py. Check the metrics URL, named batch argument, output formatting, and final execution order.

That is the bottleneck removed. Your service now overlaps cold requests, reuses warm results, and explains its runtime behavior with Linux metrics.

Secret mission

Reject Work Before Overload Spreads

Cap the service at two concurrent inference requests. Excess cold requests receive an HTTP 503 overload response while accepted work continues normally.

Clean Up Your Resources

Clean Up Your Resources

Choose between keeping your resources running, pausing them for later, or deleting them entirely. This local project has no ongoing costs.

Resources you used:

  • A running local inference service listening at 127.0.0.1:8000.
  • The cpu-inference-lab folder containing classifier.py, server.py, and load_test.py.

Keep everything running

No action is needed while you are actively testing the service. Your files remain ready for more experiments.

  • Leave the inference service running in the integrated terminal.
  • Keep the cpu-inference-lab folder open in Visual Studio Code.

Pause - I'll come back to this later

Shut down the inference service to release its local runtime resources. Your project files remain in place for your return.

  • Return to the Visual Studio Code terminal running server.py.
  • Press Ctrl+C to stop the service.
  • Keep the cpu-inference-lab folder for your return.

You'll see stopping service in the terminal. The address 127.0.0.1:8000 stops accepting requests.

Delete - I don't want to use this again

Deleting the folder is permanent. Your Python installation remains available.

  • Return to the Visual Studio Code terminal running server.py.
  • Press Ctrl+C to stop the service.

You'll see stopping service when the local process has closed.

  • Remove the cpu-inference-lab folder by running these commands:
cd ..
rm -rf cpu-inference-lab
ls

What Do These Commands Do?

  • The first command moves one directory above cpu-inference-lab.
  • The second command removes the entire local folder.
  • The final command lists the remaining items so you can verify the removal.

You should no longer see cpu-inference-lab in the output. The local inference service is now stopped.

Nice Work!

Nice Work!

You did it! You built a CPU-only support-routing service that runs AI inference locally under measured load.

You've learned how to:

  • Build a multinomial Naive Bayes classifier from local training examples. Evaluate its held-out predictions across four support categories.
  • Serve predictions through a local JSON HTTP boundary. Reproduce synchronous queueing under concurrent load.
  • Replace serial handling with request threads. Reuse repeated predictions through a bounded cache. Observe latency. Inspect CPU time. Check peak resident memory. Track context switches.
  • Secret Mission: Protect the inference path with backpressure. Return HTTP 503 overload responses when both inference slots are busy.

Ready to quiz yourself?