AI落地主流场景之一是知识库，而做知识库，必定少不了PDF文件。
传统RAG要想精准读取这些图文并茂的PDF，就需要集成PyPDF2、OpenCV、Camelot、Tesseract等多个工具，系统庞杂且低效。此外，不同 PDF 各有侧重：报告重图表、财报重表格、论文重公式，如何精准调用这些工具同样难度不低。
香港大学数据科学学院刚刚开源的RAG-Anything项目，结合开源的Milvus向量数据库，让我们逐渐看到了解决这个问题的曙光。
通过将所有功能封装在一个框架内，并引入VLM增强查询机制，RAG-Anything可以在一个框架内实现分析文本、自动理解图像、表格等多模态内容，提供更全面的答案。
下面，我们将从架构以及实验案例出发，解释为什么它能处理文本、图像、表格与公式，并且在性能上站得住脚。
01
1+3+N技术架构深度解析
RAG-Anything的核心架构可以用"1+3+N"来概括：
1个核心引擎：基于LightRAG的知识图谱构建引擎，负责实体关系抽取和向量化存储。这个引擎的特别之处在于，它不仅处理文本实体，还能理解图像中的对象、表格中的数据关系。
3大模态处理器：
- ImageModalProcessor：除了能做简单的图像识别，还能深度理解图像内容与上下文的关系
- TableModalProcessor：智能解析表格结构，理解数据之间的逻辑关系
- EquationModalProcessor：简单的符号识别基础上，能做到对数学公式的语义理解
N种解析器：支持MinerU和Docling两大解析引擎，可以根据文档类型自动选择最优解析策略。
在"1+3+N"技术架构的基础上，RAG-Anything实现了处理机制的性能突破。把传统传统RAG系统的串行处理方式（先解析文本，再处理图像，最后处理表格）升级成为了并行处理。
# 核心配置展示了并行处理的设计思路 config = RAGAnythingConfig( working_dir="./rag_storage", parser="mineru", parse_method="auto", # 自动选择最优解析策略 enable_image_processing=True, enable_table_processing=True, enable_equation_processing=True, max_workers=8 # 支持多线程并行处理 )这种并行架构带来显著效果：处理大型技术文档的速度有了明显提升。测试表明，随着CPU核心数增加，系统处理能力几乎呈线性提升，大幅缩短了文档处理时间。
除此之外，RAG-Anything还采用了****分层存储与检索优化
- 文本内容存储在传统向量数据库中
- 图像特征使用专门的视觉向量存储
- 表格数据采用结构化存储
- 公式采用语义向量化存储
这种分层设计的优势在于，系统可以为不同类型的查询使用最适合的检索策略，而非简单地采用一刀切的向量相似度搜索。
原理与架构已经讲清楚了。下面我们用一个最小可运行的示例，在 5 分钟内把文本与图像检索问答跑通。
02
五分钟快速开始（最小可运行示例）
2.1实验目标：
本实验展示如何基于RAG-Anything框架集成Milvus向量数据库以及阿里通义大模型，实现支持文本和图像处理的多模态问答系统。（展示核心代码实现部分，并非完整代码）
2.2为什么选择Milvus？
Milvus的核心优势在于其存算分离的云原生架构，带来极致的弹性伸缩能力和成本效益。通过读写分离与流批一体设计，它在保证高并发性能的同时，还能实现插入即可查的实时性能。此外，无单点故障的设计确保了企业级的高可用与高可靠性。
2.3运行步骤
环境准备
- Python 环境：Python 3.10+
- 向量数据库：Milvus 服务（MilvusLite）
- 云服务：阿里云 API 密钥（LLM + 嵌入服务）
- LLM模型：qwen-vl-max（视觉）
- Embedding模型：tongyi-embedding-vision-plus
- python -m venv .venv && source .venv/bin/activate # Windows 使用 .venvScriptsactivate - pip install -r requirements-min.txt - cp .env.example .env 并填写 DASHSCOPE_API_KEY- 运行
python minimal_[main.py](<http://main.py>)- 预期结果
- 终端打印文本问答答案与图像检索命中描述
2.4目录结构
. ├─ 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.png2.4.1项目依赖
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.02.4.2环境变量
# 阿里云 DashScope DASHSCOPE_API_KEY=your_api_key_here # 端点如官方变更，请按需替换 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 # 模型名（统一处配置） 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=11522.4.3config配置
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")) # 基础运行参数 TIMEOUT = 60 MAX_RETRIES = 22.4.4模型调用
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"]2.4.5milvus-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 out2.4.6主入口
""" 最小可运行示例： - 将一段文本 FAQ 写入 LightRAG（仅作文本语境） - 将一张图片描述向量写入 Milvus（图像检索语境） - 执行两条查询：文本问答、图像问答 """ 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) 初始化组件 llm = AliyunLLMAdapter() emb = AliyunEmbeddingAdapter() store = MilvusVectorStore() # 2) 初始化 LightRAG（仅文本检索） 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) 数据插入：文本 if SAMPLE_DOC.exists(): text = SAMPLE_[DOC.read](<http://DOC.read>)_text(encoding="utf-8") await rag.ainsert(text) print("[green]已插入文本 FAQ 到 LightRAG[/green]") else: print("[yellow]未找到 sample/docs/faq_milvus.txt[/yellow]") # 4) 数据插入：图像（描述存 Milvus） if SAMPLE_IMG.exists(): # 用 VLM 生成该图片的简要描述，作为图像语义内容 desc = await [llm.chat](<http://llm.chat>)_vlm_with_image("请简要描述图中的 Milvus 架构要点。", str(SAMPLE_IMG)) vec = await emb.embed_text(desc) # 采用文本嵌入统一维度，便于最小示例复用 store.upsert( ids=[str(uuid.uuid4())], vectors=[vec], contents=[desc], content_types=["image"], sources=[str(SAMPLE_IMG)] ) print("[green]已插入图像描述到 Milvus（content_type=image）[/green]") else: print("[yellow]未找到 sample/images/milvus_arch.png[/yellow]") # 5) 查询：文本问答（从 LightRAG） q1 = "Milvus 是否支持同时插入与搜索？请给出简短回答。" ans1 = await rag.aquery(q1, param=QueryParam(mode="hybrid")) print("\\n[bold]文本问答[/bold]") print(ans1) # 6) 查询：图像相关（从 Milvus） q2 = "Milvus 架构的关键组件有哪些？" 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]图像检索（返回图像语义描述）[/bold]") print(img_hits[0] if img_hits else []) if __name__ == "__main__": [asyncio.run](<http://asyncio.run>)(main())2.4.7测试数据集（含milvus架构图）
Milvus 的成本是多少？ Milvus 是一个 100% 免费的开源项目。 在使用 Milvus 进行生产或发布时，请遵守Apache License 2.0。 Milvus 背后的公司 Zilliz 还为那些不想构建和维护自己的分布式实例的用户提供完全托管的云版平台。Zilliz Cloud可自动维护数据的可靠性，并允许用户只为其使用付费。 Milvus 支持非 x86 架构吗？ Milvus不能在非x86平台上安装或运行。 您的 CPU 必须支持以下指令集之一才能运行 Milvus：SSE4.2、AVX、AVX2、AVX512。这些都是 x86 专用 SIMD 指令集。 Milvus 在哪里存储数据？ Milvus 处理两种类型的数据：插入数据和元数据。 插入数据（包括向量数据、标量数据和特定于 Collections 的 Schema）以增量日志的形式存储在持久存储中。Milvus 支持多种对象存储后端，包括MinIO、AWS S3、谷歌云存储（GCS）、Azure Blob 存储、阿里云 OSS 和腾讯云对象存储（COS）。 元数据在 Milvus 内部生成。每个 Milvus 模块都有自己的元数据，这些元数据存储在 etcd 中。 为什么 etcd 中没有向量数据？ etcd 存储 Milvus 模块元数据；MinIO 存储实体。 Milvus 支持同时插入和搜索数据吗？ 是的。插入操作和查询操作由两个相互独立的模块处理。从客户端的角度来看，当插入的数据进入消息队列时，插入操作符就完成了。但是，插入的数据在加载到查询节点之前是不可查询的。对于具有增量数据的增长数据段，Milvus 会自动建立临时索引，以确保高效的搜索性能，即使数据段大小未达到索引建立阈值（计算公式为dataCoord.segment.maxSize ×dataCoord.segment.sealProportion ）。你可以通过Milvus 配置文件中的配置参数queryNode.segcore.interimIndex.enableIndex 来控制这种行为--将其设置为true 可启用临时索引（默认），而将其设置为false 则会禁用临时索引。 能否在 Milvus 中插入主键重复的向量？ 可以。Milvus 不检查向量主键是否重复。 当插入主键重复的向量时，Milvus 是否将其视为更新操作？ Milvus目前不支持更新操作，也不检查实体主键是否重复。你有责任确保实体主键是唯一的，如果不是唯一的，Milvus 可能包含多个主键重复的实体。 如果出现这种情况，查询时将返回哪个数据副本仍是未知行为。这一限制将在今后的版本中修复。 自定义实体主键的最大长度是多少？ 实体主键必须是非负 64 位整数。 每次插入操作可添加的最大数据量是多少？ 插入操作的大小不得超过 1,024 MB。这是 gRPC 规定的限制。 在特定分区中搜索时，Collection 的大小会影响查询性能吗？ 不会。如果指定了搜索的分区，Milvus 只搜索指定的分区。2.4.8文本回答样例答案
2.4.9图像检索回答样例答案
到这里，我们已经能在本地跑通最小示例并理解其工作原理。
03
技术展望：多模态RAG的未来发展方向
分析RAG-Anything项目后，我们不难发现，多模态RAG技术正处于关键转折点。
三个关键技术趋势如下：
第一，模态覆盖扩大。RAG-Anything当前支持文本、图像、表格和公式，未来，建立包含视频、音频和3D模型在内的统一多模态RAG框架是大势所趋。
第二，实时处理能力提升。目前多模态RAG主要针对静态文档，但流数据处理需求增长要求支持实时文档更新和增量索引功能。
第三，边缘计算将普及。随着MilvusLite等轻量方案发展，多模态RAG将延伸至移动设备和IoT领域，创造新应用可能。
如何学习大模型 AI ？
由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。
但是具体到个人，只能说是：
“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。
这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。
我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。
我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。
第一阶段（10天）：初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。
- 大模型 AI 能干什么？
- 大模型是怎样获得「智能」的？
- 用好 AI 的核心心法
- 大模型应用业务架构
- 大模型应用技术架构
- 代码示例：向 GPT-3.5 灌入新知识
- 提示工程的意义和核心思想
- Prompt 典型构成
- 指令调优方法论
- 思维链和思维树
- Prompt 攻击和防范
- …
第二阶段（30天）：高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。
- 为什么要做 RAG
- 搭建一个简单的 ChatPDF
- 检索的基础概念
- 什么是向量表示（Embeddings）
- 向量数据库与向量检索
- 基于向量检索的 RAG
- 搭建 RAG 系统的扩展知识
- 混合检索与 RAG-Fusion 简介
- 向量模型本地部署
- …
第三阶段（30天）：模型训练
恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。
到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？
- 为什么要做 RAG
- 什么是模型
- 什么是模型训练
- 求解器 & 损失函数简介
- 小实验2：手写一个简单的神经网络并训练它
- 什么是训练/预训练/微调/轻量化微调
- Transformer结构简介
- 轻量化微调
- 实验数据集的构建
- …
第四阶段（20天）：商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。
- 硬件选型
- 带你了解全球大模型
- 使用国产大模型服务
- 搭建 OpenAI 代理
- 热身：基于阿里云 PAI 部署 Stable Diffusion
- 在本地计算机运行大模型
- 大模型的私有化部署
- 基于 vLLM 部署大模型
- 案例：如何优雅地在阿里云私有部署开源大模型
- 部署一套开源 LLM 项目
- 内容安全
- 互联网信息服务算法备案
- …
学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。
如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。
