生产级抽取：批处理、轮询与延迟管理

已弃用

本页记录的是已弃用的 Extract v1 API。新的接入请使用 Extract v2；改造已有的 v1 代码请见v1 到 v2 迁移指南。

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

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

python
typescript

pip install "llama-cloud<2.0"

npm install @llamaindex/llama-cloud

连接 Llama Cloud

python
typescript

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!,
});

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

上传文件、定义 schema、抽取结构化数据。

python
typescript

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(
    document_input_value=file_obj.id,
    config={
        "extract_options": {
            "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}")

// 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({
  document_input_value: fileObj.id,
  config: {
    extract_options: {
      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);
}

2. 多文档批量抽取

需要从许多文件抽取时，先把它们全部上传，再并发创建抽取任务、并行轮询。

python
typescript

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 = {
    "extract_options": {
        "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(
        document_input_value=file_id,
        config=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 = {
  extract_options: {
    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({
    document_input_value: fileId,
    config: 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;
}

速率限制的考量

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

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

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

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

python
typescript

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(
    document_input_value=parse_job.id,
    config={
        "extract_options": {
            "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(
    document_input_value=parse_job.id,  # Reuse the same parse result
    config={
        "extract_options": {
            "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({
  document_input_value: parseJob.id,
  config: {
    extract_options: {
      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({
  document_input_value: parseJob.id, // Reuse the same parse result
  config: {
    extract_options: {
      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);

指定页

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

python
typescript

job = client.extract.create(
    document_input_value=file_obj.id,
    config={
        "extract_options": {
            "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
        },
    },
)

const job = await client.extract.create({
  document_input_value: fileObj.id,
  config: {
    extract_options: {
      data_schema: invoiceSchema,
      extraction_target: "per_doc",
      tier: "cost_effective",
      target_pages: "1,3,5-7", // Pages 1, 3, 5, 6, 7
    },
  },
});

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

用 expand 取回元数据

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

python
typescript

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

if job.extract_metadata:
    usage = job.extract_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"],
});

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

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

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

4. 应对延迟

各档的预期延迟与费用

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

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

客户端超时模式

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

python
typescript

import time

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

    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,
        config=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,
  config: any,
  timeoutMs: number = 300_000,
  pollIntervalMs: number = 2_000
) {
  let job = await client.extract.create({
    document_input_value: fileId,
    config,
  });

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

理解 THROTTLED 状态

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

PENDING → THROTTLED → RUNNING → COMPLETED

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

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

python
typescript

job = client.extract.create(
    document_input_value=file_obj.id,
    config=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({
  document_input_value: fileObj.id,
  config: 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

5. schema 管理

用提示词生成 schema

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

python
typescript

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.data_schema)
print("Suggested config name:", generated.name)

# Use the generated schema directly in an extraction job
job = client.extract.create(
    document_input_value=file_obj.id,
    config={
        "extract_options": {
            "data_schema": generated.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.data_schema);

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

用示例文件生成 schema

上传一份有代表性的文件，让 API 分析它并给出 schema 建议。

python
typescript

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.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.data_schema);

使用前先校验 schema

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

python
typescript

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

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

if validation.valid:
    print("Schema is valid")
else:
    print(f"Schema errors: {validation.errors}")

const validation = 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"],
  },
});

if (validation.valid) {
  console.log("Schema is valid");
} else {
  console.log("Schema errors:", validation.errors);
}

串起来用

下面是一份完整的生产脚本，把批量上传、先解析再抽取、超时处理与结果收集结合起来。

python
typescript

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 = {
    "extract_options": {
        "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(
        document_input_value=parse_job.id,
        config=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 = {
  extract_options: {
    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({
    document_input_value: parseJob.id,
    config: 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 };
}

下一步做什么？

带引用与推理的数据抽取：查看抽取结果的出处与推理
为抽取自动生成 schema：更多 schema 生成的模式
抽取重复实体：表行抽取
LlamaExtract 文档：完整 API 参考

索引

数据加载

查询

存储

成本分析

应用

聊天机器人

问答

结构化数据

智能体

Chat Engines

Query Engine

连接器

文档与节点

摄取流水线

节点解析器

大语言模型

提示词

回调

节点后处理器

响应合成器

检索器

路由

结构化输出

现有数据

Presets and Modes

能力

起步

能力

Examples

Azure Deployment

Databases and Queues

LLM Integrations