Lightrag1.2.6代码解读

项目目的极简版

得到 文本块，处理为实体和关系

a. 文本块（chunks）

文本内容：“Alice and Bob are collaborating on quantum computing research.”
来源id：doc-1

b. 实体（entities）

实体名称	类型	描述	来源
Alice	person	量子物理领域的研究员	doc-1
Bob	person	数学家	doc-1
Quantum Computing	technology	利用量子力学现象进行计算的技术	doc-1

c. 关系（relationships）

关系编号	起点（src_id）	终点（tgt_id）	描述	关键词	权重	来源
1	Alice	Bob	Alice和Bob是研究伙伴	collaboration, research	1.0	doc-1
2	Alice	Quantum Computing	Alice从事量子计算研究	research, expertise	1.0	doc-1
3	Bob	Quantum Computing	Bob研究量子计算的应用	research, application	1.0	doc-1

项目结构

边看边更新

main()开始（lightrag_ollama_demo.py）

main()内容

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def main():
    # Initialize RAG instance
    rag = asyncio.run(initialize_rag())

    # Insert example text
    with open("./book.txt", "r", encoding="utf-8") as f:
        rag.insert(f.read())

    # Test different query modes
    print("\nNaive Search:")
    print(
        rag.query(
            "What are the top themes in this story?", param=QueryParam(mode="naive")
        )
    )

    print("\nLocal Search:")
    print(
        rag.query(
            "What are the top themes in this story?", param=QueryParam(mode="local")
        )
    )

    print("\nGlobal Search:")
    print(
        rag.query(
            "What are the top themes in this story?", param=QueryParam(mode="global")
        )
    )

    print("\nHybrid Search:")
    print(
        rag.query(
            "What are the top themes in this story?", param=QueryParam(mode="hybrid")
        )
    )

    # stream response
    resp = rag.query(
        "What are the top themes in this story?",
        param=QueryParam(mode="hybrid", stream=True),
    )

    if inspect.isasyncgen(resp):
        asyncio.run(print_stream(resp))
    else:
        print(resp)

初始化: rag = asyncio.run(initialize_rag())

进入initialize_rag()，（lightrag_ollama_demo.py）

1
2
3
4
5
async def initialize_rag():
	rag = LightRAG(...........)	
	await rag.initialize_storages()
	await initialize_pipeline_status()
	return rag

rag = LightRAG(………..)

（lightrag_ollama_demo.py）

初始化 LightRAG的对象各个属性，主要是llm_model_X和embedding_func，两个模型均由本例ollama提供

显式传入的参数是__init__里的，它还有__post_init__。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
rag = LightRAG(			#  lightrag_ollama_demo.py - line 22
        working_dir=WORKING_DIR,
        llm_model_func=ollama_model_complete,
        llm_model_name="deepseek-r1:7b",
        llm_model_max_async=4,
        llm_model_max_token_size=32768,
        llm_model_kwargs={
            "host": "http://localhost:6006",	# 本地ollama提供服务的端口为6006
            "options": {"num_ctx": 32768},
        },
        embedding_func=EmbeddingFunc(
            embedding_dim=768,
            max_token_size=8192,
            func=lambda texts: ollama_embed(
                texts, embed_model="nomic-embed-text", host="http://localhost:6006"
            ),
        ),
    )

await rag.initialize_storages()

（lightrag_ollama_demo.py）

每个storages的初始化都类似，其父类都时base里的Base…Storage

kg文件夹里的各种_impl里定义具体的子类（如JsonDocStatusStorage）继承了父类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
    async def initialize_storages(self):	#  lightrag.py - line 456
        """Asynchronously initialize the storages"""
        
        # StoragesStatus.CREATED即'created' str
        # self._storages_status什么时候定义的？默认值？
        # 答：__post_init__里定义的。
        # __post_init__ 是 @dataclass 提供的一个特殊方法
        # "第二初始化"，在自动__init__后执行
        # 用于在对象初始化完成后执行额外的操作。
        # post 是拉丁语前缀，表示“在……之后”
        

        if self._storages_status == StoragesStatus.CREATED:
            tasks = []
		# ====== 此时跳到下一个三级标题查看__post_init__的内容 ======
            for storage in (
                    self.full_docs,
                    self.text_chunks,
                    self.entities_vdb,
                    self.relationships_vdb,
                    self.chunks_vdb,
                    self.chunk_entity_relation_graph,
                    self.llm_response_cache,
                    self.doc_status,
            ):
                if storage:
                    # 每个存储模块的初始化
                    tasks.append(storage.initialize())

         	await asyncio.gather(*tasks)
            # 修改self._storages_status
          	self._storages_status = StoragesStatus.INITIALIZED
          	logger.debug("Initialized Storages")

__post_init__

（lightrag.py）

由于使用了@dataclass，故__init__自动实现了，而要内部再init则用__post_init__

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
    def __post_init__(self):
        from lightrag.kg.shared_storage import (
            initialize_share_data,
        )
        ...
    	# Initialize all storages
        self.key_string_value_json_storage_cls: type[BaseKVStorage] = self._get_storage_class(self.kv_storage)
        # type: ignore
        self.vector_db_storage_cls: type[BaseVectorStorage] = self._get_storage_class(
            self.vector_storage
        )  # type: ignore
        self.graph_storage_cls: type[BaseGraphStorage] = self._get_storage_class(
            self.graph_storage
        )  # type: ignore
        self.key_string_value_json_storage_cls = partial(  # type: ignore
            self.key_string_value_json_storage_cls, global_config=global_config
        )
        ...
        # line 429
        self._storages_status = StoragesStatus.CREATED

这里初始化了各个组件的，调用各种模块

就拿self.full_docs举例而言，它是 BaseKVStorage的实例化，该实例的初始化，由key_string_value_json_storage_cls 类传入 namespace和embedding_func返回

initialize_pipeline_status()

（shared_storage.py）

初始化pipeline的各种状态值

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
async def initialize_pipeline_status():
    """
    Initialize pipeline namespace with default values.
    This function is called during FASTAPI lifespan for each worker.
    """
    pipeline_namespace = await get_namespace_data("pipeline_status")    # _manager.list()

    async with get_internal_lock():
        # 通过检查必要字段，判断是否已经初始化
        if "busy" in pipeline_namespace:
            return

        # Create a shared list object for history_messages
        history_messages = _manager.list() if is_multiprocess else []
        pipeline_namespace.update(
            {
                "busy": False,  # Control concurrent processes
                "job_name": "Default Job",  # Current job name (indexing files/indexing texts)
                "job_start": None,  # Job start time
                "docs": 0,  # Total number of documents to be indexed
                "batchs": 0,  # Number of batches for processing documents
                "cur_batch": 0,  # Current processing batch
                "request_pending": False,  # Flag for pending request for processing
                "latest_message": "",  # Latest message from pipeline processing
                "history_messages": history_messages,  # 使用共享列表对象
            }
        )
        direct_log(f"Process {os.getpid()} Pipeline namespace initialized")

插入: rag.insert(f.read())

rag.insert(f.read())

-> self.ainsert(input)

-> self.apipeline_enqueue_documents

-> self.apipeline_process_enqueue_documents

进入rag.insert内

rag.insert(f.read()) -> 转到 self.ainsert(input)，input即 f.read()（字符串），

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
def insert(
        self,
        input: str | list[str],
        split_by_character: str | None = None,
        split_by_character_only: bool = False,
        ids: str | list[str] | None = None,
) -> None:
    # 创建事件循环，然后把self.ainsert加入循环，执行时传入input，其他参数都默认
    loop = always_get_an_event_loop()

    loop.run_until_complete(
        self.ainsert(input, split_by_character, split_by_character_only, ids)
    )

进入self.ainsert内

两个内容，一个是添加文档状态信息，另一个是处理文档

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
    async def ainsert(
            self,
            input: str | list[str],
            split_by_character: str | None = None,
            split_by_character_only: bool = False,
            ids: str | list[str] | None = None,
    ) -> None:
        # 将input，即文档或list of 文档 的status信息加入doc_status数据库
        await self.apipeline_enqueue_documents(input, ids)
        
        # 处理文档，创建知识库的主要部分
        await self.apipeline_process_enqueue_documents(
            split_by_character, split_by_character_only
        )

self.apipeline_enqueue_documents

最后更新于 2025-03-20

Python3内置函数 Windows操作系统安全