这本 cookbook 讲 LlamaExtract V2 的生产用法：从单个文件抽取、并发处理批次、组合「先解析再抽取」的流程、应对延迟，以及用程序管理 schema。

所有示例都用 V2 的 Extract API（client.extract），它接受一个 file_input —— 文件 ID 或 Parse 任务 ID 皆可。

• Python
• TypeScript
• Go
• Java
• CLI

pip install "llama-cloud>=2.8"

npm install @llamaindex/llama-cloud

终端窗口

go get github.com/run-llama/llama-parse-go

implementation("ai.llamaindex:llama-cloud:1.3.0")

终端窗口

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

连接 Llama Cloud

• Python
• TypeScript
• Go
• Java
• CLI

import os
from llama_cloud import LlamaCloud, AsyncLlamaCloud

client = LlamaCloud(api_key=os.environ["LLAMA_CLOUD_API_KEY"])

import LlamaCloud from "@llamaindex/llama-cloud";

const client = new LlamaCloud({
  apiKey: process.env.LLAMA_CLOUD_API_KEY!,
});

import (
    "context"

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

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

import ai.llamaindex.llamacloud.client.LlamaCloudClient;
import ai.llamaindex.llamacloud.client.okhttp.LlamaCloudOkHttpClient;

LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();

终端窗口

export LLAMA_CLOUD_API_KEY=llx-xxxxxx

1. 快速上手：单文件抽取

上传文件、定义 schema、抽取结构化数据。先下载一张样例发票：

终端窗口

mkdir -p invoices
curl -fL -o invoices/invoice_001.pdf https://media.githubusercontent.com/media/run-llama/llama-datasets/main/llama_agents/invoice-contracts/invoice-2-1.pdf

• Python
• TypeScript
• Go
• Java
• CLI

from pydantic import BaseModel, Field
from typing import Optional
import time

# Define your schema
class InvoiceData(BaseModel):
    vendor_name: str = Field(description="Name of the vendor or supplier")
    invoice_number: str = Field(description="Unique invoice identifier")
    total_amount: float = Field(description="Total amount due")
    currency: str = Field(description="Currency code (e.g. USD, EUR)")
    due_date: Optional[str] = Field(None, description="Payment due date")

# Upload the file
file_obj = client.files.create(
    file="./invoices/invoice_001.pdf",
    purpose="extract",
)

# Create an extraction job
job = client.extract.create(
    file_input=file_obj.id,
    configuration={
        "data_schema": InvoiceData.model_json_schema(),
        "extraction_target": "per_doc",
        "tier": "cost_effective",
    },
)

# Poll until complete
while job.status not in ("COMPLETED", "FAILED", "CANCELLED"):
    time.sleep(2)
    job = client.extract.get(job.id)
    print(f"Status: {job.status}")

if job.status == "COMPLETED":
    invoice = InvoiceData.model_validate(job.extract_result)
    print(f"Vendor: {invoice.vendor_name}")
    print(f"Total: {invoice.currency} {invoice.total_amount}")
else:
    print(f"Job failed: {job.error_message}")

import fs from "fs";
import LlamaCloud from "@llamaindex/llama-cloud";

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

// Upload the file
const fileObj = await client.files.create({
  file: fs.createReadStream("./invoices/invoice_001.pdf"),
  purpose: "extract",
});

// Create an extraction job
let job = await client.extract.create({
  file_input: fileObj.id,
  configuration: {
      data_schema: {
        type: "object",
        properties: {
          vendor_name: { type: "string", description: "Name of the vendor or supplier" },
          invoice_number: { type: "string", description: "Unique invoice identifier" },
          total_amount: { type: "number", description: "Total amount due" },
          currency: { type: "string", description: "Currency code (e.g. USD, EUR)" },
          due_date: { type: "string", description: "Payment due date", nullable: true },
        },
        required: ["vendor_name", "invoice_number", "total_amount", "currency"],
      },
      extraction_target: "per_doc",
      tier: "cost_effective",
    },
});

// Poll until complete
while (!["COMPLETED", "FAILED", "CANCELLED"].includes(job.status)) {
  await new Promise((r) => setTimeout(r, 2000));
  job = await client.extract.get(job.id);
  console.log(`Status: ${job.status}`);
}

if (job.status === "COMPLETED") {
  console.log("Extracted:", 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()

    // Define the schema as a JSON Schema literal
    dataSchema := map[string]*llamacloud.ExtractConfigurationDataSchemaUnionParam{
        "type": {OfString: llamacloud.String("object")},
        "properties": {OfAnyMap: map[string]any{
            "vendor_name":    map[string]any{"type": "string", "description": "Name of the vendor or supplier"},
            "invoice_number": map[string]any{"type": "string", "description": "Unique invoice identifier"},
            "total_amount":   map[string]any{"type": "number", "description": "Total amount due"},
            "currency":       map[string]any{"type": "string", "description": "Currency code (e.g. USD, EUR)"},
            "due_date":       map[string]any{"type": "string", "description": "Payment due date"},
        }},
        "required": {OfAnyArray: []any{"vendor_name", "invoice_number", "total_amount", "currency"}},
    }

    // Upload the file
    f, err := os.Open("./invoices/invoice_001.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)
    }

    // Create an extraction job
    job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
        ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
            FileInput: fileObj.ID,
            Configuration: llamacloud.ExtractConfigurationParam{
                DataSchema:       dataSchema,
                ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
                Tier:             llamacloud.ExtractConfigurationTierCostEffective,
            },
        },
    })
    if err != nil {
        log.Fatal(err)
    }

    // Poll until complete
    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.Printf("Status: %s\n", job.Status)
    }

    if job.Status == "COMPLETED" {
        fmt.Println("Extracted:", job.ExtractResult.RawJSON())
    } else {
        fmt.Printf("Job failed: %s\n", job.ErrorMessage)
    }
}

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.Arrays;
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(
                "vendor_name", Map.of("type", "string", "description", "Name of the vendor or supplier"),
                "invoice_number", Map.of("type", "string", "description", "Unique invoice identifier"),
                "total_amount", Map.of("type", "number", "description", "Total amount due"),
                "currency", Map.of("type", "string", "description", "Currency code (e.g. USD, EUR)"),
                "due_date", Map.of("type", "string", "description", "Payment due date"))))
        .putAdditionalProperty("required", JsonValue.from(Arrays.asList("vendor_name", "invoice_number", "total_amount", "currency")))
        .build();

// Upload the file
FileCreateResponse fileObj = client.files().create(
        FileCreateParams.builder()
                .file(Paths.get("./invoices/invoice_001.pdf"))
                .purpose("extract")
                .build());

// Create an extraction job
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.COST_EFFECTIVE)
                                                .build())
                                .build())
                .build());

// Poll until complete
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());
}

if (job.status().equals("COMPLETED")) {
    System.out.println(job.extractResult());
} else {
    System.out.println("Job failed: " + job.errorMessage().orElse("unknown error"));
}

终端窗口

# Define the schema as a JSON Schema literal
DATA_SCHEMA='{
  "type": "object",
  "properties": {
    "vendor_name": {"type": "string", "description": "Name of the vendor or supplier"},
    "invoice_number": {"type": "string", "description": "Unique invoice identifier"},
    "total_amount": {"type": "number", "description": "Total amount due"},
    "currency": {"type": "string", "description": "Currency code (e.g. USD, EUR)"},
    "due_date": {"type": "string", "description": "Payment due date"}
  },
  "required": ["vendor_name", "invoice_number", "total_amount", "currency"]
}'

# Upload the file
FILE_ID=$(llp files create --file ./invoices/invoice_001.pdf --purpose extract | jq -r '.id')

# Create an extraction job
JOB_ID=$(llp extract create \
  --file-input "$FILE_ID" \
  --configuration "{\"data_schema\": $DATA_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"cost_effective\"}" \
  | jq -r '.id')

# Poll until complete
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'

2. 多文档批量抽取

需要从许多文件抽取时，先把它们全部上传，再并发创建抽取任务、并行轮询。往 ./invoices 里再加一张样例发票：

终端窗口

curl -fL -o invoices/invoice_002.pdf https://media.githubusercontent.com/media/run-llama/llama-datasets/main/llama_agents/invoice-contracts/invoice-3-1.pdf

• Python
• TypeScript
• Go
• Java
• CLI

import asyncio
import time
from pathlib import Path
from pydantic import BaseModel, Field
from typing import Optional
from llama_cloud import AsyncLlamaCloud

async_client = AsyncLlamaCloud()

class InvoiceData(BaseModel):
    vendor_name: str = Field(description="Name of the vendor or supplier")
    invoice_number: str = Field(description="Unique invoice identifier")
    total_amount: float = Field(description="Total amount due")
    currency: str = Field(description="Currency code (e.g. USD, EUR)")
    due_date: Optional[str] = Field(None, description="Payment due date")

EXTRACT_CONFIG = {
    "data_schema": InvoiceData.model_json_schema(),
    "extraction_target": "per_doc",
    "tier": "cost_effective",
}

async def upload_files(file_paths: list[Path]) -> list[str]:
    """Upload files concurrently and return file IDs."""
    async def upload_one(path: Path) -> str:
        file_obj = await async_client.files.create(
            file=str(path), purpose="extract"
        )
        return file_obj.id

    file_ids = await asyncio.gather(*[upload_one(p) for p in file_paths])
    return list(file_ids)

async def extract_one(file_id: str) -> dict:
    """Create an extract job for a single file and poll until done."""
    job = await async_client.extract.create(
        file_input=file_id,
        configuration=EXTRACT_CONFIG,
    )

    while job.status not in ("COMPLETED", "FAILED", "CANCELLED"):
        await asyncio.sleep(2)
        job = await async_client.extract.get(job.id)

    if job.status == "COMPLETED":
        return {"file_id": file_id, "data": job.extract_result}
    else:
        return {"file_id": file_id, "error": job.error_message}

async def batch_extract(file_paths: list[Path], concurrency: int = 10) -> list[dict]:
    """Extract from multiple files with bounded concurrency."""
    file_ids = await upload_files(file_paths)
    semaphore = asyncio.Semaphore(concurrency)

    async def extract_with_limit(file_id: str) -> dict:
        async with semaphore:
            return await extract_one(file_id)

    results = await asyncio.gather(
        *[extract_with_limit(fid) for fid in file_ids]
    )
    return list(results)

# Usage
async def main():
    invoice_files = list(Path("./invoices").glob("*.pdf"))
    print(f"Processing {len(invoice_files)} files...")

    results = await batch_extract(invoice_files, concurrency=10)

    succeeded = [r for r in results if "data" in r]
    failed = [r for r in results if "error" in r]
    print(f"Completed: {len(succeeded)}, Failed: {len(failed)}")

    for result in succeeded:
        invoice = InvoiceData.model_validate(result["data"])
        print(f"  {invoice.vendor_name}: {invoice.currency} {invoice.total_amount}")

asyncio.run(main())

import * as fs from "fs";
import * as path from "path";

const EXTRACT_CONFIG = {
  data_schema: {
    type: "object",
    properties: {
      vendor_name: { type: "string", description: "Name of the vendor or supplier" },
      invoice_number: { type: "string", description: "Unique invoice identifier" },
      total_amount: { type: "number", description: "Total amount due" },
      currency: { type: "string", description: "Currency code" },
    },
    required: ["vendor_name", "invoice_number", "total_amount", "currency"],
  },
  extraction_target: "per_doc" as const,
  tier: "cost_effective" as const,
};

async function extractOne(fileId: string) {
  let job = await client.extract.create({
    file_input: fileId,
    configuration: EXTRACT_CONFIG,
  });

  while (!["COMPLETED", "FAILED", "CANCELLED"].includes(job.status)) {
    await new Promise((r) => setTimeout(r, 2000));
    job = await client.extract.get(job.id);
  }

  return { fileId, status: job.status, data: job.extract_result, error: job.error_message };
}

async function batchExtract(filePaths: string[], concurrency = 10) {
  // Upload all files
  const fileIds = await Promise.all(
    filePaths.map(async (fp) => {
      const fileObj = await client.files.create({
        file: fs.createReadStream(fp),
        purpose: "extract",
      });
      return fileObj.id;
    })
  );

  // Extract with bounded concurrency
  const results: any[] = [];
  for (let i = 0; i < fileIds.length; i += concurrency) {
    const batch = fileIds.slice(i, i + concurrency);
    const batchResults = await Promise.all(batch.map(extractOne));
    results.push(...batchResults);
  }
  return results;
}

我们可以用带缓冲通道做并发控制，把多个文件提交抽取：

import (
    "sync"
)

// dataSchema is defined in the Quick Start above.
extractConfig := llamacloud.ExtractConfigurationParam{
    DataSchema:       dataSchema,
    ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
    Tier:             llamacloud.ExtractConfigurationTierCostEffective,
}

type extractResult struct {
    FileID string
    Data   llamacloud.ExtractV2JobExtractResultUnion
    Err    error
}

// Upload one file and return its ID.
uploadFile := func(path string) (string, error) {
    f, err := os.Open(path)
    if err != nil {
        return "", err
    }
    defer f.Close()

    fileObj, err := client.Files.New(ctx, llamacloud.FileNewParams{
        File:    f,
        Purpose: "extract",
    })
    if err != nil {
        return "", err
    }
    return fileObj.ID, nil
}

// Create an extract job for one file and poll until done.
extractOne := func(fileID string) extractResult {
    job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
        ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
            FileInput:     fileID,
            Configuration: extractConfig,
        },
    })
    if err != nil {
        return extractResult{FileID: fileID, Err: err}
    }

    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 {
            return extractResult{FileID: fileID, Err: err}
        }
    }
    return extractResult{FileID: fileID, Data: job.ExtractResult}
}

filePaths := []string{"./invoices/invoice_001.pdf", "./invoices/invoice_002.pdf"}
sem := make(chan struct{}, 10) // Limit concurrency to 10
results := make([]extractResult, len(filePaths))

var wg sync.WaitGroup
for i, path := range filePaths {
    wg.Add(1)
    go func(i int, path string) {
        defer wg.Done()
        sem <- struct{}{}
        defer func() { <-sem }()

        fileID, err := uploadFile(path)
        if err != nil {
            results[i] = extractResult{Err: err}
            return
        }
        results[i] = extractOne(fileID)
    }(i, path)
}
wg.Wait()

succeeded, failed := 0, 0
for _, r := range results {
    if r.Err != nil {
        failed++
    } else {
        succeeded++
    }
}
fmt.Printf("Completed: %d, Failed: %d\n", succeeded, failed)

我们可以用固定线程池做并发控制，把多个文件提交抽取：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

// dataSchema is defined in the Quick Start above.
ExtractConfiguration extractConfig = ExtractConfiguration.builder()
        .dataSchema(dataSchema)
        .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
        .tier(ExtractConfiguration.Tier.COST_EFFECTIVE)
        .build();

ExecutorService pool = Executors.newFixedThreadPool(10); // Limit concurrency to 10
List<String> filePaths = Arrays.asList("./invoices/invoice_001.pdf", "./invoices/invoice_002.pdf");
List<Future<ExtractV2Job>> futures = new ArrayList<>();

for (String filePath : filePaths) {
    futures.add(pool.submit(() -> {
        FileCreateResponse fileObj = client.files().create(
                FileCreateParams.builder()
                        .file(Paths.get(filePath))
                        .purpose("extract")
                        .build());

        ExtractV2Job job = client.extract().create(
                ExtractCreateParams.builder()
                        .extractV2JobCreate(
                                ExtractV2JobCreate.builder()
                                        .fileInput(fileObj.id())
                                        .configuration(extractConfig)
                                        .build())
                        .build());

        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());
        }
        return job;
    }));
}

for (Future<ExtractV2Job> future : futures) {
    System.out.println(future.get().extractResult());
}
pool.shutdown();

我们可以用后台任务做并发控制，把多个文件提交抽取：

终端窗口

DATA_SCHEMA='{
  "type": "object",
  "properties": {
    "vendor_name": {"type": "string", "description": "Name of the vendor or supplier"},
    "invoice_number": {"type": "string", "description": "Unique invoice identifier"},
    "total_amount": {"type": "number", "description": "Total amount due"},
    "currency": {"type": "string", "description": "Currency code (e.g. USD, EUR)"},
    "due_date": {"type": "string", "description": "Payment due date"}
  },
  "required": ["vendor_name", "invoice_number", "total_amount", "currency"]
}'
EXTRACT_CONFIG="{\"data_schema\": $DATA_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"cost_effective\"}"

CONCURRENCY=10
WORK_DIR=$(mktemp -d)

# Call after starting each background job: every $CONCURRENCY jobs,
# wait for that batch to finish before starting more.
STARTED=0
throttle() {
  STARTED=$((STARTED + 1))
  if (( STARTED % CONCURRENCY == 0 )); then wait; fi
}

# Upload one file and record its ID.
upload_one() {
  llp files create --file "$1" --purpose extract \
    | jq -r '.id' > "$WORK_DIR/$(basename "$1").file_id"
}

# Create an extract job for a single file and poll until done.
extract_one() {
  local name="$1" file_id="$2" job job_id job_status

  job_id=$(llp extract create \
    --file-input "$file_id" \
    --configuration "$EXTRACT_CONFIG" \
    | jq -r '.id')

  while true; do
    job=$(llp extract get --job-id "$job_id") || break
    job_status=$(printf '%s\n' "$job" | jq -r '.status')
    case "$job_status" in COMPLETED|FAILED|CANCELLED) break ;; esac
    sleep 2
  done

  if [ "$job_status" = "COMPLETED" ]; then
    printf '%s\n' "$job" | jq -c --arg file "$name" '{file: $file, data: .extract_result}'
  else
    printf '%s\n' "$job" | jq -c --arg file "$name" '{file: $file, error: (.error_message // .)}'
  fi > "$WORK_DIR/$name.result"
}

# Upload files concurrently
INVOICE_FILES=(./invoices/*.pdf)
echo "Processing ${#INVOICE_FILES[@]} files..."
for pdf in "${INVOICE_FILES[@]}"; do
  upload_one "$pdf" &
  throttle
done
wait

# Extract from every uploaded file with bounded concurrency
STARTED=0
for pdf in "${INVOICE_FILES[@]}"; do
  name=$(basename "$pdf")
  extract_one "$name" "$(cat "$WORK_DIR/$name.file_id")" &
  throttle
done
wait

SUCCEEDED=$(jq -s '[.[] | select(.data)] | length' "$WORK_DIR"/*.result)
FAILED=$(jq -s '[.[] | select(.error)] | length' "$WORK_DIR"/*.result)
echo "Completed: $SUCCEEDED, Failed: $FAILED"

jq -r 'select(.data) | "  \(.data.vendor_name): \(.data.currency) \(.data.total_amount)"' \
  "$WORK_DIR"/*.result

速率限制的考量

• 并发度：先从 10 个并发任务起步，再按你的套餐上限调整。若看到 THROTTLED 状态，就调低并发度。
• credit：cost_effective 档每页 5 credit，agentic 档每页 15 credit。大批量之前请先算好预算。
• 文件上传：上传端点有它自己的速率限制。先把文件上传完、再创建抽取任务，把两处瓶颈分开。

3. 先解析再抽取：可组合的流程

当你需要对解析过程有更细的控制，或者想「解析一次、用不同 schema 多次抽取」时，就用先解析再抽取这套做法：先解析文档，再把 parse_job_id 传给抽取。

这样就不必为每份抽取配置重复解析同一文档，时间与 credit 都能省下。

从 CUAD 合同数据集下载一份主服务协议：

终端窗口

mkdir -p contracts
curl -fL -o contracts/master_agreement.pdf "https://huggingface.co/datasets/theatticusproject/cuad/resolve/main/CUAD_v1/full_contract_pdf/Part_III/Service/MERITLIFEINSURANCECO_06_19_2020-EX-10.%28XIV%29-MASTER%20SERVICES%20AGREEMENT.PDF"

• Python
• TypeScript
• Go
• Java
• CLI

from pydantic import BaseModel, Field
from typing import Optional
import time

# Step 1: Parse the document
parse_job = client.parsing.create(
    tier="agentic",
    version="latest",
    upload_file="./contracts/master_agreement.pdf",
)

# Wait for parse to complete
parse_result = client.parsing.wait_for_completion(
    parse_job.id, verbose=True
)
print(f"Parse complete: {parse_result.status}")

# Step 2: Extract with the first schema (contract metadata)
class ContractMetadata(BaseModel):
    parties: list[str] = Field(description="Names of all contracting parties")
    effective_date: str = Field(description="Contract effective date")
    termination_date: Optional[str] = Field(None, description="Contract end date")
    governing_law: str = Field(description="Governing law jurisdiction")

metadata_job = client.extract.create(
    file_input=parse_job.id,
    configuration={
        "data_schema": ContractMetadata.model_json_schema(),
        "extraction_target": "per_doc",
        "tier": "agentic",
        "cite_sources": True,
    },
)

while metadata_job.status not in ("COMPLETED", "FAILED", "CANCELLED"):
    time.sleep(2)
    metadata_job = client.extract.get(metadata_job.id)

metadata = ContractMetadata.model_validate(metadata_job.extract_result)
print(f"Parties: {metadata.parties}")
print(f"Governing law: {metadata.governing_law}")

# Step 3: Extract with a second schema (financial terms) from the SAME parse
class FinancialTerms(BaseModel):
    total_value: Optional[float] = Field(None, description="Total contract value")
    payment_schedule: Optional[str] = Field(None, description="Payment frequency and terms")
    penalties: Optional[str] = Field(None, description="Late payment or breach penalties")

financial_job = client.extract.create(
    file_input=parse_job.id,  # Reuse the same parse result
    configuration={
        "data_schema": FinancialTerms.model_json_schema(),
        "extraction_target": "per_doc",
        "tier": "agentic",
    },
)

while financial_job.status not in ("COMPLETED", "FAILED", "CANCELLED"):
    time.sleep(2)
    financial_job = client.extract.get(financial_job.id)

terms = FinancialTerms.model_validate(financial_job.extract_result)
print(f"Contract value: {terms.total_value}")
print(f"Payment schedule: {terms.payment_schedule}")

// Step 1: Parse the document
let parseJob = await client.parsing.create({
  tier: "agentic",
  version: "latest",
  upload_file: fs.createReadStream("./contracts/master_agreement.pdf"),
});

// Wait for parse to complete
parseJob = await client.parsing.waitForCompletion(parseJob.id);
console.log(`Parse complete: ${parseJob.status}`);

// Step 2: Extract contract metadata from parse result
let metadataJob = await client.extract.create({
  file_input: parseJob.id,
  configuration: {
      data_schema: {
        type: "object",
        properties: {
          parties: { type: "array", items: { type: "string" }, description: "Names of all contracting parties" },
          effective_date: { type: "string", description: "Contract effective date" },
          governing_law: { type: "string", description: "Governing law jurisdiction" },
        },
        required: ["parties", "effective_date", "governing_law"],
      },
      extraction_target: "per_doc",
      tier: "agentic",
      cite_sources: true,
    },
});

while (!["COMPLETED", "FAILED", "CANCELLED"].includes(metadataJob.status)) {
  await new Promise((r) => setTimeout(r, 2000));
  metadataJob = await client.extract.get(metadataJob.id);
}
console.log("Contract metadata:", metadataJob.extract_result);

// Step 3: Extract financial terms from the SAME parse result
let financialJob = await client.extract.create({
  file_input: parseJob.id, // Reuse the same parse result
  configuration: {
      data_schema: {
        type: "object",
        properties: {
          total_value: { type: "number", description: "Total contract value", nullable: true },
          payment_schedule: { type: "string", description: "Payment frequency and terms", nullable: true },
        },
      },
      extraction_target: "per_doc",
      tier: "agentic",
    },
});

while (!["COMPLETED", "FAILED", "CANCELLED"].includes(financialJob.status)) {
  await new Promise((r) => setTimeout(r, 2000));
  financialJob = await client.extract.get(financialJob.id);
}
console.log("Financial terms:", financialJob.extract_result);

// Step 1: Parse the document
f, err := os.Open("./contracts/master_agreement.pdf")
if err != nil {
    log.Fatal(err)
}
defer f.Close()

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

parseJob, err := client.Parsing.New(ctx, llamacloud.ParsingNewParams{
    Tier:    llamacloud.ParsingNewParamsTierAgentic,
    Version: llamacloud.ParsingNewParamsVersionLatest,
    FileID:  llamacloud.String(parseFile.ID),
})
if err != nil {
    log.Fatal(err)
}

// Wait for parse to complete
parseResult, err := client.Parsing.Get(ctx, parseJob.ID, llamacloud.ParsingGetParams{})
if err != nil {
    log.Fatal(err)
}
for parseResult.Job.Status != "COMPLETED" &&
    parseResult.Job.Status != "FAILED" &&
    parseResult.Job.Status != "CANCELLED" {
    time.Sleep(2 * time.Second)
    parseResult, err = client.Parsing.Get(ctx, parseJob.ID, llamacloud.ParsingGetParams{})
    if err != nil {
        log.Fatal(err)
    }
}
fmt.Printf("Parse complete: %s\n", parseResult.Job.Status)

// Step 2: Extract contract metadata from the parse result
metadataSchema := map[string]*llamacloud.ExtractConfigurationDataSchemaUnionParam{
    "type": {OfString: llamacloud.String("object")},
    "properties": {OfAnyMap: map[string]any{
        "parties":        map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "Names of all contracting parties"},
        "effective_date": map[string]any{"type": "string", "description": "Contract effective date"},
        "governing_law":  map[string]any{"type": "string", "description": "Governing law jurisdiction"},
    }},
    "required": {OfAnyArray: []any{"parties", "effective_date", "governing_law"}},
}

metadataJob, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
    ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
        FileInput: parseJob.ID,
        Configuration: llamacloud.ExtractConfigurationParam{
            DataSchema:       metadataSchema,
            ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
            Tier:             llamacloud.ExtractConfigurationTierAgentic,
            CiteSources:      llamacloud.Bool(true),
        },
    },
})
if err != nil {
    log.Fatal(err)
}

for metadataJob.Status != "COMPLETED" && metadataJob.Status != "FAILED" && metadataJob.Status != "CANCELLED" {
    time.Sleep(2 * time.Second)
    metadataJob, err = client.Extract.Get(ctx, metadataJob.ID, llamacloud.ExtractGetParams{})
    if err != nil {
        log.Fatal(err)
    }
}
fmt.Println("Contract metadata:", metadataJob.ExtractResult.RawJSON())

// Step 3: Extract financial terms from the SAME parse result
financialSchema := map[string]*llamacloud.ExtractConfigurationDataSchemaUnionParam{
    "type": {OfString: llamacloud.String("object")},
    "properties": {OfAnyMap: map[string]any{
        "total_value":      map[string]any{"type": "number", "description": "Total contract value"},
        "payment_schedule": map[string]any{"type": "string", "description": "Payment frequency and terms"},
    }},
}

financialJob, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
    ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
        FileInput: parseJob.ID, // Reuse the same parse result
        Configuration: llamacloud.ExtractConfigurationParam{
            DataSchema:       financialSchema,
            ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
            Tier:             llamacloud.ExtractConfigurationTierAgentic,
        },
    },
})
if err != nil {
    log.Fatal(err)
}

for financialJob.Status != "COMPLETED" && financialJob.Status != "FAILED" && financialJob.Status != "CANCELLED" {
    time.Sleep(2 * time.Second)
    financialJob, err = client.Extract.Get(ctx, financialJob.ID, llamacloud.ExtractGetParams{})
    if err != nil {
        log.Fatal(err)
    }
}
fmt.Println("Financial terms:", financialJob.ExtractResult.RawJSON())

import ai.llamaindex.llamacloud.models.parsing.ParsingCreateParams;
import ai.llamaindex.llamacloud.models.parsing.ParsingCreateResponse;
import ai.llamaindex.llamacloud.models.parsing.ParsingGetParams;
import ai.llamaindex.llamacloud.models.parsing.ParsingGetResponse;

// Step 1: Parse the document
FileCreateResponse parseFile = client.files().create(
        FileCreateParams.builder()
                .file(Paths.get("./contracts/master_agreement.pdf"))
                .purpose("parse")
                .build());

ParsingCreateResponse parseJob = client.parsing().create(
        ParsingCreateParams.builder()
                .tier(ParsingCreateParams.Tier.AGENTIC)
                .version(ParsingCreateParams.Version.LATEST)
                .fileId(parseFile.id())
                .build());

// Wait for parse to complete
ParsingGetResponse parseResult = client.parsing().get(
        ParsingGetParams.builder().jobId(parseJob.id()).build());
while (!parseResult.job().status().equals(ParsingGetResponse.Job.Status.COMPLETED)
        && !parseResult.job().status().equals(ParsingGetResponse.Job.Status.FAILED)
        && !parseResult.job().status().equals(ParsingGetResponse.Job.Status.CANCELLED)) {
    Thread.sleep(2000);
    parseResult = client.parsing().get(
            ParsingGetParams.builder().jobId(parseJob.id()).build());
}
System.out.println("Parse complete: " + parseResult.job().status());

// Step 2: Extract contract metadata from the parse result
ExtractConfiguration.DataSchema metadataSchema = ExtractConfiguration.DataSchema.builder()
        .putAdditionalProperty("type", JsonValue.from("object"))
        .putAdditionalProperty("properties", JsonValue.from(Map.of(
                "parties", Map.of("type", "array", "items", Map.of("type", "string"), "description", "Names of all contracting parties"),
                "effective_date", Map.of("type", "string", "description", "Contract effective date"),
                "governing_law", Map.of("type", "string", "description", "Governing law jurisdiction"))))
        .putAdditionalProperty("required", JsonValue.from(Arrays.asList("parties", "effective_date", "governing_law")))
        .build();

ExtractV2Job metadataJob = client.extract().create(
        ExtractCreateParams.builder()
                .extractV2JobCreate(
                        ExtractV2JobCreate.builder()
                                .fileInput(parseJob.id())
                                .configuration(
                                        ExtractConfiguration.builder()
                                                .dataSchema(metadataSchema)
                                                .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
                                                .tier(ExtractConfiguration.Tier.AGENTIC)
                                                .citeSources(true)
                                                .build())
                                .build())
                .build());

while (!metadataJob.status().equals("COMPLETED")
        && !metadataJob.status().equals("FAILED")
        && !metadataJob.status().equals("CANCELLED")) {
    Thread.sleep(2000);
    metadataJob = client.extract().get(ExtractGetParams.builder().jobId(metadataJob.id()).build());
}
System.out.println("Contract metadata: " + metadataJob.extractResult());

// Step 3: Extract financial terms from the SAME parse result
ExtractConfiguration.DataSchema financialSchema = ExtractConfiguration.DataSchema.builder()
        .putAdditionalProperty("type", JsonValue.from("object"))
        .putAdditionalProperty("properties", JsonValue.from(Map.of(
                "total_value", Map.of("type", "number", "description", "Total contract value"),
                "payment_schedule", Map.of("type", "string", "description", "Payment frequency and terms"))))
        .build();

ExtractV2Job financialJob = client.extract().create(
        ExtractCreateParams.builder()
                .extractV2JobCreate(
                        ExtractV2JobCreate.builder()
                                .fileInput(parseJob.id()) // Reuse the same parse result
                                .configuration(
                                        ExtractConfiguration.builder()
                                                .dataSchema(financialSchema)
                                                .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
                                                .tier(ExtractConfiguration.Tier.AGENTIC)
                                                .build())
                                .build())
                .build());

while (!financialJob.status().equals("COMPLETED")
        && !financialJob.status().equals("FAILED")
        && !financialJob.status().equals("CANCELLED")) {
    Thread.sleep(2000);
    financialJob = client.extract().get(ExtractGetParams.builder().jobId(financialJob.id()).build());
}
System.out.println("Financial terms: " + financialJob.extractResult());

终端窗口

# Step 1: Parse the document
PARSE_FILE_ID=$(llp files create --file ./contracts/master_agreement.pdf --purpose parse | jq -r '.id')

PARSE_JOB_ID=$(llp parsing create \
  --tier agentic \
  --version latest \
  --file-id "$PARSE_FILE_ID" \
  | jq -r '.id')

# Wait for parse to complete
while true; do
  STATUS=$(llp parsing get --job-id "$PARSE_JOB_ID" | jq -er '.job.status') || break
  case "$STATUS" in COMPLETED|FAILED|CANCELLED) break ;; esac
  sleep 2
done

# Step 2: Extract contract metadata from the parse result
METADATA_SCHEMA='{
  "type": "object",
  "properties": {
    "parties": {"type": "array", "items": {"type": "string"}, "description": "Names of all contracting parties"},
    "effective_date": {"type": "string", "description": "Contract effective date"},
    "governing_law": {"type": "string", "description": "Governing law jurisdiction"}
  },
  "required": ["parties", "effective_date", "governing_law"]
}'

METADATA_JOB_ID=$(llp extract create \
  --file-input "$PARSE_JOB_ID" \
  --configuration "{\"data_schema\": $METADATA_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"agentic\", \"cite_sources\": true}" \
  | jq -r '.id')

while true; do
  JOB=$(llp extract get --job-id "$METADATA_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'

# Step 3: Extract financial terms from the SAME parse result
FINANCIAL_SCHEMA='{
  "type": "object",
  "properties": {
    "total_value": {"type": "number", "description": "Total contract value"},
    "payment_schedule": {"type": "string", "description": "Payment frequency and terms"}
  }
}'

FINANCIAL_JOB_ID=$(llp extract create \
  --file-input "$PARSE_JOB_ID" \
  --configuration "{\"data_schema\": $FINANCIAL_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"agentic\"}" \
  | jq -r '.id')

while true; do
  JOB=$(llp extract get --job-id "$FINANCIAL_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'

指定页

用 target_pages 只从特定页抽取。当你只需要长文档里某些页的数据时，这很有用。

• Python
• TypeScript
• Go
• Java
• CLI

job = client.extract.create(
    file_input=file_obj.id,
    configuration={
        "data_schema": InvoiceData.model_json_schema(),
        "extraction_target": "per_doc",
        "tier": "cost_effective",
        "target_pages": "1,3,5-7",  # Pages 1, 3, 5, 6, 7
    },
)

// invoiceSchema defined earlier (see Quick Start above)
const job = await client.extract.create({
  file_input: fileObj.id,
  configuration: {
      data_schema: invoiceSchema,
      extraction_target: "per_doc",
      tier: "cost_effective",
      target_pages: "1,3,5-7", // Pages 1, 3, 5, 6, 7
    },
});

// dataSchema and fileObj are defined in the Quick Start above.
job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
    ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
        FileInput: fileObj.ID,
        Configuration: llamacloud.ExtractConfigurationParam{
            DataSchema:       dataSchema,
            ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
            Tier:             llamacloud.ExtractConfigurationTierCostEffective,
            TargetPages:      llamacloud.String("1,3,5-7"), // Pages 1, 3, 5, 6, 7
        },
    },
})

// dataSchema and fileObj are defined in the Quick Start above.
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.COST_EFFECTIVE)
                                                .targetPages("1,3,5-7") // Pages 1, 3, 5, 6, 7
                                                .build())
                                .build())
                .build());

终端窗口

llp extract create \
  --file-input "$FILE_ID" \
  --configuration "{\"data_schema\": $DATA_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"cost_effective\", \"target_pages\": \"1,3,5-7\"}"

页号从 1 开始。支持的格式有："1"、"1,3"、"1-5"、"1,3,5-7,9"。只按实际抽取的页计费。

用 expand 取回元数据

默认情况下，extract_metadata（引用、字段级置信度）与 metadata（用量统计）不包含在响应里。请用 expand 查询参数来请求它们。

• Python
• TypeScript
• Go
• Java
• CLI

# Get job with metadata
job = client.extract.get(job.id, expand=["extract_metadata", "metadata"])

if job.metadata and job.metadata.usage:
    usage = job.metadata.usage
    print(f"Pages extracted: {usage.num_pages_extracted}")
    print(f"Output tokens: {usage.num_output_tokens}")

// Get job with metadata
const job = await client.extract.get(jobId, {
  expand: ["extract_metadata", "metadata"],
});

if (job.metadata?.usage) {
  console.log(`Pages extracted: ${job.metadata.usage.num_pages_extracted}`);
}

// Get job with metadata
job, err = client.Extract.Get(ctx, job.ID, llamacloud.ExtractGetParams{
    Expand: []string{"extract_metadata", "metadata"},
})
if err != nil {
    log.Fatal(err)
}

if job.Metadata.Usage.NumPagesExtracted > 0 {
    fmt.Printf("Pages extracted: %d\n", job.Metadata.Usage.NumPagesExtracted)
}

// Get job with metadata
job = client.extract().get(
        ExtractGetParams.builder()
                .jobId(job.id())
                .addExpand("extract_metadata")
                .addExpand("metadata")
                .build());

job.metadata()
        .flatMap(metadata -> metadata.usage())
        .flatMap(usage -> usage.numPagesExtracted())
        .ifPresent(pages -> System.out.println("Pages extracted: " + pages));

终端窗口

# Get job with metadata
JOB=$(llp extract get \
  --job-id "$JOB_ID" \
  --expand extract_metadata \
  --expand metadata)

printf '%s\n' "$JOB" | jq -r '.metadata.usage.num_pages_extracted'

什么时候该用「先解析再抽取」

场景	做法
一个文件、一份 schema	直接用文件 ID 作为 file_input
一个文件、多份 schema	解析一次，再用每份 schema 以 parse 任务 ID 作为 file_input 抽取
想在抽取前检查解析质量	先解析、审阅，再抽取
一批文件用同一份 schema	用文件 ID，配合第 2 节的批量模式
只需要特定页	在 configuration 里加 target_pages

4. 应对延迟

各档的预期延迟与费用

档位	每页 credit	典型延迟	最适合
cost_effective	4	5-30 秒	高并发、较简单的文档
agentic	15	15-90 秒	复杂文档、更高准确度

延迟会随文档长度增长。一份 100 页的 PDF 在 agentic 档上的耗时，会比一张 2 页的发票久得多。

客户端超时模式

给轮询循环包上一层超时，以免无限等待。

• Python
• TypeScript
• Go
• Java
• CLI

import time

def extract_with_timeout(
    client,
    file_id: str,
    configuration: dict,
    timeout_seconds: float = 300,
    poll_interval: float = 2.0,
) -> dict:
    """Extract with a client-side timeout."""
    job = client.extract.create(
        file_input=file_id,
        configuration=configuration,
    )

    start = time.monotonic()
    while job.status not in ("COMPLETED", "FAILED", "CANCELLED"):
        elapsed = time.monotonic() - start
        if elapsed > timeout_seconds:
            raise TimeoutError(
                f"Extraction job {job.id} did not complete within "
                f"{timeout_seconds}s (last status: {job.status})"
            )
        time.sleep(poll_interval)
        job = client.extract.get(job.id)

    if job.status != "COMPLETED":
        raise RuntimeError(f"Job {job.id} ended with status: {job.status} - {job.error_message}")

    return job.extract_result

# Usage
try:
    result = extract_with_timeout(
        client,
        file_id=file_obj.id,
        configuration=EXTRACT_CONFIG,
        timeout_seconds=120,
    )
    print("Extracted:", result)
except TimeoutError as e:
    print(f"Timed out: {e}")
except RuntimeError as e:
    print(f"Failed: {e}")

async function extractWithTimeout(
  fileId: string,
  configuration: any,
  timeoutMs: number = 300_000,
  pollIntervalMs: number = 2_000
) {
  let job = await client.extract.create({
    file_input: fileId,
    configuration,
  });

  const start = Date.now();
  while (!["COMPLETED", "FAILED", "CANCELLED"].includes(job.status)) {
    if (Date.now() - start > timeoutMs) {
      throw new Error(
        `Job ${job.id} did not complete within ${timeoutMs}ms (last status: ${job.status})`
      );
    }
    await new Promise((r) => setTimeout(r, pollIntervalMs));
    job = await client.extract.get(job.id);
  }

  if (job.status !== "COMPLETED") {
    throw new Error(`Job ${job.id} failed: ${job.status} - ${job.error_message}`);
  }
  return job.extract_result;
}

// dataSchema and fileObj are defined in the Quick Start above.
extractConfig := llamacloud.ExtractConfigurationParam{
    DataSchema:       dataSchema,
    ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
    Tier:             llamacloud.ExtractConfigurationTierCostEffective,
}

extractWithTimeout := func(fileID string, timeout time.Duration) (llamacloud.ExtractV2JobExtractResultUnion, error) {
    var empty llamacloud.ExtractV2JobExtractResultUnion

    job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
        ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
            FileInput:     fileID,
            Configuration: extractConfig,
        },
    })
    if err != nil {
        return empty, err
    }

    deadline := time.Now().Add(timeout)
    for job.Status != "COMPLETED" && job.Status != "FAILED" && job.Status != "CANCELLED" {
        if time.Now().After(deadline) {
            return empty, fmt.Errorf("extraction job %s did not complete within %s (last status: %s)", job.ID, timeout, job.Status)
        }
        time.Sleep(2 * time.Second)
        job, err = client.Extract.Get(ctx, job.ID, llamacloud.ExtractGetParams{})
        if err != nil {
            return empty, err
        }
    }

    if job.Status != "COMPLETED" {
        return empty, fmt.Errorf("job %s ended with status %s: %s", job.ID, job.Status, job.ErrorMessage)
    }
    return job.ExtractResult, nil
}

result, err := extractWithTimeout(fileObj.ID, 120*time.Second)
if err != nil {
    log.Printf("Failed: %v", err)
} else {
    fmt.Println("Extracted:", result.RawJSON())
}

// dataSchema and fileObj are defined in the Quick Start above.
ExtractConfiguration extractConfig = ExtractConfiguration.builder()
        .dataSchema(dataSchema)
        .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
        .tier(ExtractConfiguration.Tier.COST_EFFECTIVE)
        .build();

ExtractV2Job job = client.extract().create(
        ExtractCreateParams.builder()
                .extractV2JobCreate(
                        ExtractV2JobCreate.builder()
                                .fileInput(fileObj.id())
                                .configuration(extractConfig)
                                .build())
                .build());

long timeoutMs = 120_000;
long start = System.currentTimeMillis();
while (!job.status().equals("COMPLETED")
        && !job.status().equals("FAILED")
        && !job.status().equals("CANCELLED")) {
    if (System.currentTimeMillis() - start > timeoutMs) {
        throw new IllegalStateException(
                "Extraction job " + job.id() + " did not complete within " + timeoutMs
                        + "ms (last status: " + job.status() + ")");
    }
    Thread.sleep(2000);
    job = client.extract().get(ExtractGetParams.builder().jobId(job.id()).build());
}

if (!job.status().equals("COMPLETED")) {
    throw new IllegalStateException(
            "Job " + job.id() + " failed: " + job.status() + " - " + job.errorMessage().orElse(""));
}
System.out.println("Extracted: " + job.extractResult());

终端窗口

TIMEOUT=120  # seconds

JOB_ID=$(llp extract create \
  --file-input "$FILE_ID" \
  --configuration "{\"data_schema\": $DATA_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"cost_effective\"}" \
  | jq -r '.id')

START=$(date +%s)
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
  if [ $(( $(date +%s) - START )) -gt "$TIMEOUT" ]; then
    echo "Timed out after ${TIMEOUT}s (last status: $STATUS)" >&2
    break
  fi
  sleep 2
done

printf '%s\n' "$JOB" | jq -r 'if .status == "COMPLETED" then .extract_result
  else "Extract job did not complete: \(.error_message // .)" end'

理解 THROTTLED 状态

系统负载较高时，你的任务可能会在进入 RUNNING 之前先处于 THROTTLED 状态。这是正常的，一旦有空余容量任务就会继续。请不要取消并重试被限流的任务 —— 那只会制造更多负载，并让你的任务被排到队列更靠后的位置。

PENDING → THROTTLED → RUNNING → COMPLETED

webhook 集成（轮询的替代方案）

对于不希望维持长期轮询连接的生产系统，可以配置 webhook，在任务完成时接收通知。

• Python
• TypeScript
• Go
• Java
• CLI

job = client.extract.create(
    file_input=file_obj.id,
    configuration=EXTRACT_CONFIG,
    webhook_configurations=[
        {
            "webhook_url": "https://your-api.example.com/webhooks/extract",
            "webhook_events": ["extract.success", "extract.error"],
            "webhook_headers": {
                "Authorization": "Bearer your-webhook-secret",
            },
        }
    ],
)

# No polling needed. Your webhook endpoint receives:
# POST https://your-api.example.com/webhooks/extract
# {
#   "event": "extract.success",
#   "job_id": "...",
#   "status": "COMPLETED",
#   ...
# }

const job = await client.extract.create({
  file_input: fileObj.id,
  configuration: EXTRACT_CONFIG,
  webhook_configurations: [
    {
      webhook_url: "https://your-api.example.com/webhooks/extract",
      webhook_events: ["extract.success", "extract.error"],
      webhook_headers: {
        Authorization: "Bearer your-webhook-secret",
      },
    },
  ],
});

// Your webhook endpoint receives a POST when the job completes

// dataSchema and fileObj are defined in the Quick Start above.
job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
    ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
        FileInput: fileObj.ID,
        Configuration: llamacloud.ExtractConfigurationParam{
            DataSchema:       dataSchema,
            ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
            Tier:             llamacloud.ExtractConfigurationTierCostEffective,
        },
        WebhookConfigurations: []llamacloud.ExtractV2JobCreateWebhookConfigurationParam{
            {
                WebhookURL:    llamacloud.String("https://your-api.example.com/webhooks/extract"),
                WebhookEvents: []string{"extract.success", "extract.error"},
                WebhookHeaders: map[string]string{
                    "Authorization": "Bearer your-webhook-secret",
                },
            },
        },
    },
})
if err != nil {
    log.Fatal(err)
}

// No polling needed. Your webhook endpoint receives a POST when the job completes.
fmt.Println("Submitted job:", job.ID)

// dataSchema and fileObj are defined in the Quick Start above.
ExtractConfiguration extractConfig = ExtractConfiguration.builder()
        .dataSchema(dataSchema)
        .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
        .tier(ExtractConfiguration.Tier.COST_EFFECTIVE)
        .build();

ExtractV2JobCreate.WebhookConfiguration webhook = ExtractV2JobCreate.WebhookConfiguration.builder()
        .webhookUrl("https://your-api.example.com/webhooks/extract")
        .addWebhookEvent(ExtractV2JobCreate.WebhookConfiguration.WebhookEvent.EXTRACT_SUCCESS)
        .addWebhookEvent(ExtractV2JobCreate.WebhookConfiguration.WebhookEvent.EXTRACT_ERROR)
        .webhookHeaders(
                ExtractV2JobCreate.WebhookConfiguration.WebhookHeaders.builder()
                        .putAdditionalProperty("Authorization", JsonValue.from("Bearer your-webhook-secret"))
                        .build())
        .build();

ExtractV2Job job = client.extract().create(
        ExtractCreateParams.builder()
                .extractV2JobCreate(
                        ExtractV2JobCreate.builder()
                                .fileInput(fileObj.id())
                                .configuration(extractConfig)
                                .addWebhookConfiguration(webhook)
                                .build())
                .build());

// No polling needed. Your webhook endpoint receives a POST when the job completes.
System.out.println("Submitted job: " + job.id());

终端窗口

llp extract create \
  --file-input "$FILE_ID" \
  --configuration "{\"data_schema\": $DATA_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"cost_effective\"}" \
  --webhook-configuration '[{"webhook_url": "https://your-api.example.com/webhooks/extract", "webhook_events": ["extract.success", "extract.error"], "webhook_headers": {"Authorization": "Bearer your-webhook-secret"}}]'

5. schema 管理

用提示词生成 schema

不想手写 JSON schema？用平实的语言描述你的需求，让 API 生成一份 schema。

• Python
• TypeScript
• Go
• Java
• CLI

generated = client.extract.generate_schema(
    prompt="Extract the company name, CEO name, founding year, and headquarters city from this annual report",
)

print("Generated schema:", generated.parameters.data_schema)
print("Suggested config name:", generated.name)

# Use the generated schema directly in an extraction job
job = client.extract.create(
    file_input=file_obj.id,
    configuration={
        "data_schema": generated.parameters.data_schema,
        "extraction_target": "per_doc",
        "tier": "cost_effective",
    },
)

const generated = await client.extract.generateSchema({
  prompt: "Extract the company name, CEO name, founding year, and headquarters city from this annual report",
});

console.log("Generated schema:", generated.parameters.data_schema);

const job = await client.extract.create({
  file_input: fileObj.id,
  configuration: {
      data_schema: generated.parameters.data_schema,
      extraction_target: "per_doc",
      tier: "cost_effective",
    },
});

import (
    "encoding/json"
)

// fileObj is defined in the Quick Start above.
generated, err := client.Extract.GenerateSchema(ctx, llamacloud.ExtractGenerateSchemaParams{
    ExtractV2SchemaGenerateRequest: llamacloud.ExtractV2SchemaGenerateRequestParam{
        Prompt: llamacloud.String("Extract the company name, CEO name, founding year, and headquarters city from this annual report"),
    },
})
if err != nil {
    log.Fatal(err)
}
fmt.Println("Generated schema:", generated.Parameters.JSON.DataSchema.Raw())

// Use the generated schema directly in an extraction job
var generatedSchema map[string]*llamacloud.ExtractConfigurationDataSchemaUnionParam
if err := json.Unmarshal([]byte(generated.Parameters.JSON.DataSchema.Raw()), &generatedSchema); err != nil {
    log.Fatal(err)
}

job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
    ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
        FileInput: fileObj.ID,
        Configuration: llamacloud.ExtractConfigurationParam{
            DataSchema:       generatedSchema,
            ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
            Tier:             llamacloud.ExtractConfigurationTierCostEffective,
        },
    },
})
if err != nil {
    log.Fatal(err)
}

import ai.llamaindex.llamacloud.models.configurations.ConfigurationCreate;
import ai.llamaindex.llamacloud.models.extract.ExtractGenerateSchemaParams;
import ai.llamaindex.llamacloud.models.extract.ExtractV2SchemaGenerateRequest;

// fileObj is defined in the Quick Start above.
ConfigurationCreate generated = client.extract().generateSchema(
        ExtractGenerateSchemaParams.builder()
                .extractV2SchemaGenerateRequest(
                        ExtractV2SchemaGenerateRequest.builder()
                                .prompt("Extract the company name, CEO name, founding year, and headquarters city from this annual report")
                                .build())
                .build());

// Use the generated schema directly in an extraction job
ExtractConfiguration.DataSchema generatedSchema = ExtractConfiguration.DataSchema.builder()
        .putAllAdditionalProperties(
                generated.parameters().extractV2().get().dataSchema()._additionalProperties())
        .build();

ExtractV2Job job = client.extract().create(
        ExtractCreateParams.builder()
                .extractV2JobCreate(
                        ExtractV2JobCreate.builder()
                                .fileInput(fileObj.id())
                                .configuration(
                                        ExtractConfiguration.builder()
                                                .dataSchema(generatedSchema)
                                                .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
                                                .tier(ExtractConfiguration.Tier.COST_EFFECTIVE)
                                                .build())
                                .build())
                .build());

终端窗口

# Generate schema from a prompt
SCHEMA=$(llp extract generate-schema \
  --prompt "Extract the company name, CEO name, founding year, and headquarters city from this annual report" \
  | jq -c '.parameters.data_schema')

echo "Generated schema: $SCHEMA"

# Use the generated schema directly in an extraction job
llp extract create \
  --file-input "$FILE_ID" \
  --configuration "{\"data_schema\": $SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"cost_effective\"}"

用示例文件生成 schema

上传一份有代表性的文件，让 API 分析它并给出 schema 建议。先下载一张待分析的样例发票：

终端窗口

mkdir -p invoices
curl -fL -o invoices/sample_invoice.pdf https://media.githubusercontent.com/media/run-llama/llama-datasets/main/llama_agents/invoice-contracts/invoice-5-1.pdf

• Python
• TypeScript
• Go
• Java
• CLI

sample_file = client.files.create(
    file="./invoices/sample_invoice.pdf",
    purpose="extract",
)

generated = client.extract.generate_schema(
    file_id=sample_file.id,
    prompt="Extract all invoice fields including line items",
)

print("Generated schema:", generated.parameters.data_schema)

const sampleFile = await client.files.create({
  file: fs.createReadStream("./invoices/sample_invoice.pdf"),
  purpose: "extract",
});

const generated = await client.extract.generateSchema({
  file_id: sampleFile.id,
  prompt: "Extract all invoice fields including line items",
});

console.log("Generated schema:", generated.parameters.data_schema);

f, err := os.Open("./invoices/sample_invoice.pdf")
if err != nil {
    log.Fatal(err)
}
defer f.Close()

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

generated, err := client.Extract.GenerateSchema(ctx, llamacloud.ExtractGenerateSchemaParams{
    ExtractV2SchemaGenerateRequest: llamacloud.ExtractV2SchemaGenerateRequestParam{
        FileID: llamacloud.String(sampleFile.ID),
        Prompt: llamacloud.String("Extract all invoice fields including line items"),
    },
})
if err != nil {
    log.Fatal(err)
}

fmt.Println("Generated schema:", generated.Parameters.JSON.DataSchema.Raw())

FileCreateResponse sampleFile = client.files().create(
        FileCreateParams.builder()
                .file(Paths.get("./invoices/sample_invoice.pdf"))
                .purpose("extract")
                .build());

ConfigurationCreate generated = client.extract().generateSchema(
        ExtractGenerateSchemaParams.builder()
                .extractV2SchemaGenerateRequest(
                        ExtractV2SchemaGenerateRequest.builder()
                                .fileId(sampleFile.id())
                                .prompt("Extract all invoice fields including line items")
                                .build())
                .build());

System.out.println("Generated schema: " + generated.parameters().extractV2().get().dataSchema());

终端窗口

FILE_ID=$(llp files create --file ./invoices/sample_invoice.pdf --purpose extract | jq -r '.id')

llp extract generate-schema \
  --file-id "$FILE_ID" \
  --prompt "Extract all invoice fields including line items" \
  | jq '.parameters.data_schema'

使用前先校验 schema

在跑抽取任务之前先把 schema 的错误挑出来。

• Python
• TypeScript
• Go
• Java
• CLI

from pydantic import BaseModel, Field

class MySchema(BaseModel):
    name: str = Field(description="Person's full name")
    age: int = Field(description="Person's age in years")

validated = client.extract.validate_schema(
    data_schema=MySchema.model_json_schema(),
)

# Returns the validated schema if valid, raises an error if invalid
print("Schema is valid:", validated.data_schema)

// Returns the validated schema if valid, throws an error if invalid
const validated = await client.extract.validateSchema({
  data_schema: {
    type: "object",
    properties: {
      name: { type: "string", description: "Person's full name" },
      age: { type: "integer", description: "Person's age in years" },
    },
    required: ["name", "age"],
  },
});

console.log("Schema is valid:", validated.data_schema);

// Returns the validated schema if valid, returns an error if invalid
validated, err := client.Extract.ValidateSchema(ctx, llamacloud.ExtractValidateSchemaParams{
    ExtractV2SchemaValidateRequest: llamacloud.ExtractV2SchemaValidateRequestParam{
        DataSchema: map[string]*llamacloud.ExtractV2SchemaValidateRequestDataSchemaUnionParam{
            "type": {OfString: llamacloud.String("object")},
            "properties": {OfAnyMap: map[string]any{
                "name": map[string]any{"type": "string", "description": "Person's full name"},
                "age":  map[string]any{"type": "integer", "description": "Person's age in years"},
            }},
            "required": {OfAnyArray: []any{"name", "age"}},
        },
    },
})
if err != nil {
    log.Fatal(err)
}
fmt.Println("Schema is valid:", validated.RawJSON())

import ai.llamaindex.llamacloud.models.extract.ExtractValidateSchemaParams;
import ai.llamaindex.llamacloud.models.extract.ExtractV2SchemaValidateRequest;
import ai.llamaindex.llamacloud.models.extract.ExtractV2SchemaValidateResponse;

ExtractV2SchemaValidateRequest.DataSchema schema = ExtractV2SchemaValidateRequest.DataSchema.builder()
        .putAdditionalProperty("type", JsonValue.from("object"))
        .putAdditionalProperty("properties", JsonValue.from(Map.of(
                "name", Map.of("type", "string", "description", "Person's full name"),
                "age", Map.of("type", "integer", "description", "Person's age in years"))))
        .putAdditionalProperty("required", JsonValue.from(Arrays.asList("name", "age")))
        .build();

// Returns the validated schema if valid, throws an error if invalid
ExtractV2SchemaValidateResponse validated = client.extract().validateSchema(
        ExtractValidateSchemaParams.builder()
                .extractV2SchemaValidateRequest(
                        ExtractV2SchemaValidateRequest.builder()
                                .dataSchema(schema)
                                .build())
                .build());

System.out.println("Schema is valid: " + validated.dataSchema());

终端窗口

# Returns the validated schema if valid, returns an error if invalid
llp extract validate-schema \
  --data-schema '{
    "type": "object",
    "properties": {
      "name": {"type": "string", "description": "Full name of the person"},
      "age": {"type": "integer", "description": "Age of the person in years"}
    },
    "required": ["name", "age"]
  }'

串起来用

下面是一份完整的生产脚本，把批量上传、先解析再抽取、超时处理与结果收集结合起来。先把两份样例供货协议下载到 ./contracts：

终端窗口

mkdir -p contracts
curl -fL -o contracts/agreement_a.pdf https://media.githubusercontent.com/media/run-llama/llama-datasets/main/llama_agents/invoice-contracts/contract-1.pdf
curl -fL -o contracts/agreement_b.pdf https://media.githubusercontent.com/media/run-llama/llama-datasets/main/llama_agents/invoice-contracts/contract-2.pdf

• Python
• TypeScript
• Go
• Java
• CLI

import asyncio
import time
from pathlib import Path
from pydantic import BaseModel, Field
from typing import Optional
from llama_cloud import AsyncLlamaCloud

async_client = AsyncLlamaCloud()

class ContractSummary(BaseModel):
    parties: list[str] = Field(description="Names of all contracting parties")
    effective_date: str = Field(description="Contract effective date")
    contract_type: str = Field(description="Type of contract (NDA, MSA, SOW, etc.)")
    total_value: Optional[float] = Field(None, description="Total contract value if specified")

EXTRACT_CONFIG = {
    "data_schema": ContractSummary.model_json_schema(),
    "extraction_target": "per_doc",
    "tier": "agentic",
    "cite_sources": True,
}

async def process_contract(file_path: Path, timeout: float = 300) -> dict:
    """Parse, then extract from a single contract with timeout."""
    # Upload
    file_obj = await async_client.files.create(
        file=str(file_path), purpose="extract"
    )

    # Parse first for higher quality
    parse_job = await async_client.parsing.create(
        tier="agentic",
        version="latest",
        file_id=file_obj.id,
    )
    parse_job = await async_client.parsing.wait_for_completion(
        parse_job.id, timeout=timeout
    )

    # Extract from parse result
    job = await async_client.extract.create(
        file_input=parse_job.id,
        configuration=EXTRACT_CONFIG,
    )

    start = time.monotonic()
    while job.status not in ("COMPLETED", "FAILED", "CANCELLED"):
        if time.monotonic() - start > timeout:
            return {"file": file_path.name, "error": f"Timeout after {timeout}s"}
        await asyncio.sleep(2)
        job = await async_client.extract.get(job.id)

    if job.status == "COMPLETED":
        return {
            "file": file_path.name,
            "data": job.extract_result,
            # Note: extract_metadata requires ?expand=extract_metadata on GET
        }
    return {"file": file_path.name, "error": job.error_message}

async def main():
    contracts = list(Path("./contracts").glob("*.pdf"))
    semaphore = asyncio.Semaphore(5)

    async def bounded(path):
        async with semaphore:
            return await process_contract(path)

    results = await asyncio.gather(*[bounded(p) for p in contracts])

    for r in results:
        if "data" in r:
            summary = ContractSummary.model_validate(r["data"])
            print(f"{r['file']}: {summary.contract_type} between {', '.join(summary.parties)}")
        else:
            print(f"{r['file']}: ERROR - {r['error']}")

asyncio.run(main())

import * as fs from "fs";
import * as path from "path";

const EXTRACT_CONFIG = {
    data_schema: {
      type: "object",
      properties: {
        parties: { type: "array", items: { type: "string" }, description: "Contracting parties" },
        effective_date: { type: "string", description: "Contract effective date" },
        contract_type: { type: "string", description: "Type of contract (NDA, MSA, SOW, etc.)" },
        total_value: { type: "number", description: "Total contract value", nullable: true },
      },
      required: ["parties", "effective_date", "contract_type"],
    },
    extraction_target: "per_doc" as const,
    tier: "agentic" as const,
    cite_sources: true,
};

async function processContract(filePath: string, timeoutMs = 300_000) {
  const fileObj = await client.files.create({
    file: fs.createReadStream(filePath),
    purpose: "extract",
  });

  // Parse first
  let parseJob = await client.parsing.create({
    tier: "agentic",
    version: "latest",
    file_id: fileObj.id,
  });
  parseJob = await client.parsing.waitForCompletion(parseJob.id);

  // Extract from parse result
  let job = await client.extract.create({
    file_input: parseJob.id,
    configuration: EXTRACT_CONFIG,
  });

  const start = Date.now();
  while (!["COMPLETED", "FAILED", "CANCELLED"].includes(job.status)) {
    if (Date.now() - start > timeoutMs) {
      return { file: path.basename(filePath), error: "Timeout" };
    }
    await new Promise((r) => setTimeout(r, 2000));
    job = await client.extract.get(job.id);
  }

  if (job.status === "COMPLETED") {
    return { file: path.basename(filePath), data: job.extract_result };
  }
  return { file: path.basename(filePath), error: job.error_message };
}

package main

import (
    "context"
    "fmt"
    "os"
    "path/filepath"
    "sync"
    "time"

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

type contractResult struct {
    File string
    Data llamacloud.ExtractV2JobExtractResultUnion
    Err  error
}

func main() {
    ctx := context.Background()
    client := llamacloud.NewClient()

    contractSchema := map[string]*llamacloud.ExtractConfigurationDataSchemaUnionParam{
        "type": {OfString: llamacloud.String("object")},
        "properties": {OfAnyMap: map[string]any{
            "parties":        map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "Names of all contracting parties"},
            "effective_date": map[string]any{"type": "string", "description": "Contract effective date"},
            "contract_type":  map[string]any{"type": "string", "description": "Type of contract (NDA, MSA, SOW, etc.)"},
            "total_value":    map[string]any{"type": "number", "description": "Total contract value if specified"},
        }},
        "required": {OfAnyArray: []any{"parties", "effective_date", "contract_type"}},
    }

    extractConfig := llamacloud.ExtractConfigurationParam{
        DataSchema:       contractSchema,
        ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
        Tier:             llamacloud.ExtractConfigurationTierAgentic,
        CiteSources:      llamacloud.Bool(true),
    }

    // Parse, then extract from a single contract with a timeout.
    processContract := func(filePath string, timeout time.Duration) contractResult {
        name := filepath.Base(filePath)

        f, err := os.Open(filePath)
        if err != nil {
            return contractResult{File: name, Err: err}
        }
        defer f.Close()

        fileObj, err := client.Files.New(ctx, llamacloud.FileNewParams{
            File:    f,
            Purpose: "extract",
        })
        if err != nil {
            return contractResult{File: name, Err: err}
        }

        // Parse first for higher quality
        parseJob, err := client.Parsing.New(ctx, llamacloud.ParsingNewParams{
            Tier:    llamacloud.ParsingNewParamsTierAgentic,
            Version: llamacloud.ParsingNewParamsVersionLatest,
            FileID:  llamacloud.String(fileObj.ID),
        })
        if err != nil {
            return contractResult{File: name, Err: err}
        }

        deadline := time.Now().Add(timeout)
        parseResult, err := client.Parsing.Get(ctx, parseJob.ID, llamacloud.ParsingGetParams{})
        if err != nil {
            return contractResult{File: name, Err: err}
        }
        for parseResult.Job.Status != "COMPLETED" &&
            parseResult.Job.Status != "FAILED" &&
            parseResult.Job.Status != "CANCELLED" {
            if time.Now().After(deadline) {
                return contractResult{File: name, Err: fmt.Errorf("parse timed out")}
            }
            time.Sleep(2 * time.Second)
            parseResult, err = client.Parsing.Get(ctx, parseJob.ID, llamacloud.ParsingGetParams{})
            if err != nil {
                return contractResult{File: name, Err: err}
            }
        }

        // Extract from the parse result
        job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
            ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
                FileInput:     parseJob.ID,
                Configuration: extractConfig,
            },
        })
        if err != nil {
            return contractResult{File: name, Err: err}
        }

        for job.Status != "COMPLETED" && job.Status != "FAILED" && job.Status != "CANCELLED" {
            if time.Now().After(deadline) {
                return contractResult{File: name, Err: fmt.Errorf("extraction timed out")}
            }
            time.Sleep(2 * time.Second)
            job, err = client.Extract.Get(ctx, job.ID, llamacloud.ExtractGetParams{})
            if err != nil {
                return contractResult{File: name, Err: err}
            }
        }

        if job.Status != "COMPLETED" {
            return contractResult{File: name, Err: fmt.Errorf("%s", job.ErrorMessage)}
        }
        return contractResult{File: name, Data: job.ExtractResult}
    }

    contracts := []string{"./contracts/agreement_a.pdf", "./contracts/agreement_b.pdf"}
    sem := make(chan struct{}, 5) // Limit concurrency to 5
    results := make([]contractResult, len(contracts))

    var wg sync.WaitGroup
    for i, path := range contracts {
        wg.Add(1)
        go func(i int, path string) {
            defer wg.Done()
            sem <- struct{}{}
            defer func() { <-sem }()
            results[i] = processContract(path, 300*time.Second)
        }(i, path)
    }
    wg.Wait()

    for _, r := range results {
        if r.Err != nil {
            fmt.Printf("%s: ERROR - %v\n", r.File, r.Err)
        } else {
            fmt.Printf("%s: %s\n", r.File, r.Data.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 ai.llamaindex.llamacloud.models.parsing.ParsingCreateParams;
import ai.llamaindex.llamacloud.models.parsing.ParsingCreateResponse;
import ai.llamaindex.llamacloud.models.parsing.ParsingGetParams;
import ai.llamaindex.llamacloud.models.parsing.ParsingGetResponse;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class BatchExtraction {
    public static void main(String[] args) throws Exception {
        // reads LLAMA_CLOUD_API_KEY from the environment
        LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();

        ExtractConfiguration.DataSchema contractSchema = ExtractConfiguration.DataSchema.builder()
                .putAdditionalProperty("type", JsonValue.from("object"))
                .putAdditionalProperty("properties", JsonValue.from(Map.of(
                        "parties", Map.of("type", "array", "items", Map.of("type", "string"), "description", "Names of all contracting parties"),
                        "effective_date", Map.of("type", "string", "description", "Contract effective date"),
                        "contract_type", Map.of("type", "string", "description", "Type of contract (NDA, MSA, SOW, etc.)"),
                        "total_value", Map.of("type", "number", "description", "Total contract value if specified"))))
                .putAdditionalProperty("required", JsonValue.from(Arrays.asList("parties", "effective_date", "contract_type")))
                .build();

        ExtractConfiguration extractConfig = ExtractConfiguration.builder()
                .dataSchema(contractSchema)
                .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
                .tier(ExtractConfiguration.Tier.AGENTIC)
                .citeSources(true)
                .build();

        ExecutorService pool = Executors.newFixedThreadPool(5); // Limit concurrency to 5
        List<String> contracts = Arrays.asList("./contracts/agreement_a.pdf", "./contracts/agreement_b.pdf");
        List<Future<ExtractV2Job>> futures = new ArrayList<>();

        for (String filePath : contracts) {
            futures.add(pool.submit(() -> {
                FileCreateResponse fileObj = client.files().create(
                        FileCreateParams.builder()
                                .file(Paths.get(filePath))
                                .purpose("extract")
                                .build());

                // Parse first for higher quality
                ParsingCreateResponse parseJob = client.parsing().create(
                        ParsingCreateParams.builder()
                                .tier(ParsingCreateParams.Tier.AGENTIC)
                                .version(ParsingCreateParams.Version.LATEST)
                                .fileId(fileObj.id())
                                .build());

                ParsingGetResponse parseResult = client.parsing().get(
                        ParsingGetParams.builder().jobId(parseJob.id()).build());
                while (!parseResult.job().status().equals(ParsingGetResponse.Job.Status.COMPLETED)
                        && !parseResult.job().status().equals(ParsingGetResponse.Job.Status.FAILED)
                        && !parseResult.job().status().equals(ParsingGetResponse.Job.Status.CANCELLED)) {
                    Thread.sleep(2000);
                    parseResult = client.parsing().get(
                            ParsingGetParams.builder().jobId(parseJob.id()).build());
                }

                // Extract from the parse result
                ExtractV2Job job = client.extract().create(
                        ExtractCreateParams.builder()
                                .extractV2JobCreate(
                                        ExtractV2JobCreate.builder()
                                                .fileInput(parseJob.id())
                                                .configuration(extractConfig)
                                                .build())
                                .build());

                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());
                }
                return job;
            }));
        }

        for (Future<ExtractV2Job> future : futures) {
            System.out.println(future.get().extractResult());
        }
        pool.shutdown();
    }
}

终端窗口

CONTRACT_SCHEMA='{
  "type": "object",
  "properties": {
    "parties": {"type": "array", "items": {"type": "string"}, "description": "Names of all contracting parties"},
    "effective_date": {"type": "string", "description": "Contract effective date"},
    "contract_type": {"type": "string", "description": "Type of contract (NDA, MSA, SOW, etc.)"},
    "total_value": {"type": "number", "description": "Total contract value if specified"}
  },
  "required": ["parties", "effective_date", "contract_type"]
}'
EXTRACT_CONFIG="{\"data_schema\": $CONTRACT_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"agentic\", \"cite_sources\": true}"

CONCURRENCY=5
TIMEOUT=300  # seconds
WORK_DIR=$(mktemp -d)

# Call after starting each background job: every $CONCURRENCY jobs,
# wait for that batch to finish before starting more.
STARTED=0
throttle() {
  STARTED=$((STARTED + 1))
  if (( STARTED % CONCURRENCY == 0 )); then wait; fi
}

# Parse, then extract from a single contract with a timeout.
process_contract() {
  local pdf="$1" name deadline file_id parse_job_id parse_status job job_id job_status
  name=$(basename "$pdf")
  deadline=$(( $(date +%s) + TIMEOUT ))

  file_id=$(llp files create --file "$pdf" --purpose extract | jq -r '.id')

  # Parse first for higher quality
  parse_job_id=$(llp parsing create \
    --tier agentic \
    --version latest \
    --file-id "$file_id" \
    | jq -r '.id')

  while true; do
    job=$(llp parsing get --job-id "$parse_job_id") || break
    parse_status=$(printf '%s\n' "$job" | jq -r '.job.status')
    case "$parse_status" in COMPLETED|FAILED|CANCELLED) break ;; esac
    if [ "$(date +%s)" -gt "$deadline" ]; then
      jq -nc --arg file "$name" --arg t "$TIMEOUT" \
        '{file: $file, error: "Parse timed out after \($t)s"}' > "$WORK_DIR/$name.result"
      return
    fi
    sleep 2
  done
  if [ "$parse_status" != "COMPLETED" ]; then
    jq -nc --arg file "$name" --arg out "$job" \
      '{file: $file, error: "Parse did not complete: \(($out | fromjson? | .job.error_message) // $out)"}' > "$WORK_DIR/$name.result"
    return
  fi

  # Extract from the parse result
  job_id=$(llp extract create \
    --file-input "$parse_job_id" \
    --configuration "$EXTRACT_CONFIG" \
    | jq -r '.id')

  while true; do
    job=$(llp extract get --job-id "$job_id") || break
    job_status=$(printf '%s\n' "$job" | jq -r '.status')
    case "$job_status" in COMPLETED|FAILED|CANCELLED) break ;; esac
    if [ "$(date +%s)" -gt "$deadline" ]; then
      jq -nc --arg file "$name" --arg t "$TIMEOUT" \
        '{file: $file, error: "Timeout after \($t)s"}' > "$WORK_DIR/$name.result"
      return
    fi
    sleep 2
  done

  # Note: extract_metadata requires --expand extract_metadata on `llp extract get`
  if [ "$job_status" = "COMPLETED" ]; then
    printf '%s\n' "$job" | jq -c --arg file "$name" '{file: $file, data: .extract_result}'
  else
    printf '%s\n' "$job" | jq -c --arg file "$name" '{file: $file, error: (.error_message // .)}'
  fi > "$WORK_DIR/$name.result"
}

for pdf in ./contracts/*.pdf; do
  process_contract "$pdf" &
  throttle
done
wait

jq -r 'if .data then
         "\(.file): \(.data.contract_type) between \(.data.parties | join(", "))"
       else
         "\(.file): ERROR - \(.error)"
       end' "$WORK_DIR"/*.result

下一步做什么？

• 带引用的数据抽取：查看抽取结果的出处
• 为抽取自动生成 schema：更多 schema 生成的模式
• LlamaExtract 文档：完整 API 参考