迁移指南：Index v1 到 v2

只有在你已有 Index v1 流水线时才需要这一页。 全新开始的话请用快速上手指南。

Index v1（pipeline、data source、data sink）已弃用。Index v2 用更简单的模型：把文件放进一个目录，在该目录上创建一个索引，然后用检索与对话 API 查询它。

本页说明哪些概念可以平移、哪些变了、v2 尚不支持什么，以及如何迁移已有的 pipeline。

迁移之前

Index v2 需要 Starter、Pro 或 Enterprise 套餐。
Python SDK：pip install "llama-cloud>=2.18"。TypeScript SDK：npm install @llamaindex/llama-cloud。
Index v2 的 SDK 方法位于顶层，即 client.indexes、client.retrieval 与 client.chat，从 Python 与 TypeScript 2.18.0、Go 与 Java 1.9.0 起可用。目录仍在 client.beta.directories 下。旧的 client.beta.indexes、client.beta.retrieval 与 client.beta.chat 仍然可用但已弃用。
v1 与 v2 不共享存储。迁移意味着把文件加入 v2，而 v2 会重新解析、切块与嵌入它们，因此要为这部分 parse credits 做预算。

概念对照

Index v1	Index v2	备注
Pipeline	Index	一个源目录对应一个索引。
Data source（连接器或文件上传）	Directory	把文件上传进目录，或从已连接文件夹（SharePoint、Google Drive、S3）同步进去。
Data sink（Pinecone、Qdrant、Postgres 等）	托管向量库，或自定义导出	v2 写入由 LlamaCloud 管理的向量库。要把数据留在自己的数据库里，就导出解析结果自行装载。见用自定义导出替代 data sink。
data source 上的 `sync_interval`	`POST /api/v1/indexes/{index_id}/sync`	由你触发同步。见定时同步索引。
`transform_config` / `embedding_config`	无	切块与嵌入由 LlamaCloud 管理，两者都不可配置。
`llama_parse_parameters`	通过 `products` 传入的已保存 Parse 配置	默认为 `cost_effective` 档。
Pipeline 文件的 `custom_metadata`	目录文件的 `metadata`	查询时可用 `custom_filters` 过滤。
`pipelines.retrieve`	`retrieval.retrieve`	混合检索，默认开启重排。
检索模式 `files_via_metadata` / `files_via_content`	`retrieval.find` / `grep` / `read`	索引内的文件操作。
Pipeline 对话	`chat` 会话	流式对话智能体，一个会话最多 10 个索引。

功能对齐

图例：✅ 支持 · 🔄 支持但方式不同 · ❌ 不支持

摄取侧的功能对齐

能力	v1	v2
通过 API 上传文件	✅	✅ `client.beta.directories.files.upload` / `.add`
SharePoint	✅	🔄 已连接文件夹，在 Web 应用里配置
Google Drive	✅	🔄 已连接文件夹，在 Web 应用里配置
Amazon S3	✅	🔄 已连接文件夹，在 Web 应用里配置
OneDrive	✅	❌
Box	✅	❌
Confluence	✅	❌
Jira	✅	❌
Azure Blob Storage	✅	❌
插入原始文本文档（`pipelines.documents.create`）	✅	❌ 请改为上传文件
逐文件自定义元数据	✅	✅ 目录文件上的 `metadata`

对于 v2 不支持的源头，用你自己的代码把文件取回来（例如开源的 LlamaIndex reader 或提供方的 SDK），再上传到目录。把这个脚本定时跑一遍，就能顶替 v1 连接器的职责。

处理侧的功能对齐

能力	v1	v2
解析设置	✅ `llama_parse_parameters`	🔄 通过 `products` 引用的已保存 Parse 配置
可配置的切块	✅ `transform_config`	❌ 固定：按句切分、每块 1024 token、重叠 200 token
自选嵌入模型 / 自带密钥	✅ OpenAI、Azure、Bedrock、Cohere、Gemini、HuggingFace	❌ 由 LlamaCloud 管理
同步时的变更检测	✅	✅ 未变化的文件会被跳过；仅元数据变化不会重新解析文件

存储侧的功能对齐

能力	v1	v2
托管向量库	✅	✅ 默认
用自己的向量数据库作为 data sink	✅ Pinecone、Qdrant、Milvus、Postgres、MongoDB Atlas、Azure AI Search、AstraDB	❌ 请改用自定义导出

检索侧的功能对齐

能力	v1	v2
稠密（向量）检索	✅	✅
混合（向量 + 关键词）	✅	✅ 默认，且权重可调
重排	✅	✅ 默认开启
元数据过滤	✅ `MetadataFilters`	🔄 `custom_filters`（`eq`、`ne`、`gt`、`lt`、`gte`、`lte`、`in`、`nin`）
按文件过滤	✅	✅ 基于 `parsed_directory_file_id` 的 `static_filters`
在索引内查找、grep 与读取文件	❌	✅ 新增
随结果返回页面截图	✅ `retrieve_page_screenshot_nodes`	🔄 建索引时传 `store_attachments=["screenshots"]`。每条结果会列出其覆盖页的截图，截图需从附件端点取回。检索本身仍是纯文本。
图形级图片检索（`retrieve_page_figure_nodes`）	✅	❌ `store_attachments=["items"]` 会存下带边界框的结构化条目，但图形不会作为独立结果返回
复合检索（一次调用查多个索引）	✅ `client.retrievers`	❌ 请分别查询每个索引再合并结果，或用一个跨多索引的对话会话
自动路由检索模式	✅	❌
对话智能体	🔄 基础	✅ 带会话的流式智能体，最多 10 个索引；当索引存了截图时它能查看页面截图
`LlamaCloudIndex` 框架集成	✅	❌ 请直接调用检索 API
块级列举（`documents/{id}/chunks`）	✅	❌

端点对照

Index v2 位于 /api/v1/indexes、/api/v1/retrieval 与 /api/v1/chat 下的 REST 路径，对应 SDK 的顶层资源 client.indexes、client.retrieval 与 client.chat。/api/v1/beta/directories 下的目录路径对应 client.beta.directories。

v1 路由	v2 路由	v2 SDK 方法
`POST /api/v1/data-sources`	`POST /api/v1/beta/directories`	`client.beta.directories.create`
`POST /api/v1/pipelines/{id}/files`	`POST /api/v1/beta/directories/{id}/files`（已有 `file_id`）或 `.../files/upload`（multipart）	`client.beta.directories.files.add` / `.upload`
`GET /api/v1/pipelines/{id}/files`	`GET /api/v1/beta/directories/{id}/files`	`client.beta.directories.files.list`
`DELETE /api/v1/pipelines/{id}/files/{file_id}`	`DELETE /api/v1/beta/directories/{id}/files/{file_id}`	`client.beta.directories.files.delete`
`POST /api/v1/data-sinks`	无	使用自定义导出
`PUT /api/v1/pipelines`（创建 / upsert）	`POST /api/v1/indexes`	`client.indexes.create`
`GET /api/v1/pipelines`	`GET /api/v1/indexes`	`client.indexes.list`
`GET /api/v1/pipelines/{id}` 与 `/status`	`GET /api/v1/indexes/{id}`	`client.indexes.get`（状态在 `metadata["status"]`）
`DELETE /api/v1/pipelines/{id}`	`DELETE /api/v1/indexes/{id}`	`client.indexes.delete`
`POST /api/v1/pipelines/{id}/sync`	`POST /api/v1/indexes/{id}/sync`	`client.indexes.sync`
`POST /api/v1/pipelines/{id}/retrieve`	`POST /api/v1/retrieval/retrieve`（`index_id` 放在请求体里）	`client.retrieval.retrieve`
`retrieve_page_screenshot_nodes`（在 `pipelines/{id}/retrieve` 上）	`GET /api/v1/beta/attachments/{attachment_name}?source_id=...`（预签名 URL）与 `GET /api/v1/beta/attachments?source_id=...`（列举）	尚无封装；请调用 REST 端点
无	`POST /api/v1/retrieval/files/find` / `grep` / `read`	`client.retrieval.find` / `grep` / `read`
`POST /api/v1/pipelines/{id}/chat`	`POST /api/v1/chat`，随后 `POST /api/v1/chat/{session_id}/messages/stream`（SSE）	`client.chat.create` / `.stream`
`POST /api/v1/retrievers/{id}/retrieve`	无	不支持

删除索引会移除它的同步与导出配置，源目录及其文件保留。

逐步迁移一条 pipeline

下面的示例使用异步 Python SDK。TypeScript SDK 方法名相同，只是改为传一个 options 对象。

第 1 步：创建目录

from llama_cloud import AsyncLlamaCloud

client = AsyncLlamaCloud()  # reads LLAMA_CLOUD_API_KEY

directory = await client.beta.directories.create(
    name="product-docs",
    description="Migrated from an Index v1 pipeline",
)

SharePoint、Google Drive 或 S3 源头： 请改为在 LlamaCloud Web 应用里创建已连接文件夹，入口是 Files 页，或建索引时的 Connect a folder。LlamaCloud 会把该目录的内容同步进来。公开 API 暂时还不能配置已连接文件夹。

第 2 步：把文件加入目录

想在迁移自己的文件之前先试下面的片段，可以把美国财政部 FY2024 财报的执行摘要下载到 docs/：

终端窗口

curl -fL --create-dirs -o docs/report.pdf https://fiscal.treasury.gov/system/files/files/reports-statements/financial-report/2024/executive-summary-2024.pdf

直接上传文件，并把你在 v1 里设置的 custom_metadata 一并带上：

from pathlib import Path

for path in Path("./docs").glob("*.pdf"):
    await client.beta.directories.files.upload(
        directory.id,
        upload_file=path,
        display_name=path.name,
        unique_id=str(path),  # stable ID, so a re-upload replaces the file instead of duplicating it
        metadata='{"department": "legal", "year": 2026}',  # JSON string on upload
    )

或者加入一个你用 client.files.create 已上传过的文件：

file_obj = await client.files.create(file=open("docs/report.pdf", "rb"), purpose="user_data")
await client.beta.directories.files.add(
    directory.id,
    file_id=file_obj.id,
    metadata={"department": "legal"},
)

第 3 步：保存一份 Parse 配置（可选）

索引默认使用 Parse 的 cost_effective 档。要替代 v1 的 llama_parse_parameters 并使用其他设置，请先保存一份 Parse 配置：

parse_config = await client.configurations.create(
    name="Index parse config",
    parameters={"product_type": "parse_v2", "version": "latest", "tier": "agentic"},
)

第 4 步：创建索引

index = await client.indexes.create(
    source_directory_id=directory.id,
    name="product-docs",
    # Omit `products` to use the default Parse configuration.
    products=[{"product_type": "parse", "product_config_id": parse_config.id}],
    # Store per-page screenshots, the v2 replacement for v1 page screenshot nodes.
    # Add "items" to also store structured items with bounding boxes.
    store_attachments=["screenshots"],
)

创建索引时就设好 store_attachments。索引创建后不可更新，所以事后想加截图，只能对同一目录再建一个新索引。

创建索引会启动它的首次同步：把目录里的每个文件解析、切块、嵌入并存入。

第 5 步：等待索引就绪

import asyncio

while True:
    idx = await client.indexes.get(index.id)
    status = idx.metadata.get("status")
    if status == "ready":
        break
    if status == "failed":
        raise RuntimeError(idx.metadata.get("error_message"))
    await asyncio.sleep(10)

只有 ready 与 failed 是稳定的状态值，其他取值一律视为仍在进行中。

第 6 步：更新检索调用

v1 的检索调用：

results = await client.pipelines.retrieve(
    pipeline_id=PIPELINE_ID,
    query="What is the refund policy?",
    dense_similarity_top_k=10,
    sparse_similarity_top_k=10,
    alpha=0.5,
    enable_reranking=True,
    rerank_top_n=5,
    search_filters={"filters": [{"key": "department", "value": "legal", "operator": "=="}]},
)
for node in results.retrieval_nodes:
    print(node.score, node.node.text)

在 v2 里同样的调用：

results = await client.retrieval.retrieve(
    index_id=index.id,
    query="What is the refund policy?",
    top_k=5,
    num_candidates=100,
    vector_pipeline_weight=0.5,
    full_text_pipeline_weight=0.5,
    rerank={"enabled": True, "top_n": 5},
    custom_filters={"department": {"operator": "eq", "value": "legal"}},
)
for r in results.results:
    print(r.score, r.rerank_score, r.content)
    print(r.static_fields.parsed_directory_file_id, r.static_fields.page_range_start)

v1 的检索参数与 v2 的对应关系如下：

v1	v2
`pipeline_id`	`index_id`
`dense_similarity_top_k` / `sparse_similarity_top_k`	`num_candidates`（每种检索的候选池）与 `top_k`（返回的结果数，上限 500）
`alpha`	`vector_pipeline_weight` / `full_text_pipeline_weight`（默认 0.5 / 0.5）
`enable_reranking`	`rerank.enabled`（默认 `true`）
`rerank_top_n`	`rerank.top_n`
`dense_similarity_cutoff`	`score_threshold`
`search_filters`（`{"filters": [...]}`）	`custom_filters`，以元数据字段为键
按文件 ID 过滤	基于 `parsed_directory_file_id` 的 `static_filters`
`retrieve_page_screenshot_nodes=True`	建索引时传 `store_attachments=["screenshots"]`，随后取 `results.results[i].static_fields.attachments`
`results.retrieval_nodes[i].node.text`	`results.results[i].content`

v2 的全部参数见检索。

第 7 步：为检索结果取回页面截图

当索引存了截图时，每条结果的 static_fields.attachments 会为该块覆盖的每一页（从 page_range_start 到 page_range_end）各列一个条目：

{"type": "screenshot", "attachment_name": "screenshots/page_7.jpg", "source_id": "dfl-..."}

GET /api/v1/beta/attachments/{attachment_name}?source_id="source_id" 会返回该图片的预签名 URL。SDK 还没封装这个端点，因此请用 HTTP 调用：

import os
import httpx

API_BASE = "https://api.cloud.llamaindex.ai"  # EU: https://api.cloud.eu.llamaindex.ai
headers = {"Authorization": f"Bearer {os.environ['LLAMA_CLOUD_API_KEY']}"}

async with httpx.AsyncClient(headers=headers) as http:
    for r in results.results:
        for att in r.static_fields.attachments or []:
            if att.type != "screenshot":
                continue
            resp = await http.get(
                f"{API_BASE}/api/v1/beta/attachments/{att.attachment_name}",
                params={"source_id": att.source_id},
            )
            resp.raise_for_status()
            image_url = resp.json()["url"]  # presigned URL to the page JPEG

GET /api/v1/beta/attachments?source_id=... 会列举某个文件存下的全部附件。

第 8 步：替代文件级检索模式

v1 的 files_via_metadata 与 files_via_content 模式对应到 v2 的文件操作：

found = await client.retrieval.find(index_id=index.id, file_name_contains="quarterly")
file_id = found.items[0].file_id

matches = await client.retrieval.grep(
    index_id=index.id, file_id=file_id, pattern="revenue|profit", context_chars=100,
)
text = await client.retrieval.read(
    index_id=index.id, file_id=file_id, offset=0, max_length=5000,
)

第 9 步：验证并切换

在切流量之前，用同一批评估查询分别跑一遍 v1 pipeline 与 v2 索引。v2 的切块与嵌入方式不同，因此得分与排序都会变化。对 v2 结果满意后，再删除 v1 的 pipeline。

用自定义导出替代 data sink

Index v2 不支持外部 data sink：Pinecone、Qdrant、Milvus、Postgres、MongoDB Atlas、Azure AI Search 与 AstraDB 都无法挂到索引上。要把数据留在自己的数据库里，请使用自定义导出。LlamaCloud 仍负责目录、解析与同步；切块、嵌入与写入由你完成。

创建一个只做解析的索引，传 vector_target="DISABLED"：

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

每次同步后，从索引的输出目录下载解析结果。源目录存放你的原始文件；索引会为每个源文件写出一个解析后的 JSON 载荷到 output_directory_id 所指的那个目录：

import httpx

idx = await client.indexes.get(index.id)

async with httpx.AsyncClient() as http:
    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)
        pages = (await http.get(presigned.url)).json()["parse"]["pages"]
        # f.id           -> parsed_directory_file_id, stable across syncs
        # f.metadata     -> the source file's metadata
        # f.updated_at   -> changes when the file is parsed again or its metadata changes

切块、嵌入并写入你的存储：
- 以 (f.id, idx.export_config_id) 为键 —— 即输出目录文件的 id 与索引的 export_config_id。插入新切块前先删掉该文件已有的记录，这样一个被重新解析的文件是替换旧切块而不是叠加。
- 只导出变化过的内容：列举输出目录时把上次导出的时刻作为 updated_at_on_or_after 传入。
- 删掉那些 parsed_directory_file_id 已不在输出目录里出现的记录。源文件被删时同步会移除其输出文件，但 LlamaCloud 无法替你清理自己的数据库。

index-v2-data-sinks 仓库为 MongoDB Atlas、PostgreSQL + pgvector、Qdrant、Pinecone、Turbopuffer 与 Azure AI Search 提供了可运行的导出器。它们基于一套共享的 Exporter 协议（export、delete_file、delete_files、list_snapshots）；挑一个离你的技术栈最近的复制即可。

设为 vector_target="DISABLED" 时没有托管向量库，因此 retrieval.*、chat 与文件操作都无内容可查 —— 检索要由你自己的数据库来服务。完整走查见自定义向量库。

定时同步索引

v1 会按 sync_interval（6、12 或 24 小时）设定的计划同步 data source。Index v2 没有内置计划：索引在创建时同步一次，此后每次由你调用 sync。

一次索引同步做了什么

把源目录与索引上次处理过的内容做比对。
解析新增与变化的文件，按内容指纹与 source_modified_at 跳过未变化的文件。仅元数据变化只会更新元数据，不会重新解析文件。
对变化的文件切块、嵌入并存入，并删除目录里已移除文件所对应的切块。

目录层面的变更（上传、删除、已连接文件夹的更新）只在下一次同步时才会到达索引。

安全地触发索引同步

每个索引一次只运行一个同步：同步进行中再次调用会返回 409 Conflict，短时间内连续调用 sync 会返回 429 Too Many Requests。请在触发同步前先读取 last_synced_at，以免把上一次同步的 ready 状态误认成这一次的：

import asyncio
from llama_cloud import ConflictError, RateLimitError

async def sync_and_wait(client, index_id, timeout_s=3600):
    before = (await client.indexes.get(index_id)).last_synced_at
    try:
        await client.indexes.sync(index_id)
    except ConflictError:
        pass  # a sync is already running; wait for it instead
    except RateLimitError:
        await asyncio.sleep(60)
        await client.indexes.sync(index_id)

    delay, waited = 2, 0
    while waited < timeout_s:
        idx = await client.indexes.get(index_id)
        status = idx.metadata.get("status")
        if status == "failed":
            raise RuntimeError(idx.metadata.get("error_message"))
        if status == "ready" and idx.last_synced_at != before:
            return idx
        await asyncio.sleep(delay)
        waited += delay
        delay = min(delay * 2, 30)
    raise TimeoutError(f"index {index_id} did not finish syncing")

各 SDK 里的同步端点见同步。

用定时任务替代 sync_interval

在你自己掌控的计划里运行 sync_and_wait，例如 cron、Kubernetes CronJob、Airflow 或定时的 GitHub Actions 工作流。每天一次或每 12 小时一次的节奏，就与多数 v1 pipeline 原本用的相当。

已连接文件夹（SharePoint、Google Drive、S3）： LlamaCloud 会周期性从源头刷新已连接文件夹，但那次刷新不会同步索引。请保留定时的 indexes.sync 调用，新增与变化的文件才能进入索引。
v2 不支持的源头： 让同一个定时任务从源头拉取变化的文件，用同样的 unique_id 上传到目录以替换旧版本，删除已不存在的文件，然后调用 sync。

迁移检查清单

[ ] 列出你的 v1 pipeline 及其 data source、data sink 与同步间隔。
[ ] 逐个对照摄取侧的功能对齐表检查 data source。除 SharePoint、Google Drive、S3 与文件上传之外的，都要准备一个上传脚本。
[ ] 对每个写入自有向量库的 pipeline，规划好自定义导出。
[ ] 若你用到页面截图检索，创建 v2 索引时就要传 store_attachments=["screenshots"]。它事后无法补加。
[ ] 若你依赖复合检索、图形级图片检索、自定义切块或指定嵌入模型，请先确认 v2 的行为能满足你，再动手迁移。
[ ] 创建目录并加入文件，把 custom_metadata 一并带过去。
[ ] 创建索引（需要时带上已保存的 Parse 配置），并等待 ready。
[ ] 按参数对照改写检索调用。
[ ] 用定时的 indexes.sync 调用替代 sync_interval。
[ ] 用你的评估查询比较 v1 与 v2 的检索质量。
[ ] 把流量切到 v2，然后删除 v1 的 pipeline。

索引

数据加载

查询

存储

成本分析

应用

聊天机器人

问答

结构化数据

智能体

Chat Engines

Query Engine

连接器

文档与节点

摄取流水线

节点解析器

大语言模型

提示词

回调

节点后处理器

响应合成器

检索器

路由

结构化输出

现有数据

Presets and Modes

能力

起步

能力

Examples

Azure Deployment

Databases and Queues

LLM Integrations