快速上手 | BytesRouter

若要走程序化路线，SDK 是推荐方式 —— 用它试不同 schema、并规模化地跑抽取。

注意

下面的示例用的是 v2 的 Extract SDK（llama-cloud）。若你需要用 Extract v1，请通过左侧栏顶部的开关切到 v1 文档。

你可以访问 Python SDK 或 Typescript SDK 的 Github 仓库。

首先获取一把 api key。你可以把它导出为环境变量方便使用，也可以在后面直接传给 client。

终端窗口

export LLAMA_CLOUD_API_KEY=llx-xxxxxx

然后安装依赖：

Python
TypeScript
Go
Java
CLI

终端窗口

pip install "llama-cloud>=2.8"

终端窗口

npm install @llamaindex/llama-cloud zod

终端窗口

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

现在库和 API key 都齐了，我们来建一个脚本文件、从文件里抽取数据。这里用的是示例里的几份简历：

终端窗口

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

import time
from pydantic import BaseModel, Field
from llama_cloud import LlamaCloud

# Define schema using Pydantic
class Resume(BaseModel):
    name: str = Field(description="Full name of candidate")
    email: str = Field(description="Email address")
    skills: list[str] = Field(description="Technical skills and technologies")

client = LlamaCloud()  # reads LLAMA_CLOUD_API_KEY from the environment

# Upload a file to extract from
file_obj = client.files.create(file="resume.pdf", purpose="extract")

# Extract data from document
job = client.extract.create(
    file_input=file_obj.id,
    configuration={
        "data_schema": Resume.model_json_schema(),
        "extraction_target": "per_doc",
        "tier": "agentic",
    },
)

# 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 { z } from 'zod';
import LlamaCloud from '@llamaindex/llama-cloud';

// Define schema using Zod
const ResumeSchema = z.object({
    name: z.string().describe("Full name of candidate"),
    email: z.string().describe("Email address"),
    skills: z.array(z.string()).describe("Technical skills and technologies"),
});

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

// Upload a file to extract from
const fileObj = await client.files.create({
  file: fs.createReadStream('resume.pdf'),
  purpose: 'extract',
});

// Extract data from document
let job = await client.extract.create({
  file_input: fileObj.id,
  configuration: {
      data_schema: z.toJSONSchema(ResumeSchema),
      extraction_target: 'per_doc',
      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);
}

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()

    // Define 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": "Full name of candidate"},
            "email": map[string]any{"type": "string", "description": "Email address"},
            "skills": map[string]any{
                "type":        "array",
                "items":       map[string]any{"type": "string"},
                "description": "Technical skills and technologies",
            },
        }},
        "required": {OfAnyArray: []any{"name", "email", "skills"}},
    }

    // 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,
                ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
                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())
}

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.HashMap;
import java.util.Map;

LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();

// Define schema as a JSON Schema literal
Map<String, Object> nameField = new HashMap<>();
nameField.put("type", "string");
nameField.put("description", "Full name of candidate");

Map<String, Object> emailField = new HashMap<>();
emailField.put("type", "string");
emailField.put("description", "Email address");

Map<String, Object> skillItems = new HashMap<>();
skillItems.put("type", "string");
Map<String, Object> skillsField = new HashMap<>();
skillsField.put("type", "array");
skillsField.put("items", skillItems);
skillsField.put("description", "Technical skills and technologies");

Map<String, Object> properties = new HashMap<>();
properties.put("name", nameField);
properties.put("email", emailField);
properties.put("skills", skillsField);

ExtractConfiguration.DataSchema dataSchema = ExtractConfiguration.DataSchema.builder()
        .putAdditionalProperty("type", JsonValue.from("object"))
        .putAdditionalProperty("properties", JsonValue.from(properties))
        .putAdditionalProperty("required", JsonValue.from(Arrays.asList("name", "email", "skills")))
        .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)
                                                .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
                                                .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());

终端窗口

# Define schema as a JSON Schema literal
DATA_SCHEMA='{
  "type": "object",
  "properties": {
    "name": {"type": "string", "description": "Full name of candidate"},
    "email": {"type": "string", "description": "Email address"},
    "skills": {"type": "array", "items": {"type": "string"}, "description": "Technical skills and technologies"}
  },
  "required": ["name", "email", "skills"]
}'

# 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, \"extraction_target\": \"per_doc\", \"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

schema 可以用 Pydantic/Zod 模型或 JSON Schema 定义。更多细节见 schema 页。

其他抽取 API

对字节或文本直接抽取

你也可以直接对原始文本调用抽取。

Python
TypeScript
Go
Java
CLI

import io
import time
from llama_cloud import LlamaCloud

client = LlamaCloud()  # reads LLAMA_CLOUD_API_KEY from the environment

source_text = "Candidate Name: Jane Doe\nEmail: jane.doe@example.com"
source_buffer = io.BytesIO(source_text.encode('utf-8'))

file_obj = client.files.create(file=source_buffer, purpose="extract", external_file_id="resume.txt")

job = client.extract.create(
    file_input=file_obj.id,
    configuration={
        "data_schema": Resume.model_json_schema(),
        "extraction_target": "per_doc",
        "tier": "agentic",
    },
)

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

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

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

const sourceText = 'Candidate Name: Jane Doe\nEmail: jane.doe@example.com';

const fileObj = await client.files.create({
  file: Buffer.from(sourceText, 'utf-8'),
  purpose: 'extract',
  external_file_id: 'resume.txt',
});

let job = await client.extract.create({
  file_input: fileObj.id,
  configuration: {
      data_schema: z.toJSONSchema(ResumeSchema),
      extraction_target: 'per_doc',
      tier: 'agentic',
  },
});

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

import (
    "strings"

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

client := llamacloud.NewClient()

sourceText := "Candidate Name: Jane Doe\nEmail: jane.doe@example.com"

fileObj, err := client.Files.New(ctx, llamacloud.FileNewParams{
    File:           strings.NewReader(sourceText),
    Purpose:        "extract",
    ExternalFileID: llamacloud.String("resume.txt"),
})
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,
            Tier:             llamacloud.ExtractConfigurationTierAgentic,
        },
    },
})
if err != nil {
    log.Fatal(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 {
        log.Fatal(err)
    }
}

import java.nio.charset.StandardCharsets;

LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();

String sourceText = "Candidate Name: Jane Doe\nEmail: jane.doe@example.com";

FileCreateResponse fileObj = client.files().create(
        FileCreateParams.builder()
                .file(sourceText.getBytes(StandardCharsets.UTF_8))
                .purpose("extract")
                .externalFileId("resume.txt")
                .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)
                                                .build())
                                .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());
}

终端窗口

printf 'Candidate Name: Jane Doe\nEmail: jane.doe@example.com' > resume.txt

FILE_ID=$(llp files create \
  --file resume.txt \
  --purpose extract \
  --external-file-id resume.txt \
  | jq -r '.id')

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

while true; do
  STATUS=$(llp extract get --job-id "$JOB_ID" | jq -er '.status') || break
  case "$STATUS" in COMPLETED|FAILED|CANCELLED) break ;; esac
  sleep 2
done

用 Parse 任务 ID 抽取

如果你已经用 LlamaParse 解析过某个文档，可以把 parse 任务 ID 直接传给抽取，而不必再上传一次文件。这会跳过重复解析，时间与 credit 都能省下。当你要对同一份文档用多份 schema 抽取时，尤其有用。

这个示例先在一份简历上跑一个 Parse 任务，再用快速上手里的 Resume schema 对其跑抽取。先下载一份存为 document.pdf：

终端窗口

curl -fL -o document.pdf https://raw.githubusercontent.com/run-llama/llama_cloud_services/main/examples/extract/data/resumes/ai_researcher.pdf

Python
TypeScript
Go
Java
CLI

import time
from llama_cloud import LlamaCloud

client = LlamaCloud()  # reads LLAMA_CLOUD_API_KEY from the environment

# Step 1: Parse the document once
parse_job = client.parsing.create(
    tier="agentic",
    version="latest",
    upload_file="./document.pdf",
)
parse_result = client.parsing.wait_for_completion(parse_job.id, verbose=True)

# Step 2: Extract using the parse job ID (no re-upload needed)
job = client.extract.create(
    file_input=parse_job.id,  # e.g. "pjb-xxxxxxxx-..."
    configuration={
        "data_schema": Resume.model_json_schema(),
        "extraction_target": "per_doc",
        "tier": "agentic",
    },
)

# 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

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

// Step 2: Extract using the parse job ID (no re-upload needed)
let job = await client.extract.create({
  file_input: parseJob.id,  // e.g. "pjb-xxxxxxxx-..."
  configuration: {
      data_schema: z.toJSONSchema(ResumeSchema),
      extraction_target: 'per_doc',
      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);
}

console.log(job.extract_result);

client := llamacloud.NewClient()

// Step 1: Parse the document once
f, err := os.Open("./document.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)
}

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)
    }
}

// Step 2: Extract using the parse job ID (no re-upload needed)
job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
    ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
        FileInput: parseJob.ID, // e.g. "pjb-xxxxxxxx-..."
        Configuration: llamacloud.ExtractConfigurationParam{
            DataSchema:       dataSchema,
            ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
            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())

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;

LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();

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

ParsingCreateResponse parseJob = client.parsing().create(
        ParsingCreateParams.builder()
                .tier(ParsingCreateParams.Tier.AGENTIC)
                .version(ParsingCreateParams.Version.LATEST)
                .fileId(parseFile.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());
}

// Step 2: Extract using the parse job ID (no re-upload needed)
ExtractV2Job job = client.extract().create(
        ExtractCreateParams.builder()
                .extractV2JobCreate(
                        ExtractV2JobCreate.builder()
                                .fileInput(parseJob.id()) // e.g. "pjb-xxxxxxxx-..."
                                .configuration(
                                        ExtractConfiguration.builder()
                                                .dataSchema(dataSchema)
                                                .extractionTarget(ExtractConfiguration.ExtractionTarget.PER_DOC)
                                                .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());

终端窗口

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

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

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 using the parse job ID (no re-upload needed)
JOB_ID=$(llp extract create \
  --file-input "$PARSE_JOB_ID" \
  --configuration "{\"data_schema\": $DATA_SCHEMA, \"extraction_target\": \"per_doc\", \"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'

file_input 字段接受文件 ID（dfl-...）或 parse 任务 ID（pjb-...）二者之一。更完整的示例（含对同一份解析结果用多份 schema 抽取）见批量抽取手册。

批处理

异步处理多个文件。先下载三份样例简历：

终端窗口

curl -fL -o resume1.pdf https://raw.githubusercontent.com/run-llama/llama_cloud_services/main/examples/extract/data/resumes/software_architect.pdf
curl -fL -o resume2.pdf https://raw.githubusercontent.com/run-llama/llama_cloud_services/main/examples/extract/data/resumes/ai_researcher.pdf
curl -fL -o resume3.pdf https://raw.githubusercontent.com/run-llama/llama_cloud_services/main/examples/extract/data/resumes/ml_engineer.pdf

Python
TypeScript
Go
Java
CLI

我们可以用信号量做并发控制，把多个文件提交抽取：

import asyncio
from llama_cloud import AsyncLlamaCloud

client = AsyncLlamaCloud()  # reads LLAMA_CLOUD_API_KEY from the environment
semaphore = asyncio.Semaphore(5)  # Limit concurrency

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

async def process_path(file_path: str):
    async with semaphore:
        file_obj = await client.files.create(file=file_path, purpose="extract")

        job = await client.extract.create(
            file_input=file_obj.id,
            configuration=EXTRACT_CONFIG,
        )

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

    return job.extract_result

async def main():
    file_paths = ["resume1.pdf", "resume2.pdf", "resume3.pdf"]
    results = await asyncio.gather(*(process_path(path) for path in file_paths))
    return results

asyncio.run(main())

我们可以用信号量做并发控制，把多个文件提交抽取：

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

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

const extractConfig = {
    data_schema: z.toJSONSchema(ResumeSchema),
    extraction_target: 'per_doc',
    tier: 'agentic',
};

// npm install async-mutex for Semaphore, or use p-limit
const semaphore = new Semaphore(5);

async function processPath(filePath: string) {
  await semaphore.acquire();
  try {
    const fileObj = await client.files.create({
      file: fs.createReadStream(filePath),
      purpose: 'extract',
    });

    let job = await client.extract.create({
      file_input: fileObj.id,
      configuration: extractConfig,
    });

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

    return job.extract_result;
  } finally {
    semaphore.release();
  }
}

const filePaths = ["resume1.pdf", "resume2.pdf", "resume3.pdf"];
const results = await Promise.all(filePaths.map(processPath));

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

import (
    "sync"
)

client := llamacloud.NewClient()
sem := make(chan struct{}, 5) // Limit concurrency

extractConfig := llamacloud.ExtractConfigurationParam{
    DataSchema:       dataSchema,
    ExtractionTarget: llamacloud.ExtractConfigurationExtractionTargetPerDoc,
    Tier:             llamacloud.ExtractConfigurationTierAgentic,
}

processPath := func(filePath string) (llamacloud.ExtractV2JobExtractResultUnion, error) {
    sem <- struct{}{}
    defer func() { <-sem }()

    var empty llamacloud.ExtractV2JobExtractResultUnion

    f, err := os.Open(filePath)
    if err != nil {
        return empty, err
    }
    defer f.Close()

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

    job, err := client.Extract.New(ctx, llamacloud.ExtractNewParams{
        ExtractV2JobCreate: llamacloud.ExtractV2JobCreateParam{
            FileInput:     fileObj.ID,
            Configuration: extractConfig,
        },
    })
    if err != nil {
        return empty, 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 empty, err
        }
    }

    return job.ExtractResult, nil
}

filePaths := []string{"resume1.pdf", "resume2.pdf", "resume3.pdf"}
results := make([]llamacloud.ExtractV2JobExtractResultUnion, len(filePaths))

var wg sync.WaitGroup
for i, path := range filePaths {
    wg.Add(1)
    go func(i int, path string) {
        defer wg.Done()
        res, err := processPath(path)
        if err != nil {
            log.Fatal(err)
        }
        results[i] = res
    }(i, path)
}
wg.Wait()

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

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

LlamaCloudClient client = LlamaCloudOkHttpClient.fromEnv();
ExecutorService pool = Executors.newFixedThreadPool(5); // Limit concurrency

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

List<String> filePaths = Arrays.asList("resume1.pdf", "resume2.pdf", "resume3.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();

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

终端窗口

EXTRACT_CONFIG="{\"data_schema\": $DATA_SCHEMA, \"extraction_target\": \"per_doc\", \"tier\": \"agentic\"}"

process_path() {
  FILE_ID=$(llp files create --file "$1" --purpose extract | jq -r '.id')

  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
    STATUS=$(printf '%s\n' "$JOB" | jq -r '.status')
    case "$STATUS" in COMPLETED|FAILED|CANCELLED) break ;; esac
    sleep 2
  done

  printf '%s\n' "$JOB" | jq -c --arg file "$1" 'if .status == "COMPLETED" then {file: $file, data: .extract_result}
    else {file: $file, error: (.error_message // .)} end' > "results/$(basename "$1").json"
}

mkdir -p results
STARTED=0
for pdf in resume1.pdf resume2.pdf resume3.pdf; do
  process_path "$pdf" &
  STARTED=$((STARTED + 1))
  if [ $((STARTED % 5)) -eq 0 ]; then wait; fi  # Limit concurrency: 5 jobs per batch
done
wait

jq -s '.' results/*.json

模式生成

你也可以用 SDK 从一段提示词或一个示例文件自动生成 JSON schema。用示例文件时，请下载 NVIDIA 的年报（Form 10-K）。基于生成 schema 的那次抽取读取 130 页中的第 1 页与第 38 页，即封面页与财年摘要页：

终端窗口

curl -fL -o sample_report.pdf https://cloud.llamaindex.ai/assets/demos/extract/nvda_10k.pdf

Python
TypeScript
Go
Java
CLI

# Generate schema from a prompt
generated = client.extract.generate_schema(
    prompt="Extract company financials including revenue, net income, and fiscal year",
)

# Generate schema from a sample file
file_obj = client.files.create(file="sample_report.pdf", purpose="extract")
generated = client.extract.generate_schema(
    file_id=file_obj.id,
    prompt="Extract key financial data",
)

# Use the generated schema in an extraction
job = client.extract.create(
    file_input=file_obj.id,
    configuration={
        "data_schema": generated.parameters.data_schema,
        "tier": "agentic",
        "target_pages": "1,38",
    },
)

// Generate schema from a prompt
const schema = await client.extract.generateSchema({
  prompt: 'Extract company financials including revenue, net income, and fiscal year',
});

// Generate schema from a sample file
const fileObj = await client.files.create({
  file: fs.createReadStream('sample_report.pdf'),
  purpose: 'extract',
});
const schemaFromFile = await client.extract.generateSchema({
  file_id: fileObj.id,
  prompt: 'Extract key financial data',
});

// Use the generated schema in an extraction
let job = await client.extract.create({
  file_input: fileObj.id,
  configuration: {
      data_schema: schemaFromFile.parameters.data_schema,
      tier: 'agentic',
      target_pages: '1,38',
  },
});

import (
    "encoding/json"
)

// Generate schema from a prompt
generated, err := client.Extract.GenerateSchema(ctx, llamacloud.ExtractGenerateSchemaParams{
    ExtractV2SchemaGenerateRequest: llamacloud.ExtractV2SchemaGenerateRequestParam{
        Prompt: llamacloud.String("Extract company financials including revenue, net income, and fiscal year"),
    },
})
if err != nil {
    log.Fatal(err)
}

// Generate schema from a sample file
f, err := os.Open("sample_report.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)
}

generated, err = client.Extract.GenerateSchema(ctx, llamacloud.ExtractGenerateSchemaParams{
    ExtractV2SchemaGenerateRequest: llamacloud.ExtractV2SchemaGenerateRequestParam{
        FileID: llamacloud.String(fileObj.ID),
        Prompt: llamacloud.String("Extract key financial data"),
    },
})
if err != nil {
    log.Fatal(err)
}

// Use the generated schema in an extraction
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,
            Tier:        llamacloud.ExtractConfigurationTierAgentic,
            TargetPages: llamacloud.String("1,38"),
        },
    },
})
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;

// Generate schema from a prompt
ConfigurationCreate generated = client.extract().generateSchema(
        ExtractGenerateSchemaParams.builder()
                .extractV2SchemaGenerateRequest(
                        ExtractV2SchemaGenerateRequest.builder()
                                .prompt("Extract company financials including revenue, net income, and fiscal year")
                                .build())
                .build());

// Generate schema from a sample file
FileCreateResponse fileObj = client.files().create(
        FileCreateParams.builder()
                .file(Paths.get("sample_report.pdf"))
                .purpose("extract")
                .build());

ConfigurationCreate schemaFromFile = client.extract().generateSchema(
        ExtractGenerateSchemaParams.builder()
                .extractV2SchemaGenerateRequest(
                        ExtractV2SchemaGenerateRequest.builder()
                                .fileId(fileObj.id())
                                .prompt("Extract key financial data")
                                .build())
                .build());

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

ExtractV2Job job = client.extract().create(
        ExtractCreateParams.builder()
                .extractV2JobCreate(
                        ExtractV2JobCreate.builder()
                                .fileInput(fileObj.id())
                                .configuration(
                                        ExtractConfiguration.builder()
                                                .dataSchema(generatedSchema)
                                                .tier(ExtractConfiguration.Tier.AGENTIC)
                                                .targetPages("1,38")
                                                .build())
                                .build())
                .build());

终端窗口

# Generate schema from a prompt
llp extract generate-schema \
  --prompt "Extract company financials including revenue, net income, and fiscal year"

# Generate schema from a sample file
FILE_ID=$(llp files create --file sample_report.pdf --purpose extract | jq -r '.id')
SCHEMA=$(llp extract generate-schema \
  --file-id "$FILE_ID" \
  --prompt "Extract key financial data" \
  | jq -c '.parameters.data_schema')

# Use the generated schema in an extraction
llp extract create \
  --file-input "$FILE_ID" \
  --configuration "{\"data_schema\": $SCHEMA, \"tier\": \"agentic\", \"target_pages\": \"1,38\"}"

索引

数据加载

查询

存储

成本分析

应用

聊天机器人

问答

结构化数据

智能体

Chat Engines

Query Engine

连接器

文档与节点

摄取流水线

节点解析器

大语言模型

提示词

回调

节点后处理器

响应合成器

检索器

路由

结构化输出

现有数据

Presets and Modes

能力

起步

能力

Examples

Azure Deployment

Databases and Queues

LLM Integrations

快速开始

定义 schema

其他抽取 API

对字节或文本直接抽取

用 Parse 任务 ID 抽取

批处理

模式生成

友情链接