diff --git a/recipe/README.md b/recipe/README.md index ebdd6036..869909ef 100644 --- a/recipe/README.md +++ b/recipe/README.md @@ -10,6 +10,12 @@ the Rust worker, export the merged weights and start the worker. - [Open-Jev-27B-v1.1 native text worker](open_jev/native.md): export the merged text backbone and trained decision head, then serve with Rust and CUDA. +- [CLM behind the frontend](clm/README.md): run CLM's own server behind the frontend on + CPU with a stub encoder, and what the response comparison has to allow for. Recipes contain setup, launch commands and examples. Reusable implementation code belongs under `src/`. + +[`compare_with_backend.py`](compare_with_backend.py) checks that the frontend returns what +the worker returned, for any recipe; [`test_compare_with_backend.py`](test_compare_with_backend.py) +covers it without a model. diff --git a/recipe/clm/README.md b/recipe/clm/README.md new file mode 100644 index 00000000..f0f77262 --- /dev/null +++ b/recipe/clm/README.md @@ -0,0 +1,92 @@ +# CLM behind the frontend + +Runs [CLM](https://github.com/Contrastive-LM/CLM)'s own `clm-serve` behind `omni-jev`, and checks that the +frontend returns what the engine returned. CLM is the second model in #9: a frozen Qwen3-8B encoder behind an +OpenAI-compatible `/v1/embeddings` endpoint, plus two projection heads and a cosine score. + +**No GPU and no 8B encoder are needed to run this.** `clm-serve` is an HTTP client of the embeddings endpoint +(`src/clm/embedder.py`), so `stub_embedder.py` can stand in for the encoder. What that exercises is the +*plumbing* — request shape, question packing, the three answer types, the serving contract, and the frontend in +front of it. It cannot tell you anything about CLM's decisions, because the vectors are not from Qwen. + +## Run it + +Three terminals, all CPU: + +```bash +# 1. a stand-in for the Qwen3-8B pooling server +python recipe/clm/stub_embedder.py --port 8090 + +# 2. CLM's own server, pointed at it (the checkpoint is 75 MB: the two heads, not the encoder) +# CLM_CKPT is the file itself; without it clm-serve looks in ~/.cache/clm and downloads. +CLM_CKPT=/tmp/CLM_v0.1-8B.pt clm-serve --port 8091 \ + --emb-url http://127.0.0.1:8090/v1/embeddings --emb-model qwen3-8b + +# 3. the frontend from #2, pointed at CLM +OMNI_JEV_BIND=127.0.0.1:8080 OMNI_JEV_BACKEND_URL=http://127.0.0.1:8091 cargo run -p omni-jev --release +``` + +```bash +python recipe/compare_with_backend.py \ + --backend http://127.0.0.1:8091 --frontend http://127.0.0.1:8080 --model clm-latest +``` + +Installing CLM without the GPU stack, since `vllm` is a hard dependency of the package but is only needed for +the encoder process: + +```bash +pip install "numpy>=1.24" requests "fastapi>=0.100" "uvicorn>=0.23" torch +pip install --no-deps "contrastive-lm @ git+https://github.com/Contrastive-LM/CLM.git" +``` + +## What passes + +The contract lines up with no adapter: status, content type and all three answer types come back through the +frontend unchanged, including CLM's `X-CLM-Latency-Ms`. + +| question | answer | +|---|---| +| `choice` | `{"type":"choice","choice":"billing","confidence":…,"probabilities":{…}}` | +| `score` | `{"type":"score","score":0.750,"confidence":…,"legend":{"0":…},"probabilities":{…}}` | +| `noul` | `{"type":"noul","noul":0.9995}` | + +## What the comparison had to learn + +`compare_with_backend.py` compared the whole response byte-for-byte, which holds for LAYA because its body is a +pure function of the request. It does not hold for an engine that reuses encoder state across requests: + +``` +same state, three times: noul=0.977197 usage.input_tokens=0 +a state not seen before: noul=0.280477 usage.input_tokens=26 +that same state again: noul=0.280477 usage.input_tokens=0 +``` + +The decision is deterministic to six decimals; `input_tokens` counts only encoder cache misses. CLM's whole +point is that candidate vectors are reusable across requests, so the field is moved by the feature that makes +it interesting. Comparing the full body reports FAIL on a correct response, and whether it does depends on +which call happened to warm the cache — so the same run can pass or fail on ordering. + +The tool now compares status, content type and the `answers` subtree, and prints the `usage` difference instead +of asserting on it. Strict equality is still the first test, so a backend whose body really is a pure function +is unaffected: + +``` +PASS department: status 200 -> 200 (usage {'billing_units': 1, 'input_tokens': 33, 'output_tokens': 0} -> {'billing_units': 1, 'input_tokens': 0, 'output_tokens': 0}) +``` + +## Open contract question + +Which response fields are allowed to differ between two otherwise identical requests? `billing_units` looks +stable; `input_tokens` does not. If the project wants the strong form — the whole body identical — then +`input_tokens` has to mean "tokens the request required" rather than "tokens this call paid for", which is a +decision for the engine, not for the frontend. + +## With a real encoder + +Replace step 1 with the upstream script and the answers become meaningful: + +```bash +GPU=0 PORT=8090 UTIL=0.35 ./serve_qwen3_8b.sh # from the CLM checkout; needs CUDA +``` + +Everything downstream is unchanged, which is the property this recipe is meant to demonstrate. diff --git a/recipe/clm/stub_embedder.py b/recipe/clm/stub_embedder.py new file mode 100644 index 00000000..abb49e4e --- /dev/null +++ b/recipe/clm/stub_embedder.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Deterministic /v1/embeddings stand-in for the Qwen3-8B pooling server. + +Why this exists: `clm-serve` is an HTTP client of an OpenAI-compatible embeddings +endpoint (`src/clm/embedder.py`), so the engine, the packing, the scoring and the +HTTP API can all be exercised without a GPU or the 8B encoder. The vectors here are +not meaningful - they are a hash-derived direction on the unit sphere - so this +verifies the *plumbing* (request shape, ordering, cache, the three question types, +the serving contract), never the model's decisions. + + python stub_embedder.py --port 8090 --dim 4096 +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import struct +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +DIM = 4096 + + +def vector(text: str, dim: int) -> list[float]: + """A stable unit vector per text: two hash digests expanded by a counter.""" + out: list[float] = [] + counter = 0 + while len(out) < dim: + digest = hashlib.sha256(f"{counter}:{text}".encode()).digest() + for i in range(0, len(digest) - 3, 4): + if len(out) == dim: + break + word = int.from_bytes(digest[i:i + 4], "big") + out.append(word / 2**31 - 1.0) + counter += 1 + norm = sum(v * v for v in out) ** 0.5 or 1.0 + return [v / norm for v in out] + + +class Handler(BaseHTTPRequestHandler): + dim = DIM + + def log_message(self, *args): # keep the run quiet + pass + + def _json(self, status: int, payload: dict) -> None: + body = json.dumps(payload).encode() + 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_GET(self) -> None: # noqa: N802 + if self.path.startswith("/v1/models"): + self._json(200, {"object": "list", "data": [{"id": "qwen3-8b", "object": "model"}]}) + else: + self._json(404, {"error": "not found"}) + + def do_POST(self) -> None: # noqa: N802 + if not self.path.startswith("/v1/embeddings"): + self._json(404, {"error": "not found"}) + return + length = int(self.headers.get("Content-Length") or 0) + body = json.loads(self.rfile.read(length) or b"{}") + texts = body.get("input") or [] + if isinstance(texts, str): + texts = [texts] + data = [] + tokens = 0 + for index, text in enumerate(texts): + vec = vector(text, Handler.dim) + raw = struct.pack(f"<{len(vec)}f", *vec) + data.append({"object": "embedding", "index": index, + "embedding": base64.b64encode(raw).decode()}) + tokens += max(1, len(text) // 4) + self._json(200, {"object": "list", "data": data, "model": body.get("model", "qwen3-8b"), + "usage": {"prompt_tokens": tokens, "total_tokens": tokens}}) + + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--port", type=int, default=8090) + parser.add_argument("--dim", type=int, default=DIM) + args = parser.parse_args() + Handler.dim = args.dim + server = ThreadingHTTPServer(("127.0.0.1", args.port), Handler) + print(f"stub embedder on http://127.0.0.1:{args.port} (dim {args.dim})", flush=True) + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/recipe/laya/compare_with_backend.py b/recipe/compare_with_backend.py similarity index 56% rename from recipe/laya/compare_with_backend.py rename to recipe/compare_with_backend.py index 0aafdf47..9f408fc9 100644 --- a/recipe/laya/compare_with_backend.py +++ b/recipe/compare_with_backend.py @@ -1,9 +1,12 @@ #!/usr/bin/env python3 -"""Check that a running omni-jev frontend returns exactly what its worker returns. +"""Check that a running omni-jev frontend returns what its worker returns. Sends the same health and decision requests to the worker directly and through -the frontend, then compares status, Content-Type and body bytes. Covers choice, -score and noul questions separately and together. Standard library only. +the frontend and compares them. Byte equality is tried first; where that fails +and both sides answered 200, the parsed ``answers`` subtrees are compared +instead, so a response whose only difference is the ``usage`` envelope still +passes. Covers choice, score and noul questions separately and together. +Standard library only. """ import argparse @@ -48,6 +51,29 @@ def fetch(base, path, body=None): return response.status, response.headers.get("Content-Type"), response.read() +def decision(body): + """The part of a response the frontend must not change: None when body is not a decision. + + ``usage`` is deliberately excluded. A backend that reuses encoder state across requests + charges only for the calls it paid for, so an identical request can report different + ``input_tokens`` depending on whether the work was already cached -- CLM does exactly + this. The answer must still be identical; the envelope is reported, not asserted. + + Comparing parsed answers rather than bytes also makes key order and whitespace + irrelevant, which is wider than "usage only" and is the intent: what the + frontend must not change is the decision, not its serialisation. A backend + whose body is a pure function is still held to byte equality, because that is + what is checked first. + """ + try: + parsed = json.loads(body) + except ValueError: + return None + if not isinstance(parsed, dict) or "answers" not in parsed: + return None + return parsed["answers"], parsed.get("usage") + + def main(): parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--backend", default="http://127.0.0.1:8000") @@ -66,8 +92,17 @@ def main(): direct = fetch(args.backend, path, body) proxied = fetch(args.frontend, path, body) ok = direct == proxied and direct[0] == 200 + note = "" + if not ok and direct[0] == 200: + # Compare the decision itself before falling back to the byte comparison. Only a + # successful status relaxes the body comparison: a 500 that happens to carry + # equal "answers" is still a failure. + left, right = decision(direct[2]), decision(proxied[2]) + if left and right and direct[:2] == proxied[:2] and left[0] == right[0]: + ok = True + note = f" (usage {left[1]} -> {right[1]})" failed += not ok - print(f"{'PASS' if ok else 'FAIL'} {name}: status {direct[0]} -> {proxied[0]}") + print(f"{'PASS' if ok else 'FAIL'} {name}: status {direct[0]} -> {proxied[0]}{note}") if body is not None: print(f" {proxied[2].decode(errors='replace')}") sys.exit(1 if failed else 0) diff --git a/recipe/laya/README.md b/recipe/laya/README.md index 254c12ca..be587eee 100644 --- a/recipe/laya/README.md +++ b/recipe/laya/README.md @@ -45,7 +45,7 @@ curl http://127.0.0.1:8080/v1/systemone \ With both services running: ```sh -python3 recipe/laya/compare_with_backend.py --model english \ +python3 recipe/compare_with_backend.py --model english \ --backend http://127.0.0.1:8000 --frontend http://127.0.0.1:8080 ``` diff --git a/recipe/laya/apple-silicon.md b/recipe/laya/apple-silicon.md index a779de94..28f5bb85 100644 --- a/recipe/laya/apple-silicon.md +++ b/recipe/laya/apple-silicon.md @@ -142,8 +142,9 @@ curl http://127.0.0.1:8080/v1/systemone \ -d '{"model":"english","state":"Please refund the duplicate charge.","questions":{"refund":{"type":"noul","instructions":"Does the customer ask for a refund?"}}}' ``` -The frontend forwards the worker's response unchanged; `compare_with_backend.py` from the -[Laya text worker](README.md#compare-responses) recipe checks that against this setup as well. +The frontend forwards the worker's response unchanged; the shared +[`compare_with_backend.py`](../compare_with_backend.py), described in the +[Laya text worker](README.md#compare-responses) recipe, checks that against this setup too. ## Test diff --git a/recipe/test_compare_with_backend.py b/recipe/test_compare_with_backend.py new file mode 100644 index 00000000..678ec3dc --- /dev/null +++ b/recipe/test_compare_with_backend.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Regression tests for compare_with_backend.py. + +Stands up two throwaway HTTP servers -- one acting as the worker, the other as the +frontend -- and checks that the comparison passes when only ``usage`` differs but still +fails when an answer or a status differs. Runs without a model, a GPU or the Rust binary. + + python recipe/test_compare_with_backend.py +""" + +from __future__ import annotations + +import json +import subprocess +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +HERE = Path(__file__).resolve().parent +TOOL = HERE / "compare_with_backend.py" + +ANSWERS = {"pick": {"type": "noul", "noul": 0.75}} +HEALTH = {"ok": True} + + +class Server(ThreadingHTTPServer): + """A worker whose response the test controls, plus a call counter.""" + + daemon_threads = True + + def __init__(self) -> None: + super().__init__(("127.0.0.1", 0), Handler) + self.calls = 0 + self.answers = ANSWERS + self.usage: dict = {"input_tokens": 0} + # When set, every other call reports this instead -- a cache that goes warm. + self.alternate_usage: dict | None = None + self.status = 200 + # The body sent with a non-200 status; normally an empty object. + self.error: dict = {} + + @property + def url(self) -> str: + return f"http://127.0.0.1:{self.server_address[1]}" + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): # keep the test output readable + pass + + def _send(self, status: int, body: bytes) -> None: + 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_GET(self) -> None: # noqa: N802 + self._send(200, json.dumps(HEALTH).encode()) + + def do_POST(self) -> None: # noqa: N802 + server: Server = self.server # type: ignore[assignment] + self.rfile.read(int(self.headers.get("Content-Length") or 0)) + server.calls += 1 + if server.status != 200: + self._send(server.status, json.dumps(server.error).encode()) + return + usage = dict(server.usage) + if server.alternate_usage is not None and server.calls % 2 == 0: + usage.update(server.alternate_usage) + self._send(200, json.dumps({"model": "m", "answers": server.answers, "usage": usage}).encode()) + + +def run(backend: str, frontend: str) -> tuple[int, str]: + proc = subprocess.run( + [sys.executable, str(TOOL), "--backend", backend, "--frontend", frontend, "--model", "m"], + capture_output=True, + text=True, + timeout=60, + ) + return proc.returncode, proc.stdout + proc.stderr + + +def serve(server: Server) -> None: + threading.Thread(target=server.serve_forever, daemon=True).start() + + +def reset(*servers: Server) -> None: + for server in servers: + server.calls = 0 + server.answers = ANSWERS + server.usage = {"input_tokens": 0} + server.alternate_usage = None + server.status = 200 + server.error = {} + + +def main() -> None: + worker, frontend = Server(), Server() + serve(worker) + serve(frontend) + results: list[tuple[str, bool, str]] = [] + + def check(name: str, condition: bool, detail: str = "") -> None: + results.append((name, condition, detail)) + + # 1. Identical responses: the strict comparison passes. + reset(worker, frontend) + code, out = run(worker.url, frontend.url) + check("identical responses pass", code == 0 and "FAIL" not in out, out) + + # 2. Only usage differs -- the cache-aware case. Must pass, and say so. + reset(worker, frontend) + worker.alternate_usage = {"input_tokens": 26} + code, out = run(worker.url, frontend.url) + check("usage-only difference passes and is reported", + code == 0 and "FAIL" not in out and "input_tokens" in out, out) + + # 3. An answer differs: must still fail. + reset(worker, frontend) + frontend.answers = {"pick": {"type": "noul", "noul": 0.1}} + code, out = run(worker.url, frontend.url) + check("a differing answer fails", code == 1 and "FAIL" in out, out) + + # 4. A status difference must fail even when the answers match. + reset(worker, frontend) + frontend.status = 503 + code, out = run(worker.url, frontend.url) + check("a status difference fails", code == 1 and "FAIL" in out, out) + + # 5. A backend that is not 200 must fail regardless of the frontend. + reset(worker, frontend) + worker.status = 500 + code, out = run(worker.url, frontend.url) + check("a non-200 backend fails", code == 1 and "FAIL" in out, out) + + # 6. Matching 500s are not a pass, even when the error body carries equal answers. + reset(worker, frontend) + worker.status = 500 + frontend.status = 500 + worker.error = {"answers": {}, "detail": "failed"} + frontend.error = {"answers": {}, "detail": "failed"} + code, out = run(worker.url, frontend.url) + check("equal answers on matching errors fail", code == 1 and "FAIL" in out, out) + + worker.shutdown() + frontend.shutdown() + + failed = 0 + for name, ok, detail in results: + print(f" {'ok ' if ok else 'FAIL'} {name}") + if not ok: + failed += 1 + print(" " + detail.strip().replace("\n", "\n ")[:400]) + print("compare_with_backend: ok" if not failed else f"compare_with_backend: {failed} FAILED") + sys.exit(1 if failed else 0) + + +if __name__ == "__main__": + main()