Milvus
Zilliz
  • Home
  • Blog

  • 간편해진 멀티모달 RAG: 20개의 개별 도구 대신 RAG-Anything + Milvus 사용

간편해진 멀티모달 RAG: 20개의 개별 도구 대신 RAG-Anything + Milvus 사용

  • Tutorials
November 25, 2025
Min Yin

멀티모달 RAG 시스템을 구축한다는 것은 OCR용, 표용, 수학 공식용, 임베딩용, 검색용 등 수십 개의 전문화된 도구를 하나로 묶는 것을 의미했습니다. 기존의 RAG 파이프라인은 텍스트용으로 설계되어 문서에 이미지, 표, 수식, 차트 및 기타 구조화된 콘텐츠가 포함되기 시작하면 툴체인은 금방 지저분해지고 관리하기 어려워졌습니다.

HKU에서 개발한RAG-Anything은 이러한 문제를 해결합니다. LightRAG를 기반으로 구축된 이 플랫폼은 다양한 콘텐츠 유형을 병렬로 구문 분석하여 통합 지식 그래프에 매핑할 수 있는 올인원 플랫폼을 제공합니다. 하지만 파이프라인을 통합하는 것은 이야기의 절반에 불과합니다. 이렇게 다양한 양식에서 증거를 검색하려면 여전히 많은 임베딩 유형을 한 번에 처리할 수 있는 빠르고 확장 가능한 벡터 검색이 필요합니다. 이것이 바로 Milvus가 필요한 이유입니다. 오픈 소스 고성능 벡터 데이터베이스인 Milvus는 여러 개의 스토리지 및 검색 솔루션이 필요하지 않습니다. 대규모 ANN 검색, 하이브리드 벡터 키워드 검색, 메타데이터 필터링, 유연한 임베딩 관리 기능을 모두 한 곳에서 지원합니다.

이 포스팅에서는 RAG-Anything과 Milvus가 어떻게 함께 작동하여 파편화된 멀티모달 툴체인을 깔끔하고 통합된 스택으로 대체하는지, 그리고 몇 단계만으로 실용적인 멀티모달 RAG Q&A 시스템을 구축하는 방법을 보여드리겠습니다.

RAG-Anything이란 무엇이며 어떻게 작동하나요?

RAG-Anything은 기존 시스템의 텍스트 전용 장벽을 허물기 위해 고안된 RAG 프레임워크입니다. 여러 전문 도구에 의존하는 대신, 혼합된 콘텐츠 유형에서 정보를 구문 분석, 처리 및 검색할 수 있는 단일 통합 환경을 제공합니다.

이 프레임워크는 텍스트, 다이어그램, 표, 수식이 포함된 문서를 지원하므로 사용자는 하나의 일관된 인터페이스를 통해 모든 양식에 걸쳐 쿼리할 수 있습니다. 따라서 학술 연구, 재무 보고, 기업 지식 관리와 같이 복합 형식의 자료가 일반적인 분야에서 특히 유용합니다.

RAG-Anything의 핵심은 문서 구문 분석→콘텐츠 분석→지식 그래프→지능형 검색이라는 다단계 멀티모달 파이프라인을 기반으로 구축됩니다. 이 아키텍처는 지능형 오케스트레이션과 교차 모달 이해를 지원하여 시스템이 단일 통합 워크플로우 내에서 다양한 콘텐츠 모달리티를 원활하게 처리할 수 있게 해줍니다.

"1 + 3 + N" 아키텍처

엔지니어링 수준에서 RAG-Anything의 기능은 "1 + 3 + N" 아키텍처를 통해 실현됩니다:

핵심 엔진

RAG-Anything의 중심에는 LightRAG에서 영감을 얻은 지식 그래프 엔진이 있습니다. 이 핵심 유닛은 멀티모달 엔티티 추출, 크로스모달 관계 매핑, 벡터화된 시맨틱 스토리지를 담당합니다. 기존의 텍스트 전용 RAG 시스템과 달리, 이 엔진은 텍스트, 이미지 내의 시각적 개체, 표에 포함된 관계형 구조에서 엔티티를 이해합니다.

3가지 모달 프로세서

RAG-Anything은 심층적인 양식별 이해를 위해 설계된 세 가지 특수 양식 프로세서를 통합합니다. 이 세 가지 프로세서가 함께 시스템의 멀티모달 분석 레이어를 구성합니다.

  • 이미지모달프로세서는 시각적 콘텐츠와 그 문맥적 의미를 해석합니다.

  • TableModalProcessor는 테이블 구조를 파싱하고 데이터 내의 논리적 및 수치적 관계를 해독합니다.

  • EquationModalProcessor는 수학적 기호와 공식의 의미를 이해합니다.

N 파서

실제 문서의 다양한 구조를 지원하기 위해 RAG-Anything은 여러 추출 엔진에 구축된 확장 가능한 구문 분석 계층을 제공합니다. 현재 MinerU와 Docling을 모두 통합하여 문서 유형과 구조적 복잡성에 따라 최적의 구문 분석기를 자동으로 선택합니다.

"1 + 3 + N" 아키텍처를 기반으로 하는 RAG-Anything은 다양한 콘텐츠 유형이 처리되는 방식을 변경하여 기존 RAG 파이프라인을 개선합니다. 텍스트, 이미지, 표를 한 번에 하나씩 처리하는 대신 한 번에 모두 처리합니다.

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

이 설계는 대용량 기술 문서의 처리 속도를 크게 높여줍니다. 벤치마크 테스트에 따르면 시스템이 더 많은 CPU 코어를 사용하면 속도가 눈에 띄게 빨라져 각 문서를 처리하는 데 필요한 시간이 급격히 줄어듭니다.

계층화된 스토리지 및 검색 최적화

RAG-Anything은 멀티모달 설계 외에도 계층화된 저장 및 검색 방식을 사용해 결과를 더욱 정확하고 효율적으로 만듭니다.

  • 텍스트는 기존의 벡터 데이터베이스에 저장됩니다.

  • 이미지는 별도의 시각적 특징 저장소에서 관리됩니다.

  • 표는 구조화된 데이터 저장소에 보관됩니다.

  • 수학 공식은 시맨틱 벡터로 변환됩니다.

각 콘텐츠 유형을 적절한 형식으로 저장함으로써 시스템은 하나의 일반적인 유사성 검색에 의존하는 대신 각 양식에 가장 적합한 검색 방법을 선택할 수 있습니다. 따라서 다양한 종류의 콘텐츠에 대해 더 빠르고 신뢰할 수 있는 결과를 얻을 수 있습니다.

Milvus가 RAG-Anything에 적합한 방법

RAG-Anything은 강력한 멀티모달 검색 기능을 제공하지만, 이를 제대로 수행하려면 모든 종류의 임베딩에 걸쳐 빠르고 확장 가능한 벡터 검색이 필요합니다. Milvus는 이 역할을 완벽하게 수행합니다.

클라우드 네이티브 아키텍처와 컴퓨팅-스토리지 분리를 통해 Milvus는 높은 확장성과 비용 효율성을 모두 제공합니다. 읽기-쓰기 분리 및 스트림-배치 통합을 지원하여 시스템이 실시간 쿼리 성능을 유지하면서 동시성이 높은 워크로드를 처리할 수 있으며, 새로운 데이터를 삽입하는 즉시 검색할 수 있습니다.

또한 Milvus는 분산형 내결함성 설계를 통해 엔터프라이즈급 안정성을 보장하므로 개별 노드에 장애가 발생하더라도 시스템이 안정적으로 유지됩니다. 따라서 프로덕션 수준의 멀티모달 RAG 배포에 매우 적합합니다.

RAG-Anything과 Milvus로 멀티모달 Q&A 시스템을 구축하는 방법

이 데모에서는 RAG-Anything 프레임워크, Milvus 벡터 데이터베이스, TongYi 임베딩 모델을 사용하여 멀티모달 Q&A 시스템을 구축하는 방법을 보여드립니다. (이 예제는 핵심 구현 코드에 중점을 두고 있으며 전체 프로덕션 설정이 아닙니다.)

실습 데모

사전 요구 사항

  • Python: 3.10 이상

  • 벡터 데이터베이스: Milvus 서비스(Milvus Lite)

  • 클라우드 서비스: 알리바바 클라우드 API 키(LLM 및 임베딩 서비스용)

  • LLM 모델: qwen-vl-max (비전 지원 모델)

임베딩 모델: 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

최소한의 작업 예제를 실행합니다:

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

예상 출력:

스크립트가 성공적으로 실행되면 터미널이 표시되어야 합니다:

  • LLM에서 생성된 텍스트 기반 Q&A 결과.

  • 쿼리에 해당하는 검색된 이미지 설명.

프로젝트 구조

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

프로젝트 종속성

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

환경 변수

# 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

구성

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

모델 호출

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

밀버스 라이트 통합

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

주요 진입점

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

이제 자체 데이터 세트로 멀티모달 RAG 시스템을 테스트할 수 있습니다.

멀티모달 RAG의 미래

더 많은 실제 데이터가 일반 텍스트를 넘어서면서 검색 증강 생성(RAG) 시스템은 진정한 멀티모달로 진화하기 시작했습니다. RAG-Anything과 같은 솔루션은 이미 텍스트, 이미지, 표, 수식 및 기타 구조화된 콘텐츠를 통합된 방식으로 처리할 수 있는 방법을 보여줍니다. 앞으로 세 가지 주요 트렌드가 멀티모달 RAG의 다음 단계를 형성할 것이라고 생각합니다:

더 많은 모달리티로 확장

RAG-Anything과 같은 현재의 프레임워크는 이미 텍스트, 이미지, 표, 수학적 표현을 처리할 수 있습니다. 다음 단계는 비디오, 오디오, 센서 데이터, 3D 모델 등 훨씬 더 풍부한 콘텐츠 유형을 지원하여 RAG 시스템이 최신 데이터의 전체 스펙트럼에서 정보를 이해하고 검색할 수 있도록 하는 것입니다.

실시간 데이터 업데이트

오늘날 대부분의 RAG 파이프라인은 비교적 정적인 데이터 소스에 의존합니다. 정보가 더욱 빠르게 변화함에 따라 미래의 시스템에는 실시간 문서 업데이트, 스트리밍 수집, 증분 색인 기능이 요구될 것입니다. 이러한 변화는 RAG가 동적인 환경에서 더욱 신속하고 시의적절하며 안정적으로 대응할 수 있게 해줄 것입니다.

RAG를 엣지 디바이스로 이동

Milvus Lite와 같은 경량 벡터 도구를 사용하면 멀티모달 RAG가 더 이상 클라우드에만 국한되지 않습니다. 엣지 디바이스와 IoT 시스템에 RAG를 배포하면 데이터가 생성되는 곳에서 더 가까운 곳에서 지능적인 검색을 수행할 수 있어 지연 시간, 개인정보 보호, 전반적인 효율성이 향상됩니다.

👉 멀티모달 RAG를 살펴볼 준비가 되셨나요?

멀티모달 파이프라인을 Milvus와 페어링하여 텍스트, 이미지 등에 걸쳐 빠르고 확장 가능한 검색을 경험해 보세요.

궁금한 점이 있거나 기능에 대해 자세히 알아보고 싶으신가요? Discord 채널에 참여하거나 GitHub에 이슈를 제출하세요. 또한 Milvus 오피스 아워를 통해 20분간의 일대일 세션을 예약하여 인사이트, 안내, 질문에 대한 답변을 얻을 수도 있습니다.

    Try Managed Milvus for Free

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

    Get Started

    Like the article? Spread the word

    계속 읽기