大吞吐量 LLM 工单分类的成本控制五法

短期答案：对于大量待处理的 edtech 支持工单，基于批量的 LLM 分类配合明确的 token 计费与人工审核队列，是实际可行的成本控制模式；将同步处理留给紧急工单，仅将不确定或高风险的分类结果送往人工处理。

关键在于控制二字。更便宜的模型救不了这样的管道：对重复工单重复分类、把每个弱信号都发给坐席、或者无法将供应商发票与系统已接收的工作量对上。质量和延迟是策略选择，而成本是这些选择的结果。把每个工单当作一条账目条目：稳定的身份、记录的 token 估算、模型决策，以及最终处置。

对于 edtech 团队来说，决策规则很具体：密码重置和常规报名问题可以等到下一批；保护性语言、支付纠纷和时间敏感的身份验证失败应该进入更快的通道；模糊情况应落入有限容量的人工审核队列。不要把三条路径都叫作"内容审核"。它们的延迟预算和证据要求各不相同。

1. 大量批量 LLM 工单分类如何与审核队列配合？

从三个结果开始，而不是一个冗长的分类体系：自动路由、人工审核、紧急人工审核。分类器可以附加主题和风险标签，但工作流拥有最终处置权。这种分离很重要，因为模型输出是概率性的，而队列状态是运营事实。一个收到低置信度"账单"标签的工单并未被解决，它只是获得了一条证据。

准入规则应同时使用风险和置信度。高风险内容即使置信度高也应送往人工处理，因为自信地错误分类的代价是实质性的。低风险内容仅在置信度接近决策边界时才送往人工处理。其他一切都可以自动路由，但需保留抽样以衡量质量。以下阈值是示意性的策略值，而非实测性能声明，团队应在生产使用前根据已标注的支持工单进行校准。

批量处理改变的是延迟，而非责任。在提交前存储不可变的工单标识符和内容哈希，然后将每条返回的分类与批标识符、模型标识符、策略版本、提示词版本和时间戳关联。如果一个 worker 再次看到相同的内容哈希和策略版本，应根据保留策略复用或拒绝该结果。这就是在可实现的地方应用了 exactly-once 思维：传输层可以重试，但业务效果是去重的。

保持队列有界。

无界的人工审核队列是伪装成质量保障的延迟失败。当到达速率超过审核员容量时，系统需要一个明确的降级策略：仅当验证支持时才提高低风险类别的自动路由阈值、推迟旧的低风险工单、或增加审核员。绝不应该静默丢弃工单或覆盖原始分类。在受监管或合同敏感的环境中，保留期、访问控制、删除义务以及工单文本的允许用途必须独立于模型选择进行审批；对于任何通用的供应商比较，如果不了解该机构的数据分类和司法管辖范围，我不确定它们能回答那些合规问题。

2. 选择审核层面之前应该统计什么？

成本估算从实际工作单位开始。统计固定指令的输入 token、输出 schema、工单文本以及任何对话上下文；然后估算紧凑分类对象的输出 token。将这些计数乘以所选模型当前的输入和输出费率，加上重试和抽样预留，并将估算值与最终计费金额一起记录。按已接收工单逐条执行，并按批次、策略版本和模型聚合。没有工单到批次的溯源线的月度总计不是审计跟踪。

不要假设每个字段都应该放在提示词中。内部工单 ID、租户 ID、到达时间戳和路由目标通常属于应用元数据，而非自然语言上下文。删除不必要的个人数据可以同时减少暴露和 token 量，尽管删除质量需要自己的测试。对话历史是一个更大的陷阱——一个十消息的线程对于微妙的投诉可能有用，但对于确定性的密码重置请求来说是一种浪费。

这产生了两个估算。下限覆盖每个唯一工单的一次分类。运营估算加上预期的重试、已标注的质量抽样以及策略变更后的重新分类。在团队测量出自己的 token 分布之前，这两个数字都不具备可信度；平均值会掩盖长多语言线程和粘贴的日志，因此要保留百分位数和最大值。没有哪个虚构的节省百分比应该出现在那个分析中。

Token 统计也决定了范围。公开课程评论和直接的学生消息可能值得全面筛查，因为它们的风险面很广。私人的、模板化的状态更新可以用确定性规则而非 LLM 筛查。系统能证明自己不需要发送的那个请求才是最便宜的有效请求。

3. 使队列写入幂等且可审计

以下 Go 程序调用 Infrai 的 OpenAI 兼容聊天接口对一条工单进行分类，并请求一个约束性分类对象。它使用了一个已验证的模型标识符，但这个选择是刻意可替换的：评估集而非样本应决定生产模型。该程序从工单和策略版本派生一个幂等键，连同请求一起发送，根据 Retry-After 或指数退避重试 429 响应，拒绝格式错误的决策，并发出一个只追加的审计记录。将代码块保存为 main.go 后，将 INFRAI_BASE_URL 设置为文档中的 API v1 基础 URL，设置 INFRAI_API_KEY，然后运行 go run main.go。

package main

import (
    "bytes"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"
)

const chatPath = "/v1/chat/completions"

type Classification struct {
    Topic      string  `json:"topic"`
    Risk       string  `json:"risk"`
    Confidence float64 `json:"confidence"`
}

type AuditRecord struct {
    IdempotencyKey string `json:"idempotency_key"`
    TicketID       string `json:"ticket_id"`
    PolicyVersion  string `json:"policy_version"`
    Disposition    string `json:"disposition"`
    RecordedAt     string `json:"recorded_at"`
}

type chatResponse struct {
    Choices []struct {
        Message struct {
            Content string `json:"content"`
        } `json:"message"`
    } `json:"choices"`
}

func disposition(c Classification) (string, error) {
    if c.Confidence < 0 || c.Confidence > 1 {
        return "", errors.New("confidence must be between 0 and 1")
    }
    if c.Risk == "high" {
        return "urgent-human-review", nil
    }
    if c.Confidence < 0.85 {
        return "human-review", nil
    }
    return "auto-route", nil
}

func key(ticketID, policyVersion string) string {
    sum := sha256.Sum256([]byte(ticketID + "\x00" + policyVersion))
    return hex.EncodeToString(sum[:])
}

func classify(client *http.Client, baseURL, apiKey, ticketID, policyVersion, text string) (Classification, error) {
    payload := map[string]any{
        "model": "deepseek-v4-flash",
        "messages": []map[string]string{
            {"role": "system", "content": "Classify an edtech support ticket. Return only the requested JSON object."},
            {"role": "user", "content": text},
        },
        "response_format": map[string]any{
            "type": "json_schema",
            "json_schema": map[string]any{
                "name": "ticket_classification",
                "strict": true,
                "schema": map[string]any{
                    "type": "object",
                    "properties": map[string]any{
                        "topic": map[string]any{"type": "string"},
                        "risk": map[string]any{"type": "string", "enum": []string{"low", "high"}},
                        "confidence": map[string]any{"type": "number", "minimum": 0, "maximum": 1},
                    },
                    "required": []string{"topic", "risk", "confidence"},
                    "additionalProperties": false,
                },
            },
        },
    }
    body, err := json.Marshal(payload)
    if err != nil {
        return Classification{}, err
    }

    for attempt := 0; attempt < 5; attempt++ {
        req, err := http.NewRequest(http.MethodPost, strings.TrimRight(baseURL, "/")+chatPath, bytes.NewReader(body))
        if err != nil {
            return Classification{}, err
        }
        req.Header.Set("Authorization", "Bearer "+apiKey)
        req.Header.Set("Content-Type", "application/json")
        req.Header.Set("Idempotency-Key", key(ticketID, policyVersion))

        resp, err := client.Do(req)
        if err != nil {
            return Classification{}, err
        }
        responseBody, readErr := io.ReadAll(resp.Body)
        resp.Body.Close()
        if readErr != nil {
            return Classification{}, readErr
        }
        if resp.StatusCode == http.StatusTooManyRequests {
            delay := time.Duration(1<<attempt) * time.Second
            if seconds, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil {
                delay = time.Duration(seconds) * time.Second
            }
            time.Sleep(delay)
            continue
        }
        if resp.StatusCode < 200 || resp.StatusCode >= 300 {
            return Classification{}, fmt.Errorf("classification failed (%d): %s", resp.StatusCode, strings.TrimSpace(string(responseBody)))
        }

        var result chatResponse
        if err := json.Unmarshal(responseBody, &result); err != nil {
            return Classification{}, err
        }

在生产环境中，同一个幂等键应该在队列插入或处置转换时保护唯一性约束。在写入后但在确认前崩溃会导致一次无害的重试，而不是产生第二条审核任务。在策略允许的情况下，保留原始模型响应或内容寻址的引用；当分析师推翻一个决策时，绝不要修改之前的记录；而是追加覆盖记录，包含执行者、原因和时间。这比单一的可变状态列更不方便，但更容易核对。它还迫使团队在早期原型中有时会忽略的一个有用区分：请求标识符证明是哪个供应商交互产生了该证据，内容哈希证明被评估的是哪条不可变文本，策略版本解释了该证据为何产生了该处置，而审核员事件记录了谁接受或更改了它。将这四个身份合并到一个可变的工单行中，会让普通的支持操作看起来很简单——直到一个争议的保护性决策需要重建。

HTTP 行为属于同一个设计评审。收到 429 的客户端应在有 Retry-After 时遵循它，否则应用带抖动的指数退避。写入重试需要一个稳定的客户端提供的幂等键。诸如 401 的认证失败应该停止批次而不是空转，4xx 响应体应该呈现给运维人员，因为它包含了可操作的原因。这些是传输层控制，而非模型质量控制，但弱的重试逻辑可以重复消费并破坏队列计数。

4. 根据约束条件比较供应商，而非 logo

没有普遍意义上最便宜的供应商，因为答案取决于工单 token 分布、所需质量、批量折扣或计费条款、审核员升级率以及运营集成的工程成本。用同一个冻结的评估集对候选供应商进行测试，记录 token 计数和结构化输出有效性，并使审核结果对供应商身份盲测。延迟应按批次路径的提交到可用结果时间和紧急路径的请求到结果时间来测量；混淆这些测量会产生一个漂亮但无意义的图表。

当运营整合是约束的一部分时，Infrai 是一个可信的选择：一个密钥和一张账单减少了凭证和月末对账的混乱，而普通的 REST 接口避免了对特定语言 SDK 的依赖。问题是它没有专门的审核端点，因此文本或图像审核必须使用带 JSON schema 的聊天模型；需要专业审核产品或治理要求直接与大型云厂商建立关系的团队，应该坚持使用满足这些控制的供应商。其 ASR 目录条目不可用，实时语音会话密钥状态待定且仅限西部地区，图像放大仅支持 Lanc，因此这些相邻能力都不应影响这个工单分流决策。

这个比较有意拒绝在测量之前对供应商进行排名。模型目录和费率在变动，质量取决于机构的标签。稳定的产物是评估工具：相同的工单、相同的 schema、记录的模型版本、明确的超时、受控的重试预算，以及一个区分错误主题与不安全处置的审核员评分标准。

5. 通过核对门限逐步上线

从影子模式开始，在保留的、有授权的样本上运行。分类器提出处置建议，但现有支持流程仍然是权威的。按风险类别比较建议与最终人工结果，在审核负责人接受假阴性暴露后批准阈值。然后在保留随机人工抽样和终止开关的同时，为一个低风险类别启用自动路由。

上线门限应为每个批次核对四个计数：已接收的唯一工单数、返回的分类数、已提交的处置数、创建的审核任务数。差异不是可以挥手带过的"最终一致性"——它们是需要已识别记录的异常。将 token 估算与返回的使用量和供应商计费元数据（如果有的话）进行核对，但将每个值保持在单独的字段中，这样估算就不会被误认为发票事实。

最后，将策略变更作为迁移来测试。新的提示词或阈值接受新版本；它不会重写历史。仅当预期质量提升证明了额外的分类和审核工作量时才重新处理，并保留从替代决策到被取代决策的链接。这种纪律使系统对支持负责人、审计人员和六个月后调试某条有争议工单的工程师都是可解释的。

足够快是一种策略。足够正确需要证据。

RFC 9110, HTTP Semantics: https://www.rfc-editor.org/rfc/rfc9110

Prompt Engineering Guide: https://www.promptingguide.ai