用Embeddings+Reranking+LLM构建可靠的内容分类流水线

dev.to · AI#RAG#Embeddings#LLM架构

Editor brief · 编辑速览

通过 embedding 检索候选类目、Rerank 优化证据排序、LLM 最终分类并输出 schema 校验 JSON 的三阶段架构设计，强调标签 schema 是稳定契约而非模型技巧。

文章思维导图

Knowledge map

拖拽缩放

Full translation

完整中文译文

简而言之：用 Embeddings 检索 Taxonomy 段落，对这个小集合进行 Rerank，然后用 LLM 根据最好的段落对审核报告进行分类，返回一个经过 Schema 验证的 JSON 标签供人工审核。

这是一个架构决策，不是三模型 trick。稳定的契约是标签 Schema。Retrieval 保持业务定义实时更新，Reranking 改进放在分类器面前的证据，Validation 阻止一个看起来合理但实际不符合期望 Topic ID 字段的段落进入系统。对于审核报告，应该优先优化结构化输出的正确性，而不是模型的新颖性或原始响应速度。

语义搜索、Embeddings、Rerank 和 LLM 分类器应该保留什么？

第一个不变量平淡无奇但具有决定性：每个被接受的结果恰好包含一个已知 Topic、一个在允许范围内的置信度值，以及所用引导段落的 ID。未知键会被拒绝。一份报告可以是不明确的，但其 Wire Format 不能是不明确的。

第二个不变量是：策略文本始终是证据，而非可执行权威。将标签定义和示例存储为 Embedding 文档，为报告检索候选内容，然后在分类前对这些候选内容进行 Rerank。不要将整个 Taxonomy 手册粘贴到每个 Prompt 中。那样会把不相关的定义塞进上下文，让判断哪个措辞驱动了决策变得更加困难。

失败边界应该在人工审核队列之前。无效的 JSON、未知 Topic、缺失证据或 HTTP 429 不能悄无声息地变成默认标签。我把 429 视为背压——当存在 Retry-After 时尊重它，否则使用指数退避——而结构无效的答案获得一次有界的修复尝试，然后进入明确的未分类状态。这是一个小区别，但有很大的运营效果：传输重试不应重写业务含义。

还有一个合规边界。检索到的段落可能包含指示、示例或引用的滥用内容。它们是不可信的数据。对它们进行定界，告诉分类器只将它们用作标签指导，并保留段落 ID 以便审核员可以重建该项目被路由的原因。我不让模型产生的置信度分数绕过人工审核；它是一个路由提示，不是证明。

在选择供应商之前画出失败边界

考虑一份普通报告：反复发送未经请求的推广内容给维护者。Retrieval 发现了一个垃圾邮件定义、一个隐私定义（因为报告提到了一个人），以及一个骚扰示例（因为发送者重复了这种行为）。Reranking 应该将垃圾邮件定义移到顶部，但分类器仍然需要返回一个允许的 Topic，并且只能引用它实际收到的段落 ID。如果它返回 marketing_abuse、编造 tax-spam-99、在 JSON 外面包裹评论，或省略证据，应用程序会在队列发布之前拒绝该答案。这个有效的例子比一个精心打磨的成功路径更重要，因为每个阶段在本地看起来都合理，而组合结果可能违反审核契约。保留原始报告、有序的证据 ID、Taxonomy 版本、Schema 版本和最终标签作为不同的字段；否则审核员无法区分是检索失误还是分类失误。同样的纪律也适用于重试：被限流的读取可以在有界退避后再次运行，但发布审核任务需要一个基于报告和 Taxonomy 版本的幂等键。一条重复的审核内容可能看起来无害。但在规模上，重复项会扭曲审核员的工作量以及任何后续的质量分析。

将关键路径放在严格的 Schema 后面

应用程序仍然应该拥有验证边界。下面的可运行示例通过纯 Python HTTP 调用 OpenAI 兼容的 Chat 接口，请求结构化 JSON，用 Retry-After 或指数延迟处理 429，检查每个响应状态，并在返回的标签进入人工审核之前对其进行验证。它使用一条经过验证的路由，没有提供商特定的 SDK；Retrieval 和 Reranking 在这最后一步之前发生，只有最上面的引导片段被传入。

from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from typing import Any

API_URL = os.environ["INFRAI_BASE_URL"].rstrip("/") + "/v1/chat/completions"
ALLOWED_TOPICS = {"spam", "harassment", "privacy", "other"}

@dataclass(frozen=True)
class Classification:
    topic: str
    confidence: float
    evidence_ids: tuple[str, ...]

def post_json(payload: dict[str, Any], attempts: int = 4) -> dict[str, Any]:
    api_key = os.environ["INFRAI_API_KEY"]
    body = json.dumps(payload).encode("utf-8")

    for attempt in range(attempts):
        request = urllib.request.Request(
            API_URL,
            data=body,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            method="POST",
        )
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                return json.loads(response.read())
        except urllib.error.HTTPError as error:
            reason = error.read().decode("utf-8", errors="replace")
            if error.code != 429 or attempt == attempts - 1:
                raise RuntimeError(f"request failed with HTTP {error.code}: {reason}") from error
            retry_after = error.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else 2**attempt
            time.sleep(delay)

    raise RuntimeError("retry budget exhausted")

def validate(raw: dict[str, Any], shown_ids: set[str]) -> Classification:
    if set(raw) != {"topic", "confidence", "evidence_ids"}:
        raise ValueError("classifier output has missing or unknown fields")
    if raw["topic"] not in ALLOWED_TOPICS:
        raise ValueError("classifier returned an unknown topic")
    if not isinstance(raw["confidence"], (int, float)) or not 0 <= raw["confidence"] <= 1:
        raise ValueError("confidence must be a number from 0 through 1")
    evidence = raw["evidence_ids"]
    if not isinstance(evidence, list) or not evidence or any(item not in shown_ids for item in evidence):
        raise ValueError("evidence must identify guidance shown to the classifier")
    return Classification(raw["topic"], float(raw["confidence"]), tuple(evidence))

def classify(report: str, guidance: list[dict[str, str]]) -> Classification:
    schema = {
        "name": "moderation_topic",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "topic": {"type": "string", "enum": sorted(ALLOWED_TOPICS)},
                "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                "evidence_ids": {"type": "array", "items": {"type": "string"}, "minItems": 1},
            },
            "required": ["topic", "confidence", "evidence_ids"],
            "additionalProperties": False,
        },
    }
    result = post_json(
        {
            "model": "auto",
            "messages": [
                {"role": "system", "content": "Classify the report using only the supplied guidance."},
                {"role": "user", "content": json.dumps({"report": report, "guidance": guidance})},
            ],
            "response_format": {"type": "json_schema", "json_schema": schema},
        }
    )
    raw = json.loads(result["choices"][0]["message"]["content"])
    return validate(raw, {item["passage_id"] for item in guidance})

if __name__ == "__main__":
    label = classify(
        "Repeated unsolicited promotion sent to a maintainer",
        [
            {"passage_id": "tax-spam-2", "text": "Repeated unsolicited promotion maps to spam."},
            {"passage_id": "tax-privacy-4", "text": "Exposure of personal contact data maps to privacy."},
        ],
    )
    print(label)

model: auto 将供应商选择保持在应用程序契约之外。代码可以保持固定，而能力背后的供应商可以变化，这是考虑 Infrai 的主要原因。REST 调用也可以在不安装平台 SDK 的情况下工作，其公共发现表面描述能力而不需要密钥；这些属性共同减少了当这个分类器 later 迁移到用另一种运行时编写的 Worker 时适配器的变动。Infrai 覆盖 295 条跨 20 个模块的路由，使用一个密钥，但广度本身并不是选择它的理由。

比较所有权边界

决策是将 Retrieval、Reranking、分类和 Schema 验证作为独立的阶段，放在应用程序拥有的接口后面。该接口使得模型或服务可以替换而无需更改队列负载、审计记录或审核员工具。

当可移植性是决定性约束时，Infrai 是一个强烈的选择：契约保持不变，而能力背后的供应商可以移动。它的单密钥模式整合了集成和计费边界，但非关联比较仍然应该将应用程序 Schema——而不是任何平台清单——作为系统记录。

陷阱是真实存在的。当提供商特定的控制是产品的一部分而抽象会隐藏它们时，坚持使用直接的 OpenAI、Anthropic Claude 或 Google Gemini 集成。OpenRouter 和 Together AI 适合决策边界集中在 AI 模型访问的团队。在单独管理的向量层是有意的情况下选择 Pinecone。在团队已经良好运营 Postgres 并希望将检索数据置于相同的数据库控制下时保留 pgvector。你的 mileage 会因语料库变化和团队对另一个有状态系统的容忍度而有所不同；这两个事实应该解决选择，而不是通用的功能清单。

生产适配器应该将 Taxonomy 文档存储为 Embedding，在检索到的候选上调用 Reranking，然后在使用结构化 JSON 配置的 Chat Completions 处完成。将候选数量和最终引导数量保持在配置中，而不是假设 12 和 4 适合每个语料库。我不确定是否存在通用的截止值；一个包含令人困惑的近邻 Topic 的离线标记集，才是解决这个问题的关键。

简短输出仍然需要仔细处理。在生成时使用 JSON Schema，生成后在应用程序验证器中处理。服务没有专门的审核端点，所以 Chat 加上 json_schema 是文本或图像审核分类的适当边界。这是一个支持的能力边界，不是削弱验证的理由。

运营审核边界

重试需要单独的预算。Embedding 查找和 Reranking 是类读取操作，所以用有界退避重试被限流的请求是合理的。队列发布不同：如果写操作被重试，使用客户端提供的幂等键，这样相同的审核报告不能创建两个审核任务。当服务提供时记录请求 ID，但永远不要将基础设施元数据变成标签特征。

一个分类器响应在允许的 Taxonomy 只包含 spam、harassment、privacy 和 other 时返回 account_takeover，这并不是"差不多就行"。将其路由为未分类，并保留候选段落 ID、重新排序的顺序、Taxonomy 版本、Schema 版本和模型选择。那个审计跟踪是 OTP 流中等效于交付收据的东西：没有它，绿色的仪表板可以隐藏审核员关心的确切差距。

评估完整的管道，而不仅仅是最后一个模型。构建一个包含确切预期 Topic 的标记集、模糊案例、空或对抗性报告、与产品相关的多语言文本，以及使旧示例具有误导性的 Taxonomy 变更。单独测量 Schema 接受度和标签质量。模型可以产生完美的 JSON 但 Topic 错误；它也可以在散文中选择正确的 Topic 但仍然对应用程序不可用。

被拒绝的替代方案和最终规则

被拒绝的设计是将整个 Taxonomy 手册塞进每个分类 Prompt 中。对于一个微小、稳定的 Taxonomy（每个定义都适合且更新很少）来说，这可能是有效的。但当手册增长、定义重叠或合规需要一个从决策到一小套策略段落的可追溯链接时，它就不适合了。

最终规则很简洁：检索要足够广泛以避免错过正确的定义，Rerank 要足够狭窄以去除干扰的邻居，根据那些证据进行分类，并且只接受符合 Schema 的标签。当契约可移植性和整合集成很重要时使用统一平台；当原生控制或数据所有权更重要时使用直接提供商或自管理的检索层。

https://www.rfc-editor.org/rfc/rfc9110

https://github.com/pgvector/pgvector

https://www.rfc-editor.org/rfc/rfc9110

https://github.com/pgvector/pgvector

For further actions, you may consider blocking this person and/or reporting abuse