Skip to content
LlamaParse 文档版本 跟随上游 源快照 455cd51e 更新于 2026-10-16

本示例演示如何在 Parse 照常返回的条目级版面框之外,拿到逐词、逐行与逐表格单元格的边界框,以及如何取回并遍历承载这些数据的 JSONL 边车文件。

在这些情况下用它:

  • 在 PDF 查看器里高亮单个词或单行,用于引用的反向定位。
  • 把抽取出的答案落地到确切的字形,而不是整个段落。
  • 做一个并排预览:从 markdown 文字悬停联动到源文档上的高亮区域。

细粒度边界框并不内联在解析结果的响应里 —— 它们住在一个独立的 JSONL 边车文件中,由结果通过一个预签名 URL 指过去。这是有意做的拆分:长文档上这个边车可能有好几 MB,而多数调用方并不需要它。整个流程就两步:带着 granular_bboxes 解析,然后按边车 URL 下载。

1. 准备

设置 API key,让各 SDK 自动读取:

终端窗口

export LLAMA_CLOUD_API_KEY="llx-..."
  • Python
  • TypeScript
  • Go
  • Java
  • CLI

安装 SDK —— 外加 httpx,下面下载边车时要用它:

终端窗口

pip install "llama-cloud>=2.8" httpx
from llama_cloud import LlamaCloud

client = LlamaCloud()  # reads LLAMA_CLOUD_API_KEY from the environment

安装 SDK:

终端窗口

npm install @llamaindex/llama-cloud
import LlamaCloud from '@llamaindex/llama-cloud';

const client = new LlamaCloud(); // reads LLAMA_CLOUD_API_KEY from the environment

安装 SDK:

终端窗口

go get github.com/run-llama/llama-parse-go
import (
  "context"

  llamacloud "github.com/run-llama/llama-parse-go"
)

ctx := context.Background()
client := llamacloud.NewClient() // reads LLAMA_CLOUD_API_KEY from the environment

把 SDK 加进你的构建:

implementation("ai.llamaindex:llama-cloud:1.3.0")
import ai.llamaindex.llamacloud.client.LlamaCloudClient;
import ai.llamaindex.llamacloud.client.okhttp.LlamaCloudOkHttpClient;

// reads LLAMA_CLOUD_API_KEY from the environment
LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();

安装 CLI:

终端窗口

go install github.com/run-llama/llama-parse-cli/cmd/llp@latest

llp 会从环境里读取 LLAMA_CLOUD_API_KEY(也可用 --api-key 传入)。

2. 带着 granular_bboxes 解析

把 output_options.granular_bboxes 设为 "word"、"line"、"cell" 的任意子集。你可以只要一个层级,也可以三个都要。Parse 会自动产出那份 JSONL 边车 —— 没有对应的 expand 值要加。

在 Python 与 TypeScript 里,SDK 会阻塞到任务结束;在 Go、Java 与 CLI 里,你要先创建任务、轮询到终态,再取结果。这里的 items 是可选的 —— 请求它是为了把内联的 items 树与边车做对照;无论你要不要它,边车 URL 本身都会自动附在结果上。

从快速上手 里下载这份报告:

终端窗口

curl -fL -o executive-summary-2024.pdf https://fiscal.treasury.gov/system/files/files/reports-statements/financial-report/2024/executive-summary-2024.pdf
  • Python
  • TypeScript
  • Go
  • Java
  • CLI
# 1) Upload the file
file = client.files.create(
    file="executive-summary-2024.pdf",
    purpose="parse",
)

# 2) Parse with word + line + cell grounding
result = client.parsing.parse(
    file_id=file.id,
    tier="agentic",
    version="latest",
    output_options={
        "granular_bboxes": ["word", "line", "cell"],
    },
    expand=["items"],
)
import fs from 'fs';

// 1) Upload the file
const file = await client.files.create({
  file: fs.createReadStream('executive-summary-2024.pdf'),
  purpose: 'parse',
});

// 2) Parse with word + line + cell grounding
const result = await client.parsing.parse({
  file_id: file.id,
  tier: 'agentic',
  version: 'latest',
  output_options: {
    granular_bboxes: ['word', 'line', 'cell'],
  },
  expand: ['items'],
});
// 1) Upload the file
f, err := os.Open("executive-summary-2024.pdf")
if err != nil {
  log.Fatal(err)
}
defer f.Close()

file, err := client.Files.New(ctx, llamacloud.FileNewParams{
  File:    f,
  Purpose: "parse",
})
if err != nil {
  log.Fatal(err)
}

// 2) Parse with word + line + cell grounding
job, err := client.Parsing.New(ctx, llamacloud.ParsingNewParams{
  FileID:  llamacloud.String(file.ID),
  Tier:    llamacloud.ParsingNewParamsTierAgentic,
  Version: llamacloud.ParsingNewParamsVersionLatest,
  OutputOptions: llamacloud.ParsingNewParamsOutputOptions{
    GranularBboxes: []string{"word", "line", "cell"},
  },
})
if err != nil {
  log.Fatal(err)
}

// expand is a GET parameter — poll until terminal, then fetch the result
getParams := llamacloud.ParsingGetParams{Expand: []string{"items"}}
result, err := client.Parsing.Get(ctx, job.ID, getParams)
if err != nil {
  log.Fatal(err)
}
for result.Job.Status != "COMPLETED" && result.Job.Status != "FAILED" && result.Job.Status != "CANCELLED" {
  time.Sleep(2 * time.Second)
  result, err = client.Parsing.Get(ctx, job.ID, getParams)
  if err != nil {
    log.Fatal(err)
  }
}
if result.Job.Status != "COMPLETED" {
  log.Fatalf("parse ended as %s", result.Job.Status)
}
// 1) Upload the file
FileCreateResponse file = client.files().create(FileCreateParams.builder()
        .file(Paths.get("executive-summary-2024.pdf"))
        .purpose("parse")
        .build());

// 2) Parse with word + line + cell grounding
ParsingCreateResponse job = client.parsing().create(ParsingCreateParams.builder()
        .fileId(file.id())
        .tier(ParsingCreateParams.Tier.AGENTIC)
        .version(ParsingCreateParams.Version.LATEST)
        .outputOptions(ParsingCreateParams.OutputOptions.builder()
                .addGranularBbox(ParsingCreateParams.OutputOptions.GranularBbox.WORD)
                .addGranularBbox(ParsingCreateParams.OutputOptions.GranularBbox.LINE)
                .addGranularBbox(ParsingCreateParams.OutputOptions.GranularBbox.CELL)
                .build())
        .build());

// expand is a query parameter — poll until terminal, then fetch the result
ParsingGetParams getParams = ParsingGetParams.builder()
        .jobId(job.id())
        .addExpand("items")
        .build();

ParsingGetResponse result = client.parsing().get(getParams);
while (!result.job().status().equals(ParsingGetResponse.Job.Status.COMPLETED)
        && !result.job().status().equals(ParsingGetResponse.Job.Status.FAILED)
        && !result.job().status().equals(ParsingGetResponse.Job.Status.CANCELLED)) {
    Thread.sleep(2000);
    result = client.parsing().get(getParams);
}
if (!result.job().status().equals(ParsingGetResponse.Job.Status.COMPLETED)) {
    throw new RuntimeException("parse ended as " + result.job().status());
}

终端窗口

# Upload the document
FILE_ID=$(llp files create \
  --file executive-summary-2024.pdf \
  --purpose parse | jq -r '.id')

# Start a parse job with word + line + cell grounding
JOB_ID=$(llp parsing create \
  --file-id "$FILE_ID" \
  --tier agentic \
  --version latest \
  --output-options.granular-bboxes '[word, line, cell]' | jq -r '.id')

# Poll until the job reaches a terminal status
while true; do
  STATUS=$(llp parsing get --job-id "$JOB_ID" | jq -er '.job.status') || break
  case "$STATUS" in COMPLETED|FAILED|CANCELLED) break ;; esac
  sleep 2
done

# The grounded-items sidecar download URL is auto-included on the result
llp parsing get --job-id "$JOB_ID" \
  | jq -r 'if .job.status == "COMPLETED" then .result_content_metadata.grounded_items.presigned_url
    else "Parse job failed: \(.job.error_message // .)" end'

在 `fast` 档上不可用

fast 档没有条目级版面,细粒度落地也就无所依附。请用 cost_effective、agentic 或 agentic_plus。

3. 找到边车的 URL

余下的步骤 —— 从结果里读出边车 URL、下载 JSONL、遍历逐词/逐行/逐单元格的落地数据 —— 用 Python 演示。result_content_metadata.grounded_items 字段与 JSONL 边车的形状在所有 SDK 上完全一致,只有字段访问与 JSON 解析的语法不同。

设了 granular_bboxes 时,结果会自动在 result_content_metadata 下带一条 grounded_items 记录。每条记录含 size_bytes、一个 exists 标记与一个 presigned_url。

sidecar = (result.result_content_metadata or {}).get("grounded_items")
if sidecar is None:
    raise RuntimeError("Sidecar missing — was `granular_bboxes` set on the parse request?")

print(f"Sidecar: {sidecar.size_bytes} bytes")
print(f"URL:     {sidecar.presigned_url}")

预签名 URL 是临时的。 请尽快下载,或再调一次 client.parsing.get(job_id=...) 以签发新地址。

4. 下载并解析 JSONL

这份边车是 JSONL —— 每行一个 JSON 对象、每页一行 —— 而不是一个大的 JSON 数组。请逐行流式读取。

import json
import httpx

response = httpx.get(sidecar.presigned_url)
response.raise_for_status()

# Each non-empty line is one page row.
pages = [json.loads(line) for line in response.text.splitlines() if line.strip()]
print(f"Pages in sidecar: {len(pages)}")

每一页的行只有两种形状:

# Success
{
    "page_number": 1,
    "page_width": 612,
    "page_height": 792,
    "success": True,
    "items": [...],
}

# Failure — grounding could not be produced for this page
{
    "page_number": 2,
    "success": False,
    "error": "...",
}

下钻之前永远先看 success:

for page in pages:
    if not page["success"]:
        print(f"Page {page['page_number']} failed: {page['error']}")
        continue
    print(f"Page {page['page_number']}: {len(page['items'])} items")

5. 遍历词级落地数据

每个 item 的形状与常规 items 响应里的 type / md / bbox 完全相同,此外多一个可选的 grounding 块。对文本型条目(段落、标题、题注),grounding 是一个 GroundedTextSupport:

{
    "source": "md",           # or "caption" — which surface the spans index into
    "lines": [
        {
            "span": [0, 11],  # [start, end) UTF-8 byte range into item[source]
            "bbox": { "x": 72.0, "y": 100.0, "w": 200.0, "h": 12.0 },
            "words": [
                {
                    "span": [0, 5],
                    "bbox": { "x": 72.0, "y": 100.0, "w": 35.0, "h": 12.0 },
                },
                # ...
            ],
        },
        # ...
    ],
}

要把第一页上的每个词高亮出来:

page = next(p for p in pages if p["success"] and p["page_number"] == 1)

for item in page["items"]:
    grounding = item.get("grounding")
    if not grounding or grounding.get("source") not in ("md", "caption"):
        continue

    source_bytes = item[grounding["source"]].encode("utf-8")
    for line in grounding["lines"]:
        for word in line.get("words", []) or []:
            start, end = word["span"]
            text = source_bytes[start:end].decode("utf-8")
            box = word["bbox"]
            print(f"  word {text!r} at ({box['x']:.0f}, {box['y']:.0f}) "
                  f"{box['w']:.0f}×{box['h']:.0f}")

这个 span 是一个 [start, end) 的 UTF-8 字节区间,指向 grounding.source 所指名的那个字段(md 或 caption)。切片前先把那个字段编码,再解码该词的字节 —— Python 的字符串下标数的是字符,不是字节。

6. 遍历表格单元格落地数据

对表格条目,grounding 则是一个 GroundedTableSupport —— 它带着逐单元格的框与区间,外加行级与列级的框:

{
    "rows": [
        # rows[row][col] is a cell or null
        [
            {
                "span": [42, 56],            # optional, into the table cell text
                "lines": [...],              # optional per-line grounding inside the cell
                "bbox": [                    # one or more boxes covering the cell
                    { "x": 100.0, "y": 200.0, "w": 50.0, "h": 16.0 },
                ],
            },
            None,                            # missing/empty cell
            # ...
        ],
        # ...
    ],
    "row_bboxes":    [[{...}], ...],         # boxes per row (a row may span multiple)
    "column_bboxes": [[{...}], ...],         # boxes per column
}

要取第一页上第 0 行、第 1 列那个单元格的 bbox:

for item in page["items"]:
    if item["type"] != "table":
        continue
    grounding = item.get("grounding")
    if not grounding or not grounding.get("rows"):
        continue

    cell = grounding["rows"][0][1]
    if cell is None:
        print("cell (0, 1) is empty")
        continue

    for box in cell.get("bbox") or []:
        print(f"cell (0, 1) box: ({box['x']:.0f}, {box['y']:.0f}) "
              f"{box['w']:.0f}×{box['h']:.0f}")

7. 把框画到页面截图上

边车里的 page_width 与 page_height 描述的是方向校正之前的原始页面视口。PDF 的 /Rotate 已经含在这个视口里。每个 bbox 的 x/y/w/h 描述的是一个字面的、未旋转的矩形;r 让它绕自身中心顺时针旋转。绝不要把 original_orientation_angle 加到 r 上。

请求 images_to_save: ["screenshot"],然后用 expand=["images_content_metadata", "metadata"] 取结果。对普通的 PDF 截图,现有的 metadata.pages[].original_orientation_angle 就描述了这页做的校正。把元数据按边车的 page_number 去匹配,而不是按数组下标、也不是按所选页范围里显示的位置。角度缺失或为 null 意味着「未知」,不是零。

下面这个辅助函数会按该页的角度把正立的 PDF 截图顺时针转回去,从而在绘制前还原出 bbox 所在的原始视口。若想反过来让截图保持正立,就把算好的 bbox 多边形按页角度的逆旋转做一次变换 —— Playground 用的是这个方向。两种做法都不会把页面方向角加进 bbox 的 r。

对齐截图之后,先在页面单位下旋转每个 bbox 的角点,再把这些角点缩放到图像像素。先缩放后旋转会在横纵缩放系数不同时让框变形。这份完整实现用了 Pillow,用 pip install Pillow 安装:

这个截图配方只讲 PDF 页,不涉及相机照片的透视校正。

import math
from PIL import ImageDraw

def bbox_polygon(box, scale_x, scale_y):
    angle = math.radians(box.get("r") or 0)
    cx, cy = box["x"] + box["w"] / 2, box["y"] + box["h"] / 2
    points = []
    for sx, sy in [(-1, -1), (1, -1), (1, 1), (-1, 1)]:
        dx, dy = sx * box["w"] / 2, sy * box["h"] / 2
        x = cx + dx * math.cos(angle) - dy * math.sin(angle)
        y = cy + dx * math.sin(angle) + dy * math.cos(angle)
        points.append((x * scale_x, y * scale_y))
    return points

def page_orientation(metadata, page_number):
    matches = [page for page in metadata["pages"] if page["page_number"] == page_number]
    if len(matches) != 1:
        raise ValueError("Exactly one metadata record is required for the source page")
    return matches[0].get("original_orientation_angle")

def render_grounding(screenshot, page_orientation_angle, page_width, page_height, boxes):
    if type(page_orientation_angle) is not int or page_orientation_angle % 90 != 0:
        raise ValueError("A known cardinal page orientation is required")
    if not all(math.isfinite(size) and size > 0 for size in (page_width, page_height)):
        raise ValueError("Page dimensions must be positive and finite")
    # Pillow uses counterclockwise angles; expand retains the entire image.
    aligned = screenshot.rotate(-page_orientation_angle, expand=True).convert("RGB")
    scale_x, scale_y = aligned.width / page_width, aligned.height / page_height
    draw = ImageDraw.Draw(aligned)
    for box in boxes:
        draw.polygon(bbox_polygon(box, scale_x, scale_y), outline="red", width=2)
    return aligned

SDK 返回的 result.metadata 是一个 model,不是字典;result.metadata.to_dict() 的形状与原始 JSON 相同。用 page_orientation(result.metadata.to_dict(), grounded_page["page_number"]) 读该页的角度。把下载好的图像、这个角度、边车给出的页面尺寸,以及词/行/单元格的 bbox 一起传给 render_grounding,再用 .save("grounding.png") 保存返回的图像。当页面方向未知时,先在绘制叠加层之前确定截图的坐标系。

另请参阅

BytesRouter 是社区翻译站点,所有译文仅供参考,一切以官方英文原文为准。