主题
本教程走一遍最常见的 Parse 场景之一：把一份财务报告里的所有表格都抽出来、载入 pandas。我们用的是财政服务局 2024 年执行摘要，与快速上手教程 用的是同一份文档，因此你可以把两个例子端到端对照着看。
本教程里的模式可以推广到任何散文与表格混杂的财务文档 —— 10-K、业绩报告、投资人演示、审计报告。
什么时候用这套模式
- 你的文档在多页上有多张表格，你要的是全部而非某一张
- 你需要表格数据作为结构化的行，而不只是 markdown 文本
- 你想把表格直接载入 pandas 做下游分析或 RAG 索引
如果你只需要某个已知页上的一张表，见解析 PDF 中的图表并用 pandas 分析 —— 那里演示的是更简单的"某一张指定表"版本。若你只需要文本或 markdown，快速上手：解析 PDF 并理解输出 就够了。
1. 准备
把 API key 设成环境变量，SDK 会自动读取：
终端窗口
export LLAMA_CLOUD_API_KEY="llx-..."安装 SDK 并构造一个 client：
- Python
- TypeScript
- Go
- Java
- CLI
终端窗口
pip install "llama-cloud>=2.8"from llama_cloud import LlamaCloud
client = LlamaCloud() # reads LLAMA_CLOUD_API_KEY from the environment终端窗口
npm install @llamaindex/llama-cloudimport LlamaCloud from '@llamaindex/llama-cloud';
import fs from 'fs';
const client = new LlamaCloud(); // reads LLAMA_CLOUD_API_KEY from the environment终端窗口
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 environmentimplementation("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();终端窗口
go install github.com/run-llama/llama-parse-cli/cmd/llp@latestllp 会从环境里读取 LLAMA_CLOUD_API_KEY（也可用 --api-key 传入）。
2. 解析文档
上传 PDF 并跑一个解析任务。我们选 agentic 以获得扎实的表格准确度，并请求 items 视图，这样就能拿到一棵可以遍历找表的结构化树。
对于篇幅长、复杂度混合的报告，请开启成本优化器。这份财政部文档有很多以文字为主的叙述页，只有少数几页是表格密集的汇总页 —— 成本优化器会自动把简单页路由到 cost_effective，只在表格页上花 premium credit。我们同时请求 metadata，以便看到哪些页被成本优化了。
下载这份报告：
终端窗口
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
file = client.files.create(
file="executive-summary-2024.pdf", # use /content/executive-summary-2024.pdf in Colab
purpose="parse",
)
result = client.parsing.parse(
file_id=file.id,
tier="agentic",
version="latest",
processing_options={
"cost_optimizer": {"enable": True},
},
expand=["markdown", "items", "metadata"],
)
print(f"Job status: {result.job.status}")
print(f"Total pages: {len(result.items.pages)}")const file = await client.files.create({
file: fs.createReadStream('executive-summary-2024.pdf'),
purpose: 'parse',
});
const result = await client.parsing.parse({
file_id: file.id,
tier: 'agentic',
version: 'latest',
processing_options: {
cost_optimizer: { enable: true },
},
expand: ['markdown', 'items', 'metadata'],
});
console.log(`Job status: ${result.job.status}`);
console.log(`Total pages: ${result.items.pages.length}`);在 Go 里 expand 是 GET 参数，所以要先创建任务、轮询到终态，再带着你要的字段取结果：
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)
}
// Enable Cost Optimizer on the create call
job, err := client.Parsing.New(ctx, llamacloud.ParsingNewParams{
FileID: llamacloud.String(file.ID),
Tier: llamacloud.ParsingNewParamsTierAgentic,
Version: llamacloud.ParsingNewParamsVersionLatest,
ProcessingOptions: llamacloud.ParsingNewParamsProcessingOptions{
CostOptimizer: llamacloud.ParsingNewParamsProcessingOptionsCostOptimizer{
Enable: llamacloud.Bool(true),
},
},
})
if err != nil {
log.Fatal(err)
}
// Request markdown, items, and metadata when you fetch the result
getParams := llamacloud.ParsingGetParams{Expand: []string{"markdown", "items", "metadata"}}
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)
}
fmt.Printf("Job status: %s\n", result.Job.Status)
fmt.Printf("Total pages: %d\n", len(result.Items.Pages))在 Java 里 expand 是查询参数，所以要先创建任务、轮询到终态，再带着你要的字段取结果：
FileCreateResponse file = client.files().create(FileCreateParams.builder()
.file(Paths.get("executive-summary-2024.pdf"))
.purpose("parse")
.build());
// Enable Cost Optimizer on the create call
ParsingCreateResponse job = client.parsing().create(ParsingCreateParams.builder()
.fileId(file.id())
.tier(ParsingCreateParams.Tier.AGENTIC)
.version(ParsingCreateParams.Version.LATEST)
.processingOptions(ParsingCreateParams.ProcessingOptions.builder()
.costOptimizer(ParsingCreateParams.ProcessingOptions.CostOptimizer.builder()
.enable(true)
.build())
.build())
.build());
// Request markdown, items, and metadata when you fetch the result
ParsingGetParams getParams = ParsingGetParams.builder()
.jobId(job.id())
.addExpand("markdown")
.addExpand("items")
.addExpand("metadata")
.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());
}
System.out.println("Job status: " + result.job().status());
System.out.println("Total pages: " + result.items().get().pages().size());在 CLI 里 expand 也是查询参数，所以要先上传、创建任务、轮询到终态，再带着你要的字段取结果：
终端窗口
# Upload the report
FILE_ID=$(llp files create \
--file executive-summary-2024.pdf \
--purpose parse | jq -r '.id')
# Start a parse job with Cost Optimizer enabled
JOB_ID=$(llp parsing create \
--file-id "$FILE_ID" \
--tier agentic \
--version latest \
--processing-options.cost-optimizer '{enable: true}' | 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
echo "Job status: $STATUS"
# Total pages
llp parsing get --job-id "$JOB_ID" --expand items | jq -r 'if .job.status == "COMPLETED" then .items.pages | length
else "Parse job failed: \(.job.error_message // .)" end'各选项在请求结构里的位置：tier 与 version 是顶层必填字段，processing_options.cost_optimizer.enable 开启逐页档位路由，而 expand 是一个顶层数组，控制返回什么。完整请求形状见配置模型。
3. 看哪些页被成本优化了
成本优化器会在逐页元数据里加上 cost_optimized 标记。检查它来确认路由是否落在你预期的位置 —— 以文字为主的叙述页应当是 True（按 cost_effective 处理），表格密集的页应当是 False（按 agentic 处理）。
- Python
- TypeScript
- Go
- Java
- CLI
for page in result.metadata.pages:
flag = "cost-optimized" if page.cost_optimized else "premium tier"
print(f"page {page.page_number}: {flag}")for (const page of result.metadata.pages) {
const flag = page.cost_optimized ? "cost-optimized" : "premium tier";
console.log(`page ${page.page_number}: ${flag}`);
}for _, page := range result.Metadata.Pages {
flag := "premium tier"
if page.CostOptimized {
flag = "cost-optimized"
}
fmt.Printf("page %d: %s\n", page.PageNumber, flag)
}for (ParsingGetResponse.Metadata.Page page : result.metadata().get().pages()) {
String flag = page.costOptimized().orElse(false) ? "cost-optimized" : "premium tier";
System.out.printf("page %d: %s%n", page.pageNumber(), flag);
}CLI 用 --expand metadata 重新请求同一个任务，并用 jq 读出逐页标记：
终端窗口
llp parsing get --job-id "$JOB_ID" --expand metadata | jq -r '
.metadata.pages[]
| "page \(.page_number): \(if .cost_optimized then "cost-optimized" else "premium tier" end)"'如果你看到一页明明有清晰表格却是 cost_optimized: True，那就不对了 —— 分类器本该把表格页路由到 premium 档。请带上这份文档提一个支持工单，好让团队调整这个启发式。关于路由如何工作的更多说明，见成本优化器。
4. 在 items 树里找出所有表格
items 视图返回逐页的结构化树，元素都带类型：标题、段落、表格、图形。表格是 item.type == "table"，带一个 rows 字段（列表的列表），以及同一份数据的 csv、html 与 md 三种表示。
遍历每一页、收集每一张表，并给每张表标上它的来源页，方便日后回溯：
- Python
- TypeScript
- Go
- Java
- CLI
all_tables = []
for page in result.items.pages:
for item in page.items:
if getattr(item, "type", None) == "table":
all_tables.append({
"page_number": page.page_number,
"rows": item.rows,
"csv": item.csv,
})
print(f"Found {len(all_tables)} tables across {len(result.items.pages)} pages.")
for t in all_tables:
n_rows = len(t["rows"])
n_cols = len(t["rows"][0]) if t["rows"] else 0
print(f" page {t['page_number']}: {n_rows} rows × {n_cols} cols")const allTables = [];
for (const page of result.items.pages) {
for (const item of page.items) {
if ("type" in item && item.type === "table" && "rows" in item) {
allTables.push({
page_number: page.page_number,
rows: item.rows,
csv: item.csv,
});
}
}
}
console.log(`Found ${allTables.length} tables across ${result.items.pages.length} pages.`);
for (const t of allTables) {
const nRows = t.rows.length;
const nCols = t.rows[0]?.length ?? 0;
console.log(` page ${t.page_number}: ${nRows} rows × ${nCols} cols`);
}type tableRecord struct {
PageNumber int64
Rows [][]*llamacloud.TableItemRowUnion
CSV string
}
var allTables []tableRecord
for _, page := range result.Items.Pages {
for _, item := range page.Items {
if item.Type == "table" {
allTables = append(allTables, tableRecord{
PageNumber: page.PageNumber,
Rows: item.Rows,
CSV: item.Csv,
})
}
}
}
fmt.Printf("Found %d tables across %d pages.\n", len(allTables), len(result.Items.Pages))
for _, t := range allTables {
nRows := len(t.Rows)
nCols := 0
if nRows > 0 {
nCols = len(t.Rows[0])
}
fmt.Printf(" page %d: %d rows x %d cols\n", t.PageNumber, nRows, nCols)
}int tableCount = 0;
java.util.List<ParsingGetResponse.Items.Page> pages = result.items().get().pages();
for (ParsingGetResponse.Items.Page page : pages) {
if (!page.isStructuredResult()) {
continue; // skip failed pages
}
ParsingGetResponse.Items.Page.StructuredResultPage structured = page.asStructuredResult();
for (ParsingGetResponse.Items.Page.StructuredResultPage.Item item : structured.items()) {
if (item.isTable()) {
tableCount++;
int nRows = item.asTable().rows().size();
int nCols = nRows > 0 ? item.asTable().rows().get(0).size() : 0;
System.out.printf(" page %d: %d rows x %d cols%n",
structured.pageNumber(), nRows, nCols);
}
}
}
System.out.printf("Found %d tables across %d pages.%n", tableCount, pages.size());终端窗口
llp parsing get --job-id "$JOB_ID" --expand items | jq -r '
.items.pages[] as $page
| $page.items[]
| select(.type == "table")
| "page \($page.page_number): \(.rows | length) rows x \(.rows[0] | length) cols"'就这份财政部 PDF 而言，第一页含那张核心的 "Financial Position & Condition" 汇总表 —— 也就是快速上手教程 里详细展示的那一张。其后的页面则是叙述文字与更小的汇总表混排。
5. 把每张表载入 pandas
到此为止的部分在任何 SDK 上都能用。从这里开始的分析是 Python + pandas，作用在第 4 步里攒出的 all_tables 列表上。还没装 pandas 的话先装上：
终端窗口
pip install pandasitem.csv 是进入 pandas 最省事的通路。Parse 会为每张表生成一份干净的 CSV 字符串，pandas.read_csv 配合一个 StringIO 包装就能直接吃进去：
import io
import pandas as pd
dataframes = []
for t in all_tables:
df = pd.read_csv(io.StringIO(t["csv"]))
df["_source_page"] = t["page_number"] # keep the source page for traceability
dataframes.append(df)
# Quick look at the first table
if dataframes:
print(dataframes[0].head())
print(f"\nColumns: {list(dataframes[0].columns)}")每个 DataFrame 都保留一个 _source_page 列，于是你可以按来源分组、带着出处写回数据仓库，或者搭一个能引用出处的 RAG 索引 —— 每一行都知道自己来自哪一页。
6. 筛出你真正关心的表
多数财务报告里只有那么几张你真正在意的“汇总”表，后面跟着一长串可以略过的附表。按内容或形状把这个列表筛一遍：
# Tables with at least 5 rows (skip tiny one-line summaries)
substantial_tables = [df for df in dataframes if len(df) >= 5]
# Or filter by column header contents
financial_tables = [
df for df in dataframes
if any("$" in str(c) or "Dollar" in str(c) for c in df.columns)
]
print(f"Substantial tables: {len(substantial_tables)}")
print(f"Financial tables (with $ in column headers): {len(financial_tables)}")如果你已经知道报告里哪几节重要，也可以按来源页筛选：
# Only tables from the executive summary section (e.g. pages 1-3)
exec_summary_tables = [
pd.read_csv(io.StringIO(t["csv"]))
for t in all_tables
if t["page_number"] in (1, 2, 3)
]7. 把表格存下来供下游使用
有了 DataFrame，剩下的就是常规 pandas 操作。几种常见写法：
# Save each table as a separate CSV with a deterministic filename
for i, t in enumerate(all_tables):
filename = f"table_p{t['page_number']:03d}_{i}.csv"
pd.read_csv(io.StringIO(t["csv"])).to_csv(filename, index=False)
# Or write them all into a single Excel workbook, one sheet per table
with pd.ExcelWriter("financial_tables.xlsx") as writer:
for i, t in enumerate(all_tables):
df = pd.read_csv(io.StringIO(t["csv"]))
sheet_name = f"page{t['page_number']}_{i}"[:31] # Excel limit
df.to_excel(writer, sheet_name=sheet_name, index=False)如果你更想让 Parse 一次性把表格导出成真正的 .xlsx 文件（而不是在 Python 里循环），就开启表格转电子表格输出，并用 expand=["xlsx_content_metadata"] 取回 —— 它会给出那个 Excel 文件的预签名地址。
拿到表格之后做什么
几个建立在你现在成果之上、性价比很高的模式：
- 能引用出处的 RAG。 靠
_source_page列，每一行都知道自己的来源页。把这些行嵌入并建索引时，把页号存进元数据，检索器就能在回答里引用具体的页。 - 模式校验。 如果你需要形状有保证的 JSON 而不是表格，就用 LlamaExtract 在同一份文档上配 Pydantic schema —— Extract 正是为此而生的。
- 跨文档比较。 把同一套模式跑在多个年度的 10-K 上，再把这些 DataFrame 拼起来、加一个
_year列，就能跟踪各项指标随时间的变化。
另请参阅
- 解析 PDF 中的图表并用 pandas 分析 —— 本教程聚焦图表的版本
- 快速上手：解析 PDF 并理解输出 —— 每种
expand视图分别返回什么，讲得细 - 成本优化器 —— 逐页的档位路由
- 档位 —— 为你的文档挑对档位
- 取回结果 —— 所有合法的
expand取值 - LlamaExtract —— 需要 JSON 形状有保证时，用按 schema 驱动的抽取