用Gemma 4构建自托管OCR

过去三年里,许多人认为 AI 越大越聪明。他们觉得参数越多,性能越好;GPU 越多,AI 就越智能。

用Gemma 4构建自托管OCR
微信 ezpoda免费咨询:AI编程 | AI模型微调| AI私有化部署
AI工具导航 | ONNX模型库 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo

过去三年里,许多人认为 AI 越大越聪明。他们觉得参数越多,性能越好;GPU 越多,AI 就越智能。

然而,这一普遍认知本周被谷歌的开放模型"Gemma 4"彻底颠覆。Gemma 是谷歌发布的一系列开放权重模型。"开放权重"意味着模型的权重数据可以自由获取。任何人都可以下载并在自己的 PC、服务器或云端运行它。

ChatGPT 和 Gemini Advanced 只能通过云端使用,而 Gemma 最大的优势在于它可以直接安装并运行在你的本地环境中。

这一切是怎么突然实现的?因为谷歌最新发布的"TurboQuant"技术,能够大幅降低大语言模型(LLM)的内存消耗。

据说它可以将内存效率降低到原来的六分之一,这有潜力显著降低 AI 的运行成本。我认为很多人都在期待"本地 LLM 终于变得更加轻量化"以及"以前无法运行的大模型现在可以在家用 GPU 上轻松运行"。

当 LLM(大规模语言模型)生成文本时,它使用一种称为"键值缓存(KV Cache)"的工作内存来存储过去已计算的 token 信息。没有它,每次生成新 token 时,都得从头重新计算。

问题在于,KV 缓存的大小会随着上下文(context 的长度)的增长而线性增加。当尝试处理长对话或长文档时,消耗 GPU 内存的正是 KV 缓存,而非模型权重。

总而言之,TurboQuant 并不是一种"神奇地大幅减少本地 LLM 整体体重的法术"。 其本质在于对"运行时不断膨胀的内存"——主要是 KV 缓存——进行强力压缩

那么,让我通过一个实时聊天机器人的快速演示来展示一切是如何运作的。

我将上传一张包含资产和负债信息的图片,你可以以任何格式上传。图片会直接显示出来。

如果你观察 Agent 的输出方式,你会发现它将文件保存到磁盘上的临时位置。文件以随机名称保存,但保留了正确的扩展名,因此格式可以被正确检测。

如果输入是 PDF,则使用 Poppler 库中的 pdftoppmPortable Pixmap)工具将每一页转换为 PNG 图像。这一步是必需的,因为视觉模型只接受图像输入。

页面以配置的 DPI(每英寸点数)进行渲染,通常为 300。较高的 DPI 可以提高准确性,但也会增加处理时间和文件大小。如果选择了页面范围,则只处理这些页面。图像临时存储,处理完毕后自动清理。

接下来,所有图像都会检查尺寸。如果图像的最大边超过最大维度限制(默认为 1536 像素),则使用高质量的 Lanczos 滤镜按比例缩放。这样既保持处理速度,又能保留足够的细节以确保文本识别的准确性。

如果文档类型设置为"auto",Agent 会快速对图像进行分类,如通用、表格、手写或扫描件。这有助于选择最佳的 OCR 提示词。如果手动选择了类型,则跳过此步骤。

之后,图像被编码并与选定的提示词一起发送到本地 Ollama API。请求使用流式传输,因此文本会在生成过程中逐步返回。Agent 将这些片段收集为最终结果,并跟踪 token 计数和处理时间等元数据。

最后,根据所选的输出类型对结果进行格式化。纯文本模式将所有内容合并,Markdown 模式添加结构,JSON 模式保留所有元数据。

这段代码将在我的 Patreon 上提供,因为它花费了我大量的时间和精力。如果你喜欢我的创作并希望看到更多类似的项目,在 Patreon 上支持我可以帮助我持续制作高质量的内容。我真心感谢你的支持。

1、Gemma 4 有何独特之处?

Gemma 4 有四种尺寸:E2B、E4B、26B、A4B 和 31B。较小的模型专为智能手机和边缘设备设计,而较大的模型则用于本地 PC 和工作站。

此外,它支持高达 256K token 的长上下文长度,并能处理超过 140 种语言。

较小尺寸支持 128K,较大尺寸支持 256K,这使得共享完整代码库或长设计文档变得非常实用。其功能也高度面向实际应用场景。

它原生支持函数调用(一种调用外部工具和 API 的机制),并且默认支持系统角色。所有模型都能处理文本和图像,较小的模型还原生支持语音。

换言之,它从一开始就不只是为聊天场景设计的,而是作为连接搜索、执行、格式化和决策的 Agent 基础设施。

不仅仅是"聪明",更是"易于集成到工作流中"。

我相信这就是 Gemma 4 的精髓所在。

虽然模型本身的智能程度很重要,但在实际应用中真正发挥作用的是三点:阅读长文本的能力、调用工具的能力、以及在本地运行的能力。

2、TurboQuant 有何独特之处?

我想在这里澄清的是,"它不是让模型本身变得更轻量化的技术"这一事实,绝不意味着它的价值很低。

相反,在本地 LLM 的实际运行中,KV 缓存才是后期变得更加显著的因素,因此减轻其负载的收益是相当可观的。

根据谷歌研究,TurboQuant 以极低位深度为目标进行压缩。对于 KV 缓存量化,它在 3.5 位/通道时实现了"绝对质量中性",在 2.5 位/通道时也仅有"边际质量下降"。

粗略来说,这个数字意味着:

  • 用于上下文保留的内存可以大幅减少。
  • 尽管如此,质量下降的可能性可以降到最低。
  • 如果 TurboQuant 在本地 LLM 中得到实现和普及,可以预期以下变化:

更容易处理长文本

这有可能使长文本摘要、代码库分析、文档输入和 RAG 等任务中处理更长的上下文变得更加容易。由于 KV 缓存随序列长度增长而增大,压缩它将使长文本操作更加实用。

在相同 GPU 上更容易维持性能

这将缓解"短文本运行良好,但长文本突然变得困难"的状况。这一改进对于约 16GB 显存的 GPU 尤为显著,因为这类 GPU 即使能加载模型本身,也常常在处理较长文本时捉襟见肘。

这与 KV 缓存压缩研究的总体趋势一致,研究报告表明内存减少可以提升吞吐量和批处理大小。

对多次执行和基于 Agent 的操作非常有效

对于需要在长时间对话、多任务处理、RAG 和代码辅助等场景下进行处理的应用,KV 缓存往往比模型本身更是瓶颈所在,因此优化这一方面的价值极高。

3、开始编码

我编写了一个函数,用于读取用户输入的类似"1–5"或"1,3,7–10"的字符串,并将其转换为整洁的页码列表。它首先按逗号拆分字符串,因此"1,3,7–10"会变成更小的片段。

每个片段会检查是否包含破折号——如果有,则将其视为一个范围,并填入起始和结束之间的所有数字。如果没有破折号,就直接取那个单独的数字。

它在构建列表时使用了集合,因此像"1–3,2"这样的重复项会被自动去除。最后,它将所有内容转换为排序后的列表,因此页码始终按顺序返回。

def parse_pages(page_str: str) -> list[int]:
    """
    Parse a page range string like '1-5' or '1,3,7-10'
    into a sorted list of 1-based page numbers.
    """
    pages = set()
    for part in page_str.split(","):
        part = part.strip()
        if "-" in part:
            start, end = part.split("-", 1)
            pages.update(range(int(start), int(end) + 1))
        else:
            pages.add(int(part))
    return sorted(pages)

我编写了一个函数,将 PDF 转换为 PNG 图像,以便 Gemma 4 能够读取它们。它首先检查 pdftoppm 是否已安装——如果没有,则打印安装提示并退出。

然后设置输出路径,使图像命名为"page-1"、"page-2"等。如果用户选择了特定页面,则为每一页运行一次 pdftoppm;否则对整个 PDF 运行一次。

-r 标志设置图像质量。转换完成后,它抓取所有"page-*.png"文件,进行排序,如果没有找到任何文件则退出。

辅助函数 extract_page_number 读取类似"page-01.png"的文件名,去掉扩展名,提取数字并转换为整数,这样应用就知道每张图像来自哪一页。

def pdf_to_images(
    pdf_path: str,
    output_dir: str,
    dpi: int = DEFAULT_DPI,
    pages: list[int] | None = None,
) -> list[Path]:
    """
    Convert a PDF to PNG images using pdftoppm (poppler).
    If `pages` is provided, only those pages are converted.
    Returns a sorted list of image paths.
    """
    if not shutil.which("pdftoppm"):
        print("Error: pdftoppm not found. Install poppler:")
        print("  macOS:  brew install poppler")
        print("  Ubuntu: sudo apt install poppler-utils")
        sys.exit(1)

    output_prefix = str(Path(output_dir) / "page")

    if pages:
        # Convert each requested page individually
        for p in pages:
            cmd = [
                "pdftoppm", "-png", "-r", str(dpi),
                "-f", str(p), "-l", str(p),
                pdf_path, output_prefix,
            ]
            subprocess.run(cmd, check=True, capture_output=True)
    else:
        # Convert all pages at once
        cmd = ["pdftoppm", "-png", "-r", str(dpi), pdf_path, output_prefix]
        subprocess.run(cmd, check=True, capture_output=True)

    images = sorted(Path(output_dir).glob("page-*.png"))

    if not images:
        print(f"Error: No page images extracted from {pdf_path}")
        sys.exit(1)

    return images

def extract_page_number(image_path: Path) -> int:
    """Extract the 1-based page number from a pdftoppm filename (e.g. page-03.png → 3)."""
    return int(image_path.stem.split("-")[-1])

接下来,我创建了一个函数,决定如何处理任何文件。它启动计时器,检查文件类型,然后走两条路径之一。对于图像,它运行 ocr_single_image 并将结果封装在一个包含文件信息、时间戳、模型名称和页码列表的字典中。

对于 PDF,它创建一个临时文件夹,使用 pdf_to_images 将页面转换为 PNG,对每张图像运行 OCR,收集结果,并构建相同的字典。临时文件夹会被自动删除。

如果文件既不是图像也不是 PDF,则打印错误并退出。无论哪种情况,函数始终返回相同的字典格式,因此 Agent 以相同的方式处理图像和 PDF。

# ── Single file processing ────────────────────────────────

def process_single_file(
    file_path: str,
    doc_type: str = "auto",
    model: str = DEFAULT_MODEL,
    dpi: int = DEFAULT_DPI,
    pages: list[int] | None = None,
    max_long_edge: int = MAX_IMAGE_LONG_EDGE,
) -> dict:
    """
    Run OCR on a single PDF or image file.
    Returns a unified result dict with per-page text and metadata.
    """
    file_path = Path(file_path)
    start_time = time.time()

    if file_path.suffix.lower() in IMAGE_EXTS:
        # Direct image OCR — single page result
        print("Recognizing image...", end="", flush=True)
        result = ocr_single_image(str(file_path), doc_type, model, max_long_edge)
        secs = result["duration_ms"] / 1000
        print(f" Done ({secs:.1f}s, {result['tokens']} tokens)")
        total_ms = (time.time() - start_time) * 1000
        return {
            "file": str(file_path.resolve()),
            "total_pages": 1,
            "processed_pages": 1,
            "model": result["model"],
            "created_at": datetime.now().isoformat(),
            "total_duration_ms": round(total_ms, 1),
            "pages": [
                {
                    "page": 1,
                    "text": result["text"],
                    "doc_type": result["doc_type"],
                    "tokens": result["tokens"],
                    "duration_ms": result["duration_ms"],
                }
            ],
        }

    elif file_path.suffix.lower() == PDF_EXT:
        # PDF: convert to images first, then OCR each page
        with tempfile.TemporaryDirectory(prefix="ocr_") as tmpdir:
            print(f"Converting PDF to images (DPI={dpi})...")
            images = pdf_to_images(str(file_path), tmpdir, dpi, pages)
            page_results = []
            print(f"Starting OCR on {len(images)} page(s)\n")

            for idx, img_path in enumerate(images):
                page_num = extract_page_number(img_path)
                print(f"  [{idx+1}/{len(images)}] Page {page_num} — recognizing...", end="", flush=True)
                result = ocr_single_image(str(img_path), doc_type, model, max_long_edge)
                secs = result["duration_ms"] / 1000
                print(f" Done ({secs:.1f}s, {result['tokens']} tokens)")
                page_results.append({
                    "page": page_num,
                    "text": result["text"],
                    "doc_type": result["doc_type"],
                    "tokens": result["tokens"],
                    "duration_ms": result["duration_ms"],
                })

            total_ms = (time.time() - start_time) * 1000
            return {
                "file": str(file_path.resolve()),
                "total_pages": len(images),
                "processed_pages": len(page_results),
                "model": model,
                "created_at": datetime.now().isoformat(),
                "total_duration_ms": round(total_ms, 1),
                "pages": page_results,
            }

    else:
        print(f"Error: Unsupported format '{file_path.suffix}'")
        sys.exit(1)

接下来,我编写了三个函数,它们都用于将结果字典转换为可读的字符串,只是风格不同。format_as_json 最简单——它将整个字典以格式化的 JSON 输出,保留时间戳和 token 计数等所有信息。

format_as_markdown 更有结构:它将输入标准化为列表,添加包含文件名、模型、页数、时间的标题,并循环遍历每一页,添加分隔线、页眉、隐藏的 HTML 注释和文本内容。

format_as_text 是最精简的——它标准化输入,只在有多个文档时添加文件名分隔符,然后直接输出原始文本,几乎没有额外格式。

在底部,FORMATTERS 将"json"、"md"或"txt"映射到对应的函数,这样 Agent 就可以用一行代码选择正确的格式化器。

# ── Output formatters ─────────────────────────────────────

def format_as_json(data: dict | list[dict]) -> str:
    """Serialize the result as pretty-printed JSON."""
    return json.dumps(data, ensure_ascii=False, indent=2)

def format_as_markdown(data: dict | list[dict]) -> str:
    """Format the result as Markdown with per-page sections."""
    items = data if isinstance(data, list) else [data]
    parts = []
    for doc in items:
        filename = Path(doc["file"]).name
        model = doc.get("model", "unknown")
        created = doc.get("created_at", "")
        total_sec = doc.get("total_duration_ms", 0) / 1000
        parts.append(f"# OCR: {filename}\n")
        parts.append(
            f"> Model: `{model}` | Pages: {doc['processed_pages']} | "
            f"Time: {total_sec:.1f}s | Date: {created}\n"
        )
        for page in doc["pages"]:
            page_sec = page.get("duration_ms", 0) / 1000
            parts.append("\n---\n")
            parts.append(f"## Page {page['page']}\n")
            parts.append(
                f"<!-- type: {page.get('doc_type', 'unknown')} | "
                f"{page_sec:.1f}s | {page.get('tokens', 0)} tokens -->\n"
            )
            parts.append(f"\n{page['text']}\n")
    return "\n".join(parts)

def format_as_text(data: dict | list[dict]) -> str:
    """Format the result as plain text, one page after another."""
    items = data if isinstance(data, list) else [data]
    parts = []
    for doc in items:
        if len(items) > 1:
            filename = Path(doc["file"]).name
            parts.append(f"{'=' * 60}")
            parts.append(f"FILE: {filename}")
            parts.append(f"{'=' * 60}\n")
        for i, page in enumerate(doc["pages"]):
            if len(doc["pages"]) > 1:
                parts.append(f"--- Page {page['page']} ---\n")
            parts.append(page["text"])
            if i < len(doc["pages"]) - 1:
                parts.append("")  # blank line between pages
    return "\n".join(parts)

# Map format name → formatter function
FORMATTERS = {
    "json": format_as_json,
    "md": format_as_markdown,
    "txt": format_as_text,
}

4、我的感想

Gemma 4 最重要的一点不在于"又多了一个顶级模型"。

我认为真正的趋势是,AI 开发正从纯粹的云端模式转向云端与本地计算相结合的混合模式。 重型推理和最终决策在云端处理,而日常辅助和内部数据处理则在本地完成。

另一方面,TurboQuant 是一项有潜力大幅提升 AI 性能和效率的技术,预计未来将受到更多关注。

它目前还不是面向普通用户的服务,但它是一项将影响 AI 未来走向的重要技术,值得持续关注 ✨


原文链接:Gemma 4 + Turboquant + RAG: Better OCR & Self-Hosted

汇智网翻译整理,转载请标明出处