从脏文档提取结构化数据的完整管道

模型的调用只是提取管道的最容易的部分，也是每个教程都会展示的部分。真正的准确率在于调用之前对文档所做的处理，以及调用之后对记录所做的断言。

管道的形态

raw file
   -> normalise to text        (PDF/HTML/OCR -> plain text, reading order fixed)
   -> chunk                    (overlapping, on structural boundaries)
   -> extract per chunk        (strict schema, temperature 0)
   -> ground                   (every quoted span must occur in the chunk)
   -> validate                 (types, ranges, cross-field invariants)
   -> repair once, or drop     (bounded; never a loop)
   -> merge + dedupe           (across chunks, by a stable key)
   -> store with provenance    (chunk id, char offsets, model, schema version)

其中有两个阶段通常被遗漏。Grounding 区分了"模型读到了"和"模型写出了看似合理的内容"。Provenance 让你在六个月后仍能回答"这个数字从哪来的"，没有它，一条错误的记录将无法被证伪。

准备工作占了大部分的准确率

阅读顺序。朴素的 PDF 文本提取按照内容在文件中出现的顺序返回内容，而不是人类阅读的顺序。双栏布局会交错出现；表格会变成一列没有表头关联的数字。如果你的准确率很差而你的 prompt 看起来没问题，打印出你实际发送的文本。这是最常见的原因，而且从模型侧是看不出来的。

按结构分块，而不是按长度分。在标题、换页或表格边界处切分，然后将小块合并到你的预算范围内。跨分块边界的字段无法被找到；几百个字符的重叠让它可以被找到两次，dedupe 会处理这种情况，而漏掉则不会。

保留偏移量。携带每个块在源文档中的起始偏移。Grounding 给你分块内的偏移；两者相加得到原始文档中的引用，这是审核者需要的。

不要去除你认为是噪音的内容。你认为的页眉、页脚和页码通常带有发票号或日期。如果非要 stripping，请在提取之后进行。

Span Grounding：零成本的检查

要求每个提取值旁边都有一个字面引文，在它之前发出，然后断言该引文是你发送的分块的子字符串。这是一个字符串比较。它不需要标签、不需要 judge model，也不需要 ground truth，而且它能捕捉到人们最担心的失败模式——一个根本不在文档中的值。

这不是正确性的证明。引文可以是真实的，但从它派生出来的值仍然可能是错误的，而且模型也可能引用了错误的真实跨度。它给你的是一个硬性底线：一条引文不存在的记录就不是一条记录。实际上，你还会得到第二个意想不到的好处——引文使人工审核变得更快，因为审核者读两行而不是十页。

比较之前在两边都规范化空白符，否则你会因为换行而失败。不要规范化大小写；模型应该是在复制。

把未 grounding 的比率作为一等指标来监控，而不是作为一行日志，因为它区分了两种 otherwise 看起来一样的问题。低且稳定的比率意味着检查在工作。对某一文档类型飙升的比率通常意味着引文是真实的而你的空白符规范化是错误的——PDF 提取中的连字、不间断空格和软连字符是常见的罪魁祸首，它们会让一个完美复制的引文在子字符串测试中失败。比率在所有类型上逐渐上升是有意思的那个，值得在任何准确率指标变动之前调查。

import json, re
from dataclasses import dataclass
from openai import OpenAI

client = OpenAI()
MODEL = "your-model-id"

SCHEMA = {
  "type": "object",
  "additionalProperties": False,
  "required": ["findings"],
  "properties": {
    "findings": {
      "type": "array",
      "items": {
        "type": "object",
        "additionalProperties": False,
        "required": ["quote", "field", "value"],
        "properties": {
          "quote": {"type": "string",
                    "description": "Text copied EXACTLY from the document, including punctuation."},
          "field": {"type": "string",
                    "enum": ["invoice_number", "total", "issue_date", "customer_name"]},
          "value": {"type": "string",
                    "description": "The normalised value. Dates as YYYY-MM-DD, totals as digits and one dot."}
        }
      }
    }
  }
}

def ws(s: str) -> str:
    return re.sub(r"\s+", " ", s).strip()

@dataclass
class Finding:
    field: str
    value: str
    quote: str
    offset: int          # into the original document

def extract_chunk(text: str, chunk_start: int) -> tuple[list[Finding], list[str]]:
    resp = client.chat.completions.create(
        model=MODEL,
        temperature=0,
        messages=[
            {"role": "system",
             "content": "Extract only what is literally printed. Quote before you answer. "
                        "If a field is not present, do not emit a finding for it."},
            {"role": "user", "content": text},
        ],
        response_format={"type": "json_schema",
                         "json_schema": {"name": "extraction", "strict": True, "schema": SCHEMA}},
    )
    choice = resp.choices[0]
    if choice.finish_reason == "length":
        raise Truncated(chunk_start)          # your bug: raise max_tokens or shrink the chunk

    findings, rejected = [], []
    haystack = ws(text)
    for f in json.loads(choice.message.content)["findings"]:
        pos = haystack.find(ws(f["quote"]))
        if pos < 0:
            rejected.append(f"ungrounded {f['field']}={f['value']!r} quote={f['quote']!r}")
            continue
        findings.append(Finding(f["field"], f["value"], f["quote"], chunk_start + pos))
    return findings, rejected

class Truncated(Exception): pass

注意这里没有的东西。没有重试循环，没有"让模型修复它"，也没有因为否则运行就会是空的而接受未 grounded 发现的回退。被拒绝的发现进入一个你计数和查看的列表；是否修复或重试是一个独立的决策，有它自己的算术，它属于这个函数之外。

重叠的分块按设计产生重复。在 (field, value) 上去重，而不是在 quote 上，因为同一个事实可能从两个地方被引用，并保留最低的偏移量，这样引用指向第一次出现的位置。

冲突是有意思的情况：两个分块给出不同的 total 值。不要默默地取第一个。根据领域的不同，正确的规则通常是"优先选择最靠近命名该字段的标题的出现"、"优先选择最后一页"或"标记待审核"。三者都是合理的，而且都比一个随机的赢家要好，因为冲突率是一个指标，而随机的赢家不是。

将 provenance 与值一起存储——分块 id、偏移量、模型 id、schema 版本。它每条记录只增加几个字节，但它是将一个需要十分钟的数据质量调查和一个需要一周的调查区分开来的关键。尤其是 schema 版本，它使得后续的迁移成为可能。

两个操作性细节决定了这条管道是否能经受住与真实语料库的接触。将工作单元设为分块，而不是文档，并给它一个幂等键 (document_hash, chunk_index, schema_version, prompt_version)。一个在三分之二处死掉的运行然后恢复而不是重启，以及在 prompt 变更后的重新运行只重新提取该变更影响的部分。其次，在 provider 侧而不是你的 worker 数量上限制并发，并将 429 视为背压而不是立即重试的错误——提取回填是发现速率限制的经典方式，而无限的重试风暴会将一个慢 job 变成失败的 job。

验证与修复：对畸形输出的处理

数组与计数：为什么模型返回十个中的七个

结构化提取中的置信度分数