Skip to content

Llama cloud

LlamaCloudIndex #

Bases: BaseManagedIndex

LlamaIndex Platform Index.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
class LlamaCloudIndex(BaseManagedIndex):
    """LlamaIndex Platform Index."""

    def __init__(
        self,
        name: str,
        nodes: Optional[List[BaseNode]] = None,
        transformations: Optional[List[TransformComponent]] = None,
        timeout: int = 60,
        project_name: str = DEFAULT_PROJECT_NAME,
        organization_id: Optional[str] = None,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        app_url: Optional[str] = None,
        show_progress: bool = False,
        callback_manager: Optional[CallbackManager] = None,
        httpx_client: Optional[httpx.Client] = None,
        async_httpx_client: Optional[httpx.AsyncClient] = None,
        **kwargs: Any,
    ) -> None:
        """Initialize the Platform Index."""
        self.name = name
        self.project_name = project_name
        self.organization_id = organization_id
        self.transformations = transformations or []

        if nodes is not None:
            # TODO: How to handle uploading nodes without running transforms on them?
            raise ValueError("LlamaCloudIndex does not support nodes on initialization")

        self._httpx_client = httpx_client
        self._async_httpx_client = async_httpx_client
        self._client = get_client(api_key, base_url, app_url, timeout, httpx_client)
        self._aclient = get_aclient(
            api_key, base_url, app_url, timeout, async_httpx_client
        )

        self._api_key = api_key
        self._base_url = base_url
        self._app_url = app_url
        self._timeout = timeout
        self._show_progress = show_progress
        self._service_context = None
        self._callback_manager = callback_manager or Settings.callback_manager

    def _wait_for_pipeline_ingestion(
        self,
        verbose: bool = False,
        raise_on_partial_success: bool = False,
    ) -> None:
        pipeline_id = self._get_pipeline_id()
        client = self._client

        if verbose:
            print("Syncing pipeline: ", end="")

        is_done = False
        while not is_done:
            status = client.pipelines.get_pipeline_status(
                pipeline_id=pipeline_id
            ).status
            if status == ManagedIngestionStatus.ERROR or (
                raise_on_partial_success
                and status == ManagedIngestionStatus.PARTIAL_SUCCESS
            ):
                raise ValueError(f"Pipeline ingestion failed for {pipeline_id}")
            elif status in [
                ManagedIngestionStatus.NOT_STARTED,
                ManagedIngestionStatus.IN_PROGRESS,
            ]:
                if verbose:
                    print(".", end="")
                time.sleep(0.5)
            else:
                is_done = True
                if verbose:
                    print("Done!")

    def _wait_for_documents_ingestion(
        self,
        doc_ids: List[str],
        verbose: bool = False,
        raise_on_error: bool = False,
    ) -> None:
        pipeline_id = self._get_pipeline_id()
        client = self._client
        if verbose:
            print("Loading data: ", end="")

        # wait until all documents are loaded
        pending_docs = set(doc_ids)
        while pending_docs:
            docs_to_remove = set()
            for doc in pending_docs:
                # we have to quote the doc id twice because it is used as a path parameter
                status = client.pipelines.get_pipeline_document_status(
                    pipeline_id=pipeline_id, document_id=quote_plus(quote_plus(doc))
                )
                if status in [
                    ManagedIngestionStatus.NOT_STARTED,
                    ManagedIngestionStatus.IN_PROGRESS,
                ]:
                    continue

                if status == ManagedIngestionStatus.ERROR:
                    if verbose:
                        print(f"Document ingestion failed for {doc}")
                    if raise_on_error:
                        raise ValueError(f"Document ingestion failed for {doc}")

                docs_to_remove.add(doc)

            pending_docs -= docs_to_remove

            if pending_docs:
                if verbose:
                    print(".", end="")
                time.sleep(0.5)

        if verbose:
            print("Done!")

        # we have to wait for pipeline ingestion because retrieval only works when
        # the pipeline status is success
        self._wait_for_pipeline_ingestion(verbose, raise_on_error)

    def _get_project_id(self) -> str:
        projects = self._client.projects.list_projects(
            organization_id=self.organization_id,
            project_name=self.project_name,
        )
        if len(projects) == 0:
            raise ValueError(
                f"Unknown project name {self.project_name}. Please confirm a "
                "managed project with this name exists."
            )
        elif len(projects) > 1:
            raise ValueError(
                f"Multiple projects found with name {self.project_name}. Please specify organization_id."
            )
        project = projects[0]

        if project.id is None:
            raise ValueError(f"No project found with name {self.project_name}")

        return project.id

    def _get_pipeline_id(self) -> str:
        project_id = self._get_project_id()
        pipelines = self._client.pipelines.search_pipelines(
            project_id=project_id,
            pipeline_name=self.name,
            pipeline_type=PipelineType.MANAGED.value,
        )
        if len(pipelines) == 0:
            raise ValueError(
                f"Unknown index name {self.name}. Please confirm a "
                "managed index with this name exists."
            )
        elif len(pipelines) > 1:
            raise ValueError(
                f"Multiple pipelines found with name {self.name} in project {self.project_name}"
            )
        pipeline = pipelines[0]

        if pipeline.id is None:
            raise ValueError(
                f"No pipeline found with name {self.name} in project {self.project_name}"
            )

        return pipeline.id

    @classmethod
    def from_documents(  # type: ignore
        cls: Type["LlamaCloudIndex"],
        documents: List[Document],
        name: str,
        transformations: Optional[List[TransformComponent]] = None,
        project_name: str = DEFAULT_PROJECT_NAME,
        organization_id: Optional[str] = None,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        app_url: Optional[str] = None,
        timeout: int = 60,
        verbose: bool = False,
        raise_on_error: bool = False,
        **kwargs: Any,
    ) -> "LlamaCloudIndex":
        """Build a LlamaCloud managed index from a sequence of documents."""
        app_url = app_url or os.environ.get("LLAMA_CLOUD_APP_URL", DEFAULT_APP_URL)
        client = get_client(api_key, base_url, app_url, timeout)

        pipeline_create = get_pipeline_create(
            name,
            client,
            PipelineType.MANAGED,
            project_name=project_name,
            transformations=transformations or default_transformations(),
            input_nodes=documents,
        )

        project = client.projects.upsert_project(
            organization_id=organization_id, request=ProjectCreate(name=project_name)
        )
        if project.id is None:
            raise ValueError(f"Failed to create/get project {project_name}")
        if verbose:
            print(f"Created project {project.id} with name {project.name}")

        pipeline = client.pipelines.upsert_pipeline(
            project_id=project.id, request=pipeline_create
        )
        if pipeline.id is None:
            raise ValueError(f"Failed to create/get pipeline {name}")
        if verbose:
            print(f"Created pipeline {pipeline.id} with name {pipeline.name}")

        index = cls(
            name,
            transformations=transformations,
            project_name=project_name,
            organization_id=project.organization_id,
            api_key=api_key,
            base_url=base_url,
            app_url=app_url,
            timeout=timeout,
            **kwargs,
        )

        # this kicks off document ingestion
        upserted_documents = client.pipelines.upsert_batch_pipeline_documents(
            pipeline_id=pipeline.id,
            request=[
                CloudDocumentCreate(
                    text=doc.text,
                    metadata=doc.metadata,
                    excluded_embed_metadata_keys=doc.excluded_embed_metadata_keys,
                    excluded_llm_metadata_keys=doc.excluded_llm_metadata_keys,
                    id=doc.id_,
                )
                for doc in documents
            ],
        )
        doc_ids = [doc.id for doc in upserted_documents]
        index._wait_for_documents_ingestion(
            doc_ids, verbose=verbose, raise_on_error=raise_on_error
        )

        print(f"Find your index at {app_url}/project/{project.id}/deploy/{pipeline.id}")

        return index

    def as_retriever(self, **kwargs: Any) -> BaseRetriever:
        """Return a Retriever for this managed index."""
        from llama_index.indices.managed.llama_cloud.retriever import (
            LlamaCloudRetriever,
        )

        similarity_top_k = kwargs.pop("similarity_top_k", None)
        dense_similarity_top_k = kwargs.pop("dense_similarity_top_k", None)
        if similarity_top_k is not None:
            dense_similarity_top_k = similarity_top_k

        return LlamaCloudRetriever(
            self.name,
            project_name=self.project_name,
            api_key=self._api_key,
            base_url=self._base_url,
            app_url=self._app_url,
            timeout=self._timeout,
            organization_id=self.organization_id,
            dense_similarity_top_k=dense_similarity_top_k,
            httpx_client=self._httpx_client,
            async_httpx_client=self._async_httpx_client,
            **kwargs,
        )

    def as_query_engine(self, **kwargs: Any) -> BaseQueryEngine:
        from llama_index.core.query_engine.retriever_query_engine import (
            RetrieverQueryEngine,
        )

        kwargs["retriever"] = self.as_retriever(**kwargs)
        return RetrieverQueryEngine.from_args(**kwargs)

    @property
    def ref_doc_info(self, batch_size: int = 100) -> Dict[str, RefDocInfo]:
        """Retrieve a dict mapping of ingested documents and their metadata. The nodes list is empty."""
        pipeline_id = self._get_pipeline_id()
        pipeline_documents: List[CloudDocument] = []
        skip = 0
        limit = batch_size
        while True:
            batch = self._client.pipelines.list_pipeline_documents(
                pipeline_id=pipeline_id,
                skip=skip,
                limit=limit,
            )
            if not batch:
                break
            pipeline_documents.extend(batch)
            skip += limit
        return {
            doc.id: RefDocInfo(metadata=doc.metadata, node_ids=[])
            for doc in pipeline_documents
        }

    def insert(
        self, document: Document, verbose: bool = False, **insert_kwargs: Any
    ) -> None:
        """Insert a document."""
        with self._callback_manager.as_trace("insert"):
            pipeline_id = self._get_pipeline_id()
            upserted_documents = self._client.pipelines.create_batch_pipeline_documents(
                pipeline_id=pipeline_id,
                request=[
                    CloudDocumentCreate(
                        text=document.text,
                        metadata=document.metadata,
                        excluded_embed_metadata_keys=document.excluded_embed_metadata_keys,
                        excluded_llm_metadata_keys=document.excluded_llm_metadata_keys,
                        id=document.id_,
                    )
                ],
            )
            upserted_document = upserted_documents[0]
            self._wait_for_documents_ingestion(
                [upserted_document.id], verbose=verbose, raise_on_error=True
            )

    def update_ref_doc(
        self, document: Document, verbose: bool = False, **update_kwargs: Any
    ) -> None:
        """Upserts a document and its corresponding nodes."""
        with self._callback_manager.as_trace("update"):
            pipeline_id = self._get_pipeline_id()
            upserted_documents = self._client.pipelines.upsert_batch_pipeline_documents(
                pipeline_id=pipeline_id,
                request=[
                    CloudDocumentCreate(
                        text=document.text,
                        metadata=document.metadata,
                        excluded_embed_metadata_keys=document.excluded_embed_metadata_keys,
                        excluded_llm_metadata_keys=document.excluded_llm_metadata_keys,
                        id=document.id_,
                    )
                ],
            )
            upserted_document = upserted_documents[0]
            self._wait_for_documents_ingestion(
                [upserted_document.id], verbose=verbose, raise_on_error=True
            )

    def refresh_ref_docs(
        self, documents: Sequence[Document], **update_kwargs: Any
    ) -> List[bool]:
        """Refresh an index with documents that have changed."""
        with self._callback_manager.as_trace("refresh"):
            pipeline_id = self._get_pipeline_id()
            upserted_documents = self._client.pipelines.upsert_batch_pipeline_documents(
                pipeline_id=pipeline_id,
                request=[
                    CloudDocumentCreate(
                        text=doc.text,
                        metadata=doc.metadata,
                        excluded_embed_metadata_keys=doc.excluded_embed_metadata_keys,
                        excluded_llm_metadata_keys=doc.excluded_llm_metadata_keys,
                        id=doc.id_,
                    )
                    for doc in documents
                ],
            )
            doc_ids = [doc.id for doc in upserted_documents]
            self._wait_for_documents_ingestion(
                doc_ids, verbose=True, raise_on_error=True
            )
            return [True] * len(doc_ids)

    def delete_ref_doc(
        self,
        ref_doc_id: str,
        delete_from_docstore: bool = False,
        verbose: bool = False,
        raise_if_not_found: bool = False,
        **delete_kwargs: Any,
    ) -> None:
        """Delete a document and its nodes by using ref_doc_id."""
        pipeline_id = self._get_pipeline_id()
        try:
            # we have to quote the ref_doc_id twice because it is used as a path parameter
            self._client.pipelines.delete_pipeline_document(
                pipeline_id=pipeline_id, document_id=quote_plus(quote_plus(ref_doc_id))
            )
        except ApiError as e:
            if e.status_code == 404 and not raise_if_not_found:
                logger.warning(f"ref_doc_id {ref_doc_id} not found, nothing deleted.")
            else:
                raise

        # we have to wait for the pipeline instead of the document, because the document is already deleted
        self._wait_for_pipeline_ingestion(
            verbose=verbose, raise_on_partial_success=False
        )

    # Nodes related methods (not implemented for LlamaCloudIndex)

    def _insert(self, nodes: Sequence[BaseNode], **insert_kwargs: Any) -> None:
        """Index-specific logic for inserting nodes to the index struct."""
        raise NotImplementedError("_insert not implemented for LlamaCloudIndex.")

    def build_index_from_nodes(self, nodes: Sequence[BaseNode]) -> None:
        """Build the index from nodes."""
        raise NotImplementedError(
            "build_index_from_nodes not implemented for LlamaCloudIndex."
        )

    def insert_nodes(self, nodes: Sequence[BaseNode], **insert_kwargs: Any) -> None:
        """Insert a set of nodes."""
        raise NotImplementedError("insert_nodes not implemented for LlamaCloudIndex.")

    def delete_nodes(
        self,
        node_ids: List[str],
        delete_from_docstore: bool = False,
        **delete_kwargs: Any,
    ) -> None:
        """Delete a set of nodes."""
        raise NotImplementedError("delete_nodes not implemented for LlamaCloudIndex.")

ref_doc_info property #

ref_doc_info: Dict[str, RefDocInfo]

Retrieve a dict mapping of ingested documents and their metadata. The nodes list is empty.

from_documents classmethod #

from_documents(documents: List[Document], name: str, transformations: Optional[List[TransformComponent]] = None, project_name: str = DEFAULT_PROJECT_NAME, organization_id: Optional[str] = None, api_key: Optional[str] = None, base_url: Optional[str] = None, app_url: Optional[str] = None, timeout: int = 60, verbose: bool = False, raise_on_error: bool = False, **kwargs: Any) -> LlamaCloudIndex

Build a LlamaCloud managed index from a sequence of documents.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
@classmethod
def from_documents(  # type: ignore
    cls: Type["LlamaCloudIndex"],
    documents: List[Document],
    name: str,
    transformations: Optional[List[TransformComponent]] = None,
    project_name: str = DEFAULT_PROJECT_NAME,
    organization_id: Optional[str] = None,
    api_key: Optional[str] = None,
    base_url: Optional[str] = None,
    app_url: Optional[str] = None,
    timeout: int = 60,
    verbose: bool = False,
    raise_on_error: bool = False,
    **kwargs: Any,
) -> "LlamaCloudIndex":
    """Build a LlamaCloud managed index from a sequence of documents."""
    app_url = app_url or os.environ.get("LLAMA_CLOUD_APP_URL", DEFAULT_APP_URL)
    client = get_client(api_key, base_url, app_url, timeout)

    pipeline_create = get_pipeline_create(
        name,
        client,
        PipelineType.MANAGED,
        project_name=project_name,
        transformations=transformations or default_transformations(),
        input_nodes=documents,
    )

    project = client.projects.upsert_project(
        organization_id=organization_id, request=ProjectCreate(name=project_name)
    )
    if project.id is None:
        raise ValueError(f"Failed to create/get project {project_name}")
    if verbose:
        print(f"Created project {project.id} with name {project.name}")

    pipeline = client.pipelines.upsert_pipeline(
        project_id=project.id, request=pipeline_create
    )
    if pipeline.id is None:
        raise ValueError(f"Failed to create/get pipeline {name}")
    if verbose:
        print(f"Created pipeline {pipeline.id} with name {pipeline.name}")

    index = cls(
        name,
        transformations=transformations,
        project_name=project_name,
        organization_id=project.organization_id,
        api_key=api_key,
        base_url=base_url,
        app_url=app_url,
        timeout=timeout,
        **kwargs,
    )

    # this kicks off document ingestion
    upserted_documents = client.pipelines.upsert_batch_pipeline_documents(
        pipeline_id=pipeline.id,
        request=[
            CloudDocumentCreate(
                text=doc.text,
                metadata=doc.metadata,
                excluded_embed_metadata_keys=doc.excluded_embed_metadata_keys,
                excluded_llm_metadata_keys=doc.excluded_llm_metadata_keys,
                id=doc.id_,
            )
            for doc in documents
        ],
    )
    doc_ids = [doc.id for doc in upserted_documents]
    index._wait_for_documents_ingestion(
        doc_ids, verbose=verbose, raise_on_error=raise_on_error
    )

    print(f"Find your index at {app_url}/project/{project.id}/deploy/{pipeline.id}")

    return index

as_retriever #

as_retriever(**kwargs: Any) -> BaseRetriever

Return a Retriever for this managed index.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def as_retriever(self, **kwargs: Any) -> BaseRetriever:
    """Return a Retriever for this managed index."""
    from llama_index.indices.managed.llama_cloud.retriever import (
        LlamaCloudRetriever,
    )

    similarity_top_k = kwargs.pop("similarity_top_k", None)
    dense_similarity_top_k = kwargs.pop("dense_similarity_top_k", None)
    if similarity_top_k is not None:
        dense_similarity_top_k = similarity_top_k

    return LlamaCloudRetriever(
        self.name,
        project_name=self.project_name,
        api_key=self._api_key,
        base_url=self._base_url,
        app_url=self._app_url,
        timeout=self._timeout,
        organization_id=self.organization_id,
        dense_similarity_top_k=dense_similarity_top_k,
        httpx_client=self._httpx_client,
        async_httpx_client=self._async_httpx_client,
        **kwargs,
    )

insert #

insert(document: Document, verbose: bool = False, **insert_kwargs: Any) -> None

Insert a document.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def insert(
    self, document: Document, verbose: bool = False, **insert_kwargs: Any
) -> None:
    """Insert a document."""
    with self._callback_manager.as_trace("insert"):
        pipeline_id = self._get_pipeline_id()
        upserted_documents = self._client.pipelines.create_batch_pipeline_documents(
            pipeline_id=pipeline_id,
            request=[
                CloudDocumentCreate(
                    text=document.text,
                    metadata=document.metadata,
                    excluded_embed_metadata_keys=document.excluded_embed_metadata_keys,
                    excluded_llm_metadata_keys=document.excluded_llm_metadata_keys,
                    id=document.id_,
                )
            ],
        )
        upserted_document = upserted_documents[0]
        self._wait_for_documents_ingestion(
            [upserted_document.id], verbose=verbose, raise_on_error=True
        )

update_ref_doc #

update_ref_doc(document: Document, verbose: bool = False, **update_kwargs: Any) -> None

Upserts a document and its corresponding nodes.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
def update_ref_doc(
    self, document: Document, verbose: bool = False, **update_kwargs: Any
) -> None:
    """Upserts a document and its corresponding nodes."""
    with self._callback_manager.as_trace("update"):
        pipeline_id = self._get_pipeline_id()
        upserted_documents = self._client.pipelines.upsert_batch_pipeline_documents(
            pipeline_id=pipeline_id,
            request=[
                CloudDocumentCreate(
                    text=document.text,
                    metadata=document.metadata,
                    excluded_embed_metadata_keys=document.excluded_embed_metadata_keys,
                    excluded_llm_metadata_keys=document.excluded_llm_metadata_keys,
                    id=document.id_,
                )
            ],
        )
        upserted_document = upserted_documents[0]
        self._wait_for_documents_ingestion(
            [upserted_document.id], verbose=verbose, raise_on_error=True
        )

refresh_ref_docs #

refresh_ref_docs(documents: Sequence[Document], **update_kwargs: Any) -> List[bool]

Refresh an index with documents that have changed.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
def refresh_ref_docs(
    self, documents: Sequence[Document], **update_kwargs: Any
) -> List[bool]:
    """Refresh an index with documents that have changed."""
    with self._callback_manager.as_trace("refresh"):
        pipeline_id = self._get_pipeline_id()
        upserted_documents = self._client.pipelines.upsert_batch_pipeline_documents(
            pipeline_id=pipeline_id,
            request=[
                CloudDocumentCreate(
                    text=doc.text,
                    metadata=doc.metadata,
                    excluded_embed_metadata_keys=doc.excluded_embed_metadata_keys,
                    excluded_llm_metadata_keys=doc.excluded_llm_metadata_keys,
                    id=doc.id_,
                )
                for doc in documents
            ],
        )
        doc_ids = [doc.id for doc in upserted_documents]
        self._wait_for_documents_ingestion(
            doc_ids, verbose=True, raise_on_error=True
        )
        return [True] * len(doc_ids)

delete_ref_doc #

delete_ref_doc(ref_doc_id: str, delete_from_docstore: bool = False, verbose: bool = False, raise_if_not_found: bool = False, **delete_kwargs: Any) -> None

Delete a document and its nodes by using ref_doc_id.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
def delete_ref_doc(
    self,
    ref_doc_id: str,
    delete_from_docstore: bool = False,
    verbose: bool = False,
    raise_if_not_found: bool = False,
    **delete_kwargs: Any,
) -> None:
    """Delete a document and its nodes by using ref_doc_id."""
    pipeline_id = self._get_pipeline_id()
    try:
        # we have to quote the ref_doc_id twice because it is used as a path parameter
        self._client.pipelines.delete_pipeline_document(
            pipeline_id=pipeline_id, document_id=quote_plus(quote_plus(ref_doc_id))
        )
    except ApiError as e:
        if e.status_code == 404 and not raise_if_not_found:
            logger.warning(f"ref_doc_id {ref_doc_id} not found, nothing deleted.")
        else:
            raise

    # we have to wait for the pipeline instead of the document, because the document is already deleted
    self._wait_for_pipeline_ingestion(
        verbose=verbose, raise_on_partial_success=False
    )

build_index_from_nodes #

build_index_from_nodes(nodes: Sequence[BaseNode]) -> None

Build the index from nodes.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
464
465
466
467
468
def build_index_from_nodes(self, nodes: Sequence[BaseNode]) -> None:
    """Build the index from nodes."""
    raise NotImplementedError(
        "build_index_from_nodes not implemented for LlamaCloudIndex."
    )

insert_nodes #

insert_nodes(nodes: Sequence[BaseNode], **insert_kwargs: Any) -> None

Insert a set of nodes.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
470
471
472
def insert_nodes(self, nodes: Sequence[BaseNode], **insert_kwargs: Any) -> None:
    """Insert a set of nodes."""
    raise NotImplementedError("insert_nodes not implemented for LlamaCloudIndex.")

delete_nodes #

delete_nodes(node_ids: List[str], delete_from_docstore: bool = False, **delete_kwargs: Any) -> None

Delete a set of nodes.

Source code in llama-index-integrations/indices/llama-index-indices-managed-llama-cloud/llama_index/indices/managed/llama_cloud/base.py
474
475
476
477
478
479
480
481
def delete_nodes(
    self,
    node_ids: List[str],
    delete_from_docstore: bool = False,
    **delete_kwargs: Any,
) -> None:
    """Delete a set of nodes."""
    raise NotImplementedError("delete_nodes not implemented for LlamaCloudIndex.")