Multimodale RAG einfach gemacht: RAG-Anything + Milvus statt 20 separater Tools

Der Aufbau eines multimodalen RAG-Systems bedeutete früher, dass man ein Dutzend spezialisierter Tools zusammenstellen musste - eines für OCR, eines für Tabellen, eines für mathematische Formeln, eines für Einbettungen, eines für die Suche und so weiter. Herkömmliche RAG-Pipelines wurden für Text entwickelt, und sobald Dokumente Bilder, Tabellen, Gleichungen, Diagramme und andere strukturierte Inhalte enthielten, wurde die Toolchain schnell unübersichtlich und unhandlich.

RAG-Anything, entwickelt von der HKU, ändert das. Es baut auf LightRAG auf und bietet eine All-in-One-Plattform, die verschiedene Inhaltstypen parallel analysieren und in einem einheitlichen Wissensgraphen abbilden kann. Doch die Vereinheitlichung der Pipeline ist nur die halbe Miete. Um Beweise über diese verschiedenen Modalitäten hinweg abzurufen, benötigen Sie immer noch eine schnelle, skalierbare Vektorsuche, die viele Einbettungstypen auf einmal verarbeiten kann. Hier kommt Milvus ins Spiel. Als quelloffene, leistungsstarke Vektordatenbank macht Milvus mehrere Speicher- und Suchlösungen überflüssig. Sie unterstützt eine umfangreiche ANN-Suche, eine hybride Abfrage von Vektor- und Schlüsselwörtern, die Filterung von Metadaten und eine flexible Verwaltung von Einbettungen - alles an einem Ort.

In diesem Beitrag werden wir aufschlüsseln, wie RAG-Anything und Milvus zusammenarbeiten, um eine fragmentierte multimodale Toolchain durch einen sauberen, einheitlichen Stack zu ersetzen - und wir werden zeigen, wie Sie ein praktisches multimodales RAG-Q&A-System mit nur wenigen Schritten aufbauen können.

Was ist RAG-Anything und wie funktioniert es?

RAG-Anything ist ein RAG-Framework, das entwickelt wurde, um die Barriere traditioneller Systeme zu durchbrechen, die nur aus Text bestehen. Anstatt sich auf mehrere spezialisierte Tools zu verlassen, bietet es eine einzige, einheitliche Umgebung, die Informationen über verschiedene Inhaltstypen hinweg analysieren, verarbeiten und abrufen kann.

Das Framework unterstützt Dokumente, die Text, Diagramme, Tabellen und mathematische Ausdrücke enthalten, und ermöglicht es den Benutzern, alle Modalitäten über eine einzige einheitliche Schnittstelle abzufragen. Dies macht es besonders nützlich in Bereichen wie der akademischen Forschung, der Finanzberichterstattung und dem Wissensmanagement in Unternehmen, wo multimodale Materialien üblich sind.

Im Kern basiert RAG-Anything auf einer mehrstufigen multimodalen Pipeline: Dokumentenparsing→Inhaltsanalyse→Wissensgraf→intelligentes Retrieval. Diese Architektur ermöglicht eine intelligente Orchestrierung und ein cross-modales Verständnis, so dass das System nahtlos verschiedene Inhaltsmodalitäten innerhalb eines einzigen integrierten Workflows verarbeiten kann.

Die "1 + 3 + N"-Architektur

Auf der technischen Ebene werden die Fähigkeiten von RAG-Anything durch seine "1 + 3 + N"-Architektur realisiert:

Die Core Engine

Im Zentrum von RAG-Anything steht eine von LightRAG inspirierte Wissensgraphen-Engine. Diese Kerneinheit ist für die multimodale Entitätsextraktion, das cross-modale Relationship Mapping und die vektorisierte semantische Speicherung verantwortlich. Im Gegensatz zu traditionellen, rein textbasierten RAG-Systemen versteht die Engine Entitäten aus Texten, visuelle Objekte in Bildern und in Tabellen eingebettete relationale Strukturen.

3 Modale Prozessoren

RAG-Anything integriert drei spezialisierte Modalitätsprozessoren, die für ein tiefes, modalitätsspezifisches Verständnis entwickelt wurden. Zusammen bilden sie die multimodale Analyseschicht des Systems.

• ImageModalProcessor interpretiert visuelle Inhalte und deren kontextuelle Bedeutung.

• TableModalProcessor analysiert Tabellenstrukturen und entschlüsselt logische und numerische Beziehungen innerhalb der Daten.

• EquationModalProcessor versteht die Semantik hinter mathematischen Symbolen und Formeln.

N-Parser

Um die vielfältigen Strukturen von realen Dokumenten zu unterstützen, bietet RAG-Anything eine erweiterbare Parsing-Schicht, die auf mehreren Extraktions-Engines aufbaut. Derzeit sind sowohl MinerU als auch Docling integriert, wobei automatisch der optimale Parser auf der Grundlage des Dokumenttyps und der strukturellen Komplexität ausgewählt wird.

Aufbauend auf der "1 + 3 + N"-Architektur verbessert RAG-Anything die traditionelle RAG-Pipeline, indem es den Umgang mit verschiedenen Inhaltstypen verändert. Anstatt Text, Bilder und Tabellen einzeln zu verarbeiten, verarbeitet das System sie alle auf einmal.

# 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
)

Dieses Design beschleunigt die Verarbeitung großer technischer Dokumente erheblich. Benchmark-Tests zeigen, dass das System deutlich schneller wird, wenn es mehr CPU-Kerne nutzt, wodurch sich die für die Verarbeitung der einzelnen Dokumente benötigte Zeit drastisch verringert.

Optimierung der Speicherung und des Abrufs auf mehreren Ebenen

Zusätzlich zu seinem multimodalen Design verwendet RAG-Anything auch einen mehrschichtigen Speicher- und Abrufansatz, um die Ergebnisse genauer und effizienter zu machen.

• Text wird in einer traditionellen Vektordatenbank gespeichert.

• Bilder werden in einem separaten visuellen Merkmalsspeicher verwaltet.

• Tabellen werden in einem strukturierten Datenspeicher aufbewahrt.

• Mathematische Formeln werden in semantische Vektoren umgewandelt.

Durch die Speicherung jedes Inhaltstyps in einem eigenen geeigneten Format kann das System die beste Abrufmethode für jede Modalität wählen, anstatt sich auf eine einzige, generische Ähnlichkeitssuche zu verlassen. Dies führt zu schnelleren und zuverlässigeren Ergebnissen für verschiedene Arten von Inhalten.

Wie sich Milvus in RAG-Anything einfügt

RAG-Anything bietet ein starkes multimodales Retrieval, aber dies erfordert eine schnelle und skalierbare Vektorsuche über alle Arten von Einbettungen. Milvus füllt diese Rolle perfekt aus.

Mit seiner Cloud-nativen Architektur und der Trennung von Datenverarbeitung und Speicher bietet Milvus sowohl hohe Skalierbarkeit als auch Kosteneffizienz. Milvus unterstützt die Trennung von Lese- und Schreibvorgängen und die Stream-Batch-Vereinigung, sodass das System Arbeitslasten mit hoher Gleichzeitigkeit bewältigen kann und gleichzeitig die Abfrageleistung in Echtzeit beibehält - neue Daten werden sofort nach dem Einfügen durchsuchbar.

Milvus gewährleistet durch sein verteiltes, fehlertolerantes Design, das das System auch dann stabil hält, wenn einzelne Knoten ausfallen, Zuverlässigkeit auf Unternehmensniveau. Damit eignet sich Milvus hervorragend für multimodale RAG-Implementierungen auf Produktionsebene.

Wie man ein multimodales Q&A-System mit RAG-Anything und Milvus aufbaut

Diese Demo zeigt, wie man ein multimodales Q&A System mit dem RAG-Anything Framework, der Milvus Vektordatenbank und dem TongYi Einbettungsmodell erstellt. (Dieses Beispiel konzentriert sich auf den Kernimplementierungscode und ist keine vollständige Produktionseinrichtung).

Hands-on-Demo

Voraussetzungen：

• Python: 3.10 oder höher

• Vektordatenbank: Milvus-Dienst (Milvus Lite)

• Cloud-Dienst: Alibaba Cloud API-Schlüssel (für LLM- und Einbettungsdienste)

• LLM-Modell: qwen-vl-max (Vision-fähiges Modell)

Einbettungsmodell: 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

Führen Sie das minimale Arbeitsbeispiel aus:

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

Erwartete Ausgabe:

Sobald das Skript erfolgreich ausgeführt wurde, sollte das Terminal angezeigt werden:

• Das vom LLM generierte textbasierte Q&A-Ergebnis.

• Die gefundene Bildbeschreibung, die der Abfrage entspricht.

Projektstruktur

.
├─ 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

Projekt-Abhängigkeiten

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

Umgebungsvariablen

# 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

Konfiguration

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

Modell-Aufruf

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"]

Milvus-Lite-Integration

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

Haupteinstiegspunkt

"""
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())

Jetzt können Sie Ihr multimodales RAG-System mit Ihrem eigenen Datensatz testen.

Die Zukunft für multimodales RAG

Da immer mehr Daten aus der realen Welt über reinen Text hinausgehen, entwickeln sich Retrieval-Augmented Generation (RAG) Systeme in Richtung echter Multimodalität weiter. Lösungen wie RAG-Anything zeigen bereits, wie Text, Bilder, Tabellen, Formeln und andere strukturierte Inhalte auf einheitliche Weise verarbeitet werden können. Mit Blick auf die Zukunft denke ich, dass drei wichtige Trends die nächste Phase der multimodalen RAG prägen werden:

Ausweitung auf weitere Modalitäten

Aktuelle Frameworks - wie RAG-Anything - können bereits Text, Bilder, Tabellen und mathematische Ausdrücke verarbeiten. Die nächste Stufe ist die Unterstützung von noch umfangreicheren Inhaltstypen, einschließlich Video, Audio, Sensordaten und 3D-Modellen, wodurch RAG-Systeme in die Lage versetzt werden, Informationen aus dem gesamten Spektrum moderner Daten zu verstehen und abzurufen.

Datenaktualisierung in Echtzeit

Die meisten RAG-Pipelines basieren heute auf relativ statischen Datenquellen. Da sich Informationen immer schneller ändern, werden künftige Systeme Dokumentenaktualisierungen in Echtzeit, Streaming-Ingestion und inkrementelle Indizierung erfordern. Durch diesen Wandel wird RAG in dynamischen Umgebungen reaktionsschneller, zeitnaher und zuverlässiger.

Verlagerung von RAG auf Edge-Geräte

Mit leichtgewichtigen Vektor-Tools wie Milvus Lite ist die multimodale RAG nicht mehr auf die Cloud beschränkt. Der Einsatz von RAG auf Edge-Geräten und IoT-Systemen ermöglicht eine intelligente Abfrage näher am Ort der Datengenerierung, wodurch Latenzzeiten, Datenschutz und Gesamteffizienz verbessert werden.

👉 Sind Sie bereit, multimodales RAG zu erkunden?

Versuchen Sie, Ihre multimodale Pipeline mit Milvus zu koppeln, und erleben Sie schnelle, skalierbare Abfragen für Text, Bilder und mehr.

Haben Sie Fragen oder möchten Sie eine Funktion genauer kennenlernen? Treten Sie unserem Discord-Kanal bei oder stellen Sie Fragen auf GitHub. Sie können auch ein 20-minütiges persönliches Gespräch buchen, um Einblicke, Anleitung und Antworten auf Ihre Fragen in den Milvus-Sprechstunden zu erhalten.