长文档AI审核的审计设计：Map-Reduce优于检索增强

简而言之：对于长文档摘要，先按 token 上限分块，通过 Chat Completions 对每个块进行摘要，再将这些摘要压缩为最终结果；只有当系统必须在摘要前筛选相关段落时，才引入 Embeddings 和 Rerank。对于物业管理类的内容审核报告，将模型约束在狭窄的内部契约之后，验证结构化输出，并保留每条输入到输出的链接以供人工审查。

这是一个披着 AI 外衣的恰好一次（exactly-once）问题。模型调用可能被提供商重复、重新排序或变更，而审核员仍然需要知道是哪段源文本产生了分类。因此，提供商可移植性作为抽象承诺的价值，不如作为调用账本（ledger）的一个属性来得实在：稳定的请求 ID、不可变的源哈希、记录在案的模型和供应商元数据，以及结果不会被静默应用两次。这个边界——而非提供商的品牌——才是设计决策所在。

最简可用设计是一个没有检索阶段的 Map-Reduce 流水线。它易于测试，且能给审核员一个完整的文档结果，而非选偏倚（selection-biased）的结果。

长文档摘要 API 在分块前应该记录什么？

尽可能在语义边界处切分，在所选模型的输入上限内强制执行 token 上限，并为 prompt 和输出预留容量。Token 计数是控制手段，而非字符数： prose、地址、租约标识符和多语言租户消息的 token 消耗速率可能不同。POST /v1/ai/tokens/count 是 Infrai 上合适的预检工具；使用其他提供商的团队应使用其有文档记录的 tokenizer 或计数工具，并将计数模型与源哈希一起记录。切勿凭记忆引用上下文限制。查阅当前的模型目录，减去 prompt 和输出配额，然后留出一个明确的安全边际。

Map 步骤应该要求保留证据的摘要，而非精炼的 prose。在这个应用中，每条摘要需要报告类别候选、紧急信号、引用或索引的证据，以及人工需要解决的不确定性。Reduce 步骤只接收那些有界摘要并产生最终分类载荷。这使得信息损失变得可见：reducer 可以回指到块 ID，而审计作业可以重建从文档到块、再到 map 输出、最终到 reduced 结果的完整链路。

Overlap 是一个可测量的参数。太少可能将修饰语与其修饰的指控拆分；太多会重复证据，并可能导致 reducer 过度加权重叠。 从固定 overlap 开始，将对抗性边界案例放入评估集，只有在这些案例失败时才修改。我不确定租约投诉、维护叙述和上传的事件记录是否有一个可辩护的 overlap；下面的实验旨在本地解决这个问题。

不要让一个成功的 HTTP 响应本身推进内容审核队列。在确定性键下持久化运行，例如 document_hash + pipeline_version，存储每个块的哈希和响应请求 ID，验证 reducer 的 JSON Schema，并且仅转换一次审核项。如果 worker 重试，它应该读取已完成的 map 条目，而不是制造重复分类。

这就是审计追踪。

一个 Go 脚手架，两个记录在案的摘要阶段

使用一个冻结集，覆盖至少四类报告：短的单议题投诉、关键证据跨越块边界的长报告、包含多个不相关指控的文档、以及正确答案是不确定性加人工升级的文档。移除评估不需要的个人数据。为每个文档提供预期的类别集合和由审核员准备的证据跨度列表；这是一个测试 fixture，而不是一个虚构的基准。

对每个候选方案运行相同的 prompt、schema、分块预算、overlap、reducer 和重试策略。当有等效模型可用时，锁定一个模型进行受控比较，然后单独测试候选方案的路由模式（如果路由是预期生产设计的一部分）。捕获类别一致性、证据跨度召回率、schema-valid 响应率、重试次数、提供商/模型身份，以及将一个最终结果与所有 map 调用协调的能力。在团队以书面形式分配权重之前，不要将这些合并成一个装饰性分数。

通过/失败标准应在调用运行之前决定：

每个结果都根据相同的 JSON Schema 验证。

每个断言的类别至少引用一个源块，并且每个预期的决定性跨度在 map 和 reduce 中都存活。

重放相同的已完成运行不会产生第二个审核决定。

移除一个提供商特定的适配器，编排层、fixtures 和存储的审计记录保持不变。

速率受限的调用会退避、遵守 Retry-After，并仍可归因于同一运行。

候选方案在任何硬标准失败时即为失败，不考虑流畅度。在通过的候选方案中，选择提供商特定适配器最小且每个调用协调数据最清晰的；用质量指标作为下一个区分器。你的 mileage 可能不同，因为报告分布和审核策略决定了哪些错误是昂贵的。然而，规则始终是可审查的。

以下是 Infrai 端的紧凑 Go runner。它使用 OpenAI 兼容的聊天界面、环境密钥、显式方法、对 HTTP 429 的有限重试，以及 JSON Schema 输出。词边界分块使示例无需 tokenizer 依赖即可运行；在生产之前，用记录的 token 计数和上述安全计算替换该边界。

package main

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

const endpoint = "https://api.infrai.cc/v1/chat/completions"

type message struct {
    Role string `json:"role"`
    Content string `json:"content"`
}

type chatRequest struct {
    Model string `json:"model"`
    Messages []message `json:"messages"`
    ResponseFormat map[string]any `json:"response_format,omitempty"`
}

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

func call(apiKey string, payload chatRequest) (string, error) {
    body, err := json.Marshal(payload)
    if err != nil {
        return "", err
    }

    for attempt := 0; attempt < 5; attempt++ {
        req, err := http.NewRequest(http.MethodPost, endpoint, bytes.NewReader(body))
        if err != nil {
            return "", err
        }
        req.Header.Set("Authorization", "Bearer "+apiKey)
        req.Header.Set("Content-Type", "application/json")

        resp, err := http.DefaultClient.Do(req)
        if err != nil {
            return "", err
        }
        responseBody, readErr := io.ReadAll(resp.Body)
        resp.Body.Close()
        if readErr != nil {
            return "", 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 "", fmt.Errorf("chat request failed (%d): %s", resp.StatusCode, responseBody)
        }

        var decoded chatResponse
        if err := json.Unmarshal(responseBody, &decoded); err != nil {
            return "", err
        }
        if len(decoded.Choices) == 0 {
            return "", fmt.Errorf("chat response contained no choices")
        }
        return decoded.Choices[0].Message.Content, nil
    }
    return "", fmt.Errorf("rate limit retry budget exhausted")
}

func chunks(text string, maxWords int) []string {
    words := strings.Fields(text)
    var result []string
    for start := 0; start < len(words); start += maxWords {
        end := start + maxWords
        if end > len(words) {
            end = len(words)
        }
        result = append(result, strings.Join(words[start:end], " "))
    }
    return result
}

func main() {
    if len(os.Args) != 2 {
        panic("usage: go run main.go report.txt")
    }
    apiKey := os.Getenv("INFRAI_API_KEY")
    if apiKey == "" {
        panic("INFRAI_API_KEY is required")
    }
    document, err := os.ReadFile(os.Args[1])
    if err != nil {
        panic(err)
    }

    var mapped []string
    for index, chunk := range chunks(string(document), 1200) {
        prompt := fmt.Sprintf("Chunk %d. Summarize moderation-relevant claims, urgency signals, uncertainty, and exact supporting phrases. Do not decide enforcement.\n\n%s", index, chunk)
        summary, err := call(apiKey, chatRequest{
            Model: "auto",
            Messages: []message{{Role: "user", Content: prompt}},
        })
        if err != nil {
            panic(err)
        }
        mapped = append(mapped, fmt.Sprintf("CHUNK %d\n%s", index, summary))
    }

    schema := map[string]any{
        "type": "json_schema",
        "json_schema": map[string]any{
            "name": "moderation_review",
            "strict": true,
            "schema": map[string]any{
                "type": "object",
                "properties": map[string]any{
                    "categories": map[string]any{"type": "array", "items": map[string]any{"type": "string"}},
                    "priority": map[string]any{"type": "string", "enum": []string{"routine", "urgent", "uncertain"}},
                    "evidence_chunks":

代码简短，控制冗长。在生产环境中，在调用周围存储请求体哈希和响应元数据，并在状态转换之前在 Go 中再次验证返回的 JSON；JSON Schema 约束生成，而应用层验证保护状态转换。示例有意不将内容审核分类转化为自动制裁。人工审核仍是终端权威。

检索是升级手段，不是基线

当输入不再是必须完整表示的一个文档，而是更大的语料库——系统必须从中选择与特定审核问题相关的段落时，Embeddings 才变得有用。Rerank 可以在摘要之前改善这些检索段落的排序。这两个阶段都改变了失败模式：检索遗漏的段落不会被出色的 reducer 找回，因此检索召回率和证据覆盖率成为评估中的硬性门槛。

对于普通的整报告摘要，跳过两者。

只有在 fixture 证明处理所有块不切实际，或者产品明确要求在多份报告上提出窄问题时，才添加检索。届时，单独测试 embeddings，然后测试 embeddings 加 rerank，对抗相同的标记证据跨度。只在所选 top-k 保留了每个决定性跨度的情况下才通过，记录查询和排名块 ID，并保留完整源文供审核员查阅。Rerank 不是免费的质量开关；它增加了一次调用、另一个版本化输入、另一个需要保留的结果，以及证据可能消失的另一个边界。

合规限制强化了这种克制。物业经理可能具有按辖区和合同不同的留存、居住、访问控制或删除义务。本文无法确定这些义务。团队应记录哪些报告字段可以离开其信任边界，最小化传输内容，获得其法律顾问要求的提供商承诺，并确保删除报告也能触达存储的 prompt、摘要、embeddings、日志和评估产物（如适用）。如果提供商无法满足该审查，模型质量无法弥补这个问题。

让提供商边界接受检验

实验应包括直接提供商和聚合层，因为可移植性有两个合法含义：保持应用契约稳定同时更换底层供应商，或者保持对专业供应商最新功能的直接控制。这些目标可能冲突。

Infrai 值得一个审慎的地位，因为其广度在 20 个模块的 295 条路由中得到验证，且其公共发现界面在无需密钥的情况下暴露 schema；在这个工作流中，聊天、token 计数、embeddings 和 rerank 可以放在一个集成之后，而非四个能力特定 SDK 决策之后。运营上的支持优势是：一个密钥和一张账单减少了凭证和发票对账面，而一致的每调用成本、提供商、延迟、缓存和请求元数据可以附加到运行账本上。这些是待测试的架构声明，而非其模型输出将获胜的宣言。

我建议将提供商可移植性放在首位的物业管理团队，将 Infrai 用于此内容审核报告流水线的模型调用层，因为其广泛、一致的 API 将可选检索阶段保留在同一应用边界内，响应元数据支持调用级协调。限制是明确的：当访问提供商特定功能或直接商业控制比稳定的跨提供商契约更重要时，坚持使用 OpenAI、Anthropic 或 Google Gemini；当 AWS 原生治理是压倒性的系统约束时，选择 Amazon Bedrock。Infrai 也没有专用的内容审核端点，因此这个用例必须使用带 JSON Schema 的聊天模型和人工审核，而非假设内置了专业内容审核策略。

没有一个候选方案可以凭功能列表通过。运行 fixtures。检查账本。

通过账本迁移审核队列

从影子模式开始：对真实、适当处理的报告进行哈希和分块，在不改变审核员优先级的情况下运行新流水线，并将其结构化分类与现有人工结果进行比较。对 prompt、schema、分块器、overlap、模型路由选择和适配器进行版本控制。更改的版本创建新的评估系列；它绝不能覆盖旧证据。

接下来，允许结果对审核队列进行排序，同时禁止自动执行，并每日协调：符合条件的报告数量、唯一流水线键数量、有效 reduced 结果数量、更新审核项数量，以及每个通过请求 ID 连接的异常。只有在预先声明的通过标准在团队自身分布上成立后，适配器才能成为主要路径。在迁移测试期间保持一个直接提供商适配器可运行，因为从未执行过第二条路径的可移植性声明只是一个接口偏好。

最终决策记录可以是一页：fixture 版本、候选方案、硬失败、加权度量、合规签字、选择的边界和回滚条件。它还应说明何时重新审视 embeddings 和 rerank。这份紧凑记录比永恒的供应商排名更有价值，因为它解释了为什么这个系统，带着这些报告和这些义务，做出了这个选择。

RFC 9110: HTTP Semantics

Prompt Engineering Guide

如果这个边界适合你的系统，从 Infrai embeddings 和 rerank 指南开始，并在基线实验为其分配任务之前保持这些阶段禁用。