金融代码审查 JSON 提取：分块 Token 预算与超时兜底方案

短答案：对于金融科技代码审查提取场景，给每个文档分块设置 token 预算和截止时间，按发现类型检索证据，并在记录租户所有权的 worker 中合并经验证的 JSON。不要让一次模型请求负责读取整个变更历史。

这是保持可审查性的最小复杂度形态：解析、计数、切分、检索、提取、验证、合并。一个长拉取请求或审计附件由此变成一组有边界的作业，而不是一次可能消耗未知量上下文并留下模糊超时信息的请求。

如何让金融科技文本转 JSON worker 将长文档保持在 token 限制内？

从输出 schema 开始，而非文档。对于代码审查助手，有用的字段可能是 severity、file、line、finding 和 evidence。每个字段组都应该有一个检索问题。"Find a possible authorization bypass" 是比 "summarize this repository" 更好的查询。单个文档可以包含变更的代码、工单、政策摘录和旧讨论；它们不应该在提取 prompt 中获得同等的篇幅。

推理前先计数。感知 tokenizer 的计数优于词数计数，但控制流是相同的：在发送前拒绝或切分超出模型和作业预算的输入。先按文件、函数、标题或段落边界切分，再应用硬上限。重叠可以保留跨边界条件，但也会重复证据并增加租户使用量，所以要测量它而不是习惯性地选择大重叠。在金融科技审查中，这个核算细节有一个实际后果：如果单个变更文件被检索用于五种发现类型，worker 必须记录五个阶段决策，即使源分块被复用，否则租户报告无法解释作业为何消耗了其配额。

每个分块保留稳定的文档版本、租户 ID、分块 ID 和源位置。嵌入对宽泛的候选通过有用；重排步骤随后可以将小的候选集与发现特定查询进行比较。最终提取调用应该只看到选定的段落和 schema。检索召回率和 JSON 有效性是不同的指标。将它们分开。

我不确定是否存在通用分块大小。正确的值取决于代码密度、注释、schema 宽度，以及一个发现是否需要两个相邻函数。一个小的标注 fixture 集可以解决这个问题：一起测量分块大小、重叠量、候选数量和截止时间，然后保留保持所需证据的最小上下文。

在调优检索之前构建有边界的 worker

下面的示例故意保持确定性。它可以在 notebook 中运行，不使用凭证，并且使核算和合并规则在添加模型之前可测试。在生产环境中，用嵌入加重排替换词法评分器，用 schema 约束的模型调用替换 extract_findings。保留作业标识、租户归属和冲突行为。

import json
import re
from dataclasses import dataclass

@dataclass(frozen=True)
class Chunk:
    chunk_id: str
    file_name: str
    start_line: int
    text: str

def split_text(file_name: str, text: str, lines_per_chunk: int = 8) -> list[Chunk]:
    lines = text.splitlines()
    return [
        Chunk(
            chunk_id=f"{file_name}:{start + 1}",
            file_name=file_name,
            start_line=start + 1,
            text="\n".join(lines[start:start + lines_per_chunk]),
        )
        for start in range(0, len(lines), lines_per_chunk)
    ]

def approximate_tokens(text: str) -> int:
    return max(1, len(re.findall(r"\w+|[^\w\s]", text)))

def retrieve(chunks: list[Chunk], query: str, limit: int = 2) -> list[Chunk]:
    terms = set(re.findall(r"[a-z0-9_]+", query.lower()))

    def score(chunk: Chunk) -> tuple[int, str]:
        words = set(re.findall(r"[a-z0-9_]+", chunk.text.lower()))
        return len(terms & words), chunk.chunk_id

    return sorted(chunks, key=score, reverse=True)[:limit]

def extract_findings(chunk: Chunk) -> list[dict]:
    findings = []
    for line_number, line in enumerate(chunk.text.splitlines(), chunk.start_line):
        if "TODO_SECURITY_REVIEW" in line:
            findings.append({
                "severity": "high",
                "file": chunk.file_name,
                "line": line_number,
                "finding": "Review the security-sensitive change",
                "evidence": line.strip(),
                "evidence_chunk": chunk.chunk_id,
            })
    return findings

def merge_findings(results: list[dict]) -> list[dict]:
    seen = set()
    merged = []
    for result in results:
        identity = (result["file"], result["line"], result["finding"])
        if identity not in seen:
            seen.add(identity)
            merged.append(result)
    return merged

def review_document(tenant_id: str, job_id: str, files: dict[str, str]) -> dict:
    chunks = [
        chunk
        for file_name, text in files.items()
        for chunk in split_text(file_name, text)
    ]
    queries = ["authorization access security", "input validation injection"]
    selected = {
        chunk.chunk_id: chunk
        for query in queries
        for chunk in retrieve(chunks, query)
    }
    input_tokens = sum(approximate_tokens(chunk.text) for chunk in selected.values())
    findings = merge_findings([
        finding
        for chunk in selected.values()
        for finding in extract_findings(chunk)
    ])
    return {
        "tenant_id": tenant_id,
        "job_id": job_id,
        "input_tokens": input_tokens,
        "findings": findings,
    }

files = {
    "payments.py": """
def approve_payment(user, payment):
    # TODO_SECURITY_REVIEW: authorization check belongs here
    return process(payment)
""".strip()
}
print(json.dumps(review_document("tenant-acme", "review-042", files), indent=2))

上面的词基计数器是一个测试 double。生产 worker 应该使用所选模型的 tokenizer 或可信的计数端点，为指令和输出预留空间，并强制执行独立的时间截止。重要的是将计数与作业一起记录。没有输入 token 计数、候选 ID 和阶段名称的超时几乎算不上事故报告。

验证属于分块边界。拒绝未知字段，强制 severity 的枚举值，并要求每个非空发现都有 evidence。格式错误的响应应该作为单个分块结果重试或隔离。它不应该强制重放整个文档，也不应该仅仅因为它是有效 JSON 就进入最终审查记录。

为什么按租户核算会改变设计

按租户成本可见性是架构需求，而非仪表板装饰。在摄取、检索、重排、提取、重试和存储中携带租户 ID。记录每个阶段的 prompt tokens、completion tokens、模型标识符、延迟、重试次数和结果。如果共享队列只报告每日总数，大客户可以悄悄消耗本应留给小租户的预算。

使用两个预算。第一个是硬作业配额：最大选定 tokens、最大分块数和最大经过时间。第二个是告警阈值，为重试和包含异常密集代码的文档留出空间。当租户超过告警阈值时，根据书面策略减少候选数量或将作业移至批处理通道。保留足够的证据来解释该决定。

这也暴露了一个常见的核算错误：只对成功提取计费。失败的检索、验证重试和放弃的请求仍然使用资源。将它们归因于同一租户和文档版本。对于金融科技审查工作流，空结果和失败结果必须保持区别；否则成本报告可能看起来健康而召回率却在崩溃。

三个词：测量各阶段。

当提取超时或证据冲突时应该发生什么？

交互式请求应该将大文档加入队列并返回作业标识。Worker 可以重试单个分块，而调用方轮询或接收完成事件。使重试替换相同（tenant_id、document_version、chunk_id、stage）键的结果。这可以防止晚到的响应作为第二个发现被追加。

重试语义需要一个刻意的边界。RFC 9110 描述了 HTTP 的幂等性和方法语义；应用程序仍然需要提供幂等性键并使其存储操作成为条件性的。只有当 worker 可以识别相同的逻辑操作时，重试才是安全的。退避应该响应速率限制，而 token 预算拒绝应该路由到切分或批处理策略，而不是不变地重试。

冲突也需要证据。如果两个分块为同一行产生不同的 severity，保留两个候选，并对相关摘录运行窄范围裁决。不要让"最后响应获胜"来决定合规相关发现。如果源缺失或矛盾，返回人类可以检查的明确审查状态。

问题是延迟。按字段检索、重排、验证和冲突通过都会增加工作，所以当产品承诺对每个文档立即给出答案时，此设计不适用。对于短 diff 使用更小的 schema 和直接有界请求；对于长审计、多文件变更和证据可追溯性比即时显示更重要的任何工作流，使用队列。

Notebook 到生产的交接检查清单

在发布之前，从重要的形态构建 fixture：跨函数拆分的发现、重复标识符、生成的文件、大 diff、缺失证据和具有严格配额的租户。固定预期字段和可接受的源位置。在衡量端到端 JSON 准确性之前评估检索召回率，然后添加超时率、p95 阶段延迟、重复率和每个租户的 tokens。我还会在只更改候选限制后，通过 worker 重放一个文档版本。如果输出发生变化，fixture 应该显示新段落是添加了证据还是仅仅取代了更好的段落；没有这种比较，较低的 token 计数可能看起来像改进，而高严重性发现却悄悄失去了其源行。这正是 notebook 到生产工作流发挥作用的地方：实验足够小以至于可以检查，但相同的标识符和指标在移入队列后仍然有效。

将敏感源文本排除在普通日志之外。尽可能存储引用和哈希，限制每个租户的并发，并使取消对 worker 可见。对超出预算的阶段发出告警："chunk 17 exceeded extraction time" 是可操作的；"the AI request timed out" 不是。

决策规则很简单。当文档很长、证据必须被引用，或租户使用必须被解释时，选择有边界的分块作业。当 diff 很小且其 token 计数已知时，选择直接请求。两种情况下，schema 验证、证据保留和按租户指标都是提取系统本身的组成部分。