主题
本示例演示如何在 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" httpxfrom llama_cloud import LlamaCloud
client = LlamaCloud() # reads LLAMA_CLOUD_API_KEY from the environment安装 SDK:
终端窗口
npm install @llamaindex/llama-cloudimport 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-goimport (
"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@latestllp 会从环境里读取 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 alignedSDK 返回的 result.metadata 是一个 model,不是字典;result.metadata.to_dict() 的形状与原始 JSON 相同。用 page_orientation(result.metadata.to_dict(), grounded_page["page_number"]) 读该页的角度。把下载好的图像、这个角度、边车给出的页面尺寸,以及词/行/单元格的 bbox 一起传给 render_grounding,再用 .save("grounding.png") 保存返回的图像。当页面方向未知时,先在绘制叠加层之前确定截图的坐标系。
另请参阅
- 配置 Parse → 细粒度边界框 —— 请求侧配置
- 响应形态 → grounded items 边车 —— 响应侧 schema 参考
- 解析 PDF 并理解输出 —— 常规的
items树(只有条目级框)