LightRAG 深度解析：为什么说它是真正适合生产环境的 RAG 框架？

用户问题 → 向量检索 → 返回 Top-K Chunk → LLM 生成回答

文档输入
    │
    ▼
┌──────────────┐
│  实体关系抽取  │ ◄── LLM 驱动，从文本中提取实体与关系
└──────┬───────┘
       │
       ▼
┌──────────────────────────────────┐
│         双层索引系统                │
│  ┌──────────┐  ┌──────────────┐  │
│  │ Low-Level │  │  High-Level  │  │
│  │ 实体级索引  │  │  主题级索引   │  │
│  └──────────┘  └──────────────┘  │
│         └──────┬───────┘         │
│                ▼                  │
│  ┌──────────────────────────┐    │
│  │  向量嵌入 + 图结构存储     │    │
│  └──────────────────────────┘    │
└──────────────────────────────────┘
       │
       ▼
┌──────────────┐
│  双层检索引擎  │  ← local / global / hybrid
└──────┬───────┘
       │
       ▼
     LLM 生成回答

# 来源：LightRAG README.md - "Selecting Query Modes" 章节
# lightrag/lightrag.py - QueryParam 数据类定义

# 来源：LightRAG README.md - "Installation" 章节
# 使用 pip 安装核心库
pip install lightrag-hku

# 安装包含 API 服务器的完整版
pip install "lightrag-hku[api]"

# 来源：LightRAG README.md - "Quick Start" 章节
# 该示例可在官方仓库 examples/ 目录下找到
import os
from lightrag import LightRAG, QueryParam
from lightrag.llm import gpt_4o_mini_complete, gpt_4o_complete

# 创建工作目录
WORKING_DIR = "./dickens"
if not os.path.exists(WORKING_DIR):
    os.mkdir(WORKING_DIR)

# 初始化 LightRAG
rag = LightRAG(
    working_dir=WORKING_DIR,
    llm_model_func=gpt_4o_mini_complete,  # 使用 gpt-4o-mini 作为 LLM
)

# 插入文档（文本字符串）
with open("./book.txt", "r", encoding="utf-8") as f:
    rag.insert(f.read())

# 执行查询，指定模式为 hybrid
result = rag.query(
    "What are the top themes in this story?",
    param=QueryParam(mode="hybrid")
)
print(result)

# 来源：LightRAG README.md - "Quick Start" 章节（示例代码）
# 以及 lightrag/lightrag.py 中定义的 ainsert / aquery 协程
import asyncio
import os
from lightrag import LightRAG, QueryParam
from lightrag.llm.openai import gpt_4o_mini_complete, openai_embed
from lightrag.kg.shared_storage import initialize_pipeline_status

WORKING_DIR = "./rag_storage"
if not os.path.exists(WORKING_DIR):
    os.mkdir(WORKING_DIR)

async def main():
    # 初始化 LightRAG 实例
    rag = LightRAG(
        working_dir=WORKING_DIR,
        embedding_func=openai_embed,
        llm_model_func=gpt_4o_mini_complete,
    )

    # 重要：必须先初始化存储！
    await rag.initialize_storages()
    await initialize_pipeline_status()

    # 异步插入文档
    await rag.ainsert("Your document text here")

    # 使用不同的查询模式
    for mode in ["naive", "local", "global", "hybrid"]:
        result = await rag.aquery(
            "What are the top themes in this story?",
            param=QueryParam(mode=mode)
        )
        print(f"[{mode}] {result}")

    # 使用完毕后清理资源
    await rag.finalize_storages()

asyncio.run(main())

# 来源：LightRAG README.md - "Quick Start" 章节（Ollama 示例）
import os
from lightrag import LightRAG, QueryParam
from lightrag.llm import ollama_model_complete, ollama_embedding
from lightrag.utils import EmbeddingFunc

WORKING_DIR = "./my_rag_project"
os.makedirs(WORKING_DIR, exist_ok=True)

rag = LightRAG(
    working_dir=WORKING_DIR,
    llm_model_func=ollama_model_complete,
    llm_model_name="gemma2:2b",          # 或 qwen2:7b 等
    llm_model_max_async=4,
    llm_model_max_token_size=32768,
    llm_model_kwargs={
        "host": "http://localhost:11434",
        "options": {"num_ctx": 32768}
    },
    embedding_func=EmbeddingFunc(
        embedding_dim=768,
        max_token_size=8192,
        func=lambda texts: ollama_embedding(
            texts,
            embed_model="nomic-embed-text",
            host="http://localhost:11434"
        ),
    ),
)

# 插入文档并查询
rag.insert(["文档内容1", "文档内容2"])
result = rag.query("你的问题", param=QueryParam(mode="hybrid"))
print(result)

# 来源：lightrag/lightrag.py - LightRAG 类定义
# 关键参数说明

rag = LightRAG(
    working_dir="./rag_storage",  # 工作目录，存储索引和缓存
    embedding_func=openai_embed,  # 嵌入函数
    llm_model_func=gpt_4o_mini_complete,  # LLM 函数
    top_k=60,                    # 检索返回的实体/关系数量（默认 60）
    chunk_token_size=1200,       # 文本分块大小（默认 1200 tokens）
    chunk_overlap_token_size=100, # 分块重叠大小
    cosine_threshold=0.4,        # 向量检索余弦相似度阈值
)

# 插入文档：支持字符串或字符串列表
rag.insert("文本内容")
rag.insert(["文档1", "文档2", "文档3"])

# 查询接口
rag.query(
    "问题",
    param=QueryParam(
        mode="hybrid",       # 可选：naive/local/global/hybrid/mix
        top_k=60,            # 检索数量
        response_type="多段摘要",  # 回答风格
    )
)