Multimodal RAG with Image Results

pip install reductoai pinecone voyageai requests boto3

import os
from reducto import Reducto

client = Reducto(api_key=os.environ["REDUCTO_API_KEY"])

with open("research_paper.pdf", "rb") as f:
    upload = client.upload(file=f)

print(f"File ID: {upload.file_id}")

result = client.parse.run(
    input=upload.file_id,
    settings={
        "return_images": ["figure", "table"]
    },
    retrieval={
        "chunking": {
            "chunk_mode": "section"
        }
    }
)

print(f"Parsed {result.usage.num_pages} pages")
print(f"Got {len(result.result.chunks)} chunks")

# Look at the first chunk
chunk = result.result.chunks[0]
print(f"Chunk has {len(chunk.blocks)} blocks")
print(f"Embed text preview: {chunk.embed[:200]}...")

figure_count = 0
table_count = 0

for chunk in result.result.chunks:
    for block in chunk.blocks:
        if block.type == "Figure":
            figure_count += 1
        elif block.type == "Table":
            table_count += 1

print(f"Found {figure_count} figures and {table_count} tables")

# Find the first figure
for chunk in result.result.chunks:
    for block in chunk.blocks:
        if block.type == "Figure":
            print(f"Type: {block.type}")
            print(f"Page: {block.bbox.page}")
            print(f"Image URL: {block.image_url[:80]}...")
            print(f"Content: {block.content[:150]}...")
            break
    else:
        continue
    break

import boto3

s3 = boto3.client("s3")
bucket_name = os.environ["S3_BUCKET_NAME"]

region = s3.get_bucket_location(Bucket=bucket_name).get("LocationConstraint")
if region is None:
    region = "us-east-1"  # us-east-1 returns None for LocationConstraint

print(f"Bucket region: {region}")

import requests

def upload_image_to_s3(image_url, s3_key):
    """Download image from Reducto and upload to S3."""
    # Download from Reducto
    response = requests.get(image_url)
    response.raise_for_status()

    # Upload to S3
    s3.put_object(
        Bucket=bucket_name,
        Key=s3_key,
        Body=response.content,
        ContentType="image/png"
    )

    # Return the permanent S3 URL
    return f"https://{bucket_name}.s3.{region}.amazonaws.com/{s3_key}"

all_items = []

for chunk_idx, chunk in enumerate(result.result.chunks):
    # Find any figures/tables in this chunk
    images_in_chunk = []
    for block in chunk.blocks:
        if block.type in ["Figure", "Table"] and block.image_url:
            image_id = f"chunk-{chunk_idx}-{block.type.lower()}-page{block.bbox.page}"
            s3_key = f"multimodal-rag/{image_id}.png"
            s3_url = upload_image_to_s3(block.image_url, s3_key)
            images_in_chunk.append({
                "s3_url": s3_url,
                "block_type": block.type,
                "page": block.bbox.page
            })

    # Get the page number from the first block
    page = chunk.blocks[0].bbox.page if chunk.blocks else 1

    # Index this chunk (with or without images)
    item = {
        "id": f"chunk-{chunk_idx}",
        "text": chunk.embed,
        "page": page,
        "has_images": len(images_in_chunk) > 0
    }

    # If this chunk has images, include the first one
    # (for chunks with multiple figures, you could store all URLs)
    if images_in_chunk:
        item["image_url"] = images_in_chunk[0]["s3_url"]
        item["block_type"] = images_in_chunk[0]["block_type"]

    all_items.append(item)

# Count what we have
chunks_with_images = sum(1 for item in all_items if item.get("has_images"))
chunks_text_only = len(all_items) - chunks_with_images

print(f"Total chunks: {len(all_items)}")
print(f"  - With images: {chunks_with_images}")
print(f"  - Text only: {chunks_text_only}")

# Find a chunk that has an image
test_item = next(item for item in all_items if item.get("image_url"))
test_url = test_item["image_url"]
print(f"Testing: {test_url}")

response = requests.head(test_url)
print(f"Status: {response.status_code}")

from pinecone import Pinecone
import voyageai

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index("multimodal-rag")

vo = voyageai.Client(api_key=os.environ["VOYAGEAI_API_KEY"])

for item in all_items:
    # Create embedding from the chunk's embed text
    embedding_response = vo.embed(
        [item["text"][:8000]],  # VoyageAI has input limits
        model="voyage-3"
    )
    embedding = embedding_response.embeddings[0]

    # Build metadata (image_url only included if chunk has images)
    metadata = {
        "text": item["text"][:1000],  # Preview for display
        "page": item["page"],
        "has_images": item["has_images"]
    }

    if item.get("image_url"):
        metadata["image_url"] = item["image_url"]
        metadata["block_type"] = item.get("block_type", "Figure")

    # Upsert to Pinecone
    index.upsert(vectors=[{
        "id": item["id"],
        "values": embedding,
        "metadata": metadata
    }])

print(f"Indexed {len(all_items)} items")

def search(query: str, top_k: int = 3):
    """Search for relevant chunks with images."""
    # Embed the query
    query_embedding = vo.embed(
        [query],
        model="voyage-3"
    ).embeddings[0]

    # Search Pinecone
    results = index.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True
    )

    return results.matches

matches = search("cell viability and treatment effects")

for match in matches:
    print(f"Score: {match.score:.3f}")
    print(f"Page: {match.metadata['page']}")
    print(f"Has image: {match.metadata.get('has_images', False)}")
    if match.metadata.get("image_url"):
        print(f"Image: {match.metadata['image_url']}")
    print(f"Text preview: {match.metadata['text'][:100]}...")
    print("---")

import os
import requests
import boto3
from pinecone import Pinecone
import voyageai
from reducto import Reducto

# Initialize clients
reducto = Reducto()
s3 = boto3.client("s3")
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
voyage = voyageai.Client()

bucket_name = os.environ["S3_BUCKET_NAME"]
index_name = "multimodal-rag"

# Get bucket region for S3 URLs
region = s3.get_bucket_location(Bucket=bucket_name).get("LocationConstraint") or "us-east-1"

def upload_image_to_s3(image_url, s3_key):
    """Download image from Reducto and upload to S3."""
    response = requests.get(image_url)
    response.raise_for_status()
    s3.put_object(Bucket=bucket_name, Key=s3_key, Body=response.content, ContentType="image/png")
    return f"https://{bucket_name}.s3.{region}.amazonaws.com/{s3_key}"

def index_document(file_path):
    """Parse document, upload images to S3, and index in Pinecone."""
    # Parse with image extraction
    with open(file_path, "rb") as f:
        upload = reducto.upload(file=f)

    result = reducto.parse.run(
        input=upload.file_id,
        settings={"return_images": ["figure", "table"]},
        retrieval={"chunking": {"chunk_mode": "section"}}
    )

    # Process chunks and upload images
    records = []
    for i, chunk in enumerate(result.result.chunks):
        image_url = None
        for block in chunk.blocks:
            if block.type in ["Figure", "Table"] and block.image_url:
                s3_key = f"{result.job_id}/chunk_{i}.png"
                image_url = upload_image_to_s3(block.image_url, s3_key)
                break

        # Embed and prepare record
        embedding = voyage.embed([chunk.embed], model="voyage-3").embeddings[0]
        records.append({
            "id": f"{result.job_id}_{i}",
            "values": embedding,
            "metadata": {
                "text": chunk.embed,
                "page": chunk.blocks[0].bbox.page if chunk.blocks else 0,
                "image_url": image_url or "",
                "has_image": image_url is not None
            }
        })

    # Upsert to Pinecone
    index = pc.Index(index_name)
    index.upsert(vectors=records)
    return len(records)

def search(query, top_k=3):
    """Search for relevant chunks."""
    embedding = voyage.embed([query], model="voyage-3").embeddings[0]
    index = pc.Index(index_name)
    results = index.query(vector=embedding, top_k=top_k, include_metadata=True)
    return results.matches

def query_with_images(query):
    """Search and prepare context for LLM."""
    matches = search(query, top_k=3)
    context = "\n\n".join([m.metadata["text"] for m in matches])
    image_urls = [m.metadata["image_url"] for m in matches if m.metadata.get("image_url")]
    return {"context": context, "images": image_urls, "query": query}

# Usage
indexed = index_document("research-paper.pdf")
print(f"Indexed {indexed} chunks")

result = query_with_images("What do the experimental results show?")
print(f"Context: {len(result['context'])} chars, Images: {len(result['images'])}")
# Pass result to your vision LLM