Мультимодальный RAG стал простым: RAG-Anything + Milvus вместо 20 отдельных инструментов

Раньше создание мультимодальной системы RAG означало объединение десятка специализированных инструментов - один для OCR, один для таблиц, один для математических формул, один для вкраплений, один для поиска и так далее. Традиционные конвейеры RAG были разработаны для работы с текстом, а когда в документы стали включать изображения, таблицы, уравнения, диаграммы и другой структурированный контент, цепочка инструментов быстро стала запутанной и неуправляемой.

RAG-Anything, разработанный в HKU, меняет эту ситуацию. Построенная на базе 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 интегрированы три специализированных процессора модальностей, предназначенные для глубокого, специфического для каждой модальности понимания. Вместе они образуют слой мультимодального анализа системы.

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

Такая конструкция значительно ускоряет работу с большими техническими документами. Бенчмарк-тесты показывают, что при использовании большего количества ядер процессора система становится заметно быстрее, что резко сокращает время обработки каждого документа.

Многоуровневая оптимизация хранения и поиска информации

Помимо мультимодального дизайна, RAG-Anything также использует многоуровневый подход к хранению и поиску информации, чтобы сделать результаты более точными и эффективными.

• Текст хранится в традиционной векторной базе данных.

• Изображения хранятся в отдельном хранилище визуальных признаков.

• Таблицы хранятся в структурированном хранилище данных.

• Математические формулы превращаются в семантические векторы.

Храня каждый тип контента в своем собственном подходящем формате, система может выбрать лучший метод поиска для каждой модальности, а не полагаться на один общий поиск по сходству. Это позволяет быстрее и надежнее получать результаты по разным видам контента.

Как Milvus вписывается в RAG-Anything

RAG-Anything обеспечивает мощный мультимодальный поиск, но для этого требуется быстрый и масштабируемый векторный поиск по всем видам вкраплений. Milvus отлично справляется с этой ролью.

Благодаря облачной нативной архитектуре и разделению вычислений и хранения данных Milvus обеспечивает высокую масштабируемость и экономичность. Он поддерживает разделение чтения и записи и унификацию потоков и пакетов, что позволяет системе справляться с высококонкурентными рабочими нагрузками, сохраняя при этом производительность запросов в реальном времени - новые данные становятся доступными для поиска сразу после вставки.

Milvus также обеспечивает надежность корпоративного уровня благодаря распределенной отказоустойчивой конструкции, которая поддерживает стабильность системы даже в случае выхода из строя отдельных узлов. Это делает ее подходящей для развертывания мультимодальных RAG на уровне производства.

Как построить мультимодальную систему вопросов и ответов с помощью RAG-Anything и Milvus

В этой демонстрации показано, как построить мультимодальную систему вопросов и ответов с использованием фреймворка RAG-Anything, векторной базы данных Milvus и модели встраивания TongYi. (Этот пример посвящен основному коду реализации и не является полноценной производственной установкой).

Практическая демонстрация

Необходимые условия：

• Python: 3.10 или выше

• База данных векторов: Сервис Milvus (Milvus Lite)

• Облачный сервис: API-ключ Alibaba Cloud (для сервисов 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:

Расширение спектра модальностей

Существующие фреймворки, такие как RAG-Anything, уже могут работать с текстом, изображениями, таблицами и математическими выражениями. Следующий рубеж - поддержка еще более богатых типов контента, включая видео, аудио, данные датчиков и 3D-модели, что позволит системам RAG понимать и извлекать информацию из всего спектра современных данных.

Обновление данных в режиме реального времени

Большинство систем RAG сегодня опираются на относительно статичные источники данных. Поскольку информация меняется все быстрее, будущие системы будут требовать обновления документов в реальном времени, потокового ввода и инкрементного индексирования. Этот переход сделает RAG более отзывчивым, своевременным и надежным в динамичных средах.

Перемещение RAG на граничные устройства

Благодаря легким векторным инструментам, таким как Milvus Lite, мультимодальная система RAG больше не ограничивается облаком. Развертывание RAG на пограничных устройствах и IoT-системах позволяет осуществлять интеллектуальное извлечение данных ближе к месту их генерации, что повышает задержку, конфиденциальность и общую эффективность.

👉 Готовы изучить мультимодальный RAG?

Попробуйте объединить свой мультимодальный конвейер с Milvus и испытайте быстрый, масштабируемый поиск по тексту, изображениям и многому другому.

У вас есть вопросы или вы хотите получить подробную информацию о какой-либо функции? Присоединяйтесь к нашему каналу Discord или создавайте проблемы на GitHub. Вы также можете забронировать 20-минутную индивидуальную сессию, чтобы получить знания, рекомендации и ответы на свои вопросы в Milvus Office Hours.