How do I add custom metadata when saving to Pinecone

[dividor]

I would like to add custom metadata to chunks when saved to pinecone with Pipeline.from_configs.

Following the 'Custom meta data extraction ...' notebook on this page, I added a step which add my custom metadata to the files in the work subfolder 'embed', however, on trying to Save to pinecone they are lost, the provided example doesn't seem to apply.

Pipeline.from_configs(
        context=ProcessorConfig(
            verbose=True, tqdm=True, num_processes=5, work_dir=WORK_DIR
        ),
        indexer_config=LocalIndexerConfig(input_path=input_dir),
        downloader_config=LocalDownloaderConfig(),
        source_connection_config=LocalConnectionConfig(),
        partitioner_config=PartitionerConfig(
            partition_by_api=True,
            api_key=os.getenv("UNSTRUCTURED_API_KEY"),
            partition_endpoint=os.getenv("UNSTRUCTURED_API_URL"),
            strategy="hi_res",
            additional_partition_args={
                "split_pdf_page": True,
                "split_pdf_allow_failed": True,
                "split_pdf_concurrency_level": 15,
                "reprocess": True,
                "extract_image_block_types": ["Image"],
                "include_metadata": True
            },
            reprocess=True
        ),
        embedder_config=embedder_config,
        #https://docs.unstructured.io/api-reference/ingest/ingest-configuration/chunking-configuration
        chunker_config=ChunkerConfig(
            chunking_strategy="by_title",
            #chunking_strategy="by_similarity",
            max_characters = MAX_CHARACTERS,
            chunk_overlap = CHUNK_OVERLAP,
            #combine_text_under_n_chars= COMBINE_TEXT_UNDER_N_CHARS
        ),
        uploader_config=LocalUploaderConfig(output_dir=INTERIM_RESULTS)
    ).run()

    # Add in custom metadata from PIA ingestion
    add_custom_metadata()

    Pipeline.from_configs(
        context=ProcessorConfig(
            verbose=True, tqdm=True, num_processes=5, work_dir=WORK_DIR
        ),
        indexer_config=LocalIndexerConfig(input_path=input_dir),
        downloader_config=LocalDownloaderConfig(),
        source_connection_config=LocalConnectionConfig(),
        partitioner_config=PartitionerConfig(
            partition_by_api=True,
            api_key=os.getenv("UNSTRUCTURED_API_KEY"),
            partition_endpoint=os.getenv("UNSTRUCTURED_API_URL"),
            strategy="hi_res",
            additional_partition_args={
                "split_pdf_page": True,
                "split_pdf_allow_failed": True,
                "split_pdf_concurrency_level": 15,
                "reprocess": True,
                "extract_image_block_types": ["Image"],
                "include_metadata": True
            },
            reprocess=True
        ),
        #https://docs.unstructured.io/api-reference/ingest/ingest-configuration/chunking-configuration
        chunker_config=ChunkerConfig(
            chunking_strategy="by_title",
            #chunking_strategy="by_similarity",
            max_characters = MAX_CHARACTERS,
            chunk_overlap = CHUNK_OVERLAP,
            #combine_text_under_n_chars= COMBINE_TEXT_UNDER_N_CHARS
        ),
        embedder_config=embedder_config,
        destination_connection_config=PineconeConnectionConfig(
            access_config=PineconeAccessConfig(
                api_key=PINECONE_API_KEY
            ),
            index_name=index_name
        ),
        stager_config=PineconeUploadStagerConfig(
            include_metadata=True
        ),
        uploader_config=PineconeUploaderConfig(
            include_metadata=True
        )
    ).run()

Where add_custom_metadata() injects the custom fields into documents there, which works nicely.

What is the recommended way to add custom metadata for saving to Pinecone?

[SantoshKumarRavi]

Hi i am facing the same issue.
is there any one know how to add custom fields in the response of final data before storing to mongodb ?

[SantoshKumarRavi]

i need to add our user details before storing for isolation

[rbiseck3]

The easiest approach for this would be to extend the existing stager and hook in your custom code for enrichment:

from unstructured_ingest.v2.processes.connectors.pinecone import PineconeUploadStager
from dataclasses import dataclass
from unstructured_ingest.v2.interfaces import FileData
from typing import Any
from pathlib import Path


@dataclass
class CustomPineconeUploadStager(PineconeUploadStager):
    def run(
        self,
        file_data: FileData,
        elements_filepath: Path,
        output_dir: Path,
        output_filename: str,
        **kwargs: Any,
    ) -> Path:
        # Custom code goes here
        return super().run(
            file_data=file_data,
            elements_filepath=elements_filepath,
            output_dir=output_dir,
            output_filename=output_filename,
            **kwargs,
        )

Then you would have to construct the pipeline directly rather than using the from_config utility to pass in this custom stager.