非结构化文档 ETL 与多模态数据

结构化 ETL：Schema 已知 → 映射 → 转换 → 加载
非结构化 ETL：Schema 未知 → 推断结构（解析/抽取）→ 归一化 → 分块 → 嵌入 → 加载

import fitz  # PyMuPDF

doc = fitz.open("contract.pdf")
page = doc[0]
text = page.get_text("text")

# 关键判断：文字层是否存在（决定是否走 OCR）
if len(text.strip()) < 50 and len(page.get_images()) > 0:
    route = "ocr"
else:
    route = "text_layer"
print(f"pages={len(doc)} route={route} chars={len(text)}")

1. 数字原生（Text-based）   有文字层，直接提取，顺序可能乱
2. 扫描件（Image-based）    无文字层，必须 OCR
3. 混合型                    部分页有文字层，部分没有

for i, page in enumerate(doc):
    t = page.get_text("text").strip()
    if len(t) < 20:
        pages_needing_ocr.append(i)

import pdfplumber

with pdfplumber.open("report.pdf") as pdf:
    page = pdf.pages[0]
    # 按 y 坐标分行，再按 x 坐标排序，近似还原阅读顺序
    words = page.extract_words(use_text_flow=False, keep_blank_chars=False)
    lines = {}
    for w in words:
        key = round(w["top"] / 5)          # 5pt 容差归并同一行
        lines.setdefault(key, []).append(w)
    ordered = [
        " ".join(x["text"] for x in sorted(lines[k], key=lambda w: w["x0"]))
        for k in sorted(lines)
    ]

# Tesseract：轻量、离线、中文需装语言包
tesseract scan.png out -l chi_sim+eng --psm 6

# PaddleOCR：中文场景精度更高，支持版面分析
python -m paddleocr --image_dir ./pages --lang ch --use_angle_cls true

from dataclasses import dataclass, field
from typing import Protocol

@dataclass
class Block:
    type: str          # title / narrative_text / table / figure
    text: str
    page: int
    bbox: tuple = field(default_factory=tuple)
    level: int = 0     # 标题层级

class Parser(Protocol):
    name: str
    version: str
    def parse(self, path: str) -> list[Block]: ...

# 不同解析器实现同一接口，产物统一为 list[Block]
PARSERS = {"pymupdf": PyMuPDFParser(), "docling": DoclingParser(), "tika": TikaParser()}

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,          # 目标 token 数（约 600~1000 字）
    chunk_overlap=120,       # 15% 重叠，避免语义在边界被截断
    separators=["\n## ", "\n### ", "\n\n", "\n", "。"],  # 优先级从高到低
    length_function=len,
)
chunks = splitter.split_text(markdown_text)

def chunk_with_breadcrumb(doc_sections):
    out = []
    for sec in doc_sections:
        breadcrumb = " > ".join(sec["path"])          # 标题路径
        for i, c in enumerate(splitter.split_text(sec["text"])):
            out.append({
                "text": f"[{breadcrumb}]\n{c}",        # 拼进正文，提升召回
                "metadata": {"breadcrumb": breadcrumb, "chunk_index": i,
                             "doc_id": sec["doc_id"], "page": sec["page"]},
            })
    return out

{
  "doc_id": "contract-2026-000123",
  "source_uri": "s3://docs/contracts/2026/000123.pdf",
  "source_hash": "sha256:9f2c...",
  "page": 7,
  "breadcrumb": "合同 > 第 3 条 > 付款方式",
  "chunk_index": 12,
  "parser": "docling@2.1.0",
  "ocr_used": false,
  "extracted_at": "2026-10-07T19:18:00+08:00",
  "acl": ["group:legal", "group:finance"],
  "language": "zh"
}

import hashlib
from tenacity import retry, wait_exponential, stop_after_attempt

@retry(wait=wait_exponential(multiplier=1, min=1, max=30), stop=stop_after_attempt(5))
def embed_batch(texts, model="bge-m3", batch_size=64):
    vectors = []
    for i in range(0, len(texts), batch_size):
        vectors.extend(client.embed(model, texts[i:i + batch_size]))
    return vectors

def needs_reembed(chunk, existing_hash):
    h = hashlib.sha256(chunk["text"].encode()).hexdigest()
    return h != existing_hash

-- 元数据表（PostgreSQL）：支撑过滤与治理
CREATE TABLE doc_chunks (
  chunk_id      TEXT PRIMARY KEY,
  doc_id        TEXT NOT NULL,
  content       TEXT NOT NULL,
  content_hash  TEXT NOT NULL,
  breadcrumb    TEXT,
  page          INT,
  embed_model   TEXT NOT NULL,
  chunk_version TEXT NOT NULL,
  acl           TEXT[],
  created_at    TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX ON doc_chunks (doc_id);
CREATE INDEX ON doc_chunks USING gin (acl);

Bronze：原始文件落地（对象存储 + 内容哈希清单）
Silver：解析产物（文本 + 版面 JSON + 表格 CSV）
Gold：分块与嵌入（chunk 表 + 向量集合）

def sync_documents(source_root, manifest_table):
    for uri, stat in scan(source_root):
        h = sha256_of(uri)
        prev = manifest_table.get(uri)
        if prev and prev["hash"] == h:
            continue                        # 内容未变，跳过
        yield {"uri": uri, "hash": h, "action": "parse"}

成本大头 = OCR 计算 + 嵌入 API 调用 + 向量存储 + 重处理

优化顺序：
1. 哈希去重，跳过未变文档（收益最大）
2. 先判断有无文字层，能不提 OCR 就不提
3. 嵌入按内容哈希缓存，跨批次复用
4. 向量量化（PQ/SQ）降低存储，牺牲少量召回
5. 分块粒度合理，避免过细导致嵌入次数翻倍