用多模态 LLM 自动化文档数据提取

我们正在 Oxlo.ai 上构建一个多模态文档提取器，它可以读取发票或表单的图像并返回结构化 JSON。这样有助于财务和运营团队自动化数据录入，而无需维护单独的 OCR 管道。

前置条件：

• 来自 https://portal.oxlo.ai 的 Oxlo.ai API 密钥
• OpenAI SDK：pip install openai
• 一个示例图像，如收据、发票或表单，保存为 PNG 或 JPEG

步骤 1：配置 Oxlo.ai 客户端

我每个项目都从实例化客户端开始。Oxlo.ai 暴露了一个完全兼容 OpenAI 的端点，所以唯一的区别是基础 URL。

from openai import OpenAI

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key="YOUR_OXLO_API_KEY"
)

步骤 2：编写系统提示

系统提示是我和模型之间的约定。我保持严格的格式要求，因为我需要确定的 JSON 输出，而不是对话式的填充。

SYSTEM_PROMPT = """You are a precise document extraction engine. Analyze the provided image and extract all visible structured data.

Rules:
- Identify the document type (invoice, receipt, form, or other).
- Extract fields such as vendor, date, total amount, line items, tax, and currency.
- Return ONLY a JSON object. No markdown, no explanations.
- If a field is missing, use null.
- Use camelCase for keys.

Example structure:
{
  "documentType": "invoice",
  "vendor": "Acme Corp",
  "date": "2024-01-15",
  "currency": "USD",
  "total": 123.45,
  "tax": 8.50,
  "lineItems": [
    {"description": "Widget A", "quantity": 2, "unitPrice": 50.00}
  ]
}
"""

步骤 3：编码图像

Oxlo.ai 接受 base64 data URI，格式与 OpenAI Vision API 相同。我编写了一个小助手函数来读取本地文件并构建消息载荷。

import base64

def encode_image(image_path):
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

def build_messages(image_path, text_query):
    b64 = encode_image(image_path)
    data_uri = f"data:image/jpeg;base64,{b64}"
    
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {
            "role": "user",
            "content": [
                {"type": "text", "text": text_query},
                {"type": "image_url", "image_url": {"url": data_uri}}
            ]
        }
    ]

步骤 4：调用 Vision 模型

我在 Oxlo.ai 上使用 Kimi K2.6，因为它在单次调用中处理视觉、推理和长上下文的能力很强。我还启用了 JSON 模式来保证可解析的输出。

def extract_document(image_path, query="Extract all data from this document."):
    messages = build_messages(image_path, query)
    
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=messages,
        response_format={"type": "json_object"},
        max_tokens=2048
    )
    
    return response.choices[0].message.content

# Example usage
result = extract_document("invoice_sample.jpg")
print(result)

下面是完整脚本。我将 API 密钥保存在环境变量中，并从命令行接受图像路径。

import os
import sys
import base64
from openai import OpenAI

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key=os.environ["OXLO_API_KEY"]
)

SYSTEM_PROMPT = """You are a precise document extraction engine. Analyze the provided image and extract all visible structured data.

Rules:
- Identify the document type (invoice, receipt, form, or other).
- Extract fields such as vendor, date, total amount, line items, tax, and currency.
- Return ONLY a JSON object. No markdown, no explanations.
- If a field is missing, use null.
- Use camelCase for keys.
"""

def encode_image(image_path):
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

def extract(image_path, query="Extract all data from this document."):
    b64 = encode_image(image_path)
    data_uri = f"data:image/jpeg;base64,{b64}"
    
    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": query},
                    {"type": "image_url", "image_url": {"url": data_uri}}
                ]
            }
        ],
        response_format={"type": "json_object"},
        max_tokens=2048
    )
    
    return response.choices[0].message.content

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python extractor.py ")
        sys.exit(1)
    
    print(extract(sys.argv[1]))

针对典型的 SaaS 发票运行它，得到的输出如下：

{
  "documentType": "invoice",
  "vendor": "CloudHosting Inc",
  "date": "2025-03-12",
  "currency": "USD",
  "total": 249.00,
  "tax": 20.75,
  "lineItems": [
    {"description": "Dedicated Server - Tier 2", "quantity": 1, "unitPrice": 199.00},
    {"description": "Bandwidth Overage", "quantity": 50, "unitPrice": 1.00}
  ]
}

两个具体的后续步骤。首先，将其包装在 FastAPI 端点中，以便其他服务可以 POST 图像并接收 JSON。其次，批量处理整个目录。由于 Oxlo.ai 采用按请求计费（参见 https://oxlo.ai/pricing），即使为每次调用添加详细的系统提示和高分辨率图像，成本也保持不变。