How can I save only part of the metadata in the node, but save all of the metadata in docstore?

[johnhu33]

How can I save only part of the metadata in the node, but save all of the metadata in docstore?

[dosubot[bot]]

To save only part of the metadata in the node but save all of the metadata in the docstore, you can follow this approach:

1. Store Partial Metadata in Node: Save the node with only the necessary metadata that fits within the constraints (e.g., 64KiB property limit).
2. Store Full Metadata in Docstore: Save the full metadata in the docstore, ensuring that each node is stored as a separate "Item" with its corresponding ref_doc_id.

Here is a simplified example of how you might implement this:

from llama_index.legacy.schema import BaseNode
from llama_index.storage.kvstore.azure import AzureKVStore
from llama_index.storage.kvstore.azure.base import ServiceMode
import json

# Initialize the AzureKVStore
kv_store = AzureKVStore(service_mode=ServiceMode.STORAGE)

# Function to save node with partial metadata
def save_node_with_partial_metadata(node: BaseNode, partial_metadata: dict):
    # Save the node with partial metadata
    node.metadata = partial_metadata
    kv_store.save_node(node)

# Function to save full metadata in docstore
def save_full_metadata_in_docstore(node_id: str, full_metadata: dict):
    # Save the full metadata in the docstore
    kv_store.save_item(node_id, full_metadata)

# Example usage
node = BaseNode(id="node1", data="some data")
partial_metadata = {"key1": "value1"}  # Partial metadata
full_metadata = {"key1": "value1", "key2": "value2", "key3": "value3"}  # Full metadata

save_node_with_partial_metadata(node, partial_metadata)
save_full_metadata_in_docstore(node.id, full_metadata)

Additionally, you can use the utility functions provided in llama_index.core.vector_stores.utils to handle the conversion between node objects and metadata dictionaries. For example, node_to_metadata_dict can be used to convert a node to a metadata dictionary, and metadata_dict_to_node can be used to convert a metadata dictionary back to a node.

Here is how you can use these utility functions:

from llama_index.core.vector_stores.utils import node_to_metadata_dict, metadata_dict_to_node

# Convert node to metadata dictionary
metadata_dict = node_to_metadata_dict(node, remove_text=True)

# Convert metadata dictionary back to node
restored_node = metadata_dict_to_node(metadata_dict)

By using these utility functions, you can ensure that the metadata is correctly handled and stored in both the node and the docstore ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

To save only part of the metadata in the node but save all of the metadata in an existing Redis vector store, you can use the node_to_metadata_dict function with the flat_metadata parameter set to True. This will ensure that the metadata is validated and stored in a flat dictionary format, which is suitable for Redis.

Here is an example of how you can achieve this:

from llama_index.legacy.schema import TextNode
from llama_index.legacy.vector_stores.utils import node_to_metadata_dict
from redis import Redis

# Create a sample node
node = TextNode(
    id="node1",
    content="This is a sample text node.",
    metadata={"author": "John Doe", "length": 100},
    ref_doc_id="doc1"
)

# Convert node to metadata dict, saving only part of the metadata
partial_metadata = {key: node.metadata[key] for key in ['author']}  # Replace 'author' with the actual keys you want to keep
node.metadata = partial_metadata

# Save the full metadata to Redis
full_metadata_dict = node_to_metadata_dict(node, flat_metadata=True)
redis_client = Redis.from_url("redis://localhost:6379")
redis_client.hmset("node1_metadata", full_metadata_dict)

In this example, the node_to_metadata_dict function is used to convert the node to a metadata dictionary, ensuring that the metadata is flat. The resulting dictionary is then saved to Redis using the hmset method. The function also ensures that the metadata dictionary is validated to be flat and contains only acceptable data types.

Additionally, if you need to restore from an existing index in Redis, you can use the following approach:

from llama_index.vector_stores.redis import RedisVectorStore
from llama_index.core import StorageContext, VectorStoreIndex
from redisvl.schema import IndexSchema

# Load the custom schema from a YAML file
custom_schema = IndexSchema.from_yaml("paul_graham.yaml")

# Create the Redis vector store with the existing schema
vector_store = RedisVectorStore(
    schema=custom_schema,
    redis_client=redis_client,
    overwrite=False
)

# Load the storage context
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# Restore the index from the existing vector store
index = VectorStoreIndex.from_vector_store(vector_store=vector_store)

This approach ensures that you can restore the index from an existing Redis vector store, maintaining the schema and metadata configurations ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[johnhu33]

@dosu please give me an example to save only part of the metadata in the node, but save all of the metadata in docstore using redis and IngestPipeline

[dosubot[bot]]

To save only part of the metadata in the node but save all of the metadata in the docstore using Redis and IngestPipeline, you can customize the ingestion pipeline to filter the metadata before saving it to the node. Here is an example of how you can achieve this:

1. Define a custom transformation to filter the metadata.
2. Add this transformation to the pipeline.

Here's how you can do it:

from llama_index.core import SimpleDirectoryReader
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.ingestion import (
    DocstoreStrategy,
    IngestionPipeline,
    IngestionCache,
)
from llama_index.storage.kvstore.redis import RedisKVStore as RedisCache
from llama_index.storage.docstore.redis import RedisDocumentStore
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.redis import RedisVectorStore
from redisvl.schema import IndexSchema

# Custom transformation to filter metadata
class MetadataFilter:
    def __call__(self, node):
        # Filter metadata to keep only specific keys
        filtered_metadata = {k: v for k, v in node.metadata.items() if k in ['key1', 'key2']}
        node.metadata = filtered_metadata
        return node

# Load documents
documents = SimpleDirectoryReader(
    "./test_redis_data", filename_as_id=True
).load_data()

# Define embedding model
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

# Define custom schema
custom_schema = IndexSchema.from_dict(
    {
        "index": {"name": "redis_vector_store", "prefix": "doc"},
        "fields": [
            {"type": "tag", "name": "id"},
            {"type": "tag", "name": "doc_id"},
            {"type": "text", "name": "text"},
            {
                "type": "vector",
                "name": "vector",
                "attrs": {
                    "dims": 384,
                    "algorithm": "hnsw",
                    "distance_metric": "cosine",
                },
            },
        ],
    }
)

# Create ingestion pipeline
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(),
        embed_model,
        MetadataFilter(),  # Add custom metadata filter transformation
    ],
    docstore=RedisDocumentStore.from_host_and_port(
        "localhost", 6379, namespace="llama_index"
    ),
    vector_store=RedisVectorStore(
        schema=custom_schema,
        redis_url="redis://localhost:6379",
    ),
    cache=IngestionCache(
        cache=RedisCache.from_host_and_port("localhost", 6379),
        collection="redis_cache",
    ),
    docstore_strategy=DocstoreStrategy.UPSERTS,
)

# Run the pipeline
nodes = pipeline.run(documents=documents)
print(f"Ingested {len(nodes)} Nodes")

In this example, the MetadataFilter class is a custom transformation that filters the metadata to keep only specific keys (key1 and key2). This transformation is added to the pipeline, ensuring that only the filtered metadata is saved in the node, while all metadata is saved in the docstore. The RedisDocumentStore is used to persist the data as Node objects are ingested, and it connects to a Redis database under the specified namespace ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[johnhu33]

@dosu the method provided would result in the following error:
File "/opt/conda/lib/python3.10/site-packages/pydantic/v1/main.py", line 341, in init
raise validation_error
pydantic.v1.error_wrappers.ValidationError: 1 validation error for IngestionPipeline
transformations -> 2
value is not a valid dict (type=type_error.dict)

[dosubot[bot]]

To fix the ValidationError in the IngestionPipeline when using a custom transformation to filter metadata, ensure that the custom transformation is correctly implemented and added to the pipeline. The transformation should be a callable object, not a dictionary.

Here is the corrected example:

from llama_index.core import SimpleDirectoryReader
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.ingestion import (
    DocstoreStrategy,
    IngestionPipeline,
    IngestionCache,
)
from llama_index.storage.kvstore.redis import RedisKVStore as RedisCache
from llama_index.storage.docstore.redis import RedisDocumentStore
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.redis import RedisVectorStore
from redisvl.schema import IndexSchema

# Custom transformation to filter metadata
class MetadataFilter:
    def __call__(self, node):
        # Filter metadata to keep only specific keys
        filtered_metadata = {k: v for k, v in node.metadata.items() if k in ['key1', 'key2']}
        node.metadata = filtered_metadata
        return node

# Load documents
documents = SimpleDirectoryReader(
    "./test_redis_data", filename_as_id=True
).load_data()

# Define embedding model
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

# Define custom schema
custom_schema = IndexSchema.from_dict(
    {
        "index": {"name": "redis_vector_store", "prefix": "doc"},
        "fields": [
            {"type": "tag", "name": "id"},
            {"type": "tag", "name": "doc_id"},
            {"type": "text", "name": "text"},
            {
                "type": "vector",
                "name": "vector",
                "attrs": {
                    "dims": 384,
                    "algorithm": "hnsw",
                    "distance_metric": "cosine",
                },
            },
        ],
    }
)

# Create ingestion pipeline
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(),
        embed_model,
        MetadataFilter(),  # Add custom metadata filter transformation
    ],
    docstore=RedisDocumentStore.from_host_and_port(
        "localhost", 6379, namespace="llama_index"
    ),
    vector_store=RedisVectorStore(
        schema=custom_schema,
        redis_url="redis://localhost:6379",
    ),
    cache=IngestionCache(
        cache=RedisCache.from_host_and_port("localhost", 6379),
        collection="redis_cache",
    ),
    docstore_strategy=DocstoreStrategy.UPSERTS,
)

# Run the pipeline
nodes = pipeline.run(documents=documents)
print(f"Ingested {len(nodes)} Nodes")

Ensure that the MetadataFilter class is correctly defined as a callable object and added to the transformations list. This should resolve the ValidationError ^[1].

To continue talking to Dosu, mention @dosu.