一个 8MB 的模型文件，比很多 App 的图标资源还小。它做不了写文章、写代码这类活，但"把这段客服消息拆成结构化字段""判断这条工单该派给谁"这种每天要跑几万次的重复判断，它能在本机、离线、几百毫秒内做完，不用 GPU，也不用起一个云端大模型服务。
这篇用一个具体任务从头走一遍：把用户发来的售后消息，抽成 {品类, 问题类型, 是否退款, 紧急度} 这样一段 JSON。做完你会拿到一条能跑的流水线，以及一套"该选 8MB 还是 29MB 档位"的判断方法。
先说清楚分工。云端大模型（比如 DeepSeek 的 Flash 这类轻量档位）知识广、推理强，适合定规则、做兜底、处理长尾。微型端侧模型适合跑量：任务固定、输出格式固定、判定边界清晰。两个配合，成本能降下来一个数量级。具体价格与额度以各家官方页面为准，这里不展开。
前置条件
- 一台开发机，macOS / Linux / Windows + WSL 均可，内存 4GB 以上足够
- Python 3.10+（具体版本要求以官方文档为准）
- 能访问 Cactus 的官方仓库与模型发布页
- 20～50 条你自己业务里的真实样本，手工标好答案
- 磁盘留出几百 MB，模型本体很小，但缓存和日志要占地方
最后一条最容易被跳过，也最影响成败。没有评估集，你没法判断该用哪个体积档，只能凭感觉。
第 1 步：先把任务写成一句话和一张表
别急着下模型。先把它写成可验收的形式：
- 输入：一段 20～500 字的用户消息（可能有错别字、口语、表情）
- 输出：固定 4 个字段的 JSON，每个字段都是枚举值
- 兜底：任一字段时间解析不出，置为
"unknown"，不允许瞎猜
```json
{
"category": ["数码", "服饰", "食品", "家居", "unknown"],
"issue": ["质量问题", "物流问题", "描述不符", "不会用", "unknown"],
"refund": ["是", "否", "unknown"],
"urgency": ["高", "中", "低"]
}
```
字段越少、枚举越短，微型模型越稳。四个字段、每个三到五个选项，是 8MB 档位能扛住的量级。如果你想要十几个字段加自由文本，直接跳到 29MB 档位，或者干脆拆成两次调用。
第 2 步：挑体积档位，别一上来就用大的
同一条模型线一般会放出几个体积档，差别主要来自参数量和量化精度。命名里的数字通常对应文件大小量级，但具体文件名和体积以官方发布页当前版本为准。
选档位的顺序是这样的：
1. 永远从最小档开始跑。8MB 能过，就不要用 29MB——体积每翻一倍，内存占用、冷启动时间、包体积都会跟着涨。
2. 用第 6 步的评估脚本给出准确率。先定一个及格线，比如字段级准确率 95%。
3. 最小档不达标，升一档重测。两档之间差距通常比想象中小，第三档才出现明显跃升，所以别一档一档试到底，可以直接跳到中档对比。
4. 最高档仍然不达标，说明任务对这个量级的模型太难，应该改任务拆分，而不是继续加大模型。
下载和校验：
```bash
mkdir -p models/needle && cd models/needle
下面的文件名与下载地址是占位，请替换为官方发布页给出的当前版本直链
curl -L -o needle-small.bin "<官方发布页 小档位直链>"
curl -L -o needle-large.bin "<官方发布页 大档位直链>"
macOS
shasum -a 256 needle-small.bin needle-large.bin
Linux
sha256sum needle-small.bin needle-large.bin
ls -lh
```
把算出来的哈希和官方页面上给出的值逐字符比对。这一步别省，模型文件下载不全的表现是"能加载但输出全是乱码"，很难排查。
第 3 步：先跑通一次推理，再谈集成
写一个薄封装，把模型加载、调用、超时收在一处。下面的模块名和类名是示意，实际以官方 README 为准，替换掉即可。
```python
needle_runner.py
import json
import time
from cactus import load_model # 模块/类名以官方文档为准
MODEL_PATH = "models/needle/needle-small.bin"
class Runner:
def __init__(self, path=MODEL_PATH):
self.model = load_model(path) # 首次加载偏慢，放进程启动时做
self.model.warmup() # 预热一次，避免首个请求超时
def generate(self, prompt: str, max_tokens: int = 128) -> str:
t0 = time.time()
out = self.model.generate(
prompt,
max_tokens=max_tokens,
temperature=0.0, # 结构化抽取任务，越低越稳
)
print(f"[needle] {time.time() - t0:.2f}s")
return out
if __name__ == "__main__":
r = Runner()
print(r.generate("你好"))
```
关键点：temperature 设成 0 或接近 0，max_tokens 卡死。微型模型一旦开始自由发挥，输出就会飘。
第 4 步：把提示词压到极短
小模型对提示词长度很敏感，上下文越长越容易漏字段。模板控制在 200 字以内：
```python
SYSTEM = """你是售后消息抽取器。只输出 JSON，不要解释，不要多余文字。
字段与取值：
category: 数码|服饰|食品|家居|unknown
issue: 质量问题|物流问题|描述不符|不会用|unknown
refund: 是|否|unknown
urgency: 高|中|低
无法判断的字段填 unknown。"""
FEWSHOT = [
("等了五天还没发货，急死了", '{"category":"unknown","issue":"物流问题","refund":"unknown","urgency":"高"}'),
("手机屏幕有个坏点，能退吗", '{"category":"数码","issue":"质量问题","refund":"是","urgency":"中"}'),
]
def build_prompt(text: str) -> str:
parts = [SYSTEM]
for q, a in FEWSHOT:
parts.append(f"输入：{q}\n输出：{a}")
parts.append(f"输入：{text}\n输出：")
return "\n".join(parts)
```
8MB 档位建议只留一到两个示例，29MB 档位可以放到三到四个。示例多了反而会让小模型"抄格式抄错位"。
第 5 步：校验 + 兜底，别信它的输出
模型返回的是字符串，不是 JSON。所有解析都要包在 try 里，并且做字段白名单校验。
```python
import json
SCHEMA = {
"category": {"数码", "服饰", "食品", "家居", "unknown"},
"issue": {"质量问题", "物流问题", "描述不符", "不会用", "unknown"},
"refund": {"是", "否", "unknown"},
"urgency": {"高", "中", "低"},
}
def parse(raw: str) -> dict:
start, end = raw.find("{"), raw.rfind("}")
if start == -1 or end == -1:
return {"error": "no_json", "raw": raw}
try:
obj = json.loads(raw[start:end + 1])
except json.JSONDecodeError:
return {"error": "bad_json", "raw": raw}
clean = {}
for k, allowed in SCHEMA.items():
v = obj.get(k, "unknown")
clean[k] = v if v in allowed else "unknown"
return clean
```
三级兜底：解析失败先重试一次（温度调高一点点换个采样）；仍失败走正则或关键词规则；规则也拿不准，写入待人工队列，或者丢给云端大模型。这条链路要提前设计好，不要等上线了再补。
第 6 步：用你自己的 20～50 条跑评估
这一步决定你选哪个档位，也是整篇里最值钱的部分。
```python
evaluate.py
import json
from needle_runner import Runner
from prompt_utils import build_prompt, parse
def run_eval(model_path: str, dataset_path: str) -> dict:
r = Runner(model_path)
data = [json.loads(l) for l in open(dataset_path, encoding="utf-8")]
total = correct = 0
bad_json = 0
field_hits = {k: 0 for k in SCHEMA}
for row in data:
out = parse(r.generate(build_prompt(row["text"])))
if "error" in out:
bad_json += 1
continue
for k in SCHEMA:
total += 1
if out.get(k) == row["label"].get(k):
correct += 1
field_hits[k] += 1
n = len(data)
return {
"samples": n,
"field_accuracy": round(correct / max(total, 1), 4),
"bad_json_rate": round(bad_json / max(n, 1), 4),
"per_field": {k: round(v / max(n, 1), 4) for k, v in field_hits.items()},
}
if __name__ == "__main__":
print(run_eval("models/needle/needle-small.bin", "data/eval.jsonl"))
```
data/eval.jsonl 每行形如 {"text": "...", "label": {...}}。跑完换模型路径再跑一次，两个结果并排放。
看两个指标：字段级准确率和 JSON 解析失败率。解析失败率超过 2%，先回去改提示词和兜底，不要急着换大模型——很多时候是格式问题，不是能力问题。
第 7 步：塞进你的应用
三种集成方式，按场景选：
- 进程内加载：Python/Node 服务里直接调 SDK。延迟低，但模型和主进程共享内存，主进程一崩全崩。
- 本地 HTTP 边车：单独起一个小服务，主应用通过
localhost调。隔离性好，推荐作为默认方案。 - 子进程调 CLI：最粗暴也最省事，适合批处理脚本，不适合高并发在线请求。
边车的最小实现：
```python
server.py
from fastapi import FastAPI
from pydantic import BaseModel
from needle_runner import Runner
from prompt_utils import build_prompt
from parse_utils import parse
app = FastAPI()
runner = Runner() # 启动时加载一次并预热
class Req(BaseModel):
text: str
@app.post("/extract")
def extract(req: Req):
raw = runner.generate(build_prompt(req.text))
result = parse(raw)
result["fallback"] = "error" in result
return result
```
```bash
uvicorn server:app --host 127.0.0.1 --port 8910
```
```bash
curl -s http://127.0.0.1:8910/extract \
-H 'Content-Type: application/json' \
-d '{"text":"买的外套尺码不对，能换货吗"}'
```
主应用侧只认 fallback 字段：为 true 就转人工或转云端大模型。整条链路对上游是透明的。
常见坑与排错
体积小不等于内存小。 模型文件 8MB，运行时还要加上 KV cache 和推理框架本身的开销，上下文开得越长涨得越多。在手机上要留足余量，具体上限以官方文档为准。
换档位要重调提示词。 大档位能吃更多示例，小档位吃不下。同一个提示词在两个档位上的表现可能完全不同，别直接套用。
输出飘。 先查 temperature，再查 max_tokens，最后看提示词里有没有"请详细说明"这类词——对抽取任务来说全是负分项。
中文数字、日期、型号容易错。 比如"三天前"、订单号、规格参数。这类字段不要交给模型，用正则后处理。
下载不完整。 表现是加载成功但输出乱码。永远校验哈希。
并发崩溃。 多数端侧运行时不是线程安全的。用单 worker + 队列，或者一进程一模型。
冷启动拖慢首请求。 在服务启动阶段做一次 warmup，别把加载放在请求路径里。
别让它干不该干的活。 需要世界知识、多步推理、长文写作的任务，微型模型做不了，硬上只会得到看起来很自信的错误答案。
版本要锁。 模型文件、提示词、评估集三者一起打标签。任何一样变了都要重跑评估，否则你无法解释线上指标的波动。
下一步建议
把评估集从 50 条扩到 200 条，覆盖错别字、表情、中英混排、超长文本这些边界。然后做分层路由：小模型先判，置信度低或 fallback 为真的走云端大模型，统计一下分流比例，你会对成本结构有全新认识。
再往下一步是用云端大模型批量产出标注，反过来打磨小模型的提示词，甚至微调一个更贴合你业务的小模型。最后是端上落地：模型文件随 App 打包还是首次启动下载、要不要做热更新、iOS 后台限制怎么绕，这些都以各平台和官方文档的当前说明为准。
整个流程的核心其实只有一句：先用最小档跑通，再用自己的数据决定要不要升级。 别人的跑分对你的业务没有参考价值，你那 50 条样本才有。