Clef-omni：用多模态决策模型替代「大模型分类」

在发布 Clef 与 Clef-flash 一周之后，Cloudflare 又推出了 Clef-omni：一个在文本与图像之外，还能吃音频和视频的决策模型。它比「又一个模型发布」更值得看，原因在输出的形状。决策模型不生成文本。你交给它一组命名字段，它在一次 prefill 里按 schema 给这些字段打分，直接返回取值。这一下就把大多数分类流水线真正会崩的那一步去掉了——把一段话解析成字段。Cloudflare 还把 Clef-flash 降到了 0.038 美元/百万输入 token，并把 Clef 迁到 SGLang 上，中间延迟快了大约 1.7 到 2 倍。如果你曾经因为「模型调用太贵」而劝自己别加分类器，现在这笔账变了。

决策模型和大语言模型是两种不同的机器，价格把这件事说得很清楚

发布了什么，以及上下文窗口的取舍

这个家族现在是三个模型。Clef 是 0.24 美元/百万输入 token，保留 64k 上下文窗口。Clef-omni 上线价 0.15 美元/百万，接受文本、图像、wav 或 mp3 音频、mp4 或 webm 视频。Clef-flash 从 0.09 降到 0.038 美元/百万，价格上已经低于 Jev，但托管版的上下文窗口从 64k 降到 24k。Cloudflare 说得很直白：他们的用量数据显示只有 0.24% 的请求超过 24k 输入 token，所以他们用一个很少被用到的能力，换了一个好得多的入门价。Hugging Face 上的权重是按 256k 上下文训练的，所以真需要长窗口可以自托管；Cloudflare 也建议这类场景改用 Clef。这个交换是合理的，但它确实是个交换——一旦你开始往里塞转录文本，24k 并不宽裕。

# Ask for named fields instead of prose. The model returns the schema, you do not parse it.
import json

def decision_request(state, questions, images=(), audio=(), videos=()):
    payload = {"model": "clef-omni", "state": state, "questions": {}}
    for name, instructions in questions.items():
        payload["questions"][name] = {"type": "noul", "instructions": instructions}
    if images:
        payload["images"] = list(images)
    if audio:
        payload["audio"] = list(audio)
    if videos:
        payload["videos"] = list(videos)
    return payload

payload = decision_request(
    state="Review the installation: a photo of the unit, an audio recording of it running, and a video of the fan.",
    questions={
        "label_visible": "Is the model and serial number label visible in the photo?",
        "sounds_normal": "Does the unit sound like it is running smoothly, without rattling or grinding?",
        "fan_running": "Is the fan running in the video?",
    },
    images=["data:image/jpeg;base64,<unit-photo>"],
    audio=["data:audio/mpeg;base64,<unit-running>"],
    videos=["data:video/mp4;base64,<fan-clip>"],
)

print(json.dumps(payload["questions"], indent=2))
print("modalities:", [k for k in ("images", "audio", "videos") if k in payload])

# Every key in the answer maps to a field you named. There is no parsing step left to get wrong.

决策模型为什么表现不一样

Clef-omni 建立在一个 Qwen3-Omni-30B-A3B-Instruct 的专家混合底座上，但 Cloudflare 保留了理解主干，丢掉了文本转语音的输出组件。这个细节解释了它的性能画像。因为模型根本不生成输出 token，转录和打标这两步被整个跳过：音频和视频帧直接映射进同一个统一序列，模型用两阶段注意力路由从内部嵌入里给每个候选取值打分。内置的词法语法保证选项语义不被破坏。训练跟 Clef 同款配方：冻结 Qwen3 主干、训练低秩适配器，并用标签平滑交叉熵损失配合 Brier 分数校准。Brier 校准这块最值得留意。一个知道自己什么时候没把握的分类器，远比一个自信但答错的分类器有用。

# A decision model is cheap and fast. The engineering question is when to escalate anyway.
DECISION_LATENCY_MS = {"text": 130, "image": 150, "audio": 400, "video": 1500}
PRICE_PER_M_INPUT = {"clef-flash": 0.038, "clef": 0.24, "clef-omni": 0.15}

def plan(modalities, confidence, confidence_floor=0.72, llm_price_per_m=2.5):
    model = "clef-omni" if {"audio", "video"} & set(modalities) else "clef-flash"
    escalate = confidence < confidence_floor
    return {
        "model": model,
        "expected_latency_ms": sum(DECISION_LATENCY_MS[m] for m in modalities),
        "escalate_to_llm": escalate,
        "why": "confidence below floor" if escalate else "decision answer is good enough",
        "decision_price_per_m_input": PRICE_PER_M_INPUT[model],
        "llm_price_per_m_input": llm_price_per_m if escalate else 0.0,
    }

print(plan(["text"], confidence=0.94))
print(plan(["image", "audio"], confidence=0.61))
print(plan(["image", "audio", "video"], confidence=0.88))

# Escalation is not a fallback for failure. It is a routing decision you set on purpose.

在 0.038 美元/百万 token 的价位上，分类不再是成本中心

延迟数据与那张跑分表

延迟是这套架构最硬的论据。纯文本决策中位约 130 毫秒，图像约 150 毫秒，音频片段几百毫秒，一段 21 秒带声视频大约 1.5 秒，全部在一次 API 调用内完成。跑分表就复杂一些，值得诚实地读。Clef-omni 在 BANKING77 宏 F1（94.8）、CLINC150+OOS（97.7）、Amazon ESCI（57.8）上领先全家族。BFCL case exact 是 98.2，对比 Clef 的 98.47、Clef-flash 的 98.76，都远高于 Jev 的 95.75。但 Clef-omni 在 Home appliances case exact 上只有 69.3，而 Clef-flash 是 97.73；When2Call 准确率 63.3，Jev 是 80.97。加上多模态并没有让它全面更强，只是让它更宽。

# Constrained answers are easy to validate. That is most of the value.
ALLOWED = {
    "label_visible": {"yes", "no", "unclear"},
    "sounds_normal": {"yes", "no", "unclear"},
    "fan_running": {"yes", "no", "unclear"},
}

def validate(answer, allowed=ALLOWED):
    problems = []
    for field, permitted in allowed.items():
        value = answer.get(field)
        if value is None:
            problems.append((field, "missing"))
        elif value not in permitted:
            problems.append((field, "out of schema: " + str(value)))
    return {"ok": not problems, "problems": problems}

print(validate({"label_visible": "yes", "sounds_normal": "no", "fan_running": "yes"}))
print(validate({"label_visible": "maybe", "sounds_normal": None, "fan_running": "yes"}))

# With an LLM you would be writing regexes against sentences. Here the check is a set membership test.

Cloudflare 内部拿它干什么

内部案例是整篇文章最容易迁移的部分。公开的 cloudflare-docs GitHub 仓库用 Clef 检测并关闭垃圾 issue。他们自家的内容管理系统 EmDash 用它审核插件库里的钓鱼。数据防泄漏团队用它扫描政府证件号这类个人信息。威胁情报团队用它辅助识别恶意域名。这些活都不光鲜，而这正是重点：它们全都是那种团队以前只有三个烂选项的活——人工审核、迟早过期的正则、或者专门拉一个 ML 团队再攒一批数据。Cloudflare 自己的说法是，决策能力下沉到了模型层，任何人都能调。对大多数团队来说，以前那条路的诚实描述是第三个选项：也就是这事干脆就不做了。

# Five million classifications a month turn a price per million into a business decision.
def annual_cost(requests_per_month, tokens_per_request, price_per_m,
                share_escalated=0.0, llm_price_per_m=2.5):
    steady = requests_per_month * tokens_per_request * (1 - share_escalated) * price_per_m
    escalated = requests_per_month * tokens_per_request * share_escalated * llm_price_per_m
    monthly = (steady + escalated) / 1_000_000
    return {"monthly_usd": round(monthly, 2), "annual_usd": round(monthly * 12, 2)}

print("all decision model:", annual_cost(5_000_000, 800, 0.038))
print("10% escalated:   ", annual_cost(5_000_000, 800, 0.038, share_escalated=0.10))
print("all LLM:         ", annual_cost(5_000_000, 800, 2.5))

# The escalation rate is the whole argument. Measure it before you quote a saving to anyone.

整个模型家族都开源权重，自托管是可选项，也是一次真实的对比

不改架构怎么接进去

Clef 兼容 Jev API，也能通过 AI Gateway 调用，所以 Cloudflare 建议的迁移方式就是换个模型 ID。但在那之前先把升级策略定下来，因为这一整个家族里真正有意思的设计问题不是「调哪个模型」，而是「答得不够好时怎么办」。能用的模式很朴素：让决策模型处理常规那部分，评估返回的置信度，只把长尾升级给大语言模型。阈值要从实测的字段级准确率里定，而不是凭感觉；而且要按字段分别跟，因为一个在「标签是否可见」上可靠、在「声音是否正常」上发虚的模型，需要的是按字段设阈值，而不是按请求设一个。

# Track confidence against outcomes. A cheap model with honest confidence is a routing signal.
from collections import defaultdict

BUCKETS = [(0.0, 0.6), (0.6, 0.8), (0.8, 0.9), (0.9, 1.01)]

def calibration(samples, buckets=BUCKETS):
    rows = defaultdict(lambda: {"n": 0, "correct": 0})
    for confidence, is_correct in samples:
        for low, high in buckets:
            if low <= confidence < high:
                rows[(low, high)]["n"] += 1
                rows[(low, high)]["correct"] += int(is_correct)
                break
    out = []
    for bucket in buckets:
        row = rows[bucket]
        if row["n"]:
            out.append({
                "confidence_bucket": bucket,
                "n": row["n"],
                "accuracy": round(row["correct"] / row["n"], 3),
            })
    return out

samples = [(0.55, False), (0.62, True), (0.71, False), (0.85, True), (0.88, True), (0.95, True), (0.97, True)]
for row in calibration(samples):
    print(row)

# If the 0.6 to 0.8 bucket only lands 50% of the time, your escalation threshold is a guess.

决定值不值得的那个数字

Cloudflare 这次降价，引出了一个大多数团队从没算过的账，因为在这之前差距并不大。每月 500 万次分类、每次约 800 token，跑 Clef-flash 大约 152 美元/月。同样的量全部发给一个 2.5 美元/百万输入 token 的大语言模型，接近 1 万美元/月。差距宽到这种程度，问题就不再是成本，而变成「在你的具体任务上准不准」——而这个问题只能靠标注几百条真实样本、两边都跑一遍来回答。先做校准，再做商业论证。一个有实测升级率的便宜分类器是可以拿去汇报的方案；一个凭感觉设升级率的便宜分类器，是未来的一次事故。