默认情况下，索引会自动完成解析、切块、嵌入，并写入一个托管的向量库。如果你更想自己掌握向量库（因为你自托管平台、用的是我们并不管理的数据库，或者只是想把数据留在自己的基础设施里），可以关掉托管导出，改为下载原始解析结果。

在这种模式下，LlamaCloud 仍负责目录管理、解析与同步编排；切块、嵌入以及写入你的数据库由你接手。

创建一个只做解析的索引

创建索引时传 vector_target="DISABLED"。流水线会退化为「仅下载」式导出，因此解析结果被写到对象存储，而不是托管向量库。

• Python
• TypeScript
• Go
• Java
• CLI

index = await client.indexes.create(
    source_directory_id=directory.id,
    vector_target="DISABLED",
)

const index = await client.indexes.create({
  source_directory_id: directory.id,
  vector_target: "DISABLED",
});

index, err := client.Indexes.New(ctx, llamacloud.IndexNewParams{
  SourceDirectoryID: directory.ID,
  VectorTarget:      llamacloud.IndexNewParamsVectorTargetDisabled,
})
if err != nil {
  log.Fatal(err)
}

import ai.llamaindex.llamacloud.models.indexes.IndexCreateParams;
import ai.llamaindex.llamacloud.models.indexes.IndexCreateResponse;

IndexCreateResponse index = client.indexes().create(
    IndexCreateParams.builder()
        .sourceDirectoryId(directory.id())
        .vectorTarget(IndexCreateParams.VectorTarget.DISABLED)
        .build());

终端窗口

INDEX_ID=$(llp beta:indexes create \
  --source-directory-id "$DIRECTORY_ID" \
  --vector-target DISABLED | jq -r '.id')

echo "Index ID: $INDEX_ID"

等待索引进入 ready 的方式与普通索引相同（见快速上手指南）。

下载解析结果

每个索引会把解析结果写到一个输出目录，与你创建的源目录是分开的。输出目录为每个源文件保存一个文件：一个 JSON 载荷，内含逐页 markdown（header、text、footer）以及每页附件（例如截图）的引用。源目录里仍是你的原始文件（PDF、DOCX 等），所以要读载荷请去输出目录。

从索引响应里以 output_directory_id 取得输出目录的 ID。同一份响应还带着 export_config_id，你在自己的数据库里给记录做键时会用到它。随后列出输出目录的文件，并从各自的预签名 URL 下载载荷。

输出目录里的每个文件都带着你在数据库中跟踪它所需的字段：

• id：即 parsed_directory_file_id。同一个源文件跨多次同步它保持不变。
• file_id：存放的 JSON 载荷。把它传给 client.files.content 以取得预签名下载地址。
• metadata：你在源文件上设置的元数据。
• updated_at：当该文件被重新解析、或其元数据发生变化时会改变。

• Python
• TypeScript
• Go
• Java
• CLI

import httpx

# The index points at its output directory and export config.
idx = await client.indexes.get(index.id)

async with httpx.AsyncClient() as http:
    # Iterating the list call walks every page of results.
    async for f in client.beta.directories.files.list(idx.output_directory_id):
        if f.file_id is None:
            continue
        presigned = await client.files.content(f.file_id)
        response = await http.get(presigned.url)
        response.raise_for_status()
        pages = response.json()["parse"]["pages"]
        # Key your records on (f.id, idx.export_config_id).
        # f.metadata holds the source file's metadata.

// The index points at its output directory and export config.
const idx = await client.indexes.get(index.id);

// Iterating the list call walks every page of results.
for await (const f of client.beta.directories.files.list(idx.output_directory_id)) {
  if (!f.file_id) continue;
  const presigned = await client.files.content(f.file_id);
  const response = await fetch(presigned.url);
  const pages = (await response.json()).parse.pages;
  // Key your records on (f.id, idx.export_config_id).
  // f.metadata holds the source file's metadata.
}

// The index points at its output directory and export config.
idx, err := client.Indexes.Get(ctx, index.ID, llamacloud.IndexGetParams{})
if err != nil {
  log.Fatal(err)
}

// ListAutoPaging walks every page of results.
iter := client.Beta.Directories.Files.ListAutoPaging(ctx, idx.OutputDirectoryID, llamacloud.BetaDirectoryFileListParams{})
for iter.Next() {
  f := iter.Current()
  if f.FileID == "" {
    continue
  }
  presigned, err := client.Files.Content(ctx, f.FileID, llamacloud.FileContentParams{})
  if err != nil {
    log.Fatal(err)
  }

  resp, err := http.Get(presigned.URL)
  if err != nil {
    log.Fatal(err)
  }

  var parsed map[string]any
  err = json.NewDecoder(resp.Body).Decode(&parsed)
  resp.Body.Close()
  if err != nil {
    log.Fatal(err)
  }
  // parsed["parse"] holds the pages. Key your records on
  // (f.ID, idx.ExportConfigID); f.Metadata holds the source file's metadata.
}
if err := iter.Err(); err != nil {
  log.Fatal(err)
}

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

import ai.llamaindex.llamacloud.models.beta.directories.files.FileListResponse;
import ai.llamaindex.llamacloud.models.indexes.IndexGetResponse;
import ai.llamaindex.llamacloud.models.files.PresignedUrl;

// The index points at its output directory and export config.
IndexGetResponse idx = client.indexes().get(index.id());
HttpClient http = HttpClient.newHttpClient();

// autoPager() walks every page of results.
for (FileListResponse f : client.beta().directories().files()
        .list(idx.outputDirectoryId()).autoPager()) {
    if (f.fileId().isEmpty()) {
        continue;
    }
    PresignedUrl presigned = client.files().content(f.fileId().get());

    HttpResponse<String> response = http.send(
        HttpRequest.newBuilder(URI.create(presigned.url())).build(),
        HttpResponse.BodyHandlers.ofString());
    String parsed = response.body();
    // parsed is the payload JSON; its "parse" field holds the pages.
    // Key your records on (f.id(), idx.exportConfigId()).
}

终端窗口

# The index points at its output directory.
OUTPUT_DIRECTORY_ID=$(llp beta:indexes get --index-id "$INDEX_ID" | jq -r '.output_directory_id')

FILE_IDS=$(llp beta:directories:files list \
  --directory-id "$OUTPUT_DIRECTORY_ID" | jq -r '.file_id // empty')

for FILE_ID in $FILE_IDS; do
  URL=$(llp files content --file-id "$FILE_ID" | jq -r '.url')
  curl -s "$URL" | jq '.parse'
done

只导出发生变化的部分

如果每次都把全部文件下载并重导一遍，会做大量重复劳动。每次同步之后，把输出目录与你上次导出的状态做个比对：

• 新增或变化的文件： 列举输出目录时把上次导出的时刻作为 updated_at_on_or_after 传入。只有被重新解析过、或元数据变过的文件会返回。
• 已移除的文件： 同步会删除已消失源文件对应的输出文件。你数据库里任何 parsed_directory_file_id 若在一次完整列举中不见了，说明它已被移除，你应当删掉相应记录。

from datetime import datetime

last_export: datetime | None = load_last_export_time()  # from your own bookkeeping
run_started = datetime.now().astimezone()

idx = await client.indexes.get(index.id)
params = {"updated_at_on_or_after": last_export} if last_export else {}

async for f in client.beta.directories.files.list(idx.output_directory_id, **params):
    ...  # download, chunk, embed, and replace this file's records

current_ids = {
    f.id async for f in client.beta.directories.files.list(idx.output_directory_id)
}
stale_ids = exported_file_ids() - current_ids  # IDs you hold for idx.export_config_id
...  # delete the records for stale_ids

save_last_export_time(run_started)

load_last_export_time、exported_file_ids 与 save_last_export_time 代表的是你自己的记账逻辑。参考实现里的 list_snapshots() 方法返回某个 sink 当前持有的文件 ID 集合。

切块、嵌入并推送到你的存储

此后步骤与任何自定义 RAG 流水线相同：

1. 拼接逐页 markdown，并用字符偏移记录页边界。
2. 用你选用的切分器切块。
3. 用你选用的嵌入模型做嵌入。
4. 把切块写入你的数据库，以 (parsed_directory_file_id, export_config_id) 为键：即输出目录文件的 id 与索引的 export_config_id。插入新切块前先删掉该文件已有的记录，这样一个被重新解析的文件是替换而不是叠加。

参考实现

index-v2-data-sinks 仓库为若干数据库提供了完整可运行的参考导出器：

• MongoDB（Atlas vector search）
• PostgreSQL + pgvector
• Qdrant
• Pinecone
• Turbopuffer
• Azure AI Search

每个导出器都处理索引/集合的创建、用于幂等重导的确定性 ID、逐文件的「先删后插」，以及供增量同步使用的快照列举。export/base.py 里共享的 Exporter 协议就是新增 sink 时要实现的契约：

class Exporter(Protocol):
    async def export(self, outputs, *, project_id, embeddings=None, embedding_model=None): ...
    async def delete_file(self, *, parsed_directory_file_id, export_config_id): ...
    async def delete_files(self, *, parsed_directory_file_ids, export_config_id): ...
    async def list_snapshots(self, *, export_config_id): ...

你可以把你的编码智能体指向这个仓库当作现成范例、复制一个你喜欢的 sink 作为起点，或针对你选的存储实现该协议。

注意事项

• 检索、对话与文件操作（client.retrieval、client.chat）查询的是托管向量库。设为 vector_target="DISABLED" 时，这些端点无内容可查 —— 检索要由你自己的数据库来服务。
• 源目录里删除文件时，你负责清理自己数据库中的相应记录。见只导出发生变化的部分。