Milvus
Zilliz
  • Home
  • Blog

  • Il RAG multimodale reso semplice: RAG-Anything + Milvus al posto di 20 strumenti separati

Il RAG multimodale reso semplice: RAG-Anything + Milvus al posto di 20 strumenti separati

  • Tutorials
November 25, 2025
Min Yin

Costruire un sistema RAG multimodale significava mettere insieme una dozzina di strumenti specializzati: uno per l'OCR, uno per le tabelle, uno per le formule matematiche, uno per gli embeddings, uno per la ricerca e così via. Le pipeline RAG tradizionali sono state progettate per il testo e quando i documenti hanno iniziato a includere immagini, tabelle, equazioni, grafici e altri contenuti strutturati, la catena di strumenti è diventata rapidamente disordinata e ingestibile.

RAG-Anything, sviluppato dalla HKU, cambia questa situazione. Basato su LightRAG, fornisce una piattaforma all-in-one in grado di analizzare diversi tipi di contenuti in parallelo e di mapparli in un knowledge graph unificato. Ma unificare la pipeline è solo metà dell'opera. Per recuperare le prove in queste diverse modalità, è necessaria una ricerca vettoriale veloce e scalabile, in grado di gestire molti tipi di incorporazioni contemporaneamente. È qui che entra in gioco Milvus. Come database vettoriale open-source ad alte prestazioni, Milvus elimina la necessità di soluzioni multiple di archiviazione e ricerca. Supporta la ricerca di RNA su larga scala, il recupero ibrido di parole vettoriali e parole chiave, il filtraggio dei metadati e la gestione flessibile degli incorporamenti, tutto in un unico posto.

In questo post spiegheremo come RAG-Anything e Milvus lavorino insieme per sostituire una catena di strumenti multimodali frammentata con uno stack pulito e unificato, e mostreremo come sia possibile costruire un pratico sistema multimodale RAG Q&A con pochi passaggi.

Cos'è RAG-Anything e come funziona

RAG-Anything è un framework RAG progettato per rompere la barriera del solo testo dei sistemi tradizionali. Invece di affidarsi a più strumenti specializzati, offre un unico ambiente unificato in grado di analizzare, elaborare e recuperare informazioni su diversi tipi di contenuto.

Il framework supporta documenti contenenti testo, diagrammi, tabelle ed espressioni matematiche, consentendo agli utenti di interrogare tutte le modalità attraverso un'unica interfaccia coesa. Ciò lo rende particolarmente utile in campi come la ricerca accademica, la rendicontazione finanziaria e la gestione della conoscenza aziendale, dove i materiali multimodali sono comuni.

RAG-Anything si basa su una pipeline multimodale a più stadi: analisi dei documenti→analisi dei contenuti→grafo della conoscenza→recupero intelligente. Questa architettura consente un'orchestrazione intelligente e una comprensione intermodale, permettendo al sistema di gestire senza problemi diverse modalità di contenuto all'interno di un unico flusso di lavoro integrato.

L'architettura "1 + 3 + N

A livello ingegneristico, le capacità di RAG-Anything sono realizzate attraverso la sua architettura "1 + 3 + N":

Il motore centrale

Al centro di RAG-Anything c'è un motore a grafo di conoscenza ispirato a LightRAG. Questa unità centrale è responsabile dell'estrazione di entità multimodali, della mappatura delle relazioni intermodali e dell'archiviazione semantica vettoriale. A differenza dei tradizionali sistemi RAG di solo testo, il motore è in grado di comprendere entità di testo, oggetti visivi all'interno di immagini e strutture relazionali incorporate in tabelle.

3 Processori modali

RAG-Anything integra tre processori modali specializzati, progettati per una comprensione profonda e specifica della modalità. Insieme, formano il livello di analisi multimodale del sistema.

  • ImageModalProcessor interpreta il contenuto visivo e il suo significato contestuale.

  • TableModalProcessor analizza le strutture delle tabelle e decodifica le relazioni logiche e numeriche all'interno dei dati.

  • EquationModalProcessor comprende la semantica dei simboli e delle formule matematiche.

Parser N

Per supportare le diverse strutture dei documenti del mondo reale, RAG-Anything fornisce un livello di parsing estensibile basato su diversi motori di estrazione. Attualmente integra sia MinerU che Docling, selezionando automaticamente il parser ottimale in base al tipo di documento e alla complessità strutturale.

Basandosi sull'architettura "1 + 3 + N", RAG-Anything migliora la pipeline RAG tradizionale cambiando il modo in cui vengono gestiti i diversi tipi di contenuto. Invece di elaborare testo, immagini e tabelle uno alla volta, il sistema li elabora tutti insieme.

# The core configuration demonstrates the parallel processing design
config = RAGAnythingConfig(
    working_dir="./rag_storage",
    parser="mineru",
    parse_method="auto",  # Automatically selects the optimal parsing strategy
    enable_image_processing=True,
    enable_table_processing=True, 
    enable_equation_processing=True,
    max_workers=8  # Supports multi-threaded parallel processing
)

Questo design accelera notevolmente la gestione di documenti tecnici di grandi dimensioni. I test di benchmark dimostrano che quando il sistema utilizza più core della CPU, diventa sensibilmente più veloce, riducendo nettamente il tempo necessario per elaborare ciascun documento.

Ottimizzazione dell'archiviazione e del recupero a strati

Oltre al design multimodale, RAG-Anything utilizza anche un approccio di archiviazione e recupero a più livelli per rendere i risultati più accurati ed efficienti.

  • Iltesto viene memorizzato in un database vettoriale tradizionale.

  • Leimmagini sono gestite in un archivio di caratteristiche visive separato.

  • Letabelle sono conservate in un archivio di dati strutturati.

  • Leformule matematiche sono trasformate in vettori semantici.

Memorizzando ogni tipo di contenuto nel proprio formato adatto, il sistema può scegliere il metodo di recupero migliore per ogni modalità, invece di affidarsi a un'unica ricerca di somiglianza generica. In questo modo si ottengono risultati più rapidi e affidabili per i diversi tipi di contenuto.

Come Milvus si integra in RAG-Anything

RAG-Anything fornisce un forte recupero multimodale, ma per farlo bene è necessaria una ricerca vettoriale rapida e scalabile su tutti i tipi di incorporazioni. Milvus svolge perfettamente questo ruolo.

Grazie alla sua architettura cloud-native e alla separazione tra calcolo e storage, Milvus offre un'elevata scalabilità ed efficienza dei costi. Supporta la separazione lettura-scrittura e l'unificazione stream-batch, consentendo al sistema di gestire carichi di lavoro ad alta liquidità mantenendo le prestazioni delle query in tempo reale: i nuovi dati diventano ricercabili subito dopo l'inserimento.

Milvus garantisce anche un'affidabilità di livello aziendale grazie al suo design distribuito e tollerante ai guasti, che mantiene il sistema stabile anche se i singoli nodi si guastano. Ciò lo rende particolarmente adatto a implementazioni RAG multimodali a livello di produzione.

Come costruire un sistema di domande e risposte multimodale con RAG-Anything e Milvus

Questa demo mostra come costruire un sistema di domande e risposte multimodale utilizzando il framework RAG-Anything, il database vettoriale Milvus e il modello di incorporamento TongYi. (Questo esempio si concentra sul codice di implementazione principale e non è una configurazione di produzione completa).

Dimostrazione pratica

Prerequisiti: Python

  • Python: 3.10 o superiore

  • Database vettoriale: Servizio Milvus (Milvus Lite)

  • Servizio cloud: Chiave API Alibaba Cloud (per i servizi LLM e embedding)

  • Modello LLM: qwen-vl-max (modello abilitato alla visione)

Modello di incorporazione: tongyi-embedding-vision-plus

- python -m venv .venv && source .venv/bin/activate  # For Windows users:  .venvScriptsactivate
- pip install -r requirements-min.txt
- cp .env.example .env #add DASHSCOPE_API_KEY

Eseguire l'esempio di lavoro minimo:

python minimal_[main.py](<http://main.py>)

Risultati attesi:

Una volta eseguito correttamente lo script, il terminale dovrebbe visualizzare:

  • Il risultato della Q&A testuale generata dall'LLM.

  • La descrizione dell'immagine recuperata corrispondente alla query.

Struttura del progetto

.
├─ requirements-min.txt
├─ .env.example
├─ [config.py](<http://config.py>)
├─ milvus_[store.py](<http://store.py>)
├─ [adapters.py](<http://adapters.py>)
├─ minimal_[main.py](<http://main.py>)
└─ sample
   ├─ docs
   │  └─ faq_milvus.txt
   └─ images
      └─ milvus_arch.png

Dipendenze del progetto

raganything
lightrag
pymilvus[lite]>=2.3.0
aiohttp>=3.8.0
orjson>=3.8.0
python-dotenv>=1.0.0
Pillow>=9.0.0
numpy>=1.21.0,<2.0.0
rich>=12.0.0

Variabili d'ambiente

# Alibaba Cloud DashScope
DASHSCOPE_API_KEY=your_api_key_here
# If the endpoint changes in future releases, please update it accordingly.
ALIYUN_LLM_URL=https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions
ALIYUN_VLM_URL=https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions
ALIYUN_EMBED_URL=https://dashscope.aliyuncs.com/api/v1/services/embeddings/text-embedding
# Model names (configure all models here for consistency)
LLM_TEXT_MODEL=qwen-max
LLM_VLM_MODEL=qwen-vl-max
EMBED_MODEL=tongyi-embedding-vision-plus
# Milvus Lite
MILVUS_URI=milvus_lite.db
MILVUS_COLLECTION=rag_multimodal_collection
EMBED_DIM=1152

Configurazione

import os
from dotenv import load_dotenv
load_dotenv()
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY", "")
LLM_TEXT_MODEL = os.getenv("LLM_TEXT_MODEL", "qwen-max")
LLM_VLM_MODEL = os.getenv("LLM_VLM_MODEL", "qwen-vl-max")
EMBED_MODEL = os.getenv("EMBED_MODEL", "tongyi-embedding-vision-plus")
ALIYUN_LLM_URL = os.getenv("ALIYUN_LLM_URL")
ALIYUN_VLM_URL = os.getenv("ALIYUN_VLM_URL")
ALIYUN_EMBED_URL = os.getenv("ALIYUN_EMBED_URL")
MILVUS_URI = os.getenv("MILVUS_URI", "milvus_lite.db")
MILVUS_COLLECTION = os.getenv("MILVUS_COLLECTION", "rag_multimodal_collection")
EMBED_DIM = int(os.getenv("EMBED_DIM", "1152"))
# Basic runtime parameters
TIMEOUT = 60
MAX_RETRIES = 2

Invocazione del modello

import os
import base64
import aiohttp
import asyncio
from typing import List, Dict, Any, Optional
from config import (
    DASHSCOPE_API_KEY, LLM_TEXT_MODEL, LLM_VLM_MODEL, EMBED_MODEL,
    ALIYUN_LLM_URL, ALIYUN_VLM_URL, ALIYUN_EMBED_URL, EMBED_DIM, TIMEOUT
)
HEADERS = {
    "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
    "Content-Type": "application/json",
}
class AliyunLLMAdapter:
    def __init__(self):
        self.text_url = ALIYUN_LLM_URL
        self.vlm_url = ALIYUN_VLM_URL
        self.text_model = LLM_TEXT_MODEL
        self.vlm_model = LLM_VLM_MODEL
    async def chat(self, prompt: str) -> str:
        payload = {
            "model": self.text_model,
            "input": {"messages": [{"role": "user", "content": prompt}]},
            "parameters": {"max_tokens": 1024, "temperature": 0.5},
        }
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=TIMEOUT)) as s:
            async with [s.post](<http://s.post>)(self.text_url, json=payload, headers=HEADERS) as r:
                r.raise_for_status()
                data = await r.json()
                return data["output"]["choices"][0]["message"]["content"]
    async def chat_vlm_with_image(self, prompt: str, image_path: str) -> str:
        with open(image_path, "rb") as f:
            image_b64 = base64.b64encode([f.read](<http://f.read>)()).decode("utf-8")
        payload = {
            "model": self.vlm_model,
            "input": {"messages": [{"role": "user", "content": [
                {"text": prompt},
                {"image": f"data:image/png;base64,{image_b64}"}
            ]}]},
            "parameters": {"max_tokens": 1024, "temperature": 0.2},
        }
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=TIMEOUT)) as s:
            async with [s.post](<http://s.post>)(self.vlm_url, json=payload, headers=HEADERS) as r:
                r.raise_for_status()
                data = await r.json()
                return data["output"]["choices"][0]["message"]["content"]
class AliyunEmbeddingAdapter:
    def __init__(self):
        self.url = ALIYUN_EMBED_URL
        self.model = EMBED_MODEL
        self.dim = EMBED_DIM
    async def embed_text(self, text: str) -> List[float]:
        payload = {
            "model": self.model,
            "input": {"texts": [text]},
            "parameters": {"text_type": "query", "dimensions": self.dim},
        }
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=TIMEOUT)) as s:
            async with [s.post](<http://s.post>)(self.url, json=payload, headers=HEADERS) as r:
                r.raise_for_status()
                data = await r.json()
                return data["output"]["embeddings"][0]["embedding"]

Integrazione di Milvus Lite

import json
import time
from typing import List, Dict, Any, Optional
from pymilvus import connections, Collection, CollectionSchema, FieldSchema, DataType, utility
from config import MILVUS_URI, MILVUS_COLLECTION, EMBED_DIM
class MilvusVectorStore:
    def __init__(self, uri: str = MILVUS_URI, collection_name: str = MILVUS_COLLECTION, dim: int = EMBED_DIM):
        self.uri = uri
        self.collection_name = collection_name
        self.dim = dim
        self.collection: Optional[Collection] = None
        self._connect_and_prepare()
    def _connect_and_prepare(self):
        connections.connect("default", uri=self.uri)
        if utility.has_collection(self.collection_name):
            self.collection = Collection(self.collection_name)
        else:
            fields = [
                FieldSchema(name="id", dtype=DataType.VARCHAR, max_length=512, is_primary=True),
                FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=self.dim),
                FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
                FieldSchema(name="content_type", dtype=DataType.VARCHAR, max_length=32),
                FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=1024),
                FieldSchema(name="ts", dtype=[DataType.INT](<http://DataType.INT>)64),
            ]
            schema = CollectionSchema(fields, "Minimal multimodal collection")
            self.collection = Collection(self.collection_name, schema)
            self.collection.create_index("vector", {
                "metric_type": "COSINE",
                "index_type": "IVF_FLAT",
                "params": {"nlist": 1024}
            })
        self.collection.load()
    def upsert(self, ids: List[str], vectors: List[List[float]], contents: List[str],
               content_types: List[str], sources: List[str]) -> None:
        data = [
            ids,
            vectors,
            contents,
            content_types,
            sources,
            [int(time.time() * 1000)] * len(ids)
        ]
        self.collection.upsert(data)
        self.collection.flush()
    def search(self, query_vectors: List[List[float]], top_k: int = 5, content_type: Optional[str] = None):
        expr = f'content_type == "{content_type}"' if content_type else None
        params = {"metric_type": "COSINE", "params": {"nprobe": 16}}
        results = [self.collection.search](<http://self.collection.search>)(
            data=query_vectors,
            anns_field="vector",
            param=params,
            limit=top_k,
            expr=expr,
            output_fields=["id", "content", "content_type", "source", "ts"]
        )
        out = []
        for hits in results:
            out.append([{
                "id": h.entity.get("id"),
                "content": h.entity.get("content"),
                "content_type": h.entity.get("content_type"),
                "source": h.entity.get("source"),
                "score": h.score
            } for h in hits])
        return out

Punto di ingresso principale

"""
Minimal Working Example:
- Insert a short text FAQ into LightRAG (text retrieval context)
- Insert an image description vector into Milvus (image retrieval context)
- Execute two example queries: one text QA and one image-based QA
"""
import asyncio
import uuid
from pathlib import Path
from rich import print
from lightrag import LightRAG, QueryParam
from lightrag.utils import EmbeddingFunc
from adapters import AliyunLLMAdapter, AliyunEmbeddingAdapter
from milvus_store import MilvusVectorStore
from config import EMBED_DIM
SAMPLE_DOC = Path("sample/docs/faq_milvus.txt")
SAMPLE_IMG = Path("sample/images/milvus_arch.png")
async def main():
    # 1) Initialize core components
    llm = AliyunLLMAdapter()
    emb = AliyunEmbeddingAdapter()
    store = MilvusVectorStore()
    # 2) Initialize LightRAG (for text-only retrieval)
    async def llm_complete(prompt: str, max_tokens: int = 1024) -> str:
        return await [llm.chat](<http://llm.chat>)(prompt)
    async def embed_func(text: str) -> list:
        return await emb.embed_text(text)
    rag = LightRAG(
        working_dir="rag_workdir_min",
        llm_model_func=llm_complete,
        embedding_func=EmbeddingFunc(
            embedding_dim=EMBED_DIM,
            max_token_size=8192,
            func=embed_func
        ),
    )
    # 3) Insert text data
    if SAMPLE_DOC.exists():
        text = SAMPLE_[DOC.read](<http://DOC.read>)_text(encoding="utf-8")
        await rag.ainsert(text)
        print("[green]Inserted FAQ text into LightRAG[/green]")
    else:
        print("[yellow] sample/docs/faq_milvus.txt not found[/yellow]")
    # 4) Insert image data (store description in Milvus)
    if SAMPLE_IMG.exists():
        # Use the VLM to generate a description as its semantic content
        desc = await [llm.chat](<http://llm.chat>)_vlm_with_image("Please briefly describe the key components of the Milvus architecture shown in the image.", str(SAMPLE_IMG))
        vec = await emb.embed_text(desc)  # Use text embeddings to maintain a consistent vector dimension, simplifying reuse
        store.upsert(
            ids=[str(uuid.uuid4())],
            vectors=[vec],
            contents=[desc],
            content_types=["image"],
            sources=[str(SAMPLE_IMG)]
        )
        print("[green]Inserted image description into Milvus（content_type=image）[/green]")
    else:
        print("[yellow] sample/images/milvus_arch.png not found[/yellow]")
    # 5) Query: Text-based QA (from LightRAG)
    q1 = "Does Milvus support simultaneous insertion and search? Give a short answer."
    ans1 = await rag.aquery(q1, param=QueryParam(mode="hybrid"))
    print("\\n[bold]Text QA[/bold]")
    print(ans1)
    # 6) Query: Image-related QA (from Milvus)
    q2 = "What are the key components of the Milvus architecture?"
    q2_vec = await emb.embed_text(q2)
    img_hits = [store.search](<http://store.search>)([q2_vec], top_k=3, content_type="image")
    print("\\n[bold]Image Retrieval (returns semantic image descriptions)[/bold]")
    print(img_hits[0] if img_hits else [])
if __name__ == "__main__":
    [asyncio.run](<http://asyncio.run>)(main())

Ora è possibile testare il sistema RAG multimodale con il proprio set di dati.

Il futuro del RAG multimodale

Man mano che un numero sempre maggiore di dati del mondo reale va oltre il testo semplice, i sistemi RAG (Retrieval-Augmented Generation) iniziano a evolvere verso una vera multimodalità. Soluzioni come RAG-Anything dimostrano già come sia possibile elaborare in modo unificato testo, immagini, tabelle, formule e altri contenuti strutturati. In prospettiva, credo che tre tendenze principali determineranno la prossima fase della RAG multimodale:

Espansione a più modalità

I framework attuali, come RAG-Anything, possono già gestire testo, immagini, tabelle ed espressioni matematiche. La prossima frontiera è il supporto di tipi di contenuto ancora più ricchi, tra cui video, audio, dati di sensori e modelli 3D, consentendo ai sistemi RAG di comprendere e recuperare informazioni dall'intero spettro dei dati moderni.

Aggiornamenti dei dati in tempo reale

La maggior parte delle pipeline RAG oggi si basa su fonti di dati relativamente statiche. Poiché le informazioni cambiano più rapidamente, i sistemi futuri richiederanno aggiornamenti dei documenti in tempo reale, ingestione in streaming e indicizzazione incrementale. Questo cambiamento renderà il RAG più reattivo, tempestivo e affidabile in ambienti dinamici.

Spostamento di RAG sui dispositivi edge

Con strumenti vettoriali leggeri come Milvus Lite, la RAG multimodale non è più confinata nel cloud. L'implementazione del RAG sui dispositivi edge e sui sistemi IoT consente di effettuare un recupero intelligente più vicino al luogo in cui i dati vengono generati, migliorando la latenza, la privacy e l'efficienza complessiva.

👉 Siete pronti a esplorare il RAG multimodale?

Provate ad abbinare la vostra pipeline multimodale a Milvus e sperimentate un recupero veloce e scalabile di testi, immagini e altro ancora.

Avete domande o volete un approfondimento su una qualsiasi funzione? Unitevi al nostro canale Discord o inviate problemi su GitHub. È anche possibile prenotare una sessione individuale di 20 minuti per ottenere approfondimenti, indicazioni e risposte alle vostre domande attraverso Milvus Office Hours.

    Try Managed Milvus for Free

    Zilliz Cloud is hassle-free, powered by Milvus and 10x faster.

    Get Started

    Like the article? Spread the word

    Continua a Leggere