Home
Blog
多模式 RAG 變得簡單：RAG-Anything + Milvus 取代 20 個獨立工具

多模式 RAG 變得簡單：RAG-Anything + Milvus 取代 20 個獨立工具

Tutorials

November 25, 2025

Min Yin

建立一個多模態的 RAG 系統曾經意味著要拼接一打專門的工具 - 一個用於 OCR、一個用於表格、一個用於數學公式、一個用於嵌入、一個用於搜尋，等等。傳統的 RAG 管道是針對文字設計的，一旦文件開始包含圖片、表格、方程式、圖表和其他結構化內容，工具鏈很快就會變得混亂且無法管理。

由香港大學開發的RAG-Anything 改變了這種情況。RAG-Anything 以 LightRAG 為基礎，提供了一個 All-in-One 平台，可以平行解析各種不同的內容類型，並將它們映射到統一的知識圖表中。但統一管道只是成功的一半。要擷取這些不同模式的證據，您仍需要一個快速、可擴充的向量搜尋，一次處理許多嵌入類型。這就是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 整合了三個專門的模態處理器，用於深入、特定模態的理解。它們共同組成了系統的多模態分析層。

ImageModalProcessor解譯視覺內容及其上下文意義。
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 建立多模式問答系統

本範例展示如何使用 RAG-Anything 框架、Milvus 向量資料庫和同義嵌入模型建立多模態問答系統。(本範例著重於核心實作程式碼，並非完整的生產設定)。

實作示範

先決條件： Python

Python3.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 產生的文字式問答結果。
與查詢對應的檢索影像描述。

專案結構

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

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

主要入口點

"""
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 的未來

隨著越來越多的真實世界資料超越純文字，Retrieval-Augmented Generation (RAG) 系統開始朝真正的多模態演進。RAG-Anything等解決方案已經展示了如何以統一的方式處理文字、影像、表格、公式和其他結構化內容。展望未來，我認為有三大趨勢將會塑造下一階段的多模態 RAG：