How to Attach Custom Metadata to Chunked Documents in a Folder-Monitoring RAG Pipeline ?

[Tejanikhil-MSR]

We're currently building a RAG pipeline using Pathway's real-time streaming capabilities. As of now, our pipeline monitors a folder containing text files and processes them accordingly. We would like to associate custom metadata with the documents before or during the chunking process. However, we haven't been able to find a way to attach this custom metadata to the chunked documents.

Is there a recommended way to add or persist custom metadata for each document while using the folder monitoring feature?

Any guidance or suggestions would be greatly appreciated.
Thanks in advance!

[szymondudycz]

Hey, connectors return the metadata in the _metadata column, stored as dict. You can then write an udf to change the _metadata column whatever you set it to will be used in the rest of the pipeline. An example of adding metadata to documents used later in the DocumentStore would be:

@pw.udf
def augment_metadata(metadata: pw.Json) -> dict:
    metadata = metadata.as_dict()
    metadata["filename"] = metadata["path"].split("/")[-1]
    return metadata

documents = pw.io.fs.read("data", format="binary", with_metadata=True)
documents = documents.with_columns(_metadata=augment_metadata(pw.this._metadata))
doc_store = DocumentStore(documents, ...)

where you can change augment_metadata to something better suited for your use case

[Tejanikhil-MSR]

Thank you very much for the quick response @szymondudycz, it was really helpful!

[Tejanikhil-MSR]

The pipeline that we want to build should be capable of processing different types of documents — including plain text files, PDFs, and JSON files. We’re looking to extend our current setup, which only handles text files, to support these additional formats.

Here's what we’ve observed so far:

• Plain text files: These can be ingested directly without any explicit schema using format="plaintext".
• PDF files: These don’t require a schema either but must be read using format="binary" and then parsed.
• JSON files: These require an explicitly defined schema to be processed.

Additionally, we’d like to preprocess these files differently based on their type (e.g., different parsing logic for PDFs vs. JSONs vs. text). However, we’re not quite sure how to structure or organize the data directories (or configuration) so that Pathway can read and dispatch processing logic appropriately based on file type.

Currently, we are using the following setup for plain text files:

data = pw.io.fs.read(DATA_DIR,
    format="plaintext",
    mode="streaming",
    with_metadata=True,
)

Could you please advise on how to extend this to handle multiple file types in the same pipeline? Specifically:

• How can we organize the data to enable type-aware reading?
• What’s the recommended way to route different formats through different preprocessing logic in Pathway?

Any guidance would be greatly appreciated!

[Tejanikhil-MSR]

we figured it out.

[zxqfd555]

Hey @Tejanikhil-MSR,
As far as I understand, your problem has been resolved. I am closing this issue. Please feel free to open a new one or to reopen this one if new problems arise.