RAG multimodal mais simples: RAG-Anything + Milvus em vez de 20 ferramentas separadas
Construir um sistema RAG multimodal costumava significar juntar uma dúzia de ferramentas especializadas - uma para OCR, uma para tabelas, uma para fórmulas matemáticas, uma para embeddings, uma para pesquisa e assim por diante. As condutas RAG tradicionais foram concebidas para texto e, quando os documentos começaram a incluir imagens, tabelas, equações, gráficos e outros conteúdos estruturados, a cadeia de ferramentas tornou-se rapidamente confusa e impossível de gerir.
O RAG-Anything, desenvolvido pela HKU, muda isso. Baseado no LightRAG, fornece uma plataforma tudo-em-um que pode analisar diversos tipos de conteúdo em paralelo e mapeá-los num gráfico de conhecimento unificado. Mas unificar o pipeline é apenas metade da história. Para recuperar provas através destas modalidades variadas, continua a ser necessária uma pesquisa vetorial rápida e escalável que possa lidar com muitos tipos de incorporação ao mesmo tempo. É aí que entra o Milvus. Sendo uma base de dados vetorial de código aberto e de elevado desempenho, o Milvus elimina a necessidade de múltiplas soluções de armazenamento e pesquisa. Suporta pesquisa ANN em grande escala, recuperação híbrida de vetor-palavra-chave, filtragem de metadados e gestão flexível de incorporação - tudo num único local.
Neste post, vamos explicar como o RAG-Anything e o Milvus trabalham juntos para substituir uma cadeia de ferramentas multimodal fragmentada por uma pilha limpa e unificada - e mostraremos como você pode criar um sistema prático de RAG Q&A multimodal com apenas algumas etapas.
O que é o RAG-Anything e como funciona
O RAG-Anything é uma estrutura RAG concebida para quebrar a barreira de apenas texto dos sistemas tradicionais. Em vez de depender de várias ferramentas especializadas, oferece um ambiente único e unificado que pode analisar, processar e recuperar informações em vários tipos de conteúdo.
A estrutura suporta documentos que contêm texto, diagramas, tabelas e expressões matemáticas, permitindo aos utilizadores consultar todas as modalidades através de uma única interface coesa. Isto torna-o particularmente útil em áreas como a investigação académica, a elaboração de relatórios financeiros e a gestão do conhecimento empresarial, onde os materiais multimodais são comuns.
Na sua essência, o RAG-Anything baseia-se num pipeline multimodal de várias fases: análise de documentos→análise de conteúdos→gráfico de conhecimentos→recuperação inteligente. Esta arquitetura permite a orquestração inteligente e a compreensão multimodal, permitindo ao sistema tratar sem problemas diversas modalidades de conteúdo num único fluxo de trabalho integrado.
A arquitetura "1 + 3 + N
Ao nível da engenharia, as capacidades do RAG-Anything são concretizadas através da sua arquitetura "1 + 3 + N":
O motor principal
No centro do RAG-Anything está um motor gráfico de conhecimento inspirado no LightRAG. Esta unidade central é responsável pela extração de entidades multimodais, mapeamento de relações intermodais e armazenamento semântico vectorizado. Ao contrário dos sistemas RAG tradicionais apenas de texto, o motor compreende entidades de texto, objectos visuais em imagens e estruturas relacionais incorporadas em tabelas.
3 Processadores modais
O RAG-Anything integra três processadores de modalidade especializados, concebidos para uma compreensão profunda e específica da modalidade. Juntos, formam a camada de análise multimodal do sistema.
O ImageModalProcessor interpreta o conteúdo visual e o seu significado contextual.
TableModalProcessor analisa estruturas de tabelas e descodifica relações lógicas e numéricas nos dados.
EquationModalProcessor compreende a semântica subjacente aos símbolos e fórmulas matemáticas.
N Analisadores
Para suportar a estrutura diversificada dos documentos do mundo real, o RAG-Anything fornece uma camada de análise extensível construída sobre vários motores de extração. Atualmente, integra tanto o MinerU como o Docling, selecionando automaticamente o analisador ideal com base no tipo de documento e na complexidade estrutural.
Com base na arquitetura "1 + 3 + N", o RAG-Anything melhora o pipeline RAG tradicional, alterando a forma como são tratados os diferentes tipos de conteúdo. Em vez de processar texto, imagens e tabelas um de cada vez, o sistema processa-os todos de uma só 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
)
Esta conceção acelera consideravelmente o tratamento de documentos técnicos de grandes dimensões. Os testes de benchmark mostram que quando o sistema utiliza mais núcleos de CPU, torna-se visivelmente mais rápido, o que reduz drasticamente o tempo necessário para processar cada documento.
Armazenamento em camadas e otimização da recuperação
Para além do seu design multimodal, o RAG-Anything também utiliza uma abordagem de armazenamento e recuperação em camadas para tornar os resultados mais precisos e eficientes.
O texto é armazenado numa base de dados vetorial tradicional.
As imagens são geridas num armazenamento de caraterísticas visuais separado.
As tabelas são guardadas num armazenamento de dados estruturado.
As fórmulas matemáticas são transformadas em vectores semânticos.
Ao armazenar cada tipo de conteúdo no seu próprio formato adequado, o sistema pode escolher o melhor método de recuperação para cada modalidade, em vez de se basear numa única pesquisa de semelhança genérica. Isto conduz a resultados mais rápidos e mais fiáveis em diferentes tipos de conteúdo.
Como é que o Milvus se enquadra no RAG-Anything
O RAG-Anything proporciona uma forte recuperação multimodal, mas para o fazer bem é necessária uma pesquisa vetorial rápida e escalável em todos os tipos de embeddings. O Milvus preenche este papel na perfeição.
Com sua arquitetura nativa da nuvem e separação entre computação e armazenamento, o Milvus oferece alta escalabilidade e eficiência de custos. Suporta separação de leitura-escrita e unificação de fluxo-lote, permitindo que o sistema lide com cargas de trabalho de alta simultaneidade, mantendo o desempenho de consulta em tempo real - novos dados tornam-se pesquisáveis imediatamente após a inserção.
O Milvus também garante fiabilidade de nível empresarial através do seu design distribuído e tolerante a falhas, que mantém o sistema estável mesmo que os nós individuais falhem. Isto torna-o ideal para implementações de RAG multimodais ao nível da produção.
Como criar um sistema multimodal de perguntas e respostas com o RAG-Anything e o Milvus
Esta demonstração mostra como construir um sistema multimodal de perguntas e respostas utilizando a estrutura RAG-Anything, a base de dados vetorial Milvus e o modelo de incorporação TongYi. (Este exemplo centra-se no código de implementação principal e não é uma configuração de produção completa).
Demonstração prática
Pré-requisitos：
Python: 3.10 ou superior
Base de dados vetorial: Serviço Milvus (Milvus Lite)
Serviço de nuvem: Chave API Alibaba Cloud (para serviços LLM e de incorporação)
Modelo LLM:
qwen-vl-max(modelo ativado por visão)
Modelo de incorporação: 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
Executar o exemplo de trabalho mínimo:
python minimal_[main.py](<http://main.py>)
Resultado esperado:
Quando o script for executado com êxito, o terminal deve exibir:
O resultado da P&R baseada em texto gerada pelo LLM.
A descrição da imagem recuperada correspondente à consulta.
Estrutura do projeto
.
├─ 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
Dependências do projeto
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
Variáveis de ambiente
# 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
Configurações
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
Invocação do 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"]
Integração do 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
Ponto 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())
Agora, pode testar o seu sistema RAG multimodal com o seu próprio conjunto de dados.
O futuro do RAG multimodal
À medida que mais dados do mundo real ultrapassam o texto simples, os sistemas Retrieval-Augmented Generation (RAG) estão a começar a evoluir para uma verdadeira multimodalidade. Soluções como o RAG-Anything já demonstram como o texto, as imagens, as tabelas, as fórmulas e outros conteúdos estruturados podem ser processados de forma unificada. Olhando para o futuro, penso que três grandes tendências irão moldar a próxima fase do RAG multimodal:
Expansão para mais modalidades
As estruturas actuais - como o RAG-Anything - já conseguem lidar com texto, imagens, tabelas e expressões matemáticas. A próxima fronteira é suportar tipos de conteúdo ainda mais ricos, incluindo vídeo, áudio, dados de sensores e modelos 3D, permitindo que os sistemas RAG compreendam e recuperem informações de todo o espetro de dados modernos.
Actualizações de dados em tempo real
Atualmente, a maioria das condutas RAG baseia-se em fontes de dados relativamente estáticas. À medida que a informação muda mais rapidamente, os sistemas futuros exigirão actualizações de documentos em tempo real, ingestão de fluxo contínuo e indexação incremental. Esta mudança tornará o RAG mais reativo, oportuno e fiável em ambientes dinâmicos.
Transferir o RAG para dispositivos periféricos
Com ferramentas vectoriais leves, como o Milvus Lite, o RAG multimodal já não está confinado à nuvem. A implantação de RAG em dispositivos de borda e sistemas IoT permite que a recuperação inteligente ocorra mais perto de onde os dados são gerados - melhorando a latência, a privacidade e a eficiência geral.
Pronto para explorar o RAG multimodal?
Tente emparelhar o seu pipeline multimodal com o Milvus e experimente uma recuperação rápida e escalável em texto, imagens e muito mais.
Tem dúvidas ou quer um mergulho profundo em qualquer recurso? Junte-se ao nosso canal Discord ou arquive problemas no GitHub. Também pode reservar uma sessão individual de 20 minutos para obter informações, orientação e respostas às suas perguntas através do Milvus Office Hours.
Try Managed Milvus for Free
Zilliz Cloud is hassle-free, powered by Milvus and 10x faster.
Get StartedLike the article? Spread the word
