The Complete Production SOP: How to Deploy and Run Marqo Multimodal Vector & Tensor Search Engine in Autonomous AI Agents

---
name: "marqo-multimodal-vector-tensor-search"
description: "Search across text and images simultaneously using Marqo multimodal embedding pipelines."
metadata:
  author: "Marqo AI"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "setup marqo multimodal index"
  - "search images with text marqo"
  - "index multimodal documents"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# Marqo Multimodal Vector & Tensor Search Engine

Deploys Marqo: configuring multimodal combinations, text-to-image and image-to-image search, custom CLIP/OpenCLIP models, filtering, and running sub-second vector queries via Python SDK.

## Execution SOP & Rules
1. Ensure local Marqo instance is reachable on http://localhost:8882 (or MARQO_URL).
2. Create index using structured settings dict with explicit treatment of multimodal_combination fields.
3. Add documents using typed dictionaries mapping image URLs and text fields.
4. Query via mq.index('name').search(q=..., filter_string=...).

import os
import marqo
from dotenv import load_dotenv

load_dotenv('.env.local')
marqo_url = os.getenv('MARQO_URL', 'http://localhost:8882')
mq = marqo.Client(url=marqo_url)

# Healthcheck probe
health = mq.health()
assert health.get('status') == 'green', f'Unhealthy Marqo node: {health}'
print(f'Connected to Marqo daemon on {marqo_url}')

import marqo

mq = marqo.Client('http://localhost:8882')
INDEX_NAME = 'ecommerce_multimodal'

settings = {
    'type': 'unstructured',
    'model': 'open_clip/ViT-B-32/laion2b_s34b_b79k',
    'treat_urls_and_pointers_as_images': True
}

existing_indexes = [idx['indexName'] for idx in mq.get_indexes()['results']]
if INDEX_NAME not in existing_indexes:
    mq.create_index(index_name=INDEX_NAME, settings_dict=settings)
    print(f'Successfully initialized {INDEX_NAME}')
else:
    print(f'Index {INDEX_NAME} already exists. Ready.')

import unittest
import marqo

class TestMarqoSearch(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.mq = marqo.Client('http://localhost:8882')
        cls.index_name = 'ecommerce_multimodal'
        cls.mq.index(cls.index_name).add_documents([
            {
                '_id': 'doc_001',
                'caption': 'Vintage distressed leather motorcycle jacket',
                'image_url': 'https://raw.githubusercontent.com/marqo-ai/marqo/mainline/examples/ImageSearchGuide/data/image1.jpg',
                'price': 120.00,
                'multimodal_field': {
                    'caption': 0.4,
                    'image_url': 0.6
                }
            }
        ], tensor_fields=['multimodal_field'])

    def test_search_execution(self):
        results = self.mq.index(self.index_name).search(
            q='distressed leather rider',
            filter_string='price < 150'
        )
        self.assertGreater(len(results['hits']), 0)
        self.assertEqual(results['hits'][0]['_id'], 'doc_001')

if __name__ == '__main__':
    unittest.main()