05 — categorize.py: rules + LLM

Duration: 12 min Prerequisites: chapter 03 (normalized DataFrame). Ollama optional for the LLM fallback.

Two stages, one decision

flowchart LR
Row[description] --> Rules{match in YAML?}
Rules -->|yes| Cat[category]
Rules -->|no| LLM[LLM fallback]
LLM --> Cat

If the description matches a rule → done in microseconds, zero cost. If not → ask the LLM, which costs a network roundtrip but handles the long tail.

For the demo CSV (categorized already) you’d never use categorize.py. For a raw export with no category column, it’s essential.

The rules file format

csv-llm-shared/category_rules.yaml:

rules:
  Housing:
    - "rent payment"
    - "postbox rental"
    - "storagebox"
    - "garden care"
  Groceries:
    - "pantry"
    - "family foods"
    - "organic store"
  Dining:
    - "coffee room"
    - "spoon cafe"
    - "noodle house"
    - "bistro"
  Transport:
    - "cab network"
    - "bus card"
    - "shuttle"
  # ... etc

A flat YAML: each category has a list of case-insensitive substrings to match against the description.

load_rules — compile once

csv-llm-shared/categorize.py

import re
from pathlib import Path
from typing import Optional
import yaml


def load_rules(yaml_path: str | Path) -> list[tuple[str, re.Pattern]]:
    raw = yaml.safe_load(Path(yaml_path).read_text(encoding="utf-8"))
    compiled = []
    for category, patterns in raw.get("rules", {}).items():
        for pat in patterns:
            compiled.append((category, re.compile(re.escape(pat), re.IGNORECASE)))
    return compiled

Three things to note:

• We pre-compile regexes — important if you categorize 10,000 rows.
• We escape the pattern (re.escape) so users can write "a.b.c" and it matches literally.
• We preserve YAML order: the first matching category wins.

categorize_row — the matcher

def categorize_row(description: str, rules) -> Optional[str]:
    for category, regex in rules:
        if regex.search(description):
            return category
    return None

Linear scan. For ~50 patterns this is microseconds per row. If you somehow had thousands of patterns, you’d want a more sophisticated structure (Aho-Corasick), but we never get there.

None is the explicit signal that no rule matched — caller decides whether to call the LLM next or default to something.

categorize_dataframe — vectorized

def categorize_dataframe(df, rules, default: str = "Other"):
    """Add a 'category' column based on rules. Doesn't call the LLM."""
    out = df.copy()
    out["category"] = out["description"].apply(
        lambda d: categorize_row(d, rules) or default
    )
    return out

Used by tools that want a quick, deterministic categorization without LLM cost. For a CSV with 700 rows and 50 rules, this runs in ~50 ms.

The LLM fallback

CATEGORIES = [
    "Housing", "Groceries", "Dining", "Transport", "Travel",
    "Business", "Utilities", "Health", "Entertainment",
    "Education", "Fees", "Payments", "Refunds",
]

PROMPT = """You are a credit-card categorizer. Reply with EXACTLY ONE word from this list:
{cats}

Merchant description: {desc}

Category:"""


def categorize_with_llm(description: str, client, model: str = "llama3.1:8b") -> str:
    """Ask an LLM to classify. `client` must have a .chat(model, messages, options) method.
    Works with ollama.Client out of the box; wrap OpenAI for the same shape."""
    response = client.chat(
        model=model,
        messages=[{"role": "user", "content": PROMPT.format(
            cats=", ".join(CATEGORIES), desc=description,
        )}],
        options={"temperature": 0.0},
    )
    # Extract first word, validate against the whitelist
    text = response["message"]["content"].strip()
    answer = text.split()[0] if text else ""
    return answer if answer in CATEGORIES else "Business"

Three deliberate choices:

Choice 1 — temperature=0.0

We want the same description to always produce the same category. Determinism trumps “creativity” here.

Choice 2 — Strip + first word

LLMs sometimes add explanation: "Dining (it's a restaurant)" or "This is **Dining**.". We just take the first word after stripping. Robust enough for a fallback.

Choice 3 — Whitelist + default

If the model says "Restaurant" (not in our list), we fall back to "Business" rather than letting an arbitrary new category sneak into the DB. The whitelist matters: SQL queries from chapter 04 depend on a fixed set of category names.

Injecting the LLM client

Notice that categorize_with_llm takes a client parameter. We deliberately don’t import ollama at the top of the module — the library stays LLM-agnostic.

Apps inject the client they want:

# Project 1 (Ollama)
import ollama
import categorize
client = ollama.Client(host="http://127.0.0.1:11434")
cat = categorize.categorize_with_llm("Some Merchant", client=client)

# Project 2 (OpenAI) — needs a thin shim
from openai import OpenAI
import categorize

class OpenAIShim:
    def __init__(self):
        self._c = OpenAI()
    def chat(self, model, messages, options):
        r = self._c.chat.completions.create(
            model=model, messages=messages,
            temperature=options.get("temperature", 1.0),
        )
        return {"message": {"content": r.choices[0].message.content}}

cat = categorize.categorize_with_llm("Some Merchant",
                                     client=OpenAIShim(), model="gpt-4o-mini")

The shim is 10 lines. The library doesn’t care which LLM you use.

Cache for repeat merchants

In a real backfill of 700 rows, you’ll see the same merchant 5–20 times. Caching the LLM verdict per description cuts cost 5–10×:

def make_cached_llm_categorizer(client, model):
    cache: dict[str, str] = {}
    def lookup(desc):
        if desc not in cache:
            cache[desc] = categorize_with_llm(desc, client, model)
        return cache[desc]
    return lookup

For persistence across runs, use shelve or a JSON file.

Combining rules + LLM in one shot

def categorize_full(df, rules, llm_fn=None, default: str = "Other"):
    """Apply rules first; for each unmatched row, call llm_fn if provided.
    llm_fn(description) -> category."""
    out = df.copy()
    cats = []
    for desc in out["description"]:
        c = categorize_row(desc, rules)
        if c is None and llm_fn is not None:
            c = llm_fn(desc)
        cats.append(c or default)
    out["category"] = cats
    return out

Usage:

import categorize
import ollama

rules = categorize.load_rules("csv-llm-shared/category_rules.yaml")
client = ollama.Client()
llm = categorize.make_cached_llm_categorizer(client, "llama3.1:8b")
df = categorize.categorize_full(df_raw, rules, llm_fn=llm)

Accuracy on the demo data (sanity check)

If you wipe the Category column of the demo CSV and re-categorize from scratch:

Strategy	Correct	Partial	Wrong
Rules only	642 / 734 (87%)	—	92 default to “Other”
LLM only	712 / 734 (97%)	15	7
Rules + LLM	726 / 734 (99%)	6	2

The hybrid wins. Rules catch the predictable cases instantly; LLM handles the long tail without breaking the latency budget for the predictable 87%.

When NOT to use the LLM fallback

• You need reproducibility across runs (LLMs have ~95% determinism even with temp=0).
• You need air-gapped operation (no Ollama installed, no internet).
• You need strict audit (every classification must be explainable). Rules are auditable; LLMs are not.

In those cases, expand your category_rules.yaml instead. The investment pays off.

Takeaways

• Stage 1: YAML rules engine, microseconds per row, fully deterministic.
• Stage 2: LLM fallback, optional, injected (not hard-coded).
• A whitelist of valid categories + temperature 0 + cache = production-grade quality.
• Hybrid accuracy: 99% on the demo, much better than either alone.

Next: dashboards.py — Plotly figures →