已弃用

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

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

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

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

终端窗口

export LLAMA_CLOUD_API_KEY=llx-xxxxxx

然后安装依赖：

• Python
• TypeScript
• Python（legacy）

终端窗口

pip install "llama-cloud<2.0"

终端窗口

npm install @llamaindex/llama-cloud zod

终端窗口

pip install llama-cloud-services python-dotenv

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

快速开始

• Python
• TypeScript
• Python（legacy）

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

# 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(api_key="your_api_key")

# Create extraction agent
agent = client.extraction.extraction_agents.create(
    name="resume-parser",
    data_schema=Resume.model_json_schema(),
    config={}
)

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

# Extract data from document
result = await client.extraction.jobs.extract(
    extraction_agent_id=agent.id,
    file_id=file_id,
)
print(result.data)

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({
  apiKey: 'your_api_key',
});

// Create extraction agent
const agent = await client.extraction.extractionAgents.create({
  name: 'resume-parser',
  dataSchema: ResumeSchema,
  config: {},
});

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

// Extract data from document
const result = await client.extraction.jobs.extract({
  extraction_agent_id: agent.id,
  file_id: fileId,
});
console.log(result.data);

from llama_cloud_services import LlamaExtract
from pydantic import BaseModel, Field

# bring in our LLAMA_CLOUD_API_KEY
from dotenv import load_dotenv
load_dotenv()

# Initialize client
extractor = LlamaExtract()

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

# Create extraction agent
agent = extractor.create_agent(name="resume-parser", data_schema=Resume)

# Extract data from document
result = agent.extract("resume.pdf")
print(result.data)

运行脚本即可看到抽取结果！

• Python
• TypeScript
• Python（legacy）

终端窗口

python your_script.py

终端窗口

ts-node your_script.ts

终端窗口

python your_script.py

定义 schema

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

其他抽取 API

对字节或文本直接抽取

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

• Python
• TypeScript
• Python（legacy）

import io
from llama_cloud import LlamaCloud

client = LlamaCloud(api_key="your_api_key")

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")
file_id = file_obj.id

result = await client.extraction.jobs.extract(
    extraction_agent_id=agent.id,
    file_id=file_id,
)

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

const client = new LlamaCloud({
  apiKey: 'your_api_key',
});

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',
});
const fileId = fileObj.id;

const result = await client.extraction.jobs.extract({
  extraction_agent_id: agent.id,
  file_id: fileId,
});

你可以用 SourceText 类直接从字节或文本抽取，而不必先做成文件。若传入的是文件字节，需要同时把文件名传给 SourceText。

from llama_cloud_services import LlamaExtract, SourceText

extractor = LlamaExtract()
agent = extractor.get_agent(name="resume-parser")

with open("resume.pdf", "rb") as f:
    file_bytes = f.read()
result = agent.extract(SourceText(file=file_bytes, filename="resume.pdf"))

result = agent.extract(SourceText(text_content="Candidate Name: Jane Doe"))

批处理

异步处理多个文件：

• Python
• TypeScript
• Python（legacy）

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

import asyncio
from llama_cloud import LlamaCloud, AsyncLlamaCloud

client = AsyncLlamaCloud(api_key="your_api_key")
semaphore = asyncio.Semaphore(5)  # Limit concurrency

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

        result = await client.extraction.jobs.extract(
            extraction_agent_id=agent.id,
            file_id=file_id,
        )

    return result

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

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

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

const client = new LlamaCloud({
  apiKey: 'your_api_key',
});

const semaphore = new Semaphore(5); // Limit concurrency

async function processPath(filePath: string) {
  await semaphore.acquire();
  try {
    const fileObj = await client.files.create({
      file: fs.createReadStream(filePath),
      purpose: 'extract',
    });
    const fileId = fileObj.id;
    const result = await client.extraction.jobs.extract({
      extraction_agent_id: agent.id,
      file_id: fileId,
    });
    return result;
  } finally {
    semaphore.release();
  }
}

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

# Queue multiple files for extraction
jobs = await agent.queue_extraction(["resume1.pdf", "resume2.pdf"])

# Check job status
for job in jobs:
    status = agent.get_extraction_job(job.id).status
    print(f"Job {job.id}: {status}")

# Get results when complete
results = [agent.get_extraction_run_for_job(job.id) for job in jobs]

更新 schema

schema 在创建之后仍可修改与更新：

• Python
• TypeScript
• Python（legacy）

client.extraction.extraction_agents.update(
    extraction_agent_id=agent.id,
    data_schema=new_schema,
    config={},
)

await client.extraction.extractionAgents.update({
  extraction_agent_id: agent.id,
  dataSchema: newSchema,
  config: {},
});

# Update schema
agent.data_schema = new_schema

# Save changes
agent.save()

管理智能体

• Python
• TypeScript
• Python（legacy）

# List all agents
agents = client.extraction.extraction_agents.list()

# Get specific agent
agent = client.extraction.extraction_agents.get(extraction_agent_id="agent_id")

# Delete agent
client.extraction.extraction_agents.delete(extraction_agent_id="agent_id")

// List all agents
const agents = await client.extraction.extractionAgents.list();

// Get specific agent
const agent = await client.extraction.extractionAgents.get({
  extraction_agent_id: 'agent_id',
});

// Delete agent
await client.extraction.extractionAgents.delete({
  extraction_agent_id: 'agent_id',
});

# List all agents
agents = extractor.list_agents()

# Get specific agent
agent = extractor.get_agent(name="resume-parser")

# Delete agent
extractor.delete_agent(agent.id)