RAG multimodal simplificado: RAG-Anything + Milvus en lugar de 20 herramientas distintas

Construir un sistema de GAR multimodal solía significar unir una docena de herramientas especializadas: una para OCR, otra para tablas, otra para fórmulas matemáticas, otra para incrustaciones, otra para búsquedas, etcétera. Los canales tradicionales de RAG se diseñaban para texto y, cuando los documentos empezaban a incluir imágenes, tablas, ecuaciones, gráficos y otros contenidos estructurados, la cadena de herramientas se volvía rápidamente desordenada e inmanejable.

RAG-Anything, desarrollado por HKU, cambia esta situación. Basada en LightRAG, ofrece una plataforma "todo en uno" capaz de analizar diversos tipos de contenido en paralelo y mapearlos en un grafo de conocimiento unificado. Pero unificar el proceso es sólo la mitad de la historia. Para recuperar pruebas en estas diversas modalidades, se necesita una búsqueda vectorial rápida y escalable que pueda manejar muchos tipos de incrustación a la vez. Ahí es donde entra Milvus. Como base de datos vectorial de código abierto y alto rendimiento, Milvus elimina la necesidad de múltiples soluciones de almacenamiento y búsqueda. Admite la búsqueda RNA a gran escala, la recuperación híbrida de vectores y palabras clave, el filtrado de metadatos y la gestión flexible de incrustaciones, todo en un solo lugar.

En este artículo, desglosaremos cómo RAG-Anything y Milvus trabajan juntos para reemplazar una cadena de herramientas multimodal fragmentada por una pila limpia y unificada, y le mostraremos cómo puede construir un práctico sistema RAG multimodal de preguntas y respuestas con sólo unos pocos pasos.

Qué es RAG-Anything y cómo funciona

RAG-Anything es un marco RAG diseñado para romper la barrera de los sistemas tradicionales basados únicamente en texto. En lugar de depender de múltiples herramientas especializadas, ofrece un único entorno unificado que puede analizar, procesar y recuperar información a través de distintos tipos de contenido.

El marco admite documentos que contengan texto, diagramas, tablas y expresiones matemáticas, lo que permite a los usuarios realizar consultas en todas las modalidades a través de una única interfaz cohesionada. Esto lo hace especialmente útil en campos como la investigación académica, los informes financieros y la gestión del conocimiento empresarial, donde los materiales multimodales son habituales.

En esencia, RAG-Anything se basa en un proceso multimodal de varias etapas: análisis sintáctico de documentos→análisis de contenidos→grafos de conocimiento→recuperación inteligente. Esta arquitectura permite la orquestación inteligente y la comprensión multimodal, lo que permite al sistema manejar sin problemas diversas modalidades de contenido dentro de un único flujo de trabajo integrado.

La arquitectura "1 + 3 + N

A nivel de ingeniería, las capacidades de RAG-Anything se materializan a través de su arquitectura "1 + 3 + N":

El motor central

En el centro de RAG-Anything se encuentra un motor de grafos de conocimiento inspirado en LightRAG. Esta unidad central es responsable de la extracción de entidades multimodales, el mapeo de relaciones intermodales y el almacenamiento semántico vectorizado. A diferencia de los sistemas RAG tradicionales basados únicamente en texto, el motor comprende entidades de texto, objetos visuales dentro de imágenes y estructuras relacionales incrustadas en tablas.

3 Procesadores modales

RAG-Anything integra tres procesadores de modalidad especializados diseñados para una comprensión profunda y específica de la modalidad. Juntos forman la capa de análisis multimodal del sistema.

• ImageModalProcessor interpreta el contenido visual y su significado contextual.

• TableModalProcessor analiza las estructuras de las tablas y descodifica las relaciones lógicas y numéricas de los datos.

• EquationModalProcessor comprende la semántica de los símbolos y fórmulas matemáticas.

N analizadores sintácticos

Para soportar la diversa estructura de los documentos del mundo real, RAG-Anything proporciona una capa de análisis extensible construida sobre múltiples motores de extracción. Actualmente, integra tanto MinerU como Docling, seleccionando automáticamente el analizador sintáctico óptimo en función del tipo de documento y la complejidad estructural.

Basándose en la arquitectura "1 + 3 + N", RAG-Anything mejora el proceso tradicional de RAG modificando el tratamiento de los distintos tipos de contenido. En lugar de procesar texto, imágenes y tablas de uno en uno, el sistema los procesa todos a la vez.

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

Este diseño acelera enormemente el tratamiento de documentos técnicos de gran tamaño. Las pruebas comparativas demuestran que cuando el sistema utiliza más núcleos de CPU, se vuelve notablemente más rápido, lo que reduce drásticamente el tiempo necesario para procesar cada documento.

Almacenamiento por capas y optimización de la recuperación

Además de su diseño multimodal, RAG-Anything también utiliza un enfoque de almacenamiento y recuperación por capas para que los resultados sean más precisos y eficientes.

• El texto se almacena en una base de datos vectorial tradicional.

• Las imágenes se gestionan en un almacén de características visuales independiente.

• Lastablas se guardan en un almacén de datos estructurados.

• Las fórmulas matemáticas se convierten en vectores semánticos.

Al almacenar cada tipo de contenido en su propio formato adecuado, el sistema puede elegir el mejor método de recuperación para cada modalidad en lugar de basarse en una única búsqueda genérica por similitud. Así se obtienen resultados más rápidos y fiables en distintos tipos de contenidos.

Cómo encaja Milvus en RAG-Anything

RAG-Anything proporciona una fuerte recuperación multimodal, pero hacerlo bien requiere una búsqueda vectorial rápida y escalable a través de todo tipo de incrustaciones. Milvus cumple esta función a la perfección.

Gracias a su arquitectura nativa en la nube y a la separación entre computación y almacenamiento, Milvus ofrece una gran escalabilidad y rentabilidad. Admite la separación de lectura-escritura y la unificación de flujo por lotes, lo que permite al sistema gestionar cargas de trabajo de alta concurrencia al tiempo que mantiene el rendimiento de las consultas en tiempo real: los nuevos datos se pueden buscar inmediatamente después de su inserción.

Milvus también garantiza una fiabilidad de nivel empresarial gracias a su diseño distribuido y tolerante a fallos, que mantiene el sistema estable incluso si fallan nodos individuales. Esto lo convierte en un sistema idóneo para las implantaciones RAG multimodales a nivel de producción.

Cómo construir un sistema multimodal de preguntas y respuestas con RAG-Anything y Milvus

Esta demostración muestra cómo construir un sistema multimodal de preguntas y respuestas utilizando el marco RAG-Anything, la base de datos vectorial Milvus y el modelo de incrustación TongYi. (Este ejemplo se centra en el código de implementación central y no es una configuración de producción completa).

Demostración práctica

Requisitos previos：

• Python 3.10 o superior

• Base de datos de vectores: Servicio Milvus (Milvus Lite)

• Servicio en la nube: Clave API de Alibaba Cloud (para servicios LLM e incrustación)

• ModeloLLM: qwen-vl-max (modelo habilitado para visión)

Modelo deincrustación: 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

Ejecutar el ejemplo de trabajo mínimo:

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

Resultados esperados:

Una vez que la secuencia de comandos se ejecuta correctamente, el terminal debe mostrar:

• El resultado de la pregunta y respuesta basada en texto generada por el LLM.

• La descripción de la imagen recuperada correspondiente a la consulta.

Estructura del proyecto

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

Dependencias del proyecto

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

Variables de entorno

# 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

Configuración

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

Invocación del modelo

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

Integración de 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

Punto de entrada principal

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

Ahora, puede probar su sistema RAG multimodal con su propio conjunto de datos.

El futuro de la GAR multimodal

A medida que más datos del mundo real van más allá del texto sin formato, los sistemas de generación mejorada de recuperación (RAG) empiezan a evolucionar hacia una verdadera multimodalidad. Soluciones como RAG-Anything ya demuestran cómo pueden procesarse de forma unificada texto, imágenes, tablas, fórmulas y otros contenidos estructurados. De cara al futuro, creo que tres tendencias principales darán forma a la próxima fase de la RAG multimodal:

Ampliación a más modalidades

Los marcos actuales, como RAG-Anything, ya permiten procesar texto, imágenes, tablas y expresiones matemáticas. La próxima frontera es la compatibilidad con tipos de contenido aún más ricos, como vídeo, audio, datos de sensores y modelos 3D, lo que permitirá a los sistemas RAG comprender y recuperar información de todo el espectro de datos modernos.

Actualización de datos en tiempo real

En la actualidad, la mayoría de los procesos GAR se basan en fuentes de datos relativamente estáticas. A medida que la información cambie con mayor rapidez, los sistemas futuros requerirán actualizaciones de documentos en tiempo real, ingestión de flujos e indexación incremental. Este cambio hará que la RAG sea más sensible, oportuna y fiable en entornos dinámicos.

Trasladar la RAG a los dispositivos periféricos

Con herramientas vectoriales ligeras como Milvus Lite, la GAR multimodal ya no está confinada a la nube. La implementación de RAG en dispositivos periféricos y sistemas IoT permite que la recuperación inteligente se produzca más cerca de donde se generan los datos, lo que mejora la latencia, la privacidad y la eficiencia general.

¿Listo para explorar la GAR multimodal?

Pruebe a emparejar su canalización multimodal con Milvus y experimente una recuperación rápida y escalable a través de texto, imágenes y más.

Tiene preguntas o desea una inmersión profunda en cualquier característica? Únase a nuestro canal Discord o presente incidencias en GitHub. También puede reservar una sesión individual de 20 minutos para obtener información, orientación y respuestas a sus preguntas a través de Milvus Office Hours.