RAG-Anything：多模态 RAG 终极形态，让 AI 真正理解你的所有文档

# 基础安装
pip install raganything

# 全功能安装
pip install 'raganything[all]'

import asyncio
from functools import partial
from raganything import RAGAnything, RAGAnythingConfig
from lightrag.llm.openai import openai_complete_if_cache, openai_embed
from lightrag.utils import EmbeddingFunc

async def main():
    # 设置 API 配置
    api_key = "your-api-key"
    base_url = "your-base-url"  # 可选

    # 创建 RAGAnything 配置
    config = RAGAnythingConfig(
        working_dir="./rag_storage",
        parser="mineru",          # 选择解析器：mineru, docling, paddleocr
        parse_method="auto",       # 解析方法：auto, ocr, txt
        enable_image_processing=True,
        enable_table_processing=True,
        enable_equation_processing=True,
    )

    # 定义 LLM 模型函数
    def llm_model_func(prompt, system_prompt=None, history_messages=[], **kwargs):
        return openai_complete_if_cache(
            "gpt-4o-mini",
            prompt,
            system_prompt=system_prompt,
            history_messages=history_messages,
            api_key=api_key,
            base_url=base_url,
            **kwargs,
        )

    # 定义视觉模型函数用于图像处理
    def vision_model_func(
        prompt, system_prompt=None, history_messages=[],
        image_data=None, messages=None, **kwargs
    ):
        if messages:
            return openai_complete_if_cache(
                "gpt-4o", "", system_prompt=None, history_messages=[],
                messages=messages, api_key=api_key, base_url=base_url, **kwargs,
            )
        elif image_data:
            return openai_complete_if_cache(
                "gpt-4o", "", system_prompt=None, history_messages=[],
                messages=[
                    {"role": "system", "content": system_prompt}
                        if system_prompt else None,
                    {"role": "user", "content": [
                        {"type": "text", "text": prompt},
                        {"type": "image_url",
                         "image_url": {
                             "url": f"data:image/jpeg;base64,{image_data}"
                         }},
                    ]} if image_data
                    else {"role": "user", "content": prompt},
                ],
                api_key=api_key, base_url=base_url, **kwargs,
            )
        else:
            return llm_model_func(prompt, system_prompt, history_messages, **kwargs)

    # 定义嵌入函数
    embedding_func = EmbeddingFunc(
        embedding_dim=3072,
        max_token_size=8192,
        func=partial(
            openai_embed.func,
            model="text-embedding-3-large",
            api_key=api_key,
            base_url=base_url,
        ),
    )

    # 初始化 RAGAnything
    rag = RAGAnything(
        config=config,
        llm_model_func=llm_model_func,
        vision_model_func=vision_model_func,
        embedding_func=embedding_func,
    )

    # 处理文档
    await rag.process_document_complete(
        file_path="your-document.pdf",
        output_dir="./output",
        parse_method="auto"
    )

    # 文本查询
    result = await rag.aquery("文档的主要内容是什么？", mode="hybrid")
    print(f"Answer: {result}")

    # 多模态查询 - 结合表格数据进行检索
    multimodal_result = await rag.aquery_with_multimodal(
        "请分析这份性能数据并与文档中的类似结果对比",
        multimodal_content=[{
            "type": "table",
            "table_data": """Method,Accuracy,Processing_Time
RAGAnything,95.2%,120ms
Traditional_RAG,87.3%,180ms
Baseline,82.1%,200ms""",
            "table_caption": "性能对比结果",
        }],
        mode="hybrid",
    )
    print(f"Answer: {multimodal_result}")

    # 多模态查询 - 结合数学公式
    equation_result = await rag.aquery_with_multimodal(
        "解释这个公式并与文档中的相关概念关联",
        multimodal_content=[{
            "type": "equation",
            "latex": r"F1 = 2 \cdot \frac{precision \cdot recall}{precision + recall}",
            "equation_caption": "F1-Score 计算公式",
        }],
        mode="hybrid",
    )
    print(f"Answer: {equation_result}")

if __name__ == "__main__":
    asyncio.run(main())