主题
本页是一张 Extract 任务的地图:任务由哪些概念构成、如何驱动它的 schema 怎么写、除 schema 之外还能设哪些选项,以及在规模化过程中保持抽取准确的做法。
逐字段的请求参考见 Extract API 参考。任务可额外返回的那些元数据见元数据扩展。
核心概念
LlamaExtract 被设计成一个灵活、可扩展的抽取平台。平台核心的概念有以下几个:
- 抽取配置:可复用的设置,包含 schema、档位与抽取选项。
- 数据 schema:用 JSON/Pydantic 格式对你要抽取的数据做结构化定义。详见下文。
- 抽取任务:用某份配置从文档里抽取结构化数据的异步任务。
- 抽取结果:一次抽取任务的结果,含抽出的数据与其他元数据。
数据 schema
数据 schema 定义你想从文档里抽取的数据结构。它是一份 JSON Schema,规定字段的名称、类型与描述,也就是你需要的那些信息的形状。
schema 本质上仍是 JSON Schema(支持完整 JSON Schema 规范的一个子集),不过我们的 Python SDK 允许你使用 Pydantic 模型,从而有更 Pythonic 的体验、带类型校验与 IDE 支持。
字段、类型、体积上限,以及 Extract 接受的那个子集 JSON Schema,都在下文的schema 设计与限制里说明。
schema 设计与限制
schema 是抽取配置中最重要的一部分。它定义了你想拿回的数据结构,而它的字段描述就是引导抽取模型的东西。
怎么定义你的 schema
schema 由字段构成。每个字段有一个名称、一个类型,并可选地有一段描述。
- 字段名 —— 使用清晰、稳定的名字,与你对数据的用法相符(例如
invoice_number、vendor_name)。它们会成为抽出 JSON 里的键。 - 字段描述 —— 描述是给底层大模型的额外上下文。不只是给人看的文档:抽取模型靠它来决定抽什么。请用描述告诉模型该字段的取值可能是什么 —— 例如这个字段的含义、通常出现在文档的什么位置、可接受的格式或示例。描述写得更好,抽取通常就更准确、更一致。
必填与可选字段
在 JSON Schema 里,一个对象的 required 数组列出必须存在的属性。列在其中即必填,未列出的即可选。用 "required": [] 可把所有属性设为可选。能否为 null 是另一回事:当某字段可能返回 null 时,请加上一个 null 分支。
这份 schema 把 invoice_number 设为必填,把 start_date 设为可选且可为 null:
{
"type": "object",
"properties": {
"invoice_number": { "type": "string" },
"start_date": {
"anyOf": [{ "type": "string" }, { "type": "null" }]
}
},
"required": ["invoice_number"]
}你也可以把这个可空字段写成 "start_date": { "type": ["string", "null"] },Extract 两种写法都接受。可空的数组与对象仍保留它们的 items 与 properties。
为保证正确性,我们建议显式写出 required。若省略 required,Extract 会为图方便把每个属性都当作必填,于是你不必把所有字段列一遍。这与标准 JSON Schema 的默认相反 —— 在标准里,省略 required 数组意味着所有属性都是可选的。
schema 的限制
LlamaExtract 只支持 JSON Schema 规范的一个子集。 虽然受限,但足以覆盖大量使用场景。
在 Extract 里,省略
required数组会让每个属性都变为必填。数据可能缺失时见必填与可选字段。在 Pydantic 里,用
Optional[T]或T | None让字段可空;若该字段同时应为可选,请把默认值设为None。根节点必须是
object类型。schema 的嵌套层数必须限制在 7 层以内。
重要的字段是键名/标题、类型与描述。用于格式化、默认值等的字段不支持。若你需要这些,可以把相应约束写进字段描述,和/或加一个后处理步骤。例如默认值可以这样实现:把字段设为可选,再把抽取结果里的 "null" 值替换成你的默认值。
其他 schema 限制:
- 属性上限:整个 schema 合计 5,000 个属性。
- 字符串内容总量上限:所有字符串(字段名、描述、枚举取值等)合计 120,000 字符。
- 原始 schema 大小上限:原始 JSON schema 字符串 150,000 字符。
若在复杂的抽取场景下触及这些上限,考虑重构抽取流程以落在这套约束内,例如分批抽取字段子集,之后再合并起来。
schema 的规模
Cost-effective、Agentic 与 Agentic Plus 支持最多 3,200 个字段的 schema。要用大 schema,Cost-effective 与 Agentic 需要 2.1 或更高版本。
规模指的是你提交的那份 schema 里的叶子字段数,也就是树最底层的标量值。它与文档无关,也与取回多少数据无关。
- 对象本身不计数,只数它里面的叶子。
{"address": {"city": ..., "zip": ...}}是 2 个字段。 - 数组只按它的元素 schema 计一次数,无论实际抽出多少项。含 40 行、每行 5 个字段的一个数组算 5 个字段,不是 200 个。
- 标量数组,例如
{"tags": ["string"]},算 1 个字段。 $ref会在每一处使用点展开。若三个字段都指向同一个含 10 个字段的Address定义,那是 30 个字段而不是 10 个。复用定义能让 schema 更好读,但不会让它更小。
一份含 30 个顶层标量、外加一张 10 列表格的 schema,就是 40 个字段。
技巧与最佳实践
- 尽量把 schema 嵌套控制在 3–4 层。
- 数据不一定存在时把字段设为可选(尤其是
boolean与int字段,缺失值的默认容易造成混淆)。 - 要抽取数量不定的实体时,用
array类型。但注意根节点不能用array类型。 - 用有描述性的字段名与详尽的字段描述。把格式说明或 few-shot 示例放进描述里。
- 最重要的是:从简单开始,边迭代边把需求加进 schema。
自动生成 schema
不必手工定义 schema,你可以用 LlamaExtract 的自动 schema 生成功能。系统能基于以下输入生成:
- 一段自然语言提示词:描述你要抽取什么数据
- 一个示例文件:上传文档,让系统从它的结构推断 schema
- 一份待改进的现有 schema:给出基础 schema,让系统优化或扩展它
这些输入可以组合使用 —— 例如同时给一个示例文件和一段提示词来引导生成。
用 REST API
终端窗口
curl -X 'POST' \
'https://api.cloud.llamaindex.ai/api/v2/extract/schema/generate?project_id={PROJECT_ID}' \
-H 'accept: application/json' \
-H "Authorization: Bearer $LLAMA_CLOUD_API_KEY" \
-H 'Content-Type: application/json' \
-d '{
"prompt": "Extract invoice details including invoice number, date, vendor name, line items with descriptions and amounts, and total amount",
"file_id": "optional-file-id-for-sample-document"
}'注意
自动生成的 schema 可能包含比你需要的更多字段。投产前请审阅生成的 schema 并删掉不必要的字段 —— 更少、定义清晰的字段通常会带来更好的抽取质量。
完整 API 文档见 LlamaExtract API 参考。
用 SDK 定义 schema
本节的示例是在一份简历上跑 Extract。先下载一份样例:
终端窗口
curl -fL -o resume.pdf https://raw.githubusercontent.com/run-llama/llama_cloud_services/main/examples/extract/data/resumes/software_architect.pdf- Python
- TypeScript
- Go
- Java
- CLI
Python SDK 可以这样安装:
终端窗口
pip install "llama-cloud>=2.8"schema 可以用 Pydantic 模型或 JSON Schema 来定义:
用 Pydantic(推荐)
from pydantic import BaseModel, Field
from typing import List, Optional
class Experience(BaseModel):
company: str = Field(description="Company name")
title: str = Field(description="Job title")
start_date: Optional[str] = Field(description="Start date of employment")
end_date: Optional[str] = Field(description="End date of employment")
class Resume(BaseModel):
name: str = Field(description="Candidate name")
experience: List[Experience] = Field(description="Work history")
schema = Resume.model_json_schema()用 JSON Schema
schema = {
"type": "object",
"properties": {
"name": {"type": "string", "description": "Candidate name"},
"experience": {
"type": "array",
"description": "Work history",
"items": {
"type": "object",
"properties": {
"company": {
"type": "string",
"description": "Company name",
},
"title": {"type": "string", "description": "Job title"},
"start_date": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "Start date of employment",
},
"end_date": {
"anyOf": [{"type": "string"}, {"type": "null"}],
"description": "End date of employment",
},
},
},
},
},
}有了 schema,就能直接用 SDK 跑抽取:
from llama_cloud import LlamaCloud
client = LlamaCloud() # reads LLAMA_CLOUD_API_KEY from the environment
file_obj = client.files.create(file="resume.pdf", purpose="extract")
job = client.extract.create(
file_input=file_obj.id,
configuration={
"data_schema": schema,
"tier": "agentic",
},
)
# Poll for completion
while job.status not in ("COMPLETED", "FAILED", "CANCELLED"):
import time; time.sleep(2)
job = client.extract.get(job.id)TypeScript SDK 可以这样安装:
终端窗口
npm install @llamaindex/llama-cloud zodschema 可以用 Zod 模型或 JSON Schema 来定义:
用 Zod(推荐)
import { z } from "zod";
const ExperienceSchema = z.object({
company: z.string().describe("Company name"),
title: z.string().describe("Job title"),
start_date: z.string().nullable().describe("Start date of employment"),
end_date: z.string().nullable().describe("End date of employment"),
});
const ResumeSchema = z.object({
name: z.string().describe("Candidate name"),
experience: z.array(ExperienceSchema).describe("Work history"),
});
const schema = z.toJSONSchema(ResumeSchema);用 JSON Schema
const schema = {
type: "object",
properties: {
name: { type: "string", description: "Candidate name" },
experience: {
type: "array",
description: "Work history",
items: {
type: "object",
properties: {
company: {
type: "string",
description: "Company name",
},
title: { type: "string", description: "Job title" },
start_date: {
anyOf: [{ type: "string" }, { type: "null" }],
description: "Start date of employment",
},
end_date: {
anyOf: [{ type: "string" }, { type: "null" }],
description: "End date of employment",
},
},
},
},
},
};有了 schema,就能直接用 SDK 跑抽取:
import fs from 'fs';
import LlamaCloud from '@llamaindex/llama-cloud';
const client = new LlamaCloud(); // reads LLAMA_CLOUD_API_KEY from the environment
const fileObj = await client.files.create({
file: fs.createReadStream('resume.pdf'),
purpose: 'extract',
});
let job = await client.extract.create({
file_input: fileObj.id,
configuration: {
data_schema: schema,
tier: 'agentic',
},
});
// Poll for completion
while (!['COMPLETED', 'FAILED', 'CANCELLED'].includes(job.status)) {
await new Promise((r) => setTimeout(r, 2000));
job = await client.extract.get(job.id);
}Go SDK 可以这样安装:
终端窗口
go get github.com/run-llama/llama-parse-goGo SDK 没有 Pydantic/Zod 的对应物 —— 把 schema 定义成一份原始 JSON Schema 字面量,再用它跑抽取:
package main
import (
"context"
"fmt"
"log"
"os"
"time"
llamacloud "github.com/run-llama/llama-parse-go"
)
func main() {
ctx := context.Background()
client := llamacloud.NewClient()
// Define the schema as a JSON Schema literal
dataSchema := map[string]*llamacloud.ExtractConfigurationDataSchemaUnionParam{
"type": {OfString: llamacloud.String("object")},
"properties": {OfAnyMap: map[string]any{
"name": map[string]any{"type": "string", "description": "Candidate name"},
"experience": map[string]any{
"type": "array",
"description": "Work history",
"items": map[string]any{
"type": "object",
"properties": map[string]any{
"company": map[string]any{"type": "string", "description": "Company name"},
"title": map[string]any{"type": "string", "description": "Job title"},
"start_date": map[string]any{
"anyOf": []any{map[string]any{"type": "string"}, map[string]any{"type": "null"}},
"description": "Start date of employment",
},
"end_date": map[string]any{
"anyOf": []any{map[string]any{"type": "string"}, map[string]any{"type": "null"}},
"description": "End date of employment",
},
},
},
},
}},
}
// Upload a file to extract from
f, err := os.Open("resume.pdf")
if err != nil {
log.Fatal(err)
}
defer f.Close()
fileObj, err := client.Files.New(ctx, llamacloud.FileNewParams{
File: f,
Purpose: "extract",
})
if err != nil {
log.Fatal(err)
}
// Extract data from document
job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
FileInput: fileObj.ID,
Configuration: llamacloud.ExtractConfigurationParam{
DataSchema: dataSchema,
Tier: llamacloud.ExtractConfigurationTierAgentic,
},
},
})
if err != nil {
log.Fatal(err)
}
// Poll for completion
for job.Status != "COMPLETED" && job.Status != "FAILED" && job.Status != "CANCELLED" {
time.Sleep(2 * time.Second)
job, err = client.Extract.Get(ctx, job.ID, llamacloud.ExtractGetParams{})
if err != nil {
log.Fatal(err)
}
}
fmt.Println(job.ExtractResult.RawJSON())
}Java SDK 可以这样安装:
implementation("ai.llamaindex:llama-cloud:1.3.0")Java SDK 没有 Pydantic/Zod 的对应物 —— 把 schema 定义成一份原始 JSON Schema 字面量,再用它跑抽取:
import ai.llamaindex.llamacloud.client.LlamaCloudClient;
import ai.llamaindex.llamacloud.client.okhttp.LlamaCloudOkHttpClient;
import ai.llamaindex.llamacloud.core.JsonValue;
import ai.llamaindex.llamacloud.models.extract.ExtractConfiguration;
import ai.llamaindex.llamacloud.models.extract.ExtractCreateParams;
import ai.llamaindex.llamacloud.models.extract.ExtractGetParams;
import ai.llamaindex.llamacloud.models.extract.ExtractV2Job;
import ai.llamaindex.llamacloud.models.extract.ExtractV2JobCreate;
import ai.llamaindex.llamacloud.models.files.FileCreateParams;
import ai.llamaindex.llamacloud.models.files.FileCreateResponse;
import java.nio.file.Paths;
import java.util.List;
import java.util.Map;
LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();
// Define the schema as a JSON Schema literal
ExtractConfiguration.DataSchema dataSchema = ExtractConfiguration.DataSchema.builder()
.putAdditionalProperty("type", JsonValue.from("object"))
.putAdditionalProperty("properties", JsonValue.from(Map.of(
"name", Map.of("type", "string", "description", "Candidate name"),
"experience", Map.of(
"type", "array",
"description", "Work history",
"items", Map.of(
"type", "object",
"properties", Map.of(
"company", Map.of("type", "string", "description", "Company name"),
"title", Map.of("type", "string", "description", "Job title"),
"start_date", Map.of(
"anyOf", List.of(Map.of("type", "string"), Map.of("type", "null")),
"description", "Start date of employment"),
"end_date", Map.of(
"anyOf", List.of(Map.of("type", "string"), Map.of("type", "null")),
"description", "End date of employment")))))))
.build();
// Upload a file to extract from
FileCreateResponse fileObj = client.files().create(
FileCreateParams.builder()
.file(Paths.get("resume.pdf"))
.purpose("extract")
.build());
// Extract data from document
ExtractV2Job job = client.extract().create(
ExtractCreateParams.builder()
.extractV2JobCreate(
ExtractV2JobCreate.builder()
.fileInput(fileObj.id())
.configuration(
ExtractConfiguration.builder()
.dataSchema(dataSchema)
.tier(ExtractConfiguration.Tier.AGENTIC)
.build())
.build())
.build());
// Poll for completion
while (!job.status().equals("COMPLETED")
&& !job.status().equals("FAILED")
&& !job.status().equals("CANCELLED")) {
Thread.sleep(2000);
job = client.extract().get(ExtractGetParams.builder().jobId(job.id()).build());
}
System.out.println(job.extractResult());CLI 可以这样安装:
终端窗口
go install github.com/run-llama/llama-parse-cli/cmd/llp@latestCLI 没有 Pydantic/Zod 的对应物 —— 把 schema 定义成一份原始 JSON Schema 字面量,再用它跑抽取:
终端窗口
# Define the schema as a JSON Schema literal
DATA_SCHEMA='{
"type": "object",
"properties": {
"name": {"type": "string", "description": "Candidate name"},
"experience": {
"type": "array",
"description": "Work history",
"items": {
"type": "object",
"properties": {
"company": {"type": "string", "description": "Company name"},
"title": {"type": "string", "description": "Job title"},
"start_date": {"anyOf": [{"type": "string"}, {"type": "null"}], "description": "Start date of employment"},
"end_date": {"anyOf": [{"type": "string"}, {"type": "null"}], "description": "End date of employment"}
}
}
}
}
}'
# Upload a file to extract from
FILE_ID=$(llp files create --file resume.pdf --purpose extract | jq -r '.id')
# Extract data from document
JOB_ID=$(llp extract create \
--file-input "$FILE_ID" \
--configuration "{\"data_schema\": $DATA_SCHEMA, \"tier\": \"agentic\"}" \
| jq -r '.id')
# Poll for completion
while true; do
JOB=$(llp extract get --job-id "$JOB_ID") || break
STATUS=$(printf '%s\n' "$JOB" | jq -r '.status')
case "$STATUS" in COMPLETED|FAILED|CANCELLED) break ;; esac
sleep 2
done
printf '%s\n' "$JOB" | jq -r 'if .status == "COMPLETED" then .extract_result
else "Extract job failed: \(.error_message // .)" end'配置选项
schema 决定取回什么;下面这些选项决定任务怎么跑 —— 由哪个档来解读文档、读哪些页、返回哪些额外元数据。
档位
抽取档决定 LlamaExtract 为把文档变成结构化数据投入多少力气。
Agentic Plus 在各类文档上提供最高的抽取质量,短小或直白的文档也一样。想要最好的结果、又能接受更高的成本与延迟时就用它。在长或复杂的文档上它可能更慢,因为它对结果投入了更多迭代功夫。
Agentic 在广泛的文档范围内平衡质量、成本与延迟,包括混合版式与表格。
Cost Effective 优先照顾低成本低延迟,适合直白的抽取、尤其是高并发场景。字段与版式可预期、歧义有限时它表现最好。
Cost-effective、Agentic 与 Agentic Plus 支持最多 3,200 个字段的 schema。
Turbo 是速度档。它自己准备文档而不另跑一次 Parse 步骤,这既是它快的原因,也意味着它不产出解析输出:Turbo 任务里没有可供取回的解析文本或 Markdown,只有你的 schema 所要的结构化结果。
选 Turbo 的理由是延迟。在它瞄准的短文档上,其他档要几十秒,它通常几秒就返回。在我们的抽取基准上,它的准确度高于 Cost Effective;若你追求极致准确,请选 Agentic 系。Turbo 按一个打包的每页费率计费、不另收解析费,而这个费率高于 Agentic:它是为速度付溢价的选项,并不是更便宜的那一个 —— 见定价。
当延迟是用户可感知的,例如交互式应用、上传即审阅的流程,或者你要跑大量短小、可预期的文档时,就该考虑 Turbo。
Turbo 有实打实的限制:
| 能力 | Turbo |
|---|---|
| 输入格式 | 仅 PDF、JPG、PNG |
| 解析 | 不跑解析步骤、也没有解析输出;parse_tier 与 parse_config_id 即使传了也会被忽略 |
| 引用 | 支持,但仅到文本级;细粒度边界框会被拒绝 |
| 置信度得分 | 支持 |
每个抽取档都有一个默认解析档,用来平衡成本与质量。当你的负载需要更低的解析成本或更高质量的文档解读时,可以在配置里用 parse_tier 覆盖它(在 playground 里位于 Advanced settings → Processing)。Turbo 是例外:它不跑解析步骤,因此没有可覆盖的解析档。默认组合与叠加费用见定价。
版本
把 configuration.version 设为形如 2.5 的版本名,或设为 latest 以使用所选档位下最新的兼容发布。也支持日期(YYYY-MM-DD),表示使用不晚于该日期的最新版本。
电子表格模式
Spreadsheet mode(spreadsheet_mode)直接从 Excel 或 CSV 工作簿(.xlsx、.xls、.csv)里抽取,操作的是它的单元格而不是把文件压平成文本。它运行在 cost_effective、agentic 与 agentic_plus 三档上,不产出引用也不产出置信度得分。Cost-effective 与 Agentic 需要 2.1 或更高版本。电子表格模式按工作簿大小以所选档的抽取费率计费,不另收 Parse 费用。见抽取定价。
当工作簿里是结构规整、形状可预期的数据、而你想把它映射到一份 schema 上时就用它 —— 比如基金工作簿里的持仓、导出的账目明细里的行项目。用 spreadsheet_mode 开启,并可用 sheet_names 把抽取限定在指定的工作表(名称要完全匹配;留空表示所有表;在电子表格上 sheet_names 取代 target_pages):
from pydantic import BaseModel, Field
class Position(BaseModel):
ticker: str = Field(description="Ticker or symbol")
market_value: float = Field(description="Position market value")
class Portfolio(BaseModel):
positions: list[Position] = Field(description="One entry per holding in the workbook")
job = client.extract.create(
file_input=file_obj.id, # an uploaded .xlsx / .xls / .csv
configuration={
"data_schema": Portfolio.model_json_schema(),
"tier": "agentic",
"spreadsheet_mode": True,
"sheet_names": ["Holdings"], # optional; omit for every sheet
},
)当你事先不知道工作簿的形状时,与其写一个有针对性的 schema,不如把所有单元格抽进一个通用的“整齐单元格”schema —— 让每个单元格带上自己的行标题路径与列标题路径:
class Cell(BaseModel):
row_hierarchy: list[str | None] # row-label path to this cell
column_hierarchy: list[str | None] # column-header path above it
value: str | None
class Table(BaseModel):
title: str
cells: list[Cell]
class Workbook(BaseModel):
tables: list[Table]在 playground 里设选项
Extract playground 会从其 Build 标签设置这些选项。各项位置见使用网页界面。
cite_sources与confidence_scores:见元数据扩展。playground 默认把两者都开启;API 默认两者都是false。parse_tier:在抽取之前解读文档的 v2 解析档,例如cost_effective、agentic或agentic_plus。见档位。Turbo 上会被忽略,因为它不跑解析步骤。system_prompt:见系统提示词。advanced_citations:哪些档位会用到它,见高级引用。webhook_configurations:任务结束时通知的 webhook。见 Webhooks。在 playground 里选的 webhook 只对当次运行生效,不会随配置保存。
系统提示词
- System Prompt:任何额外的系统级指示。注意:字段级的说明、few-shot 示例与格式要求等应当通过 schema 的字段描述来传递。
页范围与上下文窗口
页范围:用逗号分隔的页号或范围(从 1 开始计数)指定要从哪些页抽取。例如
1,3,5-7,9表示抽取第 1、3 页、第 5 到 7 页以及第 9 页;也可以用1-3,8-10抽取前三页与第 8 到 10 页。页号从 1 开始,即第一页是 1。当你只需要从大文档的特定章节抽取数据时,这个选项很有用。上下文窗口:为长文档抽取传入多少页作为上下文。当你在需要从大文档抽取、又需要周边页的上下文时很有用。它可以通过抽取档与系统提示词配置。取值较大时,周边文档能更多地保持完整,需要一次性看到跨页表格或整张发票时更有帮助;取值较小时会更激进地推进文件,需要你穷尽覆盖密集列表时更合适。
元数据扩展
提供更多元数据与洞见的额外抽取能力,见 元数据扩展 页,它涵盖:
- 引用:为抽取字段追溯来源
- 置信度得分:量化的置信度指标
这些扩展会在 extract_metadata 字段里返回额外元数据,但可能影响处理时长。
设置配置选项
在创建抽取任务时,你可以用 REST API 或 Python SDK 设置这些选项。
SDK
下面的示例读取 NVIDIA 年报的第 1 页与第 38 页,也就是封面与财年摘要。先下载它:
终端窗口
curl -fL -o nvda_10k.pdf https://cloud.llamaindex.ai/assets/demos/extract/nvda_10k.pdf- Python
- TypeScript
- Go
- Java
- CLI
首先安装 Python SDK:
终端窗口
pip install "llama-cloud>=2.8"设置各项配置的方法如下:
import time
from llama_cloud import LlamaCloud, AsyncLlamaCloud
client = LlamaCloud() # reads LLAMA_CLOUD_API_KEY from the environment
schema = {
"type": "object",
"properties": {
"company_name": {"type": "string", "description": "Name of the company"},
"revenue": {"type": "number", "description": "Annual revenue in USD"}
}
}
file_obj = client.files.create(file="nvda_10k.pdf", purpose="extract")
file_id = file_obj.id
job = client.extract.create(
file_input=file_id,
configuration={
"data_schema": schema,
"extraction_target": "per_doc", # Document extraction
"tier": "agentic", # cost_effective, agentic, agentic_plus
"version": "2.5", # Extract version name
"system_prompt": "Focus on the most recent financial data",
"target_pages": "1,38", # Extract from specific pages
"cite_sources": True, # Enable citations
"confidence_scores": True, # Enable confidence scores
},
)
# Poll for completion
while job.status not in ("COMPLETED", "FAILED", "CANCELLED"):
time.sleep(2)
job = client.extract.get(job.id)首先安装 TypeScript SDK:
终端窗口
npm install @llamaindex/llama-cloud zod设置各项配置的方法如下:
import fs from 'fs';
import LlamaCloud from '@llamaindex/llama-cloud';
const client = new LlamaCloud(); // reads LLAMA_CLOUD_API_KEY from the environment
const schema = {
type: 'object',
properties: {
company_name: { type: 'string', description: 'Name of the company' },
revenue: { type: 'number', description: 'Annual revenue in USD' },
},
};
const fileObj = await client.files.create({
file: fs.createReadStream('nvda_10k.pdf'),
purpose: 'extract',
});
const fileId = fileObj.id;
let job = await client.extract.create({
file_input: fileId,
configuration: {
data_schema: schema,
extraction_target: 'per_doc', // Document extraction
tier: 'agentic', // cost_effective, agentic, agentic_plus
version: '2.5', // Extract version name
system_prompt: 'Focus on the most recent financial data',
target_pages: '1,38', // Extract from specific pages
cite_sources: true, // Enable citations
confidence_scores: true, // Enable confidence scores
},
});
// Poll for completion
while (!['COMPLETED', 'FAILED', 'CANCELLED'].includes(job.status)) {
await new Promise((r) => setTimeout(r, 2000));
job = await client.extract.get(job.id);
}首先安装 Go SDK:
终端窗口
go get github.com/run-llama/llama-parse-go设置各项配置的方法如下:
package main
import (
"context"
"fmt"
"log"
"os"
"time"
llamacloud "github.com/run-llama/llama-parse-go"
)
func main() {
ctx := context.Background()
client := llamacloud.NewClient()
// Define schema as a JSON Schema literal
dataSchema := map[string]*llamacloud.ExtractConfigurationDataSchemaUnionParam{
"type": {OfString: llamacloud.String("object")},
"properties": {OfAnyMap: map[string]any{
"company_name": map[string]any{"type": "string", "description": "Name of the company"},
"revenue": map[string]any{"type": "number", "description": "Annual revenue in USD"},
}},
}
f, err := os.Open("nvda_10k.pdf")
if err != nil {
log.Fatal(err)
}
defer f.Close()
fileObj, err := client.Files.New(ctx, llamacloud.FileNewParams{
File: f,
Purpose: "extract",
})
if err != nil {
log.Fatal(err)
}
job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
FileInput: fileObj.ID,
Configuration: llamacloud.ExtractConfigurationParam{
DataSchema: dataSchema,
ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc, // Document extraction
Tier: llamacloud.ExtractConfigurationTierAgentic, // cost_effective, agentic
Version: llamacloud.String("2.5"), // Extract version name
SystemPrompt: llamacloud.String("Focus on the most recent financial data"),
TargetPages: llamacloud.String("1,38"), // Extract from specific pages
CiteSources: llamacloud.Bool(true), // Enable citations
ConfidenceScores: llamacloud.Bool(true), // Enable confidence scores
},
},
})
if err != nil {
log.Fatal(err)
}
// Poll for completion
for job.Status != "COMPLETED" && job.Status != "FAILED" && job.Status != "CANCELLED" {
time.Sleep(2 * time.Second)
job, err = client.Extract.Get(ctx, job.ID, llamacloud.ExtractGetParams{})
if err != nil {
log.Fatal(err)
}
}
fmt.Println(job.ExtractResult.RawJSON())
}首先安装 Java SDK:
implementation("ai.llamaindex:llama-cloud:1.3.0")设置各项配置的方法如下:
import ai.llamaindex.llamacloud.client.LlamaCloudClient;
import ai.llamaindex.llamacloud.client.okhttp.LlamaCloudOkHttpClient;
import ai.llamaindex.llamacloud.core.JsonValue;
import ai.llamaindex.llamacloud.models.extract.ExtractConfiguration;
import ai.llamaindex.llamacloud.models.extract.ExtractCreateParams;
import ai.llamaindex.llamacloud.models.extract.ExtractGetParams;
import ai.llamaindex.llamacloud.models.extract.ExtractV2Job;
import ai.llamaindex.llamacloud.models.extract.ExtractV2JobCreate;
import ai.llamaindex.llamacloud.models.files.FileCreateParams;
import ai.llamaindex.llamacloud.models.files.FileCreateResponse;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.Map;
LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();
// Define schema as a JSON Schema literal
Map<String, Object> companyNameField = new HashMap<>();
companyNameField.put("type", "string");
companyNameField.put("description", "Name of the company");
Map<String, Object> revenueField = new HashMap<>();
revenueField.put("type", "number");
revenueField.put("description", "Annual revenue in USD");
Map<String, Object> properties = new HashMap<>();
properties.put("company_name", companyNameField);
properties.put("revenue", revenueField);
ExtractConfiguration.DataSchema dataSchema = ExtractConfiguration.DataSchema.builder()
.putAdditionalProperty("type", JsonValue.from("object"))
.putAdditionalProperty("properties", JsonValue.from(properties))
.build();
// Upload a file to extract from
FileCreateResponse fileObj = client.files().create(
FileCreateParams.builder()
.file(Paths.get("nvda_10k.pdf"))
.purpose("extract")
.build());
ExtractV2Job job = client.extract().create(
ExtractCreateParams.builder()
.extractV2JobCreate(
ExtractV2JobCreate.builder()
.fileInput(fileObj.id())
.configuration(
ExtractConfiguration.builder()
.dataSchema(dataSchema)
.extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
.tier(ExtractConfiguration.Tier.AGENTIC)
.version("2.5")
.systemPrompt("Focus on the most recent financial data")
.targetPages("1,38")
.citeSources(true)
.confidenceScores(true)
.build())
.build())
.build());
// Poll for completion
while (!job.status().equals("COMPLETED")
&& !job.status().equals("FAILED")
&& !job.status().equals("CANCELLED")) {
Thread.sleep(2000);
job = client.extract().get(ExtractGetParams.builder().jobId(job.id()).build());
}
System.out.println(job.extractResult());首先安装 CLI:
终端窗口
go install github.com/run-llama/llama-parse-cli/cmd/llp@latest设置各项配置的方法如下:
终端窗口
# Define schema as a JSON Schema literal
DATA_SCHEMA='{
"type": "object",
"properties": {
"company_name": {"type": "string", "description": "Name of the company"},
"revenue": {"type": "number", "description": "Annual revenue in USD"}
}
}'
# Upload a file to extract from
FILE_ID=$(llp files create --file nvda_10k.pdf --purpose extract | jq -r '.id')
# Set various configuration options
JOB_ID=$(llp extract create \
--file-input "$FILE_ID" \
--configuration "{\"data_schema\": $DATA_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"agentic\", \"version\": \"2.5\", \"system_prompt\": \"Focus on the most recent financial data\", \"target_pages\": \"1,38\", \"cite_sources\": true, \"confidence_scores\": true}" \
| jq -r '.id')
# Poll for completion
while true; do
JOB=$(llp extract get --job-id "$JOB_ID") || break
STATUS=$(printf '%s\n' "$JOB" | jq -r '.status')
case "$STATUS" in COMPLETED|FAILED|CANCELLED) break ;; esac
sleep 2
done
printf '%s\n' "$JOB" | jq -r 'if .status == "COMPLETED" then .extract_result
else "Extract job failed: \(.error_message // .)" end'REST API
创建抽取任务时,你也可以用 REST API 设置这些选项:
终端窗口
curl -X 'POST' \
'https://api.cloud.llamaindex.ai/api/v2/extract?project_id={PROJECT_ID}' \
-H 'accept: application/json' \
-H "Authorization: Bearer $LLAMA_CLOUD_API_KEY" \
-H 'Content-Type: application/json' \
-d '{
"file_input": "{FILE_ID}",
"configuration": {
"data_schema": {
"type": "object",
"properties": {
"company_name": {"type": "string", "description": "Name of the company"},
"revenue": {"type": "number", "description": "Annual revenue in USD"}
}
},
"extraction_target": "per_doc",
"tier": "agentic",
"version": "2.5",
"system_prompt": "Focus on the most recent financial data",
"target_pages": "1,38",
"cite_sources": true,
"confidence_scores": true
}
}'配置参考
| 选项 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| 档位 | |||
tier | string | ”agentic” | 抽取档位:cost_effective、agentic(默认)、agentic_plus。也接受 turbo —— 其限制见档位。 |
version | string | ”latest” | 抽取版本名,例如 “2.5”;或填 “latest” 使用所选档位下最新的兼容发布。也支持日期(YYYY-MM-DD),表示使用不晚于该日期的最新版本。 |
| 系统提示词 | |||
system_prompt | string | null | 额外的系统级指示 |
| 页范围与上下文 | |||
target_pages | string | null | 要处理的页号或范围,逗号分隔(从 1 开始,例如 “1,3,5-7”)。页会按列出的顺序处理,因此 “3,1,2” 是先喂第 3 页、再第 1 页、再第 2 页。 |
max_pages | integer | null | 最多处理多少页 |
| 元数据扩展 | |||
cite_sources | boolean | false | 开启来源引用 |
advanced_citations | boolean | true | 更精确的引用边界框。设为 false 可让抽取更快。 |
confidence_scores | boolean | false | 开启置信度得分 |
| 电子表格 | |||
spreadsheet_mode | boolean | false | 直接读取 Excel/CSV 工作簿的单元格来抽取。cost_effective、agentic 与 agentic_plus 支持;Cost-effective 与 Agentic 需要 2.1 或更高版本。不产出引用与置信度得分。见定价。 |
sheet_names | array | null | 开启 spreadsheet_mode 时要抽取的工作表名(区分大小写、须完全匹配)。它会覆盖 target_pages;省略则抽取所有工作表。 |
性能建议
总体最佳实践
要让抽取成功率最高:
调试期先用 agentic 档:排查抽取问题时,请用
agentic档 —— 它使用能力最强的模型。若agentic能抽成功,可以再试试cost_effective,看你的场景下质量是否守得住。若连agentic都失败,问题多半出在 schema 设计上(例如字段描述含糊)。如果你的约束是墙钟延迟而非质量,请看 Turbo 档:它不跑解析步骤、并行处理各页,在 PDF 与图片输入上几秒即可返回。下面的示例在 NVIDIA 年报的第 1 页与第 38 页(封面与财年摘要)上跑 Extract。先下载它:
终端窗口
curl -fL -o nvda_10k.pdf https://cloud.llamaindex.ai/assets/demos/extract/nvda_10k.pdf- Python
- TypeScript
- Go
- Java
- CLI
import time from llama_cloud import LlamaCloud client = LlamaCloud() # reads LLAMA_CLOUD_API_KEY from the environment schema = { "type": "object", "properties": { "company_name": {"type": "string", "description": "Name of the company"}, "revenue": {"type": "number", "description": "Annual revenue in USD"}, }, } file_obj = client.files.create(file="nvda_10k.pdf", purpose="extract") # Start debugging with agentic tier job = client.extract.create( file_input=file_obj.id, configuration={ "data_schema": schema, "extraction_target": "per_doc", "tier": "agentic", "target_pages": "1,38", }, ) # Poll for completion while job.status not in ("COMPLETED", "FAILED", "CANCELLED"): time.sleep(2) job = client.extract.get(job.id) print(job.extract_result)import fs from 'fs'; import LlamaCloud from '@llamaindex/llama-cloud'; const client = new LlamaCloud(); // reads LLAMA_CLOUD_API_KEY from the environment const schema = { type: 'object', properties: { company_name: { type: 'string', description: 'Name of the company' }, revenue: { type: 'number', description: 'Annual revenue in USD' }, }, }; const fileObj = await client.files.create({ file: fs.createReadStream('nvda_10k.pdf'), purpose: 'extract', }); // Start debugging with agentic tier let job = await client.extract.create({ file_input: fileObj.id, configuration: { data_schema: schema, extraction_target: 'per_doc', tier: 'agentic', target_pages: '1,38', }, }); // Poll for completion while (!['COMPLETED', 'FAILED', 'CANCELLED'].includes(job.status)) { await new Promise((r) => setTimeout(r, 2000)); job = await client.extract.get(job.id); } console.log(job.extract_result);package main import ( "context" "fmt" "log" "os" "time" llamacloud "github.com/run-llama/llama-parse-go" ) func main() { ctx := context.Background() client := llamacloud.NewClient() schema := map[string]*llamacloud.ExtractConfigurationDataSchemaUnionParam{ "type": {OfString: llamacloud.String("object")}, "properties": {OfAnyMap: map[string]any{ "company_name": map[string]any{"type": "string", "description": "Name of the company"}, "revenue": map[string]any{"type": "number", "description": "Annual revenue in USD"}, }}, } f, err := os.Open("nvda_10k.pdf") if err != nil { log.Fatal(err) } defer f.Close() fileObj, err := client.Files.New(ctx, llamacloud.FileNewParams{ File: f, Purpose: "extract", }) if err != nil { log.Fatal(err) } // Start debugging with agentic tier job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{ ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{ FileInput: fileObj.ID, Configuration: llamacloud.ExtractConfigurationParam{ DataSchema: schema, ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc, Tier: llamacloud.ExtractConfigurationTierAgentic, TargetPages: llamacloud.String("1,38"), }, }, }) if err != nil { log.Fatal(err) } // Poll for completion for job.Status != "COMPLETED" && job.Status != "FAILED" && job.Status != "CANCELLED" { time.Sleep(2 * time.Second) job, err = client.Extract.Get(ctx, job.ID, llamacloud.ExtractGetParams{}) if err != nil { log.Fatal(err) } } fmt.Println(job.ExtractResult.RawJSON()) }import ai.llamaindex.llamacloud.client.LlamaCloudClient; import ai.llamaindex.llamacloud.client.okhttp.LlamaCloudOkHttpClient; import ai.llamaindex.llamacloud.core.JsonValue; import ai.llamaindex.llamacloud.models.extract.ExtractConfiguration; import ai.llamaindex.llamacloud.models.extract.ExtractCreateParams; import ai.llamaindex.llamacloud.models.extract.ExtractGetParams; import ai.llamaindex.llamacloud.models.extract.ExtractV2Job; import ai.llamaindex.llamacloud.models.extract.ExtractV2JobCreate; import ai.llamaindex.llamacloud.models.files.FileCreateParams; import ai.llamaindex.llamacloud.models.files.FileCreateResponse; import java.nio.file.Paths; import java.util.HashMap; import java.util.Map; public class ExtractAgenticDebug { public static void main(String[] args) throws Exception { LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv(); Map<String, Object> companyNameField = new HashMap<>(); companyNameField.put("type", "string"); companyNameField.put("description", "Name of the company"); Map<String, Object> revenueField = new HashMap<>(); revenueField.put("type", "number"); revenueField.put("description", "Annual revenue in USD"); Map<String, Object> properties = new HashMap<>(); properties.put("company_name", companyNameField); properties.put("revenue", revenueField); ExtractConfiguration.DataSchema schema = ExtractConfiguration.DataSchema.builder() .putAdditionalProperty("type", JsonValue.from("object")) .putAdditionalProperty("properties", JsonValue.from(properties)) .build(); FileCreateResponse fileObj = client.files().create( FileCreateParams.builder() .file(Paths.get("nvda_10k.pdf")) .purpose("extract") .build()); // Start debugging with agentic tier ExtractV2Job job = client.extract().create( ExtractCreateParams.builder() .extractV2JobCreate(ExtractV2JobCreate.builder() .fileInput(fileObj.id()) .configuration(ExtractConfiguration.builder() .dataSchema(schema) .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC) .tier(ExtractConfiguration.Tier.AGENTIC) .targetPages("1,38") .build()) .build()) .build()); // Poll for completion while (!job.status().equals("COMPLETED") && !job.status().equals("FAILED") && !job.status().equals("CANCELLED")) { Thread.sleep(2000); job = client.extract().get(ExtractGetParams.builder().jobId(job.id()).build()); } System.out.println(job.extractResult()); } }终端窗口
DATA_SCHEMA='{ "type": "object", "properties": { "company_name": {"type": "string", "description": "Name of the company"}, "revenue": {"type": "number", "description": "Annual revenue in USD"} } }' FILE_ID=$(llp files create --file nvda_10k.pdf --purpose extract | jq -r '.id') # Start debugging with agentic tier JOB_ID=$(llp extract create \ --file-input "$FILE_ID" \ --configuration "{\"data_schema\": $DATA_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"agentic\", \"target_pages\": \"1,38\"}" \ | jq -r '.id') # Poll for completion while true; do JOB=$(llp extract get --job-id "$JOB_ID") || break STATUS=$(printf '%s\n' "$JOB" | jq -r '.status') case "$STATUS" in COMPLETED|FAILED|CANCELLED) break ;; esac sleep 2 done printf '%s\n' "$JOB" | jq -r 'if .status == "COMPLETED" then .extract_result else "Extract job failed: \(.error_message // .)" end'先小后大、反复迭代:先用数据或 schema 的一个子集验证抽取思路,在 schema 描述上反复打磨(例如加示例、加格式说明),把准确度做上去再扩大规模。
设计清晰、聚焦的 schema:宁可要精准的短描述,也不要那种试图包办一切的冗长字段。见schema 设计与限制 与避免复杂的字段变换。
善用文档结构:用页范围、章节与切块策略来优化处理。见配置选项。
有策略地组合工具:Extract 擅长从文档里抽取信息。请把力气花在这个长处上,计算类与校验类的工作交给配套工具(例如繁重的计算放在后处理步骤里做)。
从表格与有序列表抽取
情形: 处理含表格、电子表格(CSV/Excel)或实体有序列表的文档时,你想要的是每一行、每一项都被抽出来,而不是只抽到前几行。
把行建模成一个数组字段: 把一行描述成一个对象,放进 schema 的一个数组字段里。Extract 会把所有行都返回在这个数组中,而标题、日期、合计这类文档级字段可以与它并列放在同一份 schema 里 —— 于是一个任务就同时覆盖了两者。
{
"type": "object",
"properties": {
"report_title": {"type": "string", "description": "Title of the report"},
"line_items": {
"type": "array",
"description": "Every row of the line item table, in document order",
"items": {
"type": "object",
"properties": {
"line_item": {"type": "string", "description": "Name of the line item"},
"amount": {"type": "number", "description": "Amount in USD"}
}
}
}
}
}若这张表位于长文档中已知的位置,请把 target_pages 设成那段范围。
避免复杂的字段变换
不要把业务逻辑写进字段描述里。先抽出干净的数据,再在你的应用代码中做计算。
# ❌ Problematic: Too much logic in the field description
problematic_field = {
"calculated_score": {
"type": "number",
"description": "If revenue > 1M, multiply by 0.8, else if revenue < 500K multiply by 1.2, otherwise use the base score from table 3, but only if the date is after 2020 and the category is not 'exempt'"
}
}
# ✅ Better: Simple extraction, handle logic separately
better_schema = {
"revenue": {"type": "number", "description": "Total revenue in dollars"},
"base_score": {"type": "number", "description": "Base score value from the scoring table"},
"date": {"type": "string", "description": "Date in YYYY-MM-DD format"},
"category": {"type": "string", "description": "Business category"}
}
# Then handle calculations in your application code:
def calculate_final_score(extracted_data):
revenue = extracted_data["revenue"]
if revenue > 1000000:
return extracted_data["base_score"] * 0.8
elif revenue < 500000:
return extracted_data["base_score"] * 1.2
return extracted_data["base_score"]