dsProject/dsRagAnything/T2_Query.py

import asyncio
import os

from lightrag.kg.shared_storage import initialize_pipeline_status
from raganything import RAGAnything
from lightrag import LightRAG
from Util.RagUtil import create_llm_model_func, create_vision_model_func, create_embedding_func

async def load_existing_lightrag():
    lightrag_working_dir = "./rag_storage"

    if os.path.exists(lightrag_working_dir) and os.listdir(lightrag_working_dir):
        print("✅ 发现已存在的 LightRAG 实例，正在加载...")
    else:
        print("❌ 未找到已存在的 LightRAG 实例，将创建新实例")

    llm_model_func = create_llm_model_func()
    vision_model_func = create_vision_model_func(llm_model_func)
    embedding_func = create_embedding_func()

    lightrag_instance = LightRAG(
        working_dir=lightrag_working_dir,
        llm_model_func=llm_model_func,
        embedding_func=embedding_func
    )

    await lightrag_instance.initialize_storages()
    await initialize_pipeline_status()

    rag = RAGAnything(
        lightrag=lightrag_instance,
        vision_model_func=vision_model_func,
    )

    result = await rag.aquery(
        "平台安全的保证方法有哪些？",
        mode="hybrid"
    )
    print("查询结果:", result)

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