跳到正文
DEV Community· Alex Chen·· 4 小時前AI 評分56

如何在信任免費模型方案前建立簡易 Token 計量器

Build a Tiny Token Meter Before You Trust a Free Model Tier

AI 導讀

作者為免費模型端點建立了一個以 Python 標準函式庫實作的 HTTP Token 計量代理,按上游 usage 記錄每次用量至 JSONL,並在每日用量達上限後拒絕新請求。若端點缺少 usage,計量器會把用量記作 0;而上限檢查與記錄之間的競態也可能令並發請求同時通過。

正文

It is 11:40pm and a retry loop is quietly eating someone's free tier.

Not with a crash. With a for loop that never checks what came back, a client that retries on any non-200, and a usage dashboard nobody has opened since Tuesday. By morning the quota is gone, the demo is tomorrow, and the error message says nothing useful.

That is the churn problem with free tiers, and it is what the last week of DEV posts kept circling: we are tired of AI, we are drowning in open-source AI repos, and the floor under our toys keeps moving. So here is the question I keep asking my study group. If you cannot see how many tokens a request cost, how do you know the free tier is still free?

For this case study I built one mechanism — a token meter — and put it in front of a free model endpoint. Dependency count: zero, beyond the Python standard library.

Goal, stated so it can fail

The meter should do three things and nothing else. Count tokens per request from the upstream's own usage block, append one JSON line per call to a ledger, and refuse new calls once today's total crosses a hard cap.

Three things, three ways to be wrong. Concurrency, missing usage data, and my own optimism about what the endpoint returns.

The mechanism

One file, meter.py. It is an HTTP proxy: you point your client at it, it forwards to the upstream, it records what came back.

# meter.py — a tiny token meter for one free model endpoint.
import json, os, time, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib import request as urlrequest

UPSTREAM = os.environ.get("UPSTREAM_URL", "")
TOKEN    = os.environ.get("UPSTREAM_TOKEN", "")
DAY_CAP  = int(os.environ.get("DAY_TOKEN_CAP", "200000"))
LEDGER   = os.environ.get("LEDGER_PATH", "ledger.jsonl")
PORT     = int(os.environ.get("PORT", "8080"))
STUB     = os.environ.get("METER_STUB") == "1"

_lock = threading.Lock()

def _spent_today():
    today, total = time.strftime("%Y-%m-%d"), 0
    try:
        with open(LEDGER) as f:
            for line in f:
                row = json.loads(line)
                if row["day"] == today:
                    total += row["total_tokens"]
    except FileNotFoundError:
        pass
    return total

def _record(prompt_tokens, completion_tokens):
    row = {"day": time.strftime("%Y-%m-%d"), "ts": time.time(),
           "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens,
           "total_tokens": prompt_tokens + completion_tokens}
    with _lock:
        with open(LEDGER, "a") as f:
            f.write(json.dumps(row) + "\n")
    return row

The upstream call has a stub branch, and that branch is the whole reason this article is testable at 11:40pm without spending anything.

def _call_upstream(payload):
    if STUB:
        text = payload["messages"][-1]["content"]
        return {"choices": [{"message": {"role": "assistant", "content": "stub:" + text}}],
                "usage": {"prompt_tokens": 12, "completion_tokens": 8}}

    req = urlrequest.Request(
        UPSTREAM,
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json",
                 "Authorization": f"Bearer {TOKEN}"},
        method="POST")
    with urlrequest.urlopen(req, timeout=60) as r:
        return json.loads(r.read().decode())

class Meter(BaseHTTPRequestHandler):
    def do_POST(self):
        try:
            body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        except (ValueError, KeyError) as exc:
            return self._send(400, {"error": {"type": "bad_request", "detail": str(exc)}})

        spent = _spent_today()
        if spent >= DAY_CAP:
            return self._send(429, {"error": {"type": "day_cap_reached",
                                              "spent": spent, "cap": DAY_CAP}})

        data = _call_upstream(body)
        usage = data.get("usage") or {}
        row = _record(usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0))
        data["meter"] = row
        self._send(200, data)

    def _send(self, code, obj):
        raw = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.end_headers()
        self.wfile.write(raw)

if __name__ == "__main__":
    ThreadingHTTPServer(("0.0.0.0", PORT), Meter).serve_forever()

Notice what is not there: streaming, auth, a database. Hold that thought.

Run it before you trust it

No pip install. If python3 -c "import http.server" exits quietly you have everything.

METER_STUB=1 LEDGER_PATH=/tmp/ledger.jsonl PORT=8080 python3 meter.py &

curl -s localhost:8080/v1/chat \
  -d '{"messages":[{"role":"user","content":"hi"}]}' | python3 -m json.tool

The output is deterministic by construction — same bytes every run, no network involved. You should see the stub content, a usage block, and a meter block reporting total_tokens: 20. Now cat /tmp/ledger.jsonl and you have one row. Two calls, two rows. That is the entire contract.

Then trip the cap on purpose. Restart with DAY_TOKEN_CAP=10, call once, call again. The second call returns 429 with spent: 20, cap: 10. A guard you have never seen fire is a guard you do not have.

Now break it the way reality will

Change the stub so it omits usage entirely: return {"choices": [...]} with no usage key.

Run the same curl. HTTP 200. A friendly answer. And total_tokens: 0 in your ledger, forever, no matter how many calls you make. Your cap never trips, your dashboard is a flat line, and your free tier disappears on schedule while your meter swears you spent nothing.

This is the lesson I keep relearning: metering is not counting what you sent, it is counting what the provider admits it charged. An endpoint that omits usage — or reports zeros for cached prompts — silently converts your hard cap into decoration. The fix is boring. Treat usage as untrusted input: if it is missing, either reject the response or fall back to a character-count estimate and mark the row "estimated": true. Then, once a week, compare your ledger total against whatever the provider's own dashboard shows. Two numbers that disagree are a bug report, not a rounding error.

Putting the meter on a free server

A meter that lives on your laptop is a meter that stops when the lid closes. So the useful move is to run it on the free server option instead of locally, bind it to whatever port the platform hands you via PORT, and let your notebook, your phone, and that 11:40pm retry script all talk to one address.

That is the setup I use with MonkeyCode: free model access as the upstream, the free server option as the host, this proxy in between so every call leaves a receipt. The project is open source, and the free tier is 10 million tokens as quoted to me by the operator — go verify the current figure on their page yourself, because that is exactly the kind of number that churns.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

One deployment detail worth stealing: subclass and override log_message to emit a single JSON line, or the default access log will fill your server's log viewer with what look like errors. On a free box, a scary log costs you an hour of debugging nothing.

Results, limitations, and who should walk away

With the stub, the meter is deterministic and the cap fires exactly when the arithmetic says it should. Against a live endpoint, the numbers are only as good as the usage block, and the day cap is checked before the call and recorded after it — so two requests arriving at once with five tokens left will both pass. That is a real race, not a theoretical one, and a single lock around check-and-record is the honest fix if you care.

Other things this design does not do. It has no auth, so do not expose it publicly without a shared secret header. It buffers whole responses, so streaming clients will hang — a streaming variant needs to parse the final chunk for usage. Its ledger is a JSONL file, which is fine for one student and wrong for a team. And it cannot police per-user quotas, only a global daily ceiling.

Who should skip this: anyone on a paid plan whose dashboard already breaks spend down per key, anyone running offline batch jobs where one spreadsheet beats one daemon, and anyone whose endpoint rejects non-standard client behaviour — a proxy is a client, and some gateways dislike that.

What I actually took away from this build is smaller than the code. Before you reach for a framework, write the fifteen lines that record what a single request cost, and then try to make those fifteen lines lie to you. They will, in the missing-usage case, and you will learn more from that one bad fixture than from a week of feature work.

So here is a prediction test for you. Two requests, cap at 25 tokens, ledger at 20, both arriving in the same millisecond. Which one gets the 429, and what does the ledger say afterward? If your answer is "neither, and 40," you already understand why the lock exists.

來源:DEV Community · dev.to