Construindo um sistema RAG multimodal com Elasticsearch: A história de Gotham City

# Clone the repository with the multimodal RAG code
!git clone -b https://github.com/elastic/elasticsearch-labs.git

# Navigate to the project directory
cd elasticsearch-labs/supporting-blog-content/building-multimodal-rag-with-elasticsearch-gotham

 # Install PyTorch and related libraries
!pip install torch>=2.1.0 torchvision>=0.16.0 torchaudio>=2.1.0

# Install vision processing libraries
!pip install opencv-python-headless pillow numpy

# Install the specific ImageBind fork
!pip install git+https://github.com/hkchengrex/ImageBind.git

# Install Elasticsearch and environment management
!pip install elasticsearch python-dotenv

# Input your credentials securely
import getpass

ELASTICSEARCH_URL = input("Enter the Elasticsearch endpoint url: ")
ELASTICSEARCH_API_KEY = getpass.getpass("Enter the Elasticsearch API key: ")
OPENAI_API_KEY = getpass.getpass("Enter the OpenAI API key: ")

# Configure environment variables
import os
os.environ["ELASTICSEARCH_API_KEY"] = ELASTICSEARCH_API_KEY
os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY
os.environ["ELASTICSEARCH_URL"] = ELASTICSEARCH_URL

data/
├── images/
│   ├── crime_scene1.jpg
│   ├── suspect_spotted.jpg
│   ...
├── audios/
│   ├── joker_laugh.wav
│   ...
├── texts/
│   ├── riddle.txt
│   ... 
├── depths/
│   ├── depth_suspect.png

import os

# Base directory for clues
data_dir = "data"

# List of expected files
evidences = {
    "images": ["crime_scene1.jpg","crime_scene1.jpg", "joker_alley.jpg"],
    "audios": ["joker_laugh.wav"],
    "texts": ["riddle.txt", "note2.txt”],
    "depths": ["depth_suspect.png", "jdancing-depth.png"]
}

# Create directories if they don't exist
for category, files in evidences.items():
    category_path = os.path.join(data_dir, category)
    os.makedirs(category_path, exist_ok=True)

    for file in files:
        file_path = os.path.join(category_path, file)
        if not os.path.exists(file_path):
            print(f"Warning: {file} not found in {category_path}.")

print("All files are correctly organized!")

python  stages/01-stage/files_check.py

All files are correctly organized!

Warning: joker_laugh.wav not found in data/audios/
Warning: depth_suspect.png not found in data/depths/

class EmbeddingGenerator:
    """Class for generating multimodal embeddings using ImageBind."""
    
    def __init__(self):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model = self._load_model()

    def _load_model(self):
        """Loads the ImageBind model and sets it to inference mode."""
        model = imagebind_model.imagebind_huge(pretrained=True)
        model.eval()
        model.to(self.device)
        return model

    def generate_embedding(self, input_data, modality):
        """Generates embedding for different modalities"""
        processors = {
            "vision": lambda x: data.load_and_transform_vision_data(x, self.device),
            "audio": lambda x: data.load_and_transform_audio_data(x, self.device),
            "text": lambda x: data.load_and_transform_text(x, self.device),
            "depth": self.process_depth
        }
        
        try:
            # Input type verification
            if not isinstance(input_data, list):
                raise ValueError(f"Input data must be a list. Received: {type(input_data)}")
                
            # Convert input data to a tensor format that the model can process
            # For images: [batch_size, channels, height, width] 
            # For audio: [batch_size, channels, time] 
            # For text: [batch_size, sequence_length]
            inputs = {modality: processors[modality](input_data)}
            with torch.no_grad():
                embedding = self.model(inputs)[modality]
            return embedding.squeeze(0).cpu().numpy()
        except Exception as e:
            logger.error(f"Error generating {modality} embedding: {str(e)}", exc_info=True)
            raise

input_tensor = processors[modality]([input_data], self.device)

python stages/02-stage/test_embedding_generation.py

generator = EmbeddingGenerator()
image_embedding = generator.generate_embedding("data/images/crime_scene1.jpg","vision")

print(image_embedding.shape)

(1024,)

from elasticsearch import Elasticsearch, helpers
...

class ElasticsearchManager:
    """Manages multimodal operations in Elasticsearch"""
    
    def __init__(self):
        load_dotenv()  # Load variables from .env
        self.es = self._connect_elastic()
        self.index_name = "multimodal_content"
        self._setup_index()
    
    def _connect_elastic(self):
        """Connects to Elasticsearch"""
        return Elasticsearch(
            os.getenv("ELASTICSEARCH_URL"),  # Elasticsearch endpoint
            api_key=os.getenv("ELASTICSEARCH_API_KEY")
        )
    
    def _setup_index(self):
        """Sets up the index if it doesn't exist"""
        if not self.es.indices.exists(index=self.index_name):
            mapping = {
                "mappings": {
                    "properties": {
                        "embedding": {
                            "type": "dense_vector",
                            "dims": 1024,
                            "index": True,
                            "similarity": "cosine"
                        },
                        "modality": {"type": "keyword"},
                        "content": {"type": "binary"},
                        "description": {"type": "text"},
                        "metadata": {"type": "object"},
                        "content_path": {"type": "text"}
                    }
                }
            }
            self.es.indices.create(index=self.index_name, body=mapping)
    
    def index_content(self, embedding, modality, content=None, description="", metadata=None, content_path=None):
        """Indexes multimodal content"""
        doc = {
            "embedding": embedding.tolist(),
            "modality": modality,
            "description": description,
            "metadata": metadata or {},
            "content_path": content_path
        }
        
        if content:
            doc["content"] = base64.b64encode(content).decode() if isinstance(content, bytes) else content
        
        return self.es.index(index=self.index_name, document=doc)
    
    def search_similar(self, query_embedding, modality=None, k=5):
        """Searches for similar contents"""
        query = {
            "knn": {
                "field": "embedding",
                "query_vector": query_embedding.tolist(),
                "k": k,
                "num_candidates": 100,
                "filter": [{"term": {"modality": modality}}] if modality else []
            }
        }
        
        try:
            response = self.es.search(
                index=self.index_name,
                query=query,
                size=k            
            )
            
            # Return both source data and score for each hit
            return [{
                **hit["_source"],
                "score": hit["_score"]
            } for hit in response["hits"]["hits"]]
        
        except Exception as e:
            print(f"Error: processing search_evidence: {str(e)}")
            return "Error generating search evidence"

# Example: Indexing an image from the crime scene
generator = EmbeddingGenerator()
es_manager = ElasticsearchManager(cloud_id="YOUR_CLOUD_ID", api_key="YOUR_API_KEY")

image_embedding = generator.generate_embedding("data/images/crime_scene1.jpg", "vision")

response = es_manager.index_content(
    embedding=image_embedding,
    modality="vision",
    description="Photo of the crime scene with suspicious traces",
    content_path="data/images/crime_scene1.jpg"
)
print(json.dumps(response, indent=2))

{
    "embedding": [0.12, -0.53, 0.89, ...],  
    "modality": "vision",  
    "description": "Photo of the crime scene with suspicious traces",  
    "content_path": "data/images/crime_scene1.jpg"  
}

python stages/03-stage/index_all_modalities.py

GET _cat/indices/multimodal_content?v

GET multimodal_content/_search
{
  "size": 0,
  "aggs": {
    "modalities": {
      "terms": {
        "field": "modality.keyword"
      }
    }
  }
}

GET multimodal_content/_search
{
  "size": 1,
  "query": {
    "match_all": {}
  }
}

def search_similar_evidence(self, query_embedding, k=5, modality=None):
    """Performs a kNN search to find the most similar clues."""
    
    knn_query = {
        "field": "embedding",
        "query_vector": query_embedding.tolist(),
        "k": k,
        "num_candidates": 100
    }

    query_body = {"knn": knn_query}
    if modality:
        query_body = {
            "bool": {
                "must": [
                    query_body, 
                    {"term": {"modality": modality}}
                ]
            }
        }

    try:
      results = self.es.search(
        index=self.index_name,
        query=query_body,
        _source_includes=["description", "modality", "content_path"],
        size=k
      )
    except Exception as e:
            print(f"Error processing search_evidence: {str(e)}")
            return "Error generating search evidence”

    return results["hits"]["hits"]

python stages/03-stage/search_by_audio.py

# Initialize classes
generator = EmbeddingGenerator()
es_manager = ElasticsearchManager(cloud_id="YOUR_CLOUD_ID", api_key="YOUR_API_KEY")

# Generate embedding for a suspicious audio
audio_embedding = generator.generate_embedding("data/audios/mysterious_laugh.wav", "audio")

# Search for similar evidence in Elasticsearch
similar_evidences = es_manager.search_similar_evidence(audio_embedding, k=3)

# Display the retrieved results
print("\n🔎 Similar evidence found:\n")
for i, evidence in enumerate(similar_evidences, start=1):
    description = evidence['_source']['description']
    modality = evidence['_source']['modality']
    score = evidence['_score']
    content_path = evidence['_source'].get('content_path', 'N/A')

    print(f"{i}. {description} ({modality})")
    print(f"   Similarity: {score:.4f}")
    print(f"   File path: {content_path}\n")

🔎 Similar evidence found:

1. A sinister laugh captured near the crime scene (audio)
   Similarity: 0.9985
   File path: data/audios/joker_laugh.wav

2. The Joker with green hair, white face paint, and a sinister smile in an urban night setting. (vision)
   Similarity: 0.6068
   File path: data/images/joker_laughing.png

3. Suspect dancing (vision)
   Similarity: 0.5591
   File path: data/images/jdancing.png

python stages/03-stage/search_by_text.py

# Generate embedding from text
text = "Why so serious?"
embedding_text = generator.generate_embedding([text], "text")

# Search for related evidence
similar_evidences = es_manager.search_similar(
    query_embedding=embedding_text,
    k=3
)

🔎 Similar evidence found:

1. Mysterious note found at the location (text)
   Similarity: 0.7639
   File path: data/texts/riddle.txt

2. The Joker with green hair, white face paint, and a sinister smile in an urban night setting. (vision)
   Similarity: 0.7161
   File path: data/images/joker_laughing.png

3. Why so serious (text)
   Similarity: 0.7132
   File path: data/texts/note2.txt

python stages/03-stage/search_by_image.py

# Generate embedding for a suspicious image
vision_embedding = generator.generate_embedding(["data/images/crime_scene2.jpg"], "vision")

# Search for similar evidence in Elasticsearch
similar_evidences = es_manager.search_similar(
    query_embedding=vision_embedding,
    k=3
)

🔎 Similar evidence found:

1. Photo of the crime scene: A dark, rain-soaked alley is filled with playing cards, while a sinister graffiti of the Joker laughing stands out on the brick wall. (vision)
   Similarity: 0.8258
   File path: data/images/crime_scene1.jpg

2. The Joker with green hair, white face paint, and a sinister smile in an urban night setting. (vision)
   Similarity: 0.6897
   File path: data/images/joker_laughing.png

3. Suspect dancing (vision)
   Similarity: 0.6588
   File path: data/images/jdancing.png

python stages/03-stage/search_by_depth.py

# Generate embedding for a suspicious depth map
vision_embedding = generator.generate_embedding(["data/depths/jdancing-depth.png"], "depth")

# Search for similar evidence in Elasticsearch
similar_evidences = es_manager.search_similar(
    query_embedding=vision_embedding,
    modality="vision",
    k=3
)

🔎 Similar evidence found:

1. The Joker with green hair, white face paint, and a sinister smile in an urban night setting. (vision)
   Similarity: 0.5329
   File path: data/images/joker_laughing.png

2. Photo of the crime scene: A dark, rain-soaked alley is filled with playing cards, while a sinister graffiti of the Joker laughing stands out on the brick wall. (vision)
   Similarity: 0.5053
   File path: data/images/crime_scene1.jpg

3. Suspect dancing (vision)
   Similarity: 0.4859
   File path: data/images/jdancing.png

import os
from openai import OpenAI
import logging
from dotenv import load_dotenv

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class LLMAnalyzer:
    """Evidence analyzer using GPT-4"""
    
    def __init__(self):
        load_dotenv()
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    def analyze_evidence(self, evidence_results):
        """
        Analyzes multimodal search results and generates a report
        
        Args:
            evidence_results: Dict with results by modality
            {
                'vision': [...],
                'audio': [...],
                'text': [...],
                'depth': [...]
            }
        """
        # Format evidence for the prompt
        evidence_summary = self._format_evidence(evidence_results)

        # final prompt
        prompt = f"""
You are a highly experienced forensic detective specializing in multimodal evidence analysis. Your task is to analyze the collected evidence (audio, images, text, depth maps) and conclusively determine the **prime suspect** responsible for the Gotham Central Bank case.

---

### **Collected Evidence:**
{evidence_summary}

### **Task:**
1. **Analyze all the evidence** and identify cross-modal connections.
2. **Determine the exact identity of the criminal** based on behavioral patterns, visual/auditory/textual clues, and symbolic markers.
3. **Justify your conclusion** by explaining why this suspect is definitively responsible.
4. **Assign a confidence score (0-100%)** to your conclusion.

---

### **Final Output Format (Strictly Follow This Format):**
- **Prime Suspect:** [Full Name or Alias]
- **Evidence Supporting Conclusion:** [Detailed breakdown of visual, auditory, textual, and behavioral evidence]
- **Behavioral Patterns:** [Key actions, motives, and criminal signature]
- **Confidence Level:** [0-100%]
- **Next Steps (if any):** [What additional evidence would further confirm the identity? If none, state "No further evidence required."]

If there is **insufficient evidence**, specify exactly what is missing and suggest what additional data would be needed for a conclusive identification.

This report must be **direct and definitive**--avoid speculation and provide a final, actionable determination of the suspect's identity.
"""
        try:
            response = self.client.chat.completions.create(
                model="gpt-4-turbo-preview",
                messages=[
                    {
                        "role": "system",
                        "content": "You are a forensic detective specialized in multimodal evidence analysis."
                    },
                    {"role": "user", "content": prompt_01}
                ],
                temperature=0.5,
                max_tokens=1000
            )
            
            report = response.choices[0].message.content
            logger.info("\n📋 Forensic Report Generated:")
            logger.info("=" * 50)
            logger.info(report)
            logger.info("=" * 50)
            
            return report
            
        except Exception as e:
            logger.error(f"Error generating report: {str(e)}")
            return None

python stages/04-stage/rag_crime_analyze.py

import sys
import os
sys.path.append(os.path.join(os.path.dirname(os.path.dirname(__file__)), 'src'))

from embedding_generator import EmbeddingGenerator
from elastic_manager import ElasticsearchManager
from llm_analyzer import LLMAnalyzer

import json
import logging
from dotenv import load_dotenv

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

# Initialize classes
generator = EmbeddingGenerator()
es_manager = ElasticsearchManager()

llm = LLMAnalyzer()
logger.info("✅ All components initialized successfully")
    
try:
    evidence_data = {}
    
    # Get data for each modality
    test_files = {
        'vision': 'data/images/crime_scene2.jpg',
        'audio': 'data/audios/joker_laugh.wav',
        'text': 'Why so serious?',
        'depth': 'data/depths/jdancing-depth.png'
    }
    
    logger.info("🔍 Collecting evidence...")
    for modality, test_input in test_files.items():
        try:
            if modality == 'text':
                embedding = generator.generate_embedding([test_input], modality)
            else:
                embedding = generator.generate_embedding([str(test_input)], modality)
            
            results = es_manager.search_similar(embedding, k=2)
            if results:
                evidence_data[modality] = results
                logger.info(f"✅ Data retrieved for {modality}: {len(results)} results")
            else:
                logger.warning(f"⚠️ No results found for {modality}")
                
        except Exception as e:
            logger.error(f"❌ Error retrieving {modality} data: {str(e)}")
    
    if not evidence_data:
        raise ValueError("No evidence data found in Elasticsearch!")
    
    # Test forensic report generation
    logger.info("\n📝 Generating forensic report...")
    report = llm.analyze_evidence(evidence_data)
    
    if report:
        logger.info("✅ Forensic report generated successfully")
        logger.info("\n📊 Report Preview:")
        logger.info("+" * 50)
        logger.info(report)
        logger.info("+" * 50)
    else:
        raise ValueError("Failed to generate forensic report")
        
except Exception as e:
    logger.error(f"❌ Error in analysis : {str(e)}")

**Prime Suspect:** The Joker

**Evidence Supporting Conclusion:**

- **Visual Evidence:**
  - The photo of the crime scene with playing cards scattered around and the graffiti of the Joker laughing matches the Joker's known calling cards and thematic elements. The similarity score of 0.83 indicates a high likelihood that these elements are directly associated with the Joker.
  - The image of the Joker with green hair, white face paint, and a sinister smile in an urban night setting, although with a lower similarity score of 0.69, still supports the presence or recent activity of the Joker in areas consistent with the crime scene's characteristics.

- **Auditory Evidence:**
  - The captured sinister laugh with a similarity score of 1.00 perfectly matches known audio profiles of the Joker, making it a direct auditory signature of his presence at or near the crime scene.
  - Despite the lower similarity score of 0.61, the second audio piece further corroborates the Joker's involvement through thematic consistency.

- **Textual Evidence:**
  - The mysterious note found at the location, with a similarity score of 0.76, likely contains thematic or direct references to the Joker's modus operandi or signature phrases, further implicating him in the crime.
  - The similarity score of 0.72 for the Joker's description in textual evidence reinforces the thematic connection to the crime scene.

- **Depth Evidence:**
  - Depth sensor capture of the suspect with a similarity score of 0.77 suggests a physical presence matching the Joker's known dimensions or characteristic movements.
  - The lower similarity score of 0.53 in the second depth evidence still contributes to the overall pattern of evidence pointing towards the Joker, albeit with less certainty.

**Behavioral Patterns:**
- The Joker is known for his theatrical crimes, often leaving behind a signature trail of chaos, including playing cards, sinister laughter, and thematic graffiti. These elements are not only consistent with his known criminal signature but also directly observed at the crime scene.
- His motives often include creating chaos, drawing attention to his acts, and challenging his arch-nemesis, Batman, making a high-profile bank heist fitting within his behavioral patterns.

**Confidence Level:** 95%

**Next Steps:** No further evidence required.

The combination of visual, auditory, textual, and depth evidence strongly points to the Joker as the prime suspect. The thematic consistency across multiple modes of evidence, combined with known behavioral patterns and criminal signature, leaves little doubt regarding his involvement. While there is always a small margin of uncertainty in forensic analysis, the evidence at hand provides a compelling case against the Joker with a high degree of confidence.