架构总览

相关源文件

本页面内容基于以下源文件生成：

• lightrag/lightrag.py
• lightrag/base.py
• lightrag/operate.py
• lightrag/kg/init.py
• lightrag/llm/init.py
• docs/Algorithm.md
• lightrag/utils.py
• lightrag/kg/shared_storage.py
• lightrag/exceptions.py
• lightrag/kg/json_kv_impl.py
• setup.py
• Makefile
• Dockerfile
• pyproject.toml
• docker-compose.yml
• lightrag_webui/package.json
• lightrag/tools/lightrag_visualizer/requirements.txt
• lightrag/evaluation/sample_documents/02_rag_architecture.md
• lightrag/types.py
• tests/conftest.py

LightRAG 是一个轻量级、高性能的检索增强生成（RAG）框架，采用模块化架构设计，支持多种存储后端、查询模式和 LLM 集成。该项目的核心设计理念是通过抽象层实现存储与计算的分离，通过配置驱动实现高度可定制化，同时保持简洁的 API 接口。

系统架构总览

LightRAG 的架构采用分层设计，从上到下分为应用层、核心逻辑层、存储抽象层和基础设施层。应用层提供 LightRAG 主类作为统一入口；核心逻辑层包含查询处理、文本分块、实体提取等业务逻辑；存储抽象层通过接口隔离具体实现，支持 KV 存储、向量存储、图存储和文档状态存储；基础设施层提供共享存储管理、并发控制和日志等基础能力。

正在加载图表渲染器...

架构要点说明：

1. 分层解耦：应用层通过依赖注入方式使用存储抽象层，具体存储实现可在运行时切换（lightrag/lightrag.py:165-176）
2. 存储抽象：四种存储类型（KV、向量、图、文档状态）均通过接口定义，支持 JsonKVStorage、NanoVectorDBStorage、NetworkXStorage 等多种实现（lightrag/kg/init.py:143-161）
3. 命名空间隔离：通过 workspace 参数实现多租户数据隔离，所有存储操作自动添加命名空间前缀（lightrag/kg/shared_storage.py:99-119）
4. 并发安全：使用进程间共享存储和命名空间锁保证多进程环境下的数据一致性（lightrag/kg/shared_storage.py:18-51）
5. 配置驱动：超过 50 个配置参数支持环境变量覆盖，实现零代码定制化（lightrag/lightrag.py:153-182）

核心类与配置架构

LightRAG 主类设计

LightRAG 类是整个框架的核心入口，采用 Python dataclass 装饰器定义，包含超过 60 个配置字段。该类的设计遵循"约定优于配置"原则，所有参数均有合理默认值，同时支持通过环境变量或构造参数进行覆盖。

职责边界：

• 负责：统一管理存储后端、LLM/Embedding 函数、查询参数、并发控制等全局配置；提供 ainsert、aquery 等核心 API
• 不负责：具体存储实现细节（由各 Storage 类负责）；具体 LLM 调用逻辑（由 llm_model_func 负责）

入口与关键 API：

• 构造函数：通过 @dataclass 自动生成，支持关键字参数注入
• __post_init__：初始化后的验证逻辑，包括存储实现兼容性检查、Tokenizer 初始化（lightrag/lightrag.py:474-523）

关键数据结构：

python
1@dataclass
2class LightRAG:
3    # 存储配置
4    kv_storage: str = "JsonKVStorage"
5    vector_storage: str = "NanoVectorDBStorage"
6    graph_storage: str = "NetworkXStorage"
7    doc_status_storage: str = "JsonDocStatusStorage"
8    
9    # 工作空间隔离
10    workspace: str = field(default_factory=lambda: os.getenv("WORKSPACE", ""))
11    working_dir: str = "./rag_storage"
12    
13    # 查询参数
14    top_k: int = 60
15    chunk_top_k: int = 5
16    max_entity_tokens: int = 8000
17    max_relation_tokens: int = 4000
18    max_total_tokens: int = 16000

初始化流程：

1. 调用 initialize_share_data() 初始化共享存储（lightrag/lightrag.py:499）
2. 创建工作目录（如不存在）
3. 验证四种存储实现的兼容性（lightrag/lightrag.py:506-517）
4. 检查存储相关的环境变量
5. 初始化 Tokenizer（默认使用 TiktokenTokenizer，模型为 gpt-4o-mini）

错误处理：

• 存储实现不兼容时抛出 ValueError，提示可用的实现列表
• 环境变量解析失败时使用默认值并记录警告日志
• 废弃参数（如 log_level、log_file_path）触发 UserWarning 并自动删除属性（lightrag/lightrag.py:480-497）

存储抽象层设计

多模态存储支持

LightRAG 定义了四种存储类型，每种存储都有明确的职责边界和多种实现选择：

存储实现验证机制：

verify_storage_implementation 函数在初始化时验证存储实现与类型的兼容性（lightrag/kg/init.py:143-161）：

python
1def verify_storage_implementation(storage_type: str, storage_name: str) -> None:
2    if storage_type not in STORAGE_IMPLEMENTATIONS:
3        raise ValueError(f"Unknown storage type: {storage_type}")
4    
5    storage_info = STORAGE_IMPLEMENTATIONS[storage_type]
6    if storage_name not in storage_info["implementations"]:
7        raise ValueError(
8            f"Storage implementation '{storage_name}' is not compatible with {storage_type}. "
9            f"Compatible implementations are: {', '.join(storage_info['implementations'])}"
10        )

JsonKVStorage 实现详解

JsonKVStorage 是 KV 存储的默认实现，展示了存储抽象层的具体落地方式（lightrag/kg/json_kv_impl.py:27-76）：

初始化流程：

1. 根据 workspace 参数构建隔离的文件路径（lightrag/kg/json_kv_impl.py:30-39）
2. 获取命名空间锁（get_namespace_lock）
3. 获取更新标志（get_update_flag）
4. 从 JSON 文件加载数据到共享内存

关键方法：

• get_by_id：通过 ID 获取单条记录，自动补充 create_time、update_time 字段
• get_by_ids：批量获取记录
• index_done_callback：持久化数据到 JSON 文件，处理数据清洗后的重新加载

工作空间隔离机制：

python
1if self.workspace:
2    workspace_dir = os.path.join(working_dir, self.workspace)
3else:
4    workspace_dir = working_dir
5    self.workspace = ""
6    
7self._file_name = os.path.join(workspace_dir, f"kv_store_{self.namespace}.json")

查询与检索架构

多模式查询支持

LightRAG 支持六种查询模式，通过 QueryParam 类进行配置（lightrag/base.py:85-131）：

Token 预算控制机制：

查询架构采用统一的 Token 预算控制，避免上下文溢出（lightrag/lightrag.py:191-227）：

python
1@dataclass
2class QueryParam:
3    max_entity_tokens: int = 8000    # 实体上下文预算
4    max_relation_tokens: int = 4000  # 关系上下文预算
5    max_total_tokens: int = 16000    # 总 Token 预算
6    top_k: int = 60                  # 检索实体/关系数量
7    chunk_top_k: int = 5             # 检索文本块数量

检索参数配置：

• cosine_threshold：向量检索的相似度阈值（默认 0.2）
• kg_chunk_pick_method：文本块选择策略（WEIGHT 权重优先 / VECTOR 相似度优先）
• related_chunk_number：从单个实体/关系关联的文本块数量

查询执行流程

正在加载图表渲染器...

流程要点说明：

1. 向量检索优先：先通过向量相似度检索候选实体和关系（lightrag/base.py:109-115）
2. 图结构扩展：基于候选节点获取关联的子图结构
3. 文本块关联：从实体/关系关联的 source_id 获取原始文本块
4. Token 裁剪：根据 max_entity_tokens、max_relation_tokens、max_total_tokens 裁剪上下文
5. LLM 生成：构建包含系统提示、实体、关系、文本块的完整提示词

文本处理与分块架构

Token 级别分块算法

LightRAG 采用基于 Token 的文本分块策略，核心实现在 chunking_by_token_size 函数（lightrag/operate.py:99-162）：

分块策略：

1. 默认策略：按 chunk_token_size 大小切分，相邻块之间有 chunk_overlap_token_size 重叠
2. 字符分割策略：按指定字符（如 \n\n）分割，适用于结构化文档
3. 严格字符分割：仅按字符分割，不进行二次切分（可能抛出 ChunkTokenLimitExceededError）

关键参数配置（lightrag/lightrag.py:253-304）：

python
1chunk_token_size: int = 1200           # 每个块的最大 Token 数
2chunk_overlap_token_size: int = 100    # 相邻块的重叠 Token 数
3tiktoken_model_name: str = "gpt-4o-mini"  # Tokenizer 模型
4chunking_func: Callable = chunking_by_token_size  # 自定义分块函数

分块算法实现：

python
1def chunking_by_token_size(
2    tokenizer: Tokenizer,
3    content: str,
4    split_by_character: str | None = None,
5    split_by_character_only: bool = False,
6    chunk_overlap_token_size: int = 100,
7    chunk_token_size: int = 1200,
8) -> list[dict[str, Any]]:
9    tokens = tokenizer.encode(content)
10    results = []
11    
12    if split_by_character:
13        # 字符分割逻辑
14        raw_chunks = content.split(split_by_character)
15        # ... 处理每个 chunk
16    else:
17        # Token 分割逻辑
18        for index, start in enumerate(
19            range(0, len(tokens), chunk_token_size - chunk_overlap_token_size)
20        ):
21            chunk_content = tokenizer.decode(tokens[start : start + chunk_token_size])
22            results.append({
23                "tokens": min(chunk_token_size, len(tokens) - start),
24                "content": chunk_content.strip(),
25                "chunk_order_index": index,
26            })
27    return results

错误处理：

• 当使用 split_by_character_only=True 且某个块超过 chunk_token_size 时，抛出 ChunkTokenLimitExceededError（lightrag/operate.py:121-125）
• 实体标识符超过长度限制时自动截断并记录警告日志（lightrag/operate.py:78-96）

自定义分块函数接口

LightRAG 支持注入自定义分块函数，接口定义如下（lightrag/lightrag.py:271-304）：

python
1chunking_func: Callable[
2    [Tokenizer, str, Optional[str], bool, int, int],
3    Union[List[Dict[str, Any]], Awaitable[List[Dict[str, Any]]]]
4]

参数说明：

• tokenizer：Tokenizer 实例
• content：待分块的文本
• split_by_character：分割字符
• split_by_character_only：是否仅按字符分割
• chunk_overlap_token_size：重叠大小
• chunk_token_size：块大小

返回值：每个字典包含 tokens、content、chunk_order_index 字段

LLM 与 Embedding 集成架构

LLM 模型配置

LightRAG 采用函数注入方式集成 LLM，支持任意兼容 OpenAI 接口的模型（lightrag/lightrag.py:343-377）：

关键配置：

python
1llm_model_func: Callable[..., object] | None = None  # LLM 调用函数
2llm_model_name: str = "gpt-4o-mini"                   # 模型名称
3summary_max_tokens: int = 500                         # 实体/关系描述最大 Token
4summary_context_size: int = 4096                      # LLM 响应最大 Token
5llm_model_max_async: int = 16                         # 最大并发 LLM 调用
6enable_llm_cache: bool = True                         # 启用 LLM 响应缓存

并发控制：

• llm_model_max_async 控制最大并发数，默认 16
• default_llm_timeout 控制 LLM 调用超时，默认 300 秒

Embedding 配置

Embedding 同样采用函数注入方式（lightrag/lightrag.py:309-339）：

关键配置：

python
1embedding_func: EmbeddingFunc | None = None           # Embedding 函数
2embedding_batch_num: int = 10                         # 批处理大小
3embedding_func_max_async: int = 8                     # 最大并发 Embedding 调用
4embedding_cache_config: dict = {
5    "enabled": False,
6    "similarity_threshold": 0.95,
7    "use_llm_check": False,
8}

Embedding 缓存策略：

• enabled：是否启用缓存
• similarity_threshold：相似度阈值，超过则复用缓存
• use_llm_check：是否使用 LLM 验证缓存有效性

Ollama 服务器信息配置

对于 Ollama 后端，OllamaServerInfos 类提供模型信息模拟（lightrag/base.py:41-71）：

python
1class OllamaServerInfos:
2    def __init__(self, name=None, tag=None):
3        self._lightrag_name = name or os.getenv("OLLAMA_EMULATING_MODEL_NAME", "lightrag")
4        self._lightrag_tag = tag or os.getenv("OLLAMA_EMULATING_MODEL_TAG", "latest")

共享存储与并发控制

命名空间管理机制

LightRAG 通过 workspace 参数实现多租户数据隔离，核心实现在 get_final_namespace 函数（lightrag/kg/shared_storage.py:99-119）：

python
1def get_final_namespace(namespace: str, workspace: str | None = None):
2    global _default_workspace
3    if workspace is None:
4        workspace = _default_workspace
5    
6    if workspace is None:
7        raise ValueError("Invoke namespace operation without workspace")
8    
9    final_namespace = f"{workspace}:{namespace}" if workspace else f"{namespace}"
10    return final_namespace

命名空间组合规则：

• 有 workspace：{workspace}:{namespace}
• 无 workspace：{namespace}

跨进程日志机制

direct_log 函数提供跨进程的日志输出能力，确保在多进程环境下的日志可见性（lightrag/kg/shared_storage.py:18-51）：

python
1def direct_log(message, enable_output: bool = True, level: str = "DEBUG"):
2    if not enable_output:
3        return
4    
5    # 获取当前日志级别
6    try:
7        from lightrag.utils import logger
8        current_level = logger.getEffectiveLevel()
9    except ImportError:
10        current_level = 20  # INFO
11    
12    # 级别映射
13    level_mapping = {
14        "DEBUG": 10, "INFO": 20, "WARNING": 30,
15        "ERROR": 40, "CRITICAL": 50,
16    }
17    message_level = level_mapping.get(level.upper(), logging.DEBUG)
18    
19    if message_level >= current_level:
20        print(f"{level}: {message}", file=sys.stderr, flush=True)

锁机制与调试支持

共享存储模块提供锁获取计数器用于调试（lightrag/kg/shared_storage.py:115-119）：

python
1def inc_debug_n_locks_acquired():
2    global _debug_n_locks_acquired
3    if DEBUG_LOCKS:
4        _debug_n_locks_acquired += 1
5        print(f"DEBUG: Keyed Lock acquired, total: {_debug_n_locks_acquired:>5}")

核心设计决策与取舍

1. 存储抽象层设计

决策：采用接口抽象 + 多实现模式，而非绑定单一存储后端。

理由：

• 支持从开发环境到生产环境的平滑迁移
• 允许用户根据数据规模选择合适的存储（JSON 适合小规模，Neo4j 适合大规模图谱）
• 便于测试时使用 Mock 存储

限制：

• 不同存储实现的性能特性差异较大
• 部分高级特性（如 Neo4j 的 Cypher 查询）无法通过统一接口暴露

2. 函数注入而非 SDK 绑定

决策：LLM 和 Embedding 通过函数注入，而非绑定特定 SDK。

理由：

• 支持任意 LLM 提供商（OpenAI、Azure、Ollama、本地模型）
• 便于用户封装自定义逻辑（如重试、缓存、日志）
• 避免依赖冲突

限制：

• 用户需要自行处理 API 密钥管理
• 函数签名不一致可能导致运行时错误

3. Token 级别分块而非字符级别

决策：默认使用 Token 计数进行文本分块。

理由：

• Token 是 LLM 的实际处理单位，字符计数不准确
• 避免 Token 溢出导致的 API 错误

限制：

• 需要 Tokenizer 依赖
• 非英文文本的 Token 计数可能不准确

4. 命名空间隔离而非数据库隔离

决策：通过 workspace 前缀实现多租户隔离，而非使用独立数据库。

理由：

• 简化部署和运维
• 降低资源消耗
• 适合中小规模多租户场景

限制：

• 大规模多租户场景下性能可能不足
• 无法实现严格的访问控制

5. 异步优先的 API 设计

决策：核心 API（如 ainsert、aquery）采用异步设计。

理由：

• LLM 和 Embedding 调用是 I/O 密集型
• 支持高并发场景
• 与 FastAPI 等异步框架无缝集成

限制：

• 同步调用需要额外封装
• 调试复杂度增加

技术选型表格

模块依赖关系

正在加载图表渲染器...

依赖要点说明：

1. 单向依赖：上层模块依赖下层模块，避免循环依赖
2. 接口隔离：lightrag.py 通过 kg/__init__.py 的工厂函数创建存储实例，不直接依赖具体实现
3. 共享工具：utils.py 提供日志、环境变量解析等通用功能，被所有模块依赖
4. 异常集中：所有自定义异常定义在 exceptions.py，便于统一管理

关键配置与启动流程

环境变量配置

LightRAG 支持通过环境变量覆盖所有关键配置，get_env_value 函数提供类型安全的解析（lightrag/utils.py:176-225）：

python
1def get_env_value(env_key: str, default: any, value_type: type = str, special_none: bool = False) -> any:
2    value = os.getenv(env_key)
3    if value is None:
4        return default
5    
6    # 处理 "None" 字符串
7    if special_none and value == "None":
8        return None
9    
10    # 布尔类型转换
11    if value_type is bool:
12        return value.lower() in ("true", "1", "yes", "t", "on")
13    
14    # 列表类型（JSON 解析）
15    if value_type is list:
16        import json
17        parsed_value = json.loads(value)
18        if isinstance(parsed_value, list):
19            return parsed_value
20        return default
21    
22    return value_type(value)

常用环境变量：

启动流程

1. 创建 LightRAG 实例：

   python
   1rag = LightRAG(
   2    working_dir="./rag_storage",
   3    workspace="my_project",
   4    llm_model_func=my_llm_func,
   5    embedding_func=my_embedding_func,
   6)

2. 初始化存储（如 auto_manage_storages_states=True 则自动执行）：

   python
   1await rag.initialize_storages()

3. 插入文档：

   python
   1await rag.ainsert("文档内容...")

4. 查询：

   python
   1result = await rag.aquery("查询问题...", param=QueryParam(mode="hybrid"))

5. 清理资源：

   python
   1await rag.finalize_storages()

日志系统配置

SafeStreamHandler 提供安全的日志输出，避免进程退出时的 I/O 错误（lightrag/utils.py:50-72）：

python
1class SafeStreamHandler(logging.StreamHandler):
2    def flush(self):
3        try:
4            super().flush()
5        except (ValueError, OSError):
6            pass  # 忽略已关闭的流
7    
8    def close(self):
9        try:
10            super().close()
11        except (ValueError, OSError):
12            pass

日志级别控制：通过 setup_logger 函数设置，支持文件输出和控制台输出。