Milvus
Zilliz
  • Home
  • Blog

  • RAG Multimodal Menjadi Sederhana: RAG-Apa Saja + Milvus, Bukan 20 Alat Terpisah

RAG Multimodal Menjadi Sederhana: RAG-Apa Saja + Milvus, Bukan 20 Alat Terpisah

  • Tutorials
November 25, 2025
Min Yin

Membangun sistem RAG multimodal biasanya berarti menggabungkan selusin alat khusus-satu untuk OCR, satu untuk tabel, satu untuk rumus matematika, satu untuk penyematan, satu untuk pencarian, dan seterusnya. Pipeline RAG tradisional dirancang untuk teks, dan ketika dokumen mulai menyertakan gambar, tabel, persamaan, bagan, dan konten terstruktur lainnya, rantai alat dengan cepat menjadi berantakan dan tidak terkelola.

RAG-Anything, yang dikembangkan oleh HKU, mengubahnya. Dibangun di atas LightRAG, RAG-Anything menyediakan platform All-in-One yang dapat mengurai beragam jenis konten secara paralel dan memetakannya ke dalam grafik pengetahuan terpadu. Tetapi menyatukan pipeline hanya setengah dari cerita. Untuk mengambil bukti di seluruh modalitas yang bervariasi ini, Anda masih membutuhkan pencarian vektor yang cepat dan terukur yang dapat menangani banyak jenis penyematan sekaligus. Di sinilah Milvus hadir. Sebagai basis data vektor sumber terbuka dan berkinerja tinggi, Milvus menghilangkan kebutuhan akan beberapa solusi penyimpanan dan pencarian. Milvus mendukung pencarian ANN skala besar, pengambilan kata kunci vektor hibrida, pemfilteran metadata, dan manajemen penyematan yang fleksibel-semuanya di satu tempat.

Dalam tulisan ini, kami akan menguraikan bagaimana RAG-Anything dan Milvus bekerja sama untuk menggantikan toolchain multimodal yang terfragmentasi dengan tumpukan yang bersih dan terpadu - dan kami akan menunjukkan bagaimana Anda dapat membangun sistem tanya jawab RAG multimodal yang praktis hanya dengan beberapa langkah.

Apa Itu RAG-Anything dan Cara Kerjanya

RAG-Anything adalah kerangka kerja RAG yang dirancang untuk mendobrak batasan sistem tradisional yang hanya berupa teks. Alih-alih bergantung pada beberapa alat khusus, RAG-Anything menawarkan satu lingkungan terpadu yang dapat mengurai, memproses, dan mengambil informasi di berbagai jenis konten.

Kerangka kerja ini mendukung dokumen yang berisi teks, diagram, tabel, dan ekspresi matematis, sehingga memungkinkan pengguna untuk melakukan kueri di semua modalitas melalui satu antarmuka yang kohesif. Hal ini membuatnya sangat berguna dalam bidang-bidang seperti penelitian akademis, pelaporan keuangan, dan manajemen pengetahuan perusahaan, di mana materi multimodal adalah hal yang umum.

Pada intinya, RAG-Anything dibangun di atas pipeline multimodal multitahap: penguraian dokumen → analisis konten → grafik pengetahuan → pengambilan cerdas. Arsitektur ini memungkinkan orkestrasi cerdas dan pemahaman lintas-modal, sehingga sistem dapat menangani beragam modalitas konten dengan lancar dalam satu alur kerja yang terintegrasi.

Arsitektur "1 + 3 + N"

Pada tingkat teknik, kemampuan RAG-Anything diwujudkan melalui arsitektur "1 + 3 + N":

Mesin Inti

Inti dari RAG-Anything adalah mesin grafik pengetahuan yang terinspirasi oleh LightRAG. Unit inti ini bertanggung jawab untuk ekstraksi entitas multimodal, pemetaan hubungan lintas modal, dan penyimpanan semantik vektor. Tidak seperti sistem RAG yang hanya menggunakan teks, mesin ini memahami entitas dari teks, objek visual di dalam gambar, dan struktur relasional yang tertanam di dalam tabel.

3 Prosesor Modal

RAG-Anything mengintegrasikan tiga prosesor modalitas khusus yang dirancang untuk pemahaman yang mendalam dan spesifik terhadap modalitas. Bersama-sama, mereka membentuk lapisan analisis multimodal sistem.

  • ImageModalProcessor menafsirkan konten visual dan makna kontekstualnya.

  • TableModalProcessor mengurai struktur tabel dan menerjemahkan hubungan logis dan numerik dalam data.

  • EquationModalProcessor memahami semantik di balik simbol dan rumus matematika.

Pengurai N

Untuk mendukung beragam struktur dokumen dunia nyata, RAG-Anything menyediakan lapisan penguraian yang dapat diperluas yang dibangun di atas beberapa mesin ekstraksi. Saat ini, RAG-Anything mengintegrasikan MinerU dan Docling, yang secara otomatis memilih pengurai yang optimal berdasarkan jenis dokumen dan kompleksitas struktural.

Dibangun di atas arsitektur "1 + 3 + N", RAG-Anything meningkatkan pipeline RAG tradisional dengan mengubah cara penanganan jenis konten yang berbeda. Alih-alih memproses teks, gambar, dan tabel satu per satu, sistem memproses semuanya sekaligus.

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

Desain ini sangat mempercepat penanganan dokumen teknis yang besar. Tes benchmark menunjukkan bahwa ketika sistem menggunakan lebih banyak inti CPU, sistem menjadi lebih cepat, yang secara tajam mengurangi waktu yang dibutuhkan untuk memproses setiap dokumen.

Penyimpanan Berlapis dan Optimalisasi Pengambilan

Selain desain multimodalnya, RAG-Anything juga menggunakan pendekatan penyimpanan dan pengambilan berlapis untuk membuat hasil yang lebih akurat dan efisien.

  • Teks disimpan dalam basis data vektor tradisional.

  • Gambar dikelola dalam penyimpanan fitur visual yang terpisah.

  • Tabel disimpan dalam penyimpanan data terstruktur.

  • Rumus matematika diubah menjadi vektor semantik.

Dengan menyimpan setiap jenis konten dalam format yang sesuai, sistem dapat memilih metode pencarian terbaik untuk setiap modalitas daripada mengandalkan pencarian kesamaan umum. Hal ini menghasilkan hasil yang lebih cepat dan lebih dapat diandalkan di berbagai jenis konten.

Bagaimana Milvus Cocok dengan RAG-Anything

RAG-Anything menyediakan pencarian multimodal yang kuat, tetapi untuk melakukannya dengan baik diperlukan pencarian vektor yang cepat dan terukur di semua jenis penyematan. Milvus mengisi peran ini dengan sempurna.

Dengan arsitektur cloud-native dan pemisahan komputasi-penyimpanan, Milvus memberikan skalabilitas tinggi dan efisiensi biaya. Milvus mendukung pemisahan baca-tulis dan penyatuan stream-batch, sehingga sistem dapat menangani beban kerja dengan konkurensi tinggi sambil mempertahankan kinerja kueri waktu-nyata - data baru dapat dicari segera setelah dimasukkan.

Milvus juga memastikan keandalan tingkat perusahaan melalui desain terdistribusi dan toleran terhadap kesalahan, yang menjaga sistem tetap stabil meskipun ada node yang gagal. Hal ini membuatnya sangat cocok untuk penerapan RAG multimodal tingkat produksi.

Cara Membangun Sistem Tanya Jawab Multimodal dengan RAG-Anything dan Milvus

Demo ini menunjukkan cara membangun sistem tanya jawab multimodal menggunakan kerangka kerja RAG-Anything, basis data vektor Milvus, dan model penyematan TongYi. (Contoh ini berfokus pada kode implementasi inti dan bukan merupakan penyiapan produksi penuh).

Demo Langsung

Prasyarat: Persyaratan

  • Python: 3.10 atau lebih tinggi

  • Basis Data Vektor: Layanan Milvus (Milvus Lite)

  • Layanan Cloud: Kunci API Alibaba Cloud (untuk LLM dan layanan penyematan)

  • Model LLM: qwen-vl-max (model yang mendukung visi)

Model Penyematan: 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

Jalankan contoh kerja minimal:

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

Keluaran yang diharapkan:

Setelah skrip berhasil dijalankan, terminal akan ditampilkan:

  • Hasil Tanya Jawab berbasis teks yang dihasilkan oleh LLM.

  • Deskripsi gambar yang diambil yang sesuai dengan kueri.

Struktur Proyek

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

Ketergantungan Proyek

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

Variabel Lingkungan

# 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

Konfigurasi

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

Pemanggilan Model

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

Integrasi 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

Titik Masuk Utama

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

Sekarang, Anda dapat menguji sistem RAG multimodal Anda dengan dataset Anda sendiri.

Masa Depan untuk RAG Multimodal

Seiring dengan semakin banyaknya data dunia nyata yang bergerak di luar teks biasa, sistem Retrieval-Augmented Generation (RAG) mulai berevolusi menuju multimodalitas yang sesungguhnya. Solusi seperti RAG-Anything telah menunjukkan bagaimana teks, gambar, tabel, formula, dan konten terstruktur lainnya dapat diproses dengan cara yang terpadu. Ke depannya, menurut saya ada tiga tren utama yang akan membentuk fase berikutnya dari RAG multimodal:

Memperluas ke Lebih Banyak Modalitas

Kerangka kerja yang ada saat ini-seperti RAG-Anything-sudah dapat menangani teks, gambar, tabel, dan ekspresi matematika. Perbatasan berikutnya adalah mendukung jenis konten yang lebih kaya lagi, termasuk video, audio, data sensor, dan model 3D, sehingga memungkinkan sistem RAG untuk memahami dan mengambil informasi dari spektrum penuh data modern.

Pembaruan Data Waktu Nyata

Sebagian besar pipeline RAG saat ini mengandalkan sumber data yang relatif statis. Seiring perubahan informasi yang semakin cepat, sistem masa depan akan membutuhkan pembaruan dokumen secara real-time, konsumsi streaming, dan pengindeksan tambahan. Pergeseran ini akan membuat RAG lebih responsif, tepat waktu, dan dapat diandalkan dalam lingkungan yang dinamis.

Memindahkan RAG ke Perangkat Edge

Dengan alat vektor yang ringan seperti Milvus Lite, RAG multimodal tidak lagi terbatas pada cloud. Menerapkan RAG pada perangkat edge dan sistem IoT memungkinkan pengambilan cerdas dilakukan lebih dekat dengan tempat data dihasilkan - meningkatkan latensi, privasi, dan efisiensi secara keseluruhan.

👉 Siap menjelajahi RAG multimodal?

Coba pasangkan pipeline multimodal Anda dengan Milvus dan rasakan pengambilan yang cepat dan terukur di seluruh teks, gambar, dan banyak lagi.

Ada pertanyaan atau ingin mendalami fitur apa pun? Bergabunglah dengan saluran Discord kami atau ajukan pertanyaan di GitHub. Anda juga dapat memesan sesi tatap muka selama 20 menit untuk mendapatkan wawasan, panduan, dan jawaban atas pertanyaan Anda melalui 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

    Terus Baca