RAG multimodal simplifié : RAG-Anything + Milvus au lieu de 20 outils distincts
Construire un système RAG multimodal signifiait auparavant assembler une douzaine d'outils spécialisés - un pour l'OCR, un pour les tableaux, un pour les formules mathématiques, un pour les embeddings, un pour la recherche, et ainsi de suite. Les pipelines RAG traditionnels étaient conçus pour le texte, et dès que les documents commençaient à inclure des images, des tableaux, des équations, des graphiques et d'autres contenus structurés, la chaîne d'outils devenait rapidement désordonnée et ingérable.
RAG-Anything, développé par HKU, change cela. Basé sur LightRAG, il fournit une plateforme tout-en-un capable d'analyser divers types de contenus en parallèle et de les cartographier dans un graphe de connaissances unifié. Mais l'unification du pipeline n'est que la moitié de l'histoire. Pour extraire des preuves à travers ces différentes modalités, vous avez toujours besoin d'une recherche vectorielle rapide et évolutive qui peut traiter de nombreux types d'intégration à la fois. C'est là que Milvus entre en jeu. En tant que base de données vectorielles haute performance à code source ouvert, Milvus élimine la nécessité de recourir à de multiples solutions de stockage et de recherche. Il prend en charge la recherche ANN à grande échelle, la recherche hybride vecteur-mot-clé, le filtrage des métadonnées et la gestion flexible des incorporations, le tout en un seul endroit.
Dans cet article, nous expliquerons comment RAG-Anything et Milvus fonctionnent ensemble pour remplacer une chaîne d'outils multimodale fragmentée par une pile propre et unifiée, et nous montrerons comment vous pouvez construire un système RAG Q&A multimodal pratique en quelques étapes seulement.
Qu'est-ce que RAG-Anything et comment fonctionne-t-il ?
RAG-Anything est un cadre RAG conçu pour briser la barrière du texte seul des systèmes traditionnels. Au lieu de s'appuyer sur plusieurs outils spécialisés, il offre un environnement unique et unifié capable d'analyser, de traiter et d'extraire des informations à partir de différents types de contenu.
Le cadre prend en charge les documents contenant du texte, des diagrammes, des tableaux et des expressions mathématiques, ce qui permet aux utilisateurs d'effectuer des recherches dans toutes les modalités au moyen d'une interface unique et cohérente. Cela le rend particulièrement utile dans des domaines tels que la recherche universitaire, les rapports financiers et la gestion des connaissances d'entreprise, où les documents multimodaux sont courants.
RAG-Anything repose sur un pipeline multimodal à plusieurs étapes : analyse de documents→ analyse de contenu→ graphe de connaissances→ recherche intelligente. Cette architecture permet une orchestration intelligente et une compréhension multimodale, ce qui permet au système de traiter en toute transparence diverses modalités de contenu dans le cadre d'un flux de travail intégré unique.
L'architecture "1 + 3 + N
Au niveau de l'ingénierie, les capacités de RAG-Anything sont réalisées grâce à son architecture "1 + 3 + N" :
Le moteur central
Au centre de RAG-Anything se trouve un moteur de graphe de connaissances inspiré de LightRAG. Cette unité centrale est responsable de l'extraction multimodale des entités, de la mise en correspondance des relations intermodales et du stockage sémantique vectorisé. Contrairement aux systèmes RAG traditionnels basés sur le texte uniquement, le moteur comprend les entités du texte, les objets visuels dans les images et les structures relationnelles intégrées dans les tableaux.
3 Processeurs modaux
RAG-Anything intègre trois processeurs de modalité spécialisés, conçus pour une compréhension approfondie et spécifique à chaque modalité. Ensemble, ils forment la couche d'analyse multimodale du système.
ImageModalProcessor interprète le contenu visuel et sa signification contextuelle.
TableModalProcessor analyse les structures des tableaux et décode les relations logiques et numériques au sein des données.
EquationModalProcessor comprend la sémantique des symboles et des formules mathématiques.
N Analyseurs
Pour prendre en charge la structure diversifiée des documents du monde réel, RAG-Anything fournit une couche d'analyse syntaxique extensible fondée sur plusieurs moteurs d'extraction. Actuellement, il intègre à la fois MinerU et Docling, sélectionnant automatiquement l'analyseur optimal en fonction du type de document et de sa complexité structurelle.
S'appuyant sur l'architecture "1 + 3 + N", RAG-Anything améliore le pipeline RAG traditionnel en modifiant la manière dont les différents types de contenu sont traités. Au lieu de traiter le texte, les images et les tableaux un par un, le système les traite tous en même temps.
# 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
)
Cette conception accélère considérablement le traitement des documents techniques volumineux. Les tests d'étalonnage montrent que lorsque le système utilise davantage de cœurs d'unité centrale, il devient nettement plus rapide, ce qui réduit considérablement le temps nécessaire au traitement de chaque document.
Optimisation du stockage et de la recherche en couches
En plus de sa conception multimodale, RAG-Anything utilise également une approche de stockage et de récupération en couches pour rendre les résultats plus précis et plus efficaces.
Letexte est stocké dans une base de données vectorielle traditionnelle.
Lesimages sont gérées dans une base de données visuelle distincte.
Lestableaux sont conservés dans une base de données structurée.
Lesformules mathématiques sont transformées en vecteurs sémantiques.
En stockant chaque type de contenu dans son propre format, le système peut choisir la meilleure méthode de recherche pour chaque modalité au lieu de s'appuyer sur une recherche de similarité unique et générique. Cela permet d'obtenir des résultats plus rapides et plus fiables pour différents types de contenu.
Comment Milvus s'intègre dans RAG-Anything
RAG-Anything permet une récupération multimodale efficace, mais pour bien faire, il faut une recherche vectorielle rapide et évolutive pour tous les types d'encastrements. Milvus remplit parfaitement ce rôle.
Grâce à son architecture cloud-native et à la séparation calcul-stockage, Milvus offre à la fois une grande évolutivité et un bon rapport coût-efficacité. Il prend en charge la séparation lecture-écriture et l'unification par lots de flux, ce qui permet au système de gérer des charges de travail à forte concomitance tout en maintenant des performances de requête en temps réel - les nouvelles données deviennent consultables immédiatement après leur insertion.
Milvus garantit également une fiabilité de niveau entreprise grâce à sa conception distribuée et tolérante aux pannes, qui maintient le système stable même si des nœuds individuels tombent en panne. Il est donc parfaitement adapté aux déploiements de RAG multimodaux au niveau de la production.
Comment construire un système multimodal de questions-réponses avec RAG-Anything et Milvus
Cette démo montre comment construire un système de questions-réponses multimodal en utilisant le cadre RAG-Anything, la base de données vectorielle Milvus et le modèle d'intégration TongYi. (Cet exemple se concentre sur le code d'implémentation de base et n'est pas une installation de production complète).
Démonstration pratique
Prérequis：
Python : 3.10 ou supérieur
Base de données vectorielle : Service Milvus (Milvus Lite)
Service Cloud : Clé API Alibaba Cloud (pour les services LLM et d'intégration)
Modèle LLM :
qwen-vl-max(modèle basé sur la vision)
Modèle d'intégration: 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
Exécuter l'exemple de travail minimal :
python minimal_[main.py](<http://main.py>)
Résultat attendu :
Une fois le script exécuté avec succès, le terminal devrait afficher :
le résultat de la question-réponse textuelle générée par le LLM.
La description de l'image récupérée correspondant à la requête.
Structure du projet
.
├─ 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
Dépendances du projet
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
Variables d'environnement
# 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
Configuration du projet
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
Invocation du modèle
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"]
Intégration de 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
Point d'entrée principal
"""
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())
Vous pouvez maintenant tester votre système RAG multimodal avec votre propre ensemble de données.
L'avenir de la RAG multimodale
À mesure que les données du monde réel dépassent le simple texte, les systèmes RAG (Retrieval-Augmented Generation) commencent à évoluer vers une véritable multimodalité. Des solutions comme RAG-Anything démontrent déjà comment le texte, les images, les tableaux, les formules et d'autres contenus structurés peuvent être traités de manière unifiée. À l'avenir, je pense que trois grandes tendances façonneront la prochaine phase de la RAG multimodale :
Extension à d'autres modalités
Les cadres actuels, tels que RAG-Anything, peuvent déjà traiter du texte, des images, des tableaux et des expressions mathématiques. La prochaine étape consistera à prendre en charge des types de contenu encore plus riches, notamment la vidéo, l'audio, les données de capteurs et les modèles 3D, ce qui permettra aux systèmes RAG de comprendre et d'extraire des informations de l'ensemble du spectre des données modernes.
Mises à jour des données en temps réel
La plupart des pipelines RAG actuels s'appuient sur des sources de données relativement statiques. L'information évoluant plus rapidement, les futurs systèmes nécessiteront des mises à jour de documents en temps réel, une ingestion en continu et une indexation incrémentale. Cette évolution rendra le RAG plus réactif, plus opportun et plus fiable dans les environnements dynamiques.
Transférer le RAG vers les appareils périphériques
Avec des outils vectoriels légers tels que Milvus Lite, la GCR multimodale n'est plus confinée au nuage. Le déploiement de la RAG sur des dispositifs de périphérie et des systèmes IoT permet une récupération intelligente plus proche de l'endroit où les données sont générées, ce qui améliore la latence, la confidentialité et l'efficacité globale.
👉 Prêt à explorer la RAG multimodale ?
Essayez de coupler votre pipeline multimodal avec Milvus et faites l'expérience d'une recherche rapide et évolutive sur du texte, des images et bien plus encore.
Vous avez des questions ou souhaitez approfondir une fonctionnalité ? Rejoignez notre canal Discord ou déposez des questions sur GitHub. Vous pouvez également réserver une session individuelle de 20 minutes pour obtenir des informations, des conseils et des réponses à vos questions dans le cadre des Milvus Office Hours.
Try Managed Milvus for Free
Zilliz Cloud is hassle-free, powered by Milvus and 10x faster.
Get StartedLike the article? Spread the word
