Quick Start · Jope Inference Server

⏱ 約 30 分鐘跑完🐍 Python 3.11+📦 FastAPI · pyzmq · msgpack

兩支 process。一支跑 HTTP 提供管理端 endpoint（載入模型、觸發 training、health check）；另一支跑 ZMQ REP socket 處理即時 predict。先用 50 行以內 Python 把兩邊 mock 起來，用 test client 驗證，之後再換成真的 PLS+Ridge 模型。

REST · 管理面

Cold path

:5556

載入模型 · POST /model/load
啟動 training · POST /training/start
查進度 · GET /training/{id}
健康檢查 · GET /health

套件: FastAPI + uvicorn

ZMQ · 推論面

Hot path

:5555

Heartbeat · ping / pong
即時預測 · predict / predict_reply
模型資訊 · model_info
Latency · ≤ 20 ms p95 來回

套件: pyzmq + msgpack

安裝套件

四個 library，一行 install。

 1pip install fastapi uvicorn[standard] pyzmq msgpack scikit-learn numpy

fastapi · REST framework — 用 decorator 寫 endpoint，自動產生 OpenAPI schema
uvicorn[standard] · 跑 FastAPI 用的 ASGI server
pyzmq · Python 的 ZMQ binding — 我們只用 REP socket
msgpack · ZMQ 上的 binary 序列化格式（spec 規定用這個，不是 JSON）

Management REST · 25 行

最小的 FastAPI app，提供 /health 和 /model/list。

 1# 檔名: management_api.py
 2# 執行: uvicorn management_api:app --host 0.0.0.0 --port 5556
 3from fastapi import FastAPI
 4
 5app = FastAPI(title="Jope Inference · Management API")
 6
 7@app.get("/health")
 8def health():
 9    return {
10        "status": "ok",
11        "model_loaded": True,
12        "active_version": "v5",
13        "uptime_seconds": 0,
14        "server_version": "0.1.0",
15        "protocol_version": 1,
16        "python_version": "3.11.4",
17    }
18
19@app.get("/model/list")
20def list_models():
21    return {
22        "active": "v5",
23        "models": [{"version": "v5", "status": "active",
24                    "trained_at": "2026-03-15T10:00:00Z"}],
25    }

驗證

 1curl http://localhost:5556/health
 2curl http://localhost:5556/model/list

回傳的 JSON 形狀應該跟 REST reference 上的一致。

Predict worker · 25 行

Mock REP socket，把 ping 和 predict 的回應寫死。

 1# 檔名: inference_worker.py
 2# 執行: python inference_worker.py
 3import time, uuid, zmq, msgpack
 4
 5ctx  = zmq.Context(io_threads=1)
 6sock = ctx.socket(zmq.REP)
 7sock.bind("tcp://0.0.0.0:5555")
 8print("Inference worker ready on :5555")
 9
10while True:
11    req = msgpack.unpackb(sock.recv(), raw=False)
12
13    if req["type"] == "ping":
14        reply_body  = {"server_version": "0.1.0", "protocol_version": 1,
15                       "model_version": "v5", "uptime_seconds": 0}
16        reply_type  = "pong"
17
18    elif req["type"] == "predict":
19        # TODO: 之後換成你的 PLS+Ridge 模型
20        reply_body  = {"concentrations": {"EPA": 5.23, "DHA": 3.19, "DPA": 1.82},
21                       "confidence":     {"EPA": 0.95, "DHA": 0.92, "DPA": 0.88},
22                       "model_version": "v5", "inference_ms": 8.3}
23        reply_type  = "predict_reply"
24
25    sock.send(msgpack.packb({
26        "v": 1, "type": reply_type,
27        "id": str(uuid.uuid4()),
28        "correlation_id": req["id"],
29        "ts": time.time(), "body": reply_body, "error": None,
30    }, use_bin_type=True))

驗證 · 另開一個 terminal 跑這段 client

 1# 檔名: test_client.py
 2import time, uuid, zmq, msgpack
 3
 4ctx  = zmq.Context()
 5sock = ctx.socket(zmq.REQ)
 6sock.connect("tcp://localhost:5555")
 7
 8sock.send(msgpack.packb({
 9    "v": 1, "type": "predict",
10    "id": str(uuid.uuid4()), "ts": time.time(),
11    "body": {
12        "spectrum": {
13            "wavenumbers":  [200.0 + i*1.5 for i in range(2048)],
14            "intensities":  [0.0] * 2048,
15            "channel": 1, "integration_ms": 1000, "scan_seq": 1,
16        },
17        "context": {"batch_id": "DEV-001", "port": "extract-E1", "column_index": 3},
18    },
19}, use_bin_type=True))
20
21reply = msgpack.unpackb(sock.recv(), raw=False)
22print("Got:", reply["body"])

預期輸出：Got: {'concentrations': {'EPA': 5.23, 'DHA': 3.19, 'DPA': 1.82}, ...}

從 spec 產 server stub

一行指令產出 FastAPI 骨架，route 和 schema 都接好。

 1# 一行指令從 OpenAPI spec 產出 server stub
 2npx @openapitools/openapi-generator-cli generate \
 3  -i https://jope-docs.pages.dev/specs/openapi.yaml \
 4  -g python-fastapi \
 5  -o ./server-stub

產出的 ./server-stub 底下有 models/（每個 schema 對應一個 Pydantic class）和 apis/（每個 tag 一個 APIRouter）。打開產出的 main.py，把邏輯接進 handler 就完成 scaffolding。

Hosted mock 做線路測試

契約跟正式 server 一樣。任一邊還沒好時都可以用。

這個 docs 站內建一支 mock Cloudflare Pages Function，回應契約跟正式 server 一致。 Python 端還沒好、但 C# Operator Console 要先驗，可以對 mock 打；反過來，要確認 Python client 送的 request 形狀對不對，也可以對 mock 驗。

GET https://jope-docs.pages.dev/mock-api/healthGET https://jope-docs.pages.dev/mock-api/model/listPOST https://jope-docs.pages.dev/mock-api/model/loadPOST https://jope-docs.pages.dev/mock-api/training/startGET https://jope-docs.pages.dev/mock-api/training/{id}

或者直接到 REST reference 頁、在每個 operation 上點 Try It，直接從瀏覽器打 mock 看回應。

實作細節提醒

容易漏掉的幾個點，先放在這。

Envelope 的 v 永遠是 1。Client 在 handshake 會檢查 version，對不上就拒收。明確寫 v: 1。
correlation_id 要原樣回傳。Console 靠這個欄位把 reply 配對到 request，漏掉的話每個 predict 看起來都像 timeout。
Pack 時加 use_bin_type=True。MessagePack 有兩種 string 模式，C# client 等新版（bin），漏掉會造成非 ASCII 字編碼不一致。
REP socket 長連，不要每 request 重連。ZMQ socket 本來就是設計給長連用的。每封訊息重連，20 ms 的 latency budget 都會被 socket 建立吃光。
Deterministic 錯誤回 retryable: false。MODEL_NOT_LOADED、INVALID_SPECTRUM、SPECTRUM_OUT_OF_RANGE 這類 retry 也沒用，回 false 讓 client 不浪費 cycle。

進一步

REST Reference

30 分鐘把 server 跑起來