diff --git a/README.md b/README.md deleted file mode 100644 index 42de1c1c..00000000 --- a/README.md +++ /dev/null @@ -1,198 +0,0 @@ -# LightRAG: Simple and Fast Retrieval-Augmented Generation -![请添加图片描述](https://i-blog.csdnimg.cn/direct/567139f1a36e4564abc63ce5c12b6271.jpeg) - - - - - - -This repository hosts the code of LightRAG. The structure of this code is based on [nano-graphrag](https://github.com/gusye1234/nano-graphrag). -![请添加图片描述](https://i-blog.csdnimg.cn/direct/b2aaf634151b4706892693ffb43d9093.png) -## Install - -* Install from source - -```bash -cd LightRAG -pip install -e . -``` -* Install from PyPI -```bash -pip install lightrag-hku -``` - -## Quick Start - -* Set OpenAI API key in environment: `export OPENAI_API_KEY="sk-...".` -* Download the demo text "A Christmas Carol by Charles Dickens" -```bash -curl https://raw.githubusercontent.com/gusye1234/nano-graphrag/main/tests/mock_data.txt > ./book.txt -``` -Use the below python snippet: - -```python -from lightrag import LightRAG, QueryParam - -rag = LightRAG(working_dir="./dickens") - -with open("./book.txt") as f: - rag.insert(f.read()) - -# Perform naive search -print(rag.query("What are the top themes in this story?", param=QueryParam(mode="naive"))) - -# Perform local search -print(rag.query("What are the top themes in this story?", param=QueryParam(mode="local"))) - -# Perform global search -print(rag.query("What are the top themes in this story?", param=QueryParam(mode="global"))) - -# Perform hybird search -print(rag.query("What are the top themes in this story?", param=QueryParam(mode="hybird"))) -``` -Batch Insert -```python -rag.insert(["TEXT1", "TEXT2",...]) -``` -Incremental Insert - -```python -rag = LightRAG(working_dir="./dickens") - -with open("./newText.txt") as f: - rag.insert(f.read()) -``` -## Evaluation -### Dataset -The dataset used in LightRAG can be download from [TommyChien/UltraDomain](https://huggingface.co/datasets/TommyChien/UltraDomain). - -### Generate Query -LightRAG uses the following prompt to generate high-level queries, with the corresponding code located in `example/generate_query.py`. -```python -Given the following description of a dataset: - -{description} - -Please identify 5 potential users who would engage with this dataset. For each user, list 5 tasks they would perform with this dataset. Then, for each (user, task) combination, generate 5 questions that require a high-level understanding of the entire dataset. - -Output the results in the following structure: -- User 1: [user description] - - Task 1: [task description] - - Question 1: - - Question 2: - - Question 3: - - Question 4: - - Question 5: - - Task 2: [task description] - ... - - Task 5: [task description] -- User 2: [user description] - ... -- User 5: [user description] - ... -``` - - ### Batch Eval -To evaluate the performance of two RAG systems on high-level queries, LightRAG uses the following prompt, with the specific code available in `example/batch_eval.py`. -```python ----Role--- -You are an expert tasked with evaluating two answers to the same question based on three criteria: **Comprehensiveness**, **Diversity**, and **Empowerment**. ----Goal--- -You will evaluate two answers to the same question based on three criteria: **Comprehensiveness**, **Diversity**, and **Empowerment**. - -- **Comprehensiveness**: How much detail does the answer provide to cover all aspects and details of the question? -- **Diversity**: How varied and rich is the answer in providing different perspectives and insights on the question? -- **Empowerment**: How well does the answer help the reader understand and make informed judgments about the topic? - -For each criterion, choose the better answer (either Answer 1 or Answer 2) and explain why. Then, select an overall winner based on these three categories. - -Here is the question: -{query} - -Here are the two answers: - -**Answer 1:** -{answer1} - -**Answer 2:** -{answer2} - -Evaluate both answers using the three criteria listed above and provide detailed explanations for each criterion. - -Output your evaluation in the following JSON format: - -{{ - "Comprehensiveness": {{ - "Winner": "[Answer 1 or Answer 2]", - "Explanation": "[Provide explanation here]" - }}, - "Empowerment": {{ - "Winner": "[Answer 1 or Answer 2]", - "Explanation": "[Provide explanation here]" - }}, - "Overall Winner": {{ - "Winner": "[Answer 1 or Answer 2]", - "Explanation": "[Summarize why this answer is the overall winner based on the three criteria]" - }} -}} -``` -### Overall Performance Table -### Overall Performance Table -| | **Agriculture** | | **CS** | | **Legal** | | **Mix** | | -|----------------------|-------------------------|-----------------------|-----------------------|-----------------------|-----------------------|-----------------------|-----------------------|-----------------------| -| | NaiveRAG | **LightRAG** | NaiveRAG | **LightRAG** | NaiveRAG | **LightRAG** | NaiveRAG | **LightRAG** | -| **Comprehensiveness** | 32.69% | **67.31%** | 35.44% | **64.56%** | 19.05% | **80.95%** | 36.36% | **63.64%** | -| **Diversity** | 24.09% | **75.91%** | 35.24% | **64.76%** | 10.98% | **89.02%** | 30.76% | **69.24%** | -| **Empowerment** | 31.35% | **68.65%** | 35.48% | **64.52%** | 17.59% | **82.41%** | 40.95% | **59.05%** | -| **Overall** | 33.30% | **66.70%** | 34.76% | **65.24%** | 17.46% | **82.54%** | 37.59% | **62.40%** | -| | RQ-RAG | **LightRAG** | RQ-RAG | **LightRAG** | RQ-RAG | **LightRAG** | RQ-RAG | **LightRAG** | -| **Comprehensiveness** | 32.05% | **67.95%** | 39.30% | **60.70%** | 18.57% | **81.43%** | 38.89% | **61.11%** | -| **Diversity** | 29.44% | **70.56%** | 38.71% | **61.29%** | 15.14% | **84.86%** | 28.50% | **71.50%** | -| **Empowerment** | 32.51% | **67.49%** | 37.52% | **62.48%** | 17.80% | **82.20%** | 43.96% | **56.04%** | -| **Overall** | 33.29% | **66.71%** | 39.03% | **60.97%** | 17.80% | **82.20%** | 39.61% | **60.39%** | -| | HyDE | **LightRAG** | HyDE | **LightRAG** | HyDE | **LightRAG** | HyDE | **LightRAG** | -| **Comprehensiveness** | 24.39% | **75.61%** | 36.49% | **63.51%** | 27.68% | **72.32%** | 42.17% | **57.83%** | -| **Diversity** | 24.96% | **75.34%** | 37.41% | **62.59%** | 18.79% | **81.21%** | 30.88% | **69.12%** | -| **Empowerment** | 24.89% | **75.11%** | 34.99% | **65.01%** | 26.99% | **73.01%** | **45.61%** | **54.39%** | -| **Overall** | 23.17% | **76.83%** | 35.67% | **64.33%** | 27.68% | **72.32%** | 42.72% | **57.28%** | -| | GraphRAG | **LightRAG** | GraphRAG | **LightRAG** | GraphRAG | **LightRAG** | GraphRAG | **LightRAG** | -| **Comprehensiveness** | 45.56% | **54.44%** | 45.98% | **54.02%** | 47.13% | **52.87%** | **51.86%** | 48.14% | -| **Diversity** | 19.65% | **80.35%** | 39.64% | **60.36%** | 25.55% | **74.45%** | 35.87% | **64.13%** | -| **Empowerment** | 36.69% | **63.31%** | 45.09% | **54.91%** | 42.81% | **57.19%** | **52.94%** | 47.06% | -| **Overall** | 43.62% | **56.38%** | 45.98% | **54.02%** | 45.70% | **54.30%** | **51.86%** | 48.14% | - -## Code Structure - -```python -. -├── examples -│ ├── batch_eval.py -│ ├── generate_query.py -│ ├── insert.py -│ └── query.py -├── lightrag -│ ├── __init__.py -│ ├── base.py -│ ├── lightrag.py -│ ├── llm.py -│ ├── operate.py -│ ├── prompt.py -│ ├── storage.py -│ └── utils.jpeg -├── LICENSE -├── README.md -├── requirements.txt -└── setup.py -``` -## Citation - -``` -@article{guo2024lightrag, -title={LightRAG: Simple and Fast Retrieval-Augmented Generation}, -author={Zirui Guo and Lianghao Xia and Yanhua Yu and Tu Ao and Chao Huang}, -year={2024}, -eprint={2410.05779}, -archivePrefix={arXiv}, -primaryClass={cs.IR} -} -``` diff --git a/lightrag/__init__.py b/lightrag/__init__.py deleted file mode 100644 index dc497cd4..00000000 --- a/lightrag/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -from .lightrag import LightRAG, QueryParam - -__version__ = "0.0.2" -__author__ = "Zirui Guo" -__url__ = "https://github.com/HKUDS/GraphEdit" diff --git a/lightrag/base.py b/lightrag/base.py deleted file mode 100644 index 9c0422fe..00000000 --- a/lightrag/base.py +++ /dev/null @@ -1,116 +0,0 @@ -from dataclasses import dataclass, field -from typing import TypedDict, Union, Literal, Generic, TypeVar - -import numpy as np - -from .utils import EmbeddingFunc - -TextChunkSchema = TypedDict( - "TextChunkSchema", - {"tokens": int, "content": str, "full_doc_id": str, "chunk_order_index": int}, -) - -T = TypeVar("T") - -@dataclass -class QueryParam: - mode: Literal["local", "global", "hybird", "naive"] = "global" - only_need_context: bool = False - response_type: str = "Multiple Paragraphs" - top_k: int = 60 - max_token_for_text_unit: int = 4000 - max_token_for_global_context: int = 4000 - max_token_for_local_context: int = 4000 - - -@dataclass -class StorageNameSpace: - namespace: str - global_config: dict - - async def index_done_callback(self): - """commit the storage operations after indexing""" - pass - - async def query_done_callback(self): - """commit the storage operations after querying""" - pass - -@dataclass -class BaseVectorStorage(StorageNameSpace): - embedding_func: EmbeddingFunc - meta_fields: set = field(default_factory=set) - - async def query(self, query: str, top_k: int) -> list[dict]: - raise NotImplementedError - - async def upsert(self, data: dict[str, dict]): - """Use 'content' field from value for embedding, use key as id. - If embedding_func is None, use 'embedding' field from value - """ - raise NotImplementedError - -@dataclass -class BaseKVStorage(Generic[T], StorageNameSpace): - async def all_keys(self) -> list[str]: - raise NotImplementedError - - async def get_by_id(self, id: str) -> Union[T, None]: - raise NotImplementedError - - async def get_by_ids( - self, ids: list[str], fields: Union[set[str], None] = None - ) -> list[Union[T, None]]: - raise NotImplementedError - - async def filter_keys(self, data: list[str]) -> set[str]: - """return un-exist keys""" - raise NotImplementedError - - async def upsert(self, data: dict[str, T]): - raise NotImplementedError - - async def drop(self): - raise NotImplementedError - - -@dataclass -class BaseGraphStorage(StorageNameSpace): - async def has_node(self, node_id: str) -> bool: - raise NotImplementedError - - async def has_edge(self, source_node_id: str, target_node_id: str) -> bool: - raise NotImplementedError - - async def node_degree(self, node_id: str) -> int: - raise NotImplementedError - - async def edge_degree(self, src_id: str, tgt_id: str) -> int: - raise NotImplementedError - - async def get_node(self, node_id: str) -> Union[dict, None]: - raise NotImplementedError - - async def get_edge( - self, source_node_id: str, target_node_id: str - ) -> Union[dict, None]: - raise NotImplementedError - - async def get_node_edges( - self, source_node_id: str - ) -> Union[list[tuple[str, str]], None]: - raise NotImplementedError - - async def upsert_node(self, node_id: str, node_data: dict[str, str]): - raise NotImplementedError - - async def upsert_edge( - self, source_node_id: str, target_node_id: str, edge_data: dict[str, str] - ): - raise NotImplementedError - - async def clustering(self, algorithm: str): - raise NotImplementedError - - async def embed_nodes(self, algorithm: str) -> tuple[np.ndarray, list[str]]: - raise NotImplementedError("Node embedding is not used in lightrag.") \ No newline at end of file diff --git a/lightrag/lightrag.py b/lightrag/lightrag.py deleted file mode 100644 index 836fda9e..00000000 --- a/lightrag/lightrag.py +++ /dev/null @@ -1,300 +0,0 @@ -import asyncio -import os -from dataclasses import asdict, dataclass, field -from datetime import datetime -from functools import partial -from typing import Type, cast - -from .llm import gpt_4o_complete, gpt_4o_mini_complete, openai_embedding -from .operate import ( - chunking_by_token_size, - extract_entities, - local_query, - global_query, - hybird_query, - naive_query, -) - -from .storage import ( - JsonKVStorage, - NanoVectorDBStorage, - NetworkXStorage, -) -from .utils import ( - EmbeddingFunc, - compute_mdhash_id, - limit_async_func_call, - convert_response_to_json, - logger, - set_logger, -) -from .base import ( - BaseGraphStorage, - BaseKVStorage, - BaseVectorStorage, - StorageNameSpace, - QueryParam, -) - -def always_get_an_event_loop() -> asyncio.AbstractEventLoop: - try: - # If there is already an event loop, use it. - loop = asyncio.get_event_loop() - except RuntimeError: - # If in a sub-thread, create a new event loop. - logger.info("Creating a new event loop in a sub-thread.") - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - return loop - -@dataclass -class LightRAG: - working_dir: str = field( - default_factory=lambda: f"./lightrag_cache_{datetime.now().strftime('%Y-%m-%d-%H:%M:%S')}" - ) - - # text chunking - chunk_token_size: int = 1200 - chunk_overlap_token_size: int = 100 - tiktoken_model_name: str = "gpt-4o-mini" - - # entity extraction - entity_extract_max_gleaning: int = 1 - entity_summary_to_max_tokens: int = 500 - - # node embedding - node_embedding_algorithm: str = "node2vec" - node2vec_params: dict = field( - default_factory=lambda: { - "dimensions": 1536, - "num_walks": 10, - "walk_length": 40, - "num_walks": 10, - "window_size": 2, - "iterations": 3, - "random_seed": 3, - } - ) - - # text embedding - embedding_func: EmbeddingFunc = field(default_factory=lambda: openai_embedding) - embedding_batch_num: int = 32 - embedding_func_max_async: int = 16 - - # LLM - llm_model_func: callable = gpt_4o_mini_complete - llm_model_max_token_size: int = 32768 - llm_model_max_async: int = 16 - - # storage - key_string_value_json_storage_cls: Type[BaseKVStorage] = JsonKVStorage - vector_db_storage_cls: Type[BaseVectorStorage] = NanoVectorDBStorage - vector_db_storage_cls_kwargs: dict = field(default_factory=dict) - graph_storage_cls: Type[BaseGraphStorage] = NetworkXStorage - enable_llm_cache: bool = True - - # extension - addon_params: dict = field(default_factory=dict) - convert_response_to_json_func: callable = convert_response_to_json - - def __post_init__(self): - log_file = os.path.join(self.working_dir, "lightrag.log") - set_logger(log_file) - logger.info(f"Logger initialized for working directory: {self.working_dir}") - - _print_config = ",\n ".join([f"{k} = {v}" for k, v in asdict(self).items()]) - logger.debug(f"LightRAG init with param:\n {_print_config}\n") - - if not os.path.exists(self.working_dir): - logger.info(f"Creating working directory {self.working_dir}") - os.makedirs(self.working_dir) - - self.full_docs = self.key_string_value_json_storage_cls( - namespace="full_docs", global_config=asdict(self) - ) - - self.text_chunks = self.key_string_value_json_storage_cls( - namespace="text_chunks", global_config=asdict(self) - ) - - self.llm_response_cache = ( - self.key_string_value_json_storage_cls( - namespace="llm_response_cache", global_config=asdict(self) - ) - if self.enable_llm_cache - else None - ) - self.chunk_entity_relation_graph = self.graph_storage_cls( - namespace="chunk_entity_relation", global_config=asdict(self) - ) - self.embedding_func = limit_async_func_call(self.embedding_func_max_async)( - self.embedding_func - ) - self.entities_vdb = ( - self.vector_db_storage_cls( - namespace="entities", - global_config=asdict(self), - embedding_func=self.embedding_func, - meta_fields={"entity_name"} - ) - ) - self.relationships_vdb = ( - self.vector_db_storage_cls( - namespace="relationships", - global_config=asdict(self), - embedding_func=self.embedding_func, - meta_fields={"src_id", "tgt_id"} - ) - ) - self.chunks_vdb = ( - self.vector_db_storage_cls( - namespace="chunks", - global_config=asdict(self), - embedding_func=self.embedding_func, - ) - ) - - self.llm_model_func = limit_async_func_call(self.llm_model_max_async)( - partial(self.llm_model_func, hashing_kv=self.llm_response_cache) - ) - - def insert(self, string_or_strings): - loop = always_get_an_event_loop() - return loop.run_until_complete(self.ainsert(string_or_strings)) - - async def ainsert(self, string_or_strings): - try: - if isinstance(string_or_strings, str): - string_or_strings = [string_or_strings] - - new_docs = { - compute_mdhash_id(c.strip(), prefix="doc-"): {"content": c.strip()} - for c in string_or_strings - } - _add_doc_keys = await self.full_docs.filter_keys(list(new_docs.keys())) - new_docs = {k: v for k, v in new_docs.items() if k in _add_doc_keys} - if not len(new_docs): - logger.warning(f"All docs are already in the storage") - return - logger.info(f"[New Docs] inserting {len(new_docs)} docs") - - inserting_chunks = {} - for doc_key, doc in new_docs.items(): - chunks = { - compute_mdhash_id(dp["content"], prefix="chunk-"): { - **dp, - "full_doc_id": doc_key, - } - for dp in chunking_by_token_size( - doc["content"], - overlap_token_size=self.chunk_overlap_token_size, - max_token_size=self.chunk_token_size, - tiktoken_model=self.tiktoken_model_name, - ) - } - inserting_chunks.update(chunks) - _add_chunk_keys = await self.text_chunks.filter_keys( - list(inserting_chunks.keys()) - ) - inserting_chunks = { - k: v for k, v in inserting_chunks.items() if k in _add_chunk_keys - } - if not len(inserting_chunks): - logger.warning(f"All chunks are already in the storage") - return - logger.info(f"[New Chunks] inserting {len(inserting_chunks)} chunks") - - await self.chunks_vdb.upsert(inserting_chunks) - - logger.info("[Entity Extraction]...") - maybe_new_kg = await extract_entities( - inserting_chunks, - knwoledge_graph_inst=self.chunk_entity_relation_graph, - entity_vdb=self.entities_vdb, - relationships_vdb=self.relationships_vdb, - global_config=asdict(self), - ) - if maybe_new_kg is None: - logger.warning("No new entities and relationships found") - return - self.chunk_entity_relation_graph = maybe_new_kg - - await self.full_docs.upsert(new_docs) - await self.text_chunks.upsert(inserting_chunks) - finally: - await self._insert_done() - - async def _insert_done(self): - tasks = [] - for storage_inst in [ - self.full_docs, - self.text_chunks, - self.llm_response_cache, - self.entities_vdb, - self.relationships_vdb, - self.chunks_vdb, - self.chunk_entity_relation_graph, - ]: - if storage_inst is None: - continue - tasks.append(cast(StorageNameSpace, storage_inst).index_done_callback()) - await asyncio.gather(*tasks) - - def query(self, query: str, param: QueryParam = QueryParam()): - loop = always_get_an_event_loop() - return loop.run_until_complete(self.aquery(query, param)) - - async def aquery(self, query: str, param: QueryParam = QueryParam()): - if param.mode == "local": - response = await local_query( - query, - self.chunk_entity_relation_graph, - self.entities_vdb, - self.relationships_vdb, - self.text_chunks, - param, - asdict(self), - ) - elif param.mode == "global": - response = await global_query( - query, - self.chunk_entity_relation_graph, - self.entities_vdb, - self.relationships_vdb, - self.text_chunks, - param, - asdict(self), - ) - elif param.mode == "hybird": - response = await hybird_query( - query, - self.chunk_entity_relation_graph, - self.entities_vdb, - self.relationships_vdb, - self.text_chunks, - param, - asdict(self), - ) - elif param.mode == "naive": - response = await naive_query( - query, - self.chunks_vdb, - self.text_chunks, - param, - asdict(self), - ) - else: - raise ValueError(f"Unknown mode {param.mode}") - await self._query_done() - return response - - - async def _query_done(self): - tasks = [] - for storage_inst in [self.llm_response_cache]: - if storage_inst is None: - continue - tasks.append(cast(StorageNameSpace, storage_inst).index_done_callback()) - await asyncio.gather(*tasks) - - diff --git a/lightrag/llm.py b/lightrag/llm.py deleted file mode 100644 index ee700a10..00000000 --- a/lightrag/llm.py +++ /dev/null @@ -1,88 +0,0 @@ -import os -import numpy as np -from openai import AsyncOpenAI, APIConnectionError, RateLimitError, Timeout -from tenacity import ( - retry, - stop_after_attempt, - wait_exponential, - retry_if_exception_type, -) - -from .base import BaseKVStorage -from .utils import compute_args_hash, wrap_embedding_func_with_attrs - -@retry( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=10), - retry=retry_if_exception_type((RateLimitError, APIConnectionError, Timeout)), -) -async def openai_complete_if_cache( - model, prompt, system_prompt=None, history_messages=[], **kwargs -) -> str: - openai_async_client = AsyncOpenAI() - hashing_kv: BaseKVStorage = kwargs.pop("hashing_kv", None) - messages = [] - if system_prompt: - messages.append({"role": "system", "content": system_prompt}) - messages.extend(history_messages) - messages.append({"role": "user", "content": prompt}) - if hashing_kv is not None: - args_hash = compute_args_hash(model, messages) - if_cache_return = await hashing_kv.get_by_id(args_hash) - if if_cache_return is not None: - return if_cache_return["return"] - - response = await openai_async_client.chat.completions.create( - model=model, messages=messages, **kwargs - ) - - if hashing_kv is not None: - await hashing_kv.upsert( - {args_hash: {"return": response.choices[0].message.content, "model": model}} - ) - return response.choices[0].message.content - -async def gpt_4o_complete( - prompt, system_prompt=None, history_messages=[], **kwargs -) -> str: - return await openai_complete_if_cache( - "gpt-4o", - prompt, - system_prompt=system_prompt, - history_messages=history_messages, - **kwargs, - ) - - -async def gpt_4o_mini_complete( - prompt, system_prompt=None, history_messages=[], **kwargs -) -> str: - return await openai_complete_if_cache( - "gpt-4o-mini", - prompt, - system_prompt=system_prompt, - history_messages=history_messages, - **kwargs, - ) - -@wrap_embedding_func_with_attrs(embedding_dim=1536, max_token_size=8192) -@retry( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=10), - retry=retry_if_exception_type((RateLimitError, APIConnectionError, Timeout)), -) -async def openai_embedding(texts: list[str]) -> np.ndarray: - openai_async_client = AsyncOpenAI() - response = await openai_async_client.embeddings.create( - model="text-embedding-3-small", input=texts, encoding_format="float" - ) - return np.array([dp.embedding for dp in response.data]) - -if __name__ == "__main__": - import asyncio - - async def main(): - result = await gpt_4o_mini_complete('How are you?') - print(result) - - asyncio.run(main()) diff --git a/lightrag/operate.py b/lightrag/operate.py deleted file mode 100644 index 2d3271da..00000000 --- a/lightrag/operate.py +++ /dev/null @@ -1,944 +0,0 @@ -import asyncio -import json -import re -from typing import Union -from collections import Counter, defaultdict - -from .utils import ( - logger, - clean_str, - compute_mdhash_id, - decode_tokens_by_tiktoken, - encode_string_by_tiktoken, - is_float_regex, - list_of_list_to_csv, - pack_user_ass_to_openai_messages, - split_string_by_multi_markers, - truncate_list_by_token_size, -) -from .base import ( - BaseGraphStorage, - BaseKVStorage, - BaseVectorStorage, - TextChunkSchema, - QueryParam, -) -from .prompt import GRAPH_FIELD_SEP, PROMPTS - -def chunking_by_token_size( - content: str, overlap_token_size=128, max_token_size=1024, tiktoken_model="gpt-4o" -): - tokens = encode_string_by_tiktoken(content, model_name=tiktoken_model) - results = [] - for index, start in enumerate( - range(0, len(tokens), max_token_size - overlap_token_size) - ): - chunk_content = decode_tokens_by_tiktoken( - tokens[start : start + max_token_size], model_name=tiktoken_model - ) - results.append( - { - "tokens": min(max_token_size, len(tokens) - start), - "content": chunk_content.strip(), - "chunk_order_index": index, - } - ) - return results - -async def _handle_entity_relation_summary( - entity_or_relation_name: str, - description: str, - global_config: dict, -) -> str: - use_llm_func: callable = global_config["llm_model_func"] - llm_max_tokens = global_config["llm_model_max_token_size"] - tiktoken_model_name = global_config["tiktoken_model_name"] - summary_max_tokens = global_config["entity_summary_to_max_tokens"] - - tokens = encode_string_by_tiktoken(description, model_name=tiktoken_model_name) - if len(tokens) < summary_max_tokens: # No need for summary - return description - prompt_template = PROMPTS["summarize_entity_descriptions"] - use_description = decode_tokens_by_tiktoken( - tokens[:llm_max_tokens], model_name=tiktoken_model_name - ) - context_base = dict( - entity_name=entity_or_relation_name, - description_list=use_description.split(GRAPH_FIELD_SEP), - ) - use_prompt = prompt_template.format(**context_base) - logger.debug(f"Trigger summary: {entity_or_relation_name}") - summary = await use_llm_func(use_prompt, max_tokens=summary_max_tokens) - return summary - - -async def _handle_single_entity_extraction( - record_attributes: list[str], - chunk_key: str, -): - if record_attributes[0] != '"entity"' or len(record_attributes) < 4: - return None - # add this record as a node in the G - entity_name = clean_str(record_attributes[1].upper()) - if not entity_name.strip(): - return None - entity_type = clean_str(record_attributes[2].upper()) - entity_description = clean_str(record_attributes[3]) - entity_source_id = chunk_key - return dict( - entity_name=entity_name, - entity_type=entity_type, - description=entity_description, - source_id=entity_source_id, - ) - - -async def _handle_single_relationship_extraction( - record_attributes: list[str], - chunk_key: str, -): - if record_attributes[0] != '"relationship"' or len(record_attributes) < 5: - return None - # add this record as edge - source = clean_str(record_attributes[1].upper()) - target = clean_str(record_attributes[2].upper()) - edge_description = clean_str(record_attributes[3]) - - edge_keywords = clean_str(record_attributes[4]) - edge_source_id = chunk_key - weight = ( - float(record_attributes[-1]) if is_float_regex(record_attributes[-1]) else 1.0 - ) - return dict( - src_id=source, - tgt_id=target, - weight=weight, - description=edge_description, - keywords=edge_keywords, - source_id=edge_source_id, - ) - - -async def _merge_nodes_then_upsert( - entity_name: str, - nodes_data: list[dict], - knwoledge_graph_inst: BaseGraphStorage, - global_config: dict, -): - already_entitiy_types = [] - already_source_ids = [] - already_description = [] - - already_node = await knwoledge_graph_inst.get_node(entity_name) - if already_node is not None: - already_entitiy_types.append(already_node["entity_type"]) - already_source_ids.extend( - split_string_by_multi_markers(already_node["source_id"], [GRAPH_FIELD_SEP]) - ) - already_description.append(already_node["description"]) - - entity_type = sorted( - Counter( - [dp["entity_type"] for dp in nodes_data] + already_entitiy_types - ).items(), - key=lambda x: x[1], - reverse=True, - )[0][0] - description = GRAPH_FIELD_SEP.join( - sorted(set([dp["description"] for dp in nodes_data] + already_description)) - ) - source_id = GRAPH_FIELD_SEP.join( - set([dp["source_id"] for dp in nodes_data] + already_source_ids) - ) - description = await _handle_entity_relation_summary( - entity_name, description, global_config - ) - node_data = dict( - entity_type=entity_type, - description=description, - source_id=source_id, - ) - await knwoledge_graph_inst.upsert_node( - entity_name, - node_data=node_data, - ) - node_data["entity_name"] = entity_name - return node_data - - -async def _merge_edges_then_upsert( - src_id: str, - tgt_id: str, - edges_data: list[dict], - knwoledge_graph_inst: BaseGraphStorage, - global_config: dict, -): - already_weights = [] - already_source_ids = [] - already_description = [] - already_keywords = [] - - if await knwoledge_graph_inst.has_edge(src_id, tgt_id): - already_edge = await knwoledge_graph_inst.get_edge(src_id, tgt_id) - already_weights.append(already_edge["weight"]) - already_source_ids.extend( - split_string_by_multi_markers(already_edge["source_id"], [GRAPH_FIELD_SEP]) - ) - already_description.append(already_edge["description"]) - already_keywords.extend( - split_string_by_multi_markers(already_edge["keywords"], [GRAPH_FIELD_SEP]) - ) - - weight = sum([dp["weight"] for dp in edges_data] + already_weights) - description = GRAPH_FIELD_SEP.join( - sorted(set([dp["description"] for dp in edges_data] + already_description)) - ) - keywords = GRAPH_FIELD_SEP.join( - sorted(set([dp["keywords"] for dp in edges_data] + already_keywords)) - ) - source_id = GRAPH_FIELD_SEP.join( - set([dp["source_id"] for dp in edges_data] + already_source_ids) - ) - for need_insert_id in [src_id, tgt_id]: - if not (await knwoledge_graph_inst.has_node(need_insert_id)): - await knwoledge_graph_inst.upsert_node( - need_insert_id, - node_data={ - "source_id": source_id, - "description": description, - "entity_type": '"UNKNOWN"', - }, - ) - description = await _handle_entity_relation_summary( - (src_id, tgt_id), description, global_config - ) - await knwoledge_graph_inst.upsert_edge( - src_id, - tgt_id, - edge_data=dict( - weight=weight, - description=description, - keywords=keywords, - source_id=source_id, - ), - ) - - edge_data = dict( - src_id=src_id, - tgt_id=tgt_id, - description=description, - keywords=keywords, - ) - - return edge_data - -async def extract_entities( - chunks: dict[str, TextChunkSchema], - knwoledge_graph_inst: BaseGraphStorage, - entity_vdb: BaseVectorStorage, - relationships_vdb: BaseVectorStorage, - global_config: dict, -) -> Union[BaseGraphStorage, None]: - use_llm_func: callable = global_config["llm_model_func"] - entity_extract_max_gleaning = global_config["entity_extract_max_gleaning"] - - ordered_chunks = list(chunks.items()) - - entity_extract_prompt = PROMPTS["entity_extraction"] - context_base = dict( - tuple_delimiter=PROMPTS["DEFAULT_TUPLE_DELIMITER"], - record_delimiter=PROMPTS["DEFAULT_RECORD_DELIMITER"], - completion_delimiter=PROMPTS["DEFAULT_COMPLETION_DELIMITER"], - entity_types=",".join(PROMPTS["DEFAULT_ENTITY_TYPES"]), - ) - continue_prompt = PROMPTS["entiti_continue_extraction"] - if_loop_prompt = PROMPTS["entiti_if_loop_extraction"] - - already_processed = 0 - already_entities = 0 - already_relations = 0 - - async def _process_single_content(chunk_key_dp: tuple[str, TextChunkSchema]): - nonlocal already_processed, already_entities, already_relations - chunk_key = chunk_key_dp[0] - chunk_dp = chunk_key_dp[1] - content = chunk_dp["content"] - hint_prompt = entity_extract_prompt.format(**context_base, input_text=content) - final_result = await use_llm_func(hint_prompt) - - history = pack_user_ass_to_openai_messages(hint_prompt, final_result) - for now_glean_index in range(entity_extract_max_gleaning): - glean_result = await use_llm_func(continue_prompt, history_messages=history) - - history += pack_user_ass_to_openai_messages(continue_prompt, glean_result) - final_result += glean_result - if now_glean_index == entity_extract_max_gleaning - 1: - break - - if_loop_result: str = await use_llm_func( - if_loop_prompt, history_messages=history - ) - if_loop_result = if_loop_result.strip().strip('"').strip("'").lower() - if if_loop_result != "yes": - break - - records = split_string_by_multi_markers( - final_result, - [context_base["record_delimiter"], context_base["completion_delimiter"]], - ) - - maybe_nodes = defaultdict(list) - maybe_edges = defaultdict(list) - for record in records: - record = re.search(r"\((.*)\)", record) - if record is None: - continue - record = record.group(1) - record_attributes = split_string_by_multi_markers( - record, [context_base["tuple_delimiter"]] - ) - if_entities = await _handle_single_entity_extraction( - record_attributes, chunk_key - ) - if if_entities is not None: - maybe_nodes[if_entities["entity_name"]].append(if_entities) - continue - - if_relation = await _handle_single_relationship_extraction( - record_attributes, chunk_key - ) - if if_relation is not None: - maybe_edges[(if_relation["src_id"], if_relation["tgt_id"])].append( - if_relation - ) - already_processed += 1 - already_entities += len(maybe_nodes) - already_relations += len(maybe_edges) - now_ticks = PROMPTS["process_tickers"][ - already_processed % len(PROMPTS["process_tickers"]) - ] - print( - f"{now_ticks} Processed {already_processed} chunks, {already_entities} entities(duplicated), {already_relations} relations(duplicated)\r", - end="", - flush=True, - ) - return dict(maybe_nodes), dict(maybe_edges) - - # use_llm_func is wrapped in ascynio.Semaphore, limiting max_async callings - results = await asyncio.gather( - *[_process_single_content(c) for c in ordered_chunks] - ) - print() # clear the progress bar - maybe_nodes = defaultdict(list) - maybe_edges = defaultdict(list) - for m_nodes, m_edges in results: - for k, v in m_nodes.items(): - maybe_nodes[k].extend(v) - for k, v in m_edges.items(): - maybe_edges[tuple(sorted(k))].extend(v) - all_entities_data = await asyncio.gather( - *[ - _merge_nodes_then_upsert(k, v, knwoledge_graph_inst, global_config) - for k, v in maybe_nodes.items() - ] - ) - all_relationships_data = await asyncio.gather( - *[ - _merge_edges_then_upsert(k[0], k[1], v, knwoledge_graph_inst, global_config) - for k, v in maybe_edges.items() - ] - ) - if not len(all_entities_data): - logger.warning("Didn't extract any entities, maybe your LLM is not working") - return None - if not len(all_relationships_data): - logger.warning("Didn't extract any relationships, maybe your LLM is not working") - return None - - if entity_vdb is not None: - data_for_vdb = { - compute_mdhash_id(dp["entity_name"], prefix="ent-"): { - "content": dp["entity_name"] + dp["description"], - "entity_name": dp["entity_name"], - } - for dp in all_entities_data - } - await entity_vdb.upsert(data_for_vdb) - - if relationships_vdb is not None: - data_for_vdb = { - compute_mdhash_id(dp["src_id"] + dp["tgt_id"], prefix="rel-"): { - "src_id": dp["src_id"], - "tgt_id": dp["tgt_id"], - "content": dp["keywords"] + dp["src_id"] + dp["tgt_id"] + dp["description"], - } - for dp in all_relationships_data - } - await relationships_vdb.upsert(data_for_vdb) - - return knwoledge_graph_inst - -async def local_query( - query, - knowledge_graph_inst: BaseGraphStorage, - entities_vdb: BaseVectorStorage, - relationships_vdb: BaseVectorStorage, - text_chunks_db: BaseKVStorage[TextChunkSchema], - query_param: QueryParam, - global_config: dict, -) -> str: - use_model_func = global_config["llm_model_func"] - - kw_prompt_temp = PROMPTS["keywords_extraction"] - kw_prompt = kw_prompt_temp.format(query=query) - result = await use_model_func(kw_prompt) - - try: - keywords_data = json.loads(result) - keywords = keywords_data.get("low_level_keywords", []) - keywords = ', '.join(keywords) - except json.JSONDecodeError as e: - # Handle parsing error - print(f"JSON parsing error: {e}") - return PROMPTS["fail_response"] - - context = await _build_local_query_context( - keywords, - knowledge_graph_inst, - entities_vdb, - text_chunks_db, - query_param, - ) - if query_param.only_need_context: - return context - if context is None: - return PROMPTS["fail_response"] - sys_prompt_temp = PROMPTS["rag_response"] - sys_prompt = sys_prompt_temp.format( - context_data=context, response_type=query_param.response_type - ) - response = await use_model_func( - query, - system_prompt=sys_prompt, - ) - return response - -async def _build_local_query_context( - query, - knowledge_graph_inst: BaseGraphStorage, - entities_vdb: BaseVectorStorage, - text_chunks_db: BaseKVStorage[TextChunkSchema], - query_param: QueryParam, -): - results = await entities_vdb.query(query, top_k=query_param.top_k) - if not len(results): - return None - node_datas = await asyncio.gather( - *[knowledge_graph_inst.get_node(r["entity_name"]) for r in results] - ) - if not all([n is not None for n in node_datas]): - logger.warning("Some nodes are missing, maybe the storage is damaged") - node_degrees = await asyncio.gather( - *[knowledge_graph_inst.node_degree(r["entity_name"]) for r in results] - ) - node_datas = [ - {**n, "entity_name": k["entity_name"], "rank": d} - for k, n, d in zip(results, node_datas, node_degrees) - if n is not None - ] - use_text_units = await _find_most_related_text_unit_from_entities( - node_datas, query_param, text_chunks_db, knowledge_graph_inst - ) - use_relations = await _find_most_related_edges_from_entities( - node_datas, query_param, knowledge_graph_inst - ) - logger.info( - f"Local query uses {len(node_datas)} entites, {len(use_relations)} relations, {len(use_text_units)} text units" - ) - entites_section_list = [["id", "entity", "type", "description", "rank"]] - for i, n in enumerate(node_datas): - entites_section_list.append( - [ - i, - n["entity_name"], - n.get("entity_type", "UNKNOWN"), - n.get("description", "UNKNOWN"), - n["rank"], - ] - ) - entities_context = list_of_list_to_csv(entites_section_list) - - relations_section_list = [ - ["id", "source", "target", "description", "keywords", "weight", "rank"] - ] - for i, e in enumerate(use_relations): - relations_section_list.append( - [ - i, - e["src_tgt"][0], - e["src_tgt"][1], - e["description"], - e["keywords"], - e["weight"], - e["rank"], - ] - ) - relations_context = list_of_list_to_csv(relations_section_list) - - text_units_section_list = [["id", "content"]] - for i, t in enumerate(use_text_units): - text_units_section_list.append([i, t["content"]]) - text_units_context = list_of_list_to_csv(text_units_section_list) - return f""" ------Entities----- -```csv -{entities_context} -``` ------Relationships----- -```csv -{relations_context} -``` ------Sources----- -```csv -{text_units_context} -``` -""" - -async def _find_most_related_text_unit_from_entities( - node_datas: list[dict], - query_param: QueryParam, - text_chunks_db: BaseKVStorage[TextChunkSchema], - knowledge_graph_inst: BaseGraphStorage, -): - text_units = [ - split_string_by_multi_markers(dp["source_id"], [GRAPH_FIELD_SEP]) - for dp in node_datas - ] - edges = await asyncio.gather( - *[knowledge_graph_inst.get_node_edges(dp["entity_name"]) for dp in node_datas] - ) - all_one_hop_nodes = set() - for this_edges in edges: - if not this_edges: - continue - all_one_hop_nodes.update([e[1] for e in this_edges]) - all_one_hop_nodes = list(all_one_hop_nodes) - all_one_hop_nodes_data = await asyncio.gather( - *[knowledge_graph_inst.get_node(e) for e in all_one_hop_nodes] - ) - all_one_hop_text_units_lookup = { - k: set(split_string_by_multi_markers(v["source_id"], [GRAPH_FIELD_SEP])) - for k, v in zip(all_one_hop_nodes, all_one_hop_nodes_data) - if v is not None - } - all_text_units_lookup = {} - for index, (this_text_units, this_edges) in enumerate(zip(text_units, edges)): - for c_id in this_text_units: - if c_id in all_text_units_lookup: - continue - relation_counts = 0 - for e in this_edges: - if ( - e[1] in all_one_hop_text_units_lookup - and c_id in all_one_hop_text_units_lookup[e[1]] - ): - relation_counts += 1 - all_text_units_lookup[c_id] = { - "data": await text_chunks_db.get_by_id(c_id), - "order": index, - "relation_counts": relation_counts, - } - if any([v is None for v in all_text_units_lookup.values()]): - logger.warning("Text chunks are missing, maybe the storage is damaged") - all_text_units = [ - {"id": k, **v} for k, v in all_text_units_lookup.items() if v is not None - ] - all_text_units = sorted( - all_text_units, key=lambda x: (x["order"], -x["relation_counts"]) - ) - all_text_units = truncate_list_by_token_size( - all_text_units, - key=lambda x: x["data"]["content"], - max_token_size=query_param.max_token_for_text_unit, - ) - all_text_units: list[TextChunkSchema] = [t["data"] for t in all_text_units] - return all_text_units - -async def _find_most_related_edges_from_entities( - node_datas: list[dict], - query_param: QueryParam, - knowledge_graph_inst: BaseGraphStorage, -): - all_related_edges = await asyncio.gather( - *[knowledge_graph_inst.get_node_edges(dp["entity_name"]) for dp in node_datas] - ) - all_edges = set() - for this_edges in all_related_edges: - all_edges.update([tuple(sorted(e)) for e in this_edges]) - all_edges = list(all_edges) - all_edges_pack = await asyncio.gather( - *[knowledge_graph_inst.get_edge(e[0], e[1]) for e in all_edges] - ) - all_edges_degree = await asyncio.gather( - *[knowledge_graph_inst.edge_degree(e[0], e[1]) for e in all_edges] - ) - all_edges_data = [ - {"src_tgt": k, "rank": d, **v} - for k, v, d in zip(all_edges, all_edges_pack, all_edges_degree) - if v is not None - ] - all_edges_data = sorted( - all_edges_data, key=lambda x: (x["rank"], x["weight"]), reverse=True - ) - all_edges_data = truncate_list_by_token_size( - all_edges_data, - key=lambda x: x["description"], - max_token_size=query_param.max_token_for_global_context, - ) - return all_edges_data - -async def global_query( - query, - knowledge_graph_inst: BaseGraphStorage, - entities_vdb: BaseVectorStorage, - relationships_vdb: BaseVectorStorage, - text_chunks_db: BaseKVStorage[TextChunkSchema], - query_param: QueryParam, - global_config: dict, -) -> str: - use_model_func = global_config["llm_model_func"] - - kw_prompt_temp = PROMPTS["keywords_extraction"] - kw_prompt = kw_prompt_temp.format(query=query) - result = await use_model_func(kw_prompt) - - try: - keywords_data = json.loads(result) - keywords = keywords_data.get("high_level_keywords", []) - keywords = ', '.join(keywords) - except json.JSONDecodeError as e: - # Handle parsing error - print(f"JSON parsing error: {e}") - return PROMPTS["fail_response"] - - context = await _build_global_query_context( - keywords, - knowledge_graph_inst, - entities_vdb, - relationships_vdb, - text_chunks_db, - query_param, - ) - - if query_param.only_need_context: - return context - if context is None: - return PROMPTS["fail_response"] - - sys_prompt_temp = PROMPTS["rag_response"] - sys_prompt = sys_prompt_temp.format( - context_data=context, response_type=query_param.response_type - ) - response = await use_model_func( - query, - system_prompt=sys_prompt, - ) - return response - -async def _build_global_query_context( - keywords, - knowledge_graph_inst: BaseGraphStorage, - entities_vdb: BaseVectorStorage, - relationships_vdb: BaseVectorStorage, - text_chunks_db: BaseKVStorage[TextChunkSchema], - query_param: QueryParam, -): - results = await relationships_vdb.query(keywords, top_k=query_param.top_k) - - if not len(results): - return None - - edge_datas = await asyncio.gather( - *[knowledge_graph_inst.get_edge(r["src_id"], r["tgt_id"]) for r in results] - ) - - if not all([n is not None for n in edge_datas]): - logger.warning("Some edges are missing, maybe the storage is damaged") - edge_degree = await asyncio.gather( - *[knowledge_graph_inst.edge_degree(r["src_id"], r["tgt_id"]) for r in results] - ) - edge_datas = [ - {"src_id": k["src_id"], "tgt_id": k["tgt_id"], "rank": d, **v} - for k, v, d in zip(results, edge_datas, edge_degree) - if v is not None - ] - edge_datas = sorted( - edge_datas, key=lambda x: (x["rank"], x["weight"]), reverse=True - ) - edge_datas = truncate_list_by_token_size( - edge_datas, - key=lambda x: x["description"], - max_token_size=query_param.max_token_for_global_context, - ) - - use_entities = await _find_most_related_entities_from_relationships( - edge_datas, query_param, knowledge_graph_inst - ) - use_text_units = await _find_related_text_unit_from_relationships( - edge_datas, query_param, text_chunks_db, knowledge_graph_inst - ) - logger.info( - f"Global query uses {len(use_entities)} entites, {len(edge_datas)} relations, {len(use_text_units)} text units" - ) - relations_section_list = [ - ["id", "source", "target", "description", "keywords", "weight", "rank"] - ] - for i, e in enumerate(edge_datas): - relations_section_list.append( - [ - i, - e["src_id"], - e["tgt_id"], - e["description"], - e["keywords"], - e["weight"], - e["rank"], - ] - ) - relations_context = list_of_list_to_csv(relations_section_list) - - entites_section_list = [["id", "entity", "type", "description", "rank"]] - for i, n in enumerate(use_entities): - entites_section_list.append( - [ - i, - n["entity_name"], - n.get("entity_type", "UNKNOWN"), - n.get("description", "UNKNOWN"), - n["rank"], - ] - ) - entities_context = list_of_list_to_csv(entites_section_list) - - text_units_section_list = [["id", "content"]] - for i, t in enumerate(use_text_units): - text_units_section_list.append([i, t["content"]]) - text_units_context = list_of_list_to_csv(text_units_section_list) - - return f""" ------Entities----- -```csv -{entities_context} -``` ------Relationships----- -```csv -{relations_context} -``` ------Sources----- -```csv -{text_units_context} -``` -""" - -async def _find_most_related_entities_from_relationships( - edge_datas: list[dict], - query_param: QueryParam, - knowledge_graph_inst: BaseGraphStorage, -): - entity_names = set() - for e in edge_datas: - entity_names.add(e["src_id"]) - entity_names.add(e["tgt_id"]) - - node_datas = await asyncio.gather( - *[knowledge_graph_inst.get_node(entity_name) for entity_name in entity_names] - ) - - node_degrees = await asyncio.gather( - *[knowledge_graph_inst.node_degree(entity_name) for entity_name in entity_names] - ) - node_datas = [ - {**n, "entity_name": k, "rank": d} - for k, n, d in zip(entity_names, node_datas, node_degrees) - ] - - node_datas = truncate_list_by_token_size( - node_datas, - key=lambda x: x["description"], - max_token_size=query_param.max_token_for_local_context, - ) - - return node_datas - -async def _find_related_text_unit_from_relationships( - edge_datas: list[dict], - query_param: QueryParam, - text_chunks_db: BaseKVStorage[TextChunkSchema], - knowledge_graph_inst: BaseGraphStorage, -): - - text_units = [ - split_string_by_multi_markers(dp["source_id"], [GRAPH_FIELD_SEP]) - for dp in edge_datas - ] - - all_text_units_lookup = {} - - for index, unit_list in enumerate(text_units): - for c_id in unit_list: - if c_id not in all_text_units_lookup: - all_text_units_lookup[c_id] = { - "data": await text_chunks_db.get_by_id(c_id), - "order": index, - } - - if any([v is None for v in all_text_units_lookup.values()]): - logger.warning("Text chunks are missing, maybe the storage is damaged") - all_text_units = [ - {"id": k, **v} for k, v in all_text_units_lookup.items() if v is not None - ] - all_text_units = sorted( - all_text_units, key=lambda x: x["order"] - ) - all_text_units = truncate_list_by_token_size( - all_text_units, - key=lambda x: x["data"]["content"], - max_token_size=query_param.max_token_for_text_unit, - ) - all_text_units: list[TextChunkSchema] = [t["data"] for t in all_text_units] - - return all_text_units - -async def hybird_query( - query, - knowledge_graph_inst: BaseGraphStorage, - entities_vdb: BaseVectorStorage, - relationships_vdb: BaseVectorStorage, - text_chunks_db: BaseKVStorage[TextChunkSchema], - query_param: QueryParam, - global_config: dict, -) -> str: - use_model_func = global_config["llm_model_func"] - - kw_prompt_temp = PROMPTS["keywords_extraction"] - kw_prompt = kw_prompt_temp.format(query=query) - result = await use_model_func(kw_prompt) - - try: - keywords_data = json.loads(result) - hl_keywords = keywords_data.get("high_level_keywords", []) - ll_keywords = keywords_data.get("low_level_keywords", []) - hl_keywords = ', '.join(hl_keywords) - ll_keywords = ', '.join(ll_keywords) - except json.JSONDecodeError as e: - # Handle parsing error - print(f"JSON parsing error: {e}") - return PROMPTS["fail_response"] - - low_level_context = await _build_local_query_context( - ll_keywords, - knowledge_graph_inst, - entities_vdb, - text_chunks_db, - query_param, - ) - - high_level_context = await _build_global_query_context( - hl_keywords, - knowledge_graph_inst, - entities_vdb, - relationships_vdb, - text_chunks_db, - query_param, - ) - - context = combine_contexts(high_level_context, low_level_context) - - if query_param.only_need_context: - return context - if context is None: - return PROMPTS["fail_response"] - - sys_prompt_temp = PROMPTS["rag_response"] - sys_prompt = sys_prompt_temp.format( - context_data=context, response_type=query_param.response_type - ) - response = await use_model_func( - query, - system_prompt=sys_prompt, - ) - return response - -def combine_contexts(high_level_context, low_level_context): - # Function to extract entities, relationships, and sources from context strings - def extract_sections(context): - entities_match = re.search(r'-----Entities-----\s*```csv\s*(.*?)\s*```', context, re.DOTALL) - relationships_match = re.search(r'-----Relationships-----\s*```csv\s*(.*?)\s*```', context, re.DOTALL) - sources_match = re.search(r'-----Sources-----\s*```csv\s*(.*?)\s*```', context, re.DOTALL) - - entities = entities_match.group(1) if entities_match else '' - relationships = relationships_match.group(1) if relationships_match else '' - sources = sources_match.group(1) if sources_match else '' - - return entities, relationships, sources - - # Extract sections from both contexts - hl_entities, hl_relationships, hl_sources = extract_sections(high_level_context) - ll_entities, ll_relationships, ll_sources = extract_sections(low_level_context) - - # Combine and deduplicate the entities - combined_entities_set = set(filter(None, hl_entities.strip().split('\n') + ll_entities.strip().split('\n'))) - combined_entities = '\n'.join(combined_entities_set) - - # Combine and deduplicate the relationships - combined_relationships_set = set(filter(None, hl_relationships.strip().split('\n') + ll_relationships.strip().split('\n'))) - combined_relationships = '\n'.join(combined_relationships_set) - - # Combine and deduplicate the sources - combined_sources_set = set(filter(None, hl_sources.strip().split('\n') + ll_sources.strip().split('\n'))) - combined_sources = '\n'.join(combined_sources_set) - - # Format the combined context - return f""" ------Entities----- -```csv -{combined_entities} ------Relationships----- -{combined_relationships} ------Sources----- -{combined_sources} -""" - -async def naive_query( - query, - chunks_vdb: BaseVectorStorage, - text_chunks_db: BaseKVStorage[TextChunkSchema], - query_param: QueryParam, - global_config: dict, -): - use_model_func = global_config["llm_model_func"] - results = await chunks_vdb.query(query, top_k=query_param.top_k) - if not len(results): - return PROMPTS["fail_response"] - chunks_ids = [r["id"] for r in results] - chunks = await text_chunks_db.get_by_ids(chunks_ids) - - maybe_trun_chunks = truncate_list_by_token_size( - chunks, - key=lambda x: x["content"], - max_token_size=query_param.max_token_for_text_unit, - ) - logger.info(f"Truncate {len(chunks)} to {len(maybe_trun_chunks)} chunks") - section = "--New Chunk--\n".join([c["content"] for c in maybe_trun_chunks]) - if query_param.only_need_context: - return section - sys_prompt_temp = PROMPTS["naive_rag_response"] - sys_prompt = sys_prompt_temp.format( - content_data=section, response_type=query_param.response_type - ) - response = await use_model_func( - query, - system_prompt=sys_prompt, - ) - return response - diff --git a/lightrag/prompt.py b/lightrag/prompt.py deleted file mode 100644 index 5d28e49c..00000000 --- a/lightrag/prompt.py +++ /dev/null @@ -1,256 +0,0 @@ -GRAPH_FIELD_SEP = "" - -PROMPTS = {} - -PROMPTS["DEFAULT_TUPLE_DELIMITER"] = "<|>" -PROMPTS["DEFAULT_RECORD_DELIMITER"] = "##" -PROMPTS["DEFAULT_COMPLETION_DELIMITER"] = "<|COMPLETE|>" -PROMPTS["process_tickers"] = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"] - -PROMPTS["DEFAULT_ENTITY_TYPES"] = ["organization", "person", "geo", "event"] - -PROMPTS[ - "entity_extraction" -] = """-Goal- -Given a text document that is potentially relevant to this activity and a list of entity types, identify all entities of those types from the text and all relationships among the identified entities. - --Steps- -1. Identify all entities. For each identified entity, extract the following information: -- entity_name: Name of the entity, capitalized -- entity_type: One of the following types: [{entity_types}] -- entity_description: Comprehensive description of the entity's attributes and activities -Format each entity as ("entity"{tuple_delimiter}{tuple_delimiter}{tuple_delimiter} - -2. From the entities identified in step 1, identify all pairs of (source_entity, target_entity) that are *clearly related* to each other. -For each pair of related entities, extract the following information: -- source_entity: name of the source entity, as identified in step 1 -- target_entity: name of the target entity, as identified in step 1 -- relationship_description: explanation as to why you think the source entity and the target entity are related to each other -- relationship_strength: a numeric score indicating strength of the relationship between the source entity and target entity -- relationship_keywords: one or more high-level key words that summarize the overarching nature of the relationship, focusing on concepts or themes rather than specific details -Format each relationship as ("relationship"{tuple_delimiter}{tuple_delimiter}{tuple_delimiter}{tuple_delimiter}{tuple_delimiter}) - -3. Identify high-level key words that summarize the main concepts, themes, or topics of the entire text. These should capture the overarching ideas present in the document. -Format the content-level key words as ("content_keywords"{tuple_delimiter}) - -4. Return output in English as a single list of all the entities and relationships identified in steps 1 and 2. Use **{record_delimiter}** as the list delimiter. - -5. When finished, output {completion_delimiter} - -###################### --Examples- -###################### -Example 1: - -Entity_types: [person, technology, mission, organization, location] -Text: -while Alex clenched his jaw, the buzz of frustration dull against the backdrop of Taylor's authoritarian certainty. It was this competitive undercurrent that kept him alert, the sense that his and Jordan's shared commitment to discovery was an unspoken rebellion against Cruz's narrowing vision of control and order. - -Then Taylor did something unexpected. They paused beside Jordan and, for a moment, observed the device with something akin to reverence. “If this tech can be understood..." Taylor said, their voice quieter, "It could change the game for us. For all of us.” - -The underlying dismissal earlier seemed to falter, replaced by a glimpse of reluctant respect for the gravity of what lay in their hands. Jordan looked up, and for a fleeting heartbeat, their eyes locked with Taylor's, a wordless clash of wills softening into an uneasy truce. - -It was a small transformation, barely perceptible, but one that Alex noted with an inward nod. They had all been brought here by different paths -################ -Output: -("entity"{tuple_delimiter}"Alex"{tuple_delimiter}"person"{tuple_delimiter}"Alex is a character who experiences frustration and is observant of the dynamics among other characters."){record_delimiter} -("entity"{tuple_delimiter}"Taylor"{tuple_delimiter}"person"{tuple_delimiter}"Taylor is portrayed with authoritarian certainty and shows a moment of reverence towards a device, indicating a change in perspective."){record_delimiter} -("entity"{tuple_delimiter}"Jordan"{tuple_delimiter}"person"{tuple_delimiter}"Jordan shares a commitment to discovery and has a significant interaction with Taylor regarding a device."){record_delimiter} -("entity"{tuple_delimiter}"Cruz"{tuple_delimiter}"person"{tuple_delimiter}"Cruz is associated with a vision of control and order, influencing the dynamics among other characters."){record_delimiter} -("entity"{tuple_delimiter}"The Device"{tuple_delimiter}"technology"{tuple_delimiter}"The Device is central to the story, with potential game-changing implications, and is revered by Taylor."){record_delimiter} -("relationship"{tuple_delimiter}"Alex"{tuple_delimiter}"Taylor"{tuple_delimiter}"Alex is affected by Taylor's authoritarian certainty and observes changes in Taylor's attitude towards the device."{tuple_delimiter}"power dynamics, perspective shift"{tuple_delimiter}7){record_delimiter} -("relationship"{tuple_delimiter}"Alex"{tuple_delimiter}"Jordan"{tuple_delimiter}"Alex and Jordan share a commitment to discovery, which contrasts with Cruz's vision."{tuple_delimiter}"shared goals, rebellion"{tuple_delimiter}6){record_delimiter} -("relationship"{tuple_delimiter}"Taylor"{tuple_delimiter}"Jordan"{tuple_delimiter}"Taylor and Jordan interact directly regarding the device, leading to a moment of mutual respect and an uneasy truce."{tuple_delimiter}"conflict resolution, mutual respect"{tuple_delimiter}8){record_delimiter} -("relationship"{tuple_delimiter}"Jordan"{tuple_delimiter}"Cruz"{tuple_delimiter}"Jordan's commitment to discovery is in rebellion against Cruz's vision of control and order."{tuple_delimiter}"ideological conflict, rebellion"{tuple_delimiter}5){record_delimiter} -("relationship"{tuple_delimiter}"Taylor"{tuple_delimiter}"The Device"{tuple_delimiter}"Taylor shows reverence towards the device, indicating its importance and potential impact."{tuple_delimiter}"reverence, technological significance"{tuple_delimiter}9){record_delimiter} -("content_keywords"{tuple_delimiter}"power dynamics, ideological conflict, discovery, rebellion"){completion_delimiter} -############################# -Example 2: - -Entity_types: [person, technology, mission, organization, location] -Text: -They were no longer mere operatives; they had become guardians of a threshold, keepers of a message from a realm beyond stars and stripes. This elevation in their mission could not be shackled by regulations and established protocols—it demanded a new perspective, a new resolve. - -Tension threaded through the dialogue of beeps and static as communications with Washington buzzed in the background. The team stood, a portentous air enveloping them. It was clear that the decisions they made in the ensuing hours could redefine humanity's place in the cosmos or condemn them to ignorance and potential peril. - -Their connection to the stars solidified, the group moved to address the crystallizing warning, shifting from passive recipients to active participants. Mercer's latter instincts gained precedence— the team's mandate had evolved, no longer solely to observe and report but to interact and prepare. A metamorphosis had begun, and Operation: Dulce hummed with the newfound frequency of their daring, a tone set not by the earthly -############# -Output: -("entity"{tuple_delimiter}"Washington"{tuple_delimiter}"location"{tuple_delimiter}"Washington is a location where communications are being received, indicating its importance in the decision-making process."){record_delimiter} -("entity"{tuple_delimiter}"Operation: Dulce"{tuple_delimiter}"mission"{tuple_delimiter}"Operation: Dulce is described as a mission that has evolved to interact and prepare, indicating a significant shift in objectives and activities."){record_delimiter} -("entity"{tuple_delimiter}"The team"{tuple_delimiter}"organization"{tuple_delimiter}"The team is portrayed as a group of individuals who have transitioned from passive observers to active participants in a mission, showing a dynamic change in their role."){record_delimiter} -("relationship"{tuple_delimiter}"The team"{tuple_delimiter}"Washington"{tuple_delimiter}"The team receives communications from Washington, which influences their decision-making process."{tuple_delimiter}"decision-making, external influence"{tuple_delimiter}7){record_delimiter} -("relationship"{tuple_delimiter}"The team"{tuple_delimiter}"Operation: Dulce"{tuple_delimiter}"The team is directly involved in Operation: Dulce, executing its evolved objectives and activities."{tuple_delimiter}"mission evolution, active participation"{tuple_delimiter}9){completion_delimiter} -("content_keywords"{tuple_delimiter}"mission evolution, decision-making, active participation, cosmic significance"){completion_delimiter} -############################# -Example 3: - -Entity_types: [person, role, technology, organization, event, location, concept] -Text: -their voice slicing through the buzz of activity. "Control may be an illusion when facing an intelligence that literally writes its own rules," they stated stoically, casting a watchful eye over the flurry of data. - -"It's like it's learning to communicate," offered Sam Rivera from a nearby interface, their youthful energy boding a mix of awe and anxiety. "This gives talking to strangers' a whole new meaning." - -Alex surveyed his team—each face a study in concentration, determination, and not a small measure of trepidation. "This might well be our first contact," he acknowledged, "And we need to be ready for whatever answers back." - -Together, they stood on the edge of the unknown, forging humanity's response to a message from the heavens. The ensuing silence was palpable—a collective introspection about their role in this grand cosmic play, one that could rewrite human history. - -The encrypted dialogue continued to unfold, its intricate patterns showing an almost uncanny anticipation -############# -Output: -("entity"{tuple_delimiter}"Sam Rivera"{tuple_delimiter}"person"{tuple_delimiter}"Sam Rivera is a member of a team working on communicating with an unknown intelligence, showing a mix of awe and anxiety."){record_delimiter} -("entity"{tuple_delimiter}"Alex"{tuple_delimiter}"person"{tuple_delimiter}"Alex is the leader of a team attempting first contact with an unknown intelligence, acknowledging the significance of their task."){record_delimiter} -("entity"{tuple_delimiter}"Control"{tuple_delimiter}"concept"{tuple_delimiter}"Control refers to the ability to manage or govern, which is challenged by an intelligence that writes its own rules."){record_delimiter} -("entity"{tuple_delimiter}"Intelligence"{tuple_delimiter}"concept"{tuple_delimiter}"Intelligence here refers to an unknown entity capable of writing its own rules and learning to communicate."){record_delimiter} -("entity"{tuple_delimiter}"First Contact"{tuple_delimiter}"event"{tuple_delimiter}"First Contact is the potential initial communication between humanity and an unknown intelligence."){record_delimiter} -("entity"{tuple_delimiter}"Humanity's Response"{tuple_delimiter}"event"{tuple_delimiter}"Humanity's Response is the collective action taken by Alex's team in response to a message from an unknown intelligence."){record_delimiter} -("relationship"{tuple_delimiter}"Sam Rivera"{tuple_delimiter}"Intelligence"{tuple_delimiter}"Sam Rivera is directly involved in the process of learning to communicate with the unknown intelligence."{tuple_delimiter}"communication, learning process"{tuple_delimiter}9){record_delimiter} -("relationship"{tuple_delimiter}"Alex"{tuple_delimiter}"First Contact"{tuple_delimiter}"Alex leads the team that might be making the First Contact with the unknown intelligence."{tuple_delimiter}"leadership, exploration"{tuple_delimiter}10){record_delimiter} -("relationship"{tuple_delimiter}"Alex"{tuple_delimiter}"Humanity's Response"{tuple_delimiter}"Alex and his team are the key figures in Humanity's Response to the unknown intelligence."{tuple_delimiter}"collective action, cosmic significance"{tuple_delimiter}8){record_delimiter} -("relationship"{tuple_delimiter}"Control"{tuple_delimiter}"Intelligence"{tuple_delimiter}"The concept of Control is challenged by the Intelligence that writes its own rules."{tuple_delimiter}"power dynamics, autonomy"{tuple_delimiter}7){record_delimiter} -("content_keywords"{tuple_delimiter}"first contact, control, communication, cosmic significance"){completion_delimiter} -############################# --Real Data- -###################### -Entity_types: {entity_types} -Text: {input_text} -###################### -Output: -""" - -PROMPTS[ - "summarize_entity_descriptions" -] = """You are a helpful assistant responsible for generating a comprehensive summary of the data provided below. -Given one or two entities, and a list of descriptions, all related to the same entity or group of entities. -Please concatenate all of these into a single, comprehensive description. Make sure to include information collected from all the descriptions. -If the provided descriptions are contradictory, please resolve the contradictions and provide a single, coherent summary. -Make sure it is written in third person, and include the entity names so we the have full context. - -####### --Data- -Entities: {entity_name} -Description List: {description_list} -####### -Output: -""" - -PROMPTS[ - "entiti_continue_extraction" -] = """MANY entities were missed in the last extraction. Add them below using the same format: -""" - -PROMPTS[ - "entiti_if_loop_extraction" -] = """It appears some entities may have still been missed. Answer YES | NO if there are still entities that need to be added. -""" - -PROMPTS["fail_response"] = "Sorry, I'm not able to provide an answer to that question." - -PROMPTS[ - "rag_response" -] = """---Role--- - -You are a helpful assistant responding to questions about data in the tables provided. - - ----Goal--- - -Generate a response of the target length and format that responds to the user's question, summarizing all information in the input data tables appropriate for the response length and format, and incorporating any relevant general knowledge. -If you don't know the answer, just say so. Do not make anything up. -Do not include information where the supporting evidence for it is not provided. - ----Target response length and format--- - -{response_type} - - ----Data tables--- - -{context_data} - - ----Goal--- - -Generate a response of the target length and format that responds to the user's question, summarizing all information in the input data tables appropriate for the response length and format, and incorporating any relevant general knowledge. - -If you don't know the answer, just say so. Do not make anything up. - -Do not include information where the supporting evidence for it is not provided. - - ----Target response length and format--- - -{response_type} - -Add sections and commentary to the response as appropriate for the length and format. Style the response in markdown. -""" - -PROMPTS["keywords_extraction"] = """---Role--- - -You are a helpful assistant tasked with identifying both high-level and low-level keywords in the user's query. - ----Goal--- - -Given the query, list both high-level and low-level keywords. High-level keywords focus on overarching concepts or themes, while low-level keywords focus on specific entities, details, or concrete terms. - ----Instructions--- - -- Output the keywords in JSON format. -- The JSON should have two keys: - - "high_level_keywords" for overarching concepts or themes. - - "low_level_keywords" for specific entities or details. - -###################### --Examples- -###################### -Example 1: - -Query: "How does international trade influence global economic stability?" -################ -Output: -{{ - "high_level_keywords": ["International trade", "Global economic stability", "Economic impact"], - "low_level_keywords": ["Trade agreements", "Tariffs", "Currency exchange", "Imports", "Exports"] -}} -############################# -Example 2: - -Query: "What are the environmental consequences of deforestation on biodiversity?" -################ -Output: -{{ - "high_level_keywords": ["Environmental consequences", "Deforestation", "Biodiversity loss"], - "low_level_keywords": ["Species extinction", "Habitat destruction", "Carbon emissions", "Rainforest", "Ecosystem"] -}} -############################# -Example 3: - -Query: "What is the role of education in reducing poverty?" -################ -Output: -{{ - "high_level_keywords": ["Education", "Poverty reduction", "Socioeconomic development"], - "low_level_keywords": ["School access", "Literacy rates", "Job training", "Income inequality"] -}} -############################# --Real Data- -###################### -Query: {query} -###################### -Output: - -""" - -PROMPTS[ - "naive_rag_response" -] = """You're a helpful assistant -Below are the knowledge you know: -{content_data} ---- -If you don't know the answer or if the provided knowledge do not contain sufficient information to provide an answer, just say so. Do not make anything up. -Generate a response of the target length and format that responds to the user's question, summarizing all information in the input data tables appropriate for the response length and format, and incorporating any relevant general knowledge. -If you don't know the answer, just say so. Do not make anything up. -Do not include information where the supporting evidence for it is not provided. ----Target response length and format--- -{response_type} -""" diff --git a/lightrag/storage.py b/lightrag/storage.py deleted file mode 100644 index 2f2bb7d8..00000000 --- a/lightrag/storage.py +++ /dev/null @@ -1,246 +0,0 @@ -import asyncio -import html -import json -import os -from collections import defaultdict -from dataclasses import dataclass, field -from typing import Any, Union, cast -import pickle -import hnswlib -import networkx as nx -import numpy as np -from nano_vectordb import NanoVectorDB -import xxhash - -from .utils import load_json, logger, write_json -from .base import ( - BaseGraphStorage, - BaseKVStorage, - BaseVectorStorage, -) - -@dataclass -class JsonKVStorage(BaseKVStorage): - def __post_init__(self): - working_dir = self.global_config["working_dir"] - self._file_name = os.path.join(working_dir, f"kv_store_{self.namespace}.json") - self._data = load_json(self._file_name) or {} - logger.info(f"Load KV {self.namespace} with {len(self._data)} data") - - async def all_keys(self) -> list[str]: - return list(self._data.keys()) - - async def index_done_callback(self): - write_json(self._data, self._file_name) - - async def get_by_id(self, id): - return self._data.get(id, None) - - async def get_by_ids(self, ids, fields=None): - if fields is None: - return [self._data.get(id, None) for id in ids] - return [ - ( - {k: v for k, v in self._data[id].items() if k in fields} - if self._data.get(id, None) - else None - ) - for id in ids - ] - - async def filter_keys(self, data: list[str]) -> set[str]: - return set([s for s in data if s not in self._data]) - - async def upsert(self, data: dict[str, dict]): - left_data = {k: v for k, v in data.items() if k not in self._data} - self._data.update(left_data) - return left_data - - async def drop(self): - self._data = {} - -@dataclass -class NanoVectorDBStorage(BaseVectorStorage): - cosine_better_than_threshold: float = 0.2 - - def __post_init__(self): - - self._client_file_name = os.path.join( - self.global_config["working_dir"], f"vdb_{self.namespace}.json" - ) - self._max_batch_size = self.global_config["embedding_batch_num"] - self._client = NanoVectorDB( - self.embedding_func.embedding_dim, storage_file=self._client_file_name - ) - self.cosine_better_than_threshold = self.global_config.get( - "cosine_better_than_threshold", self.cosine_better_than_threshold - ) - - async def upsert(self, data: dict[str, dict]): - logger.info(f"Inserting {len(data)} vectors to {self.namespace}") - if not len(data): - logger.warning("You insert an empty data to vector DB") - return [] - list_data = [ - { - "__id__": k, - **{k1: v1 for k1, v1 in v.items() if k1 in self.meta_fields}, - } - for k, v in data.items() - ] - contents = [v["content"] for v in data.values()] - batches = [ - contents[i : i + self._max_batch_size] - for i in range(0, len(contents), self._max_batch_size) - ] - embeddings_list = await asyncio.gather( - *[self.embedding_func(batch) for batch in batches] - ) - embeddings = np.concatenate(embeddings_list) - for i, d in enumerate(list_data): - d["__vector__"] = embeddings[i] - results = self._client.upsert(datas=list_data) - return results - - async def query(self, query: str, top_k=5): - embedding = await self.embedding_func([query]) - embedding = embedding[0] - results = self._client.query( - query=embedding, - top_k=top_k, - better_than_threshold=self.cosine_better_than_threshold, - ) - results = [ - {**dp, "id": dp["__id__"], "distance": dp["__metrics__"]} for dp in results - ] - return results - - async def index_done_callback(self): - self._client.save() - -@dataclass -class NetworkXStorage(BaseGraphStorage): - @staticmethod - def load_nx_graph(file_name) -> nx.Graph: - if os.path.exists(file_name): - return nx.read_graphml(file_name) - return None - - @staticmethod - def write_nx_graph(graph: nx.Graph, file_name): - logger.info( - f"Writing graph with {graph.number_of_nodes()} nodes, {graph.number_of_edges()} edges" - ) - nx.write_graphml(graph, file_name) - - @staticmethod - def stable_largest_connected_component(graph: nx.Graph) -> nx.Graph: - """Refer to https://github.com/microsoft/graphrag/index/graph/utils/stable_lcc.py - Return the largest connected component of the graph, with nodes and edges sorted in a stable way. - """ - from graspologic.utils import largest_connected_component - - graph = graph.copy() - graph = cast(nx.Graph, largest_connected_component(graph)) - node_mapping = {node: html.unescape(node.upper().strip()) for node in graph.nodes()} # type: ignore - graph = nx.relabel_nodes(graph, node_mapping) - return NetworkXStorage._stabilize_graph(graph) - - @staticmethod - def _stabilize_graph(graph: nx.Graph) -> nx.Graph: - """Refer to https://github.com/microsoft/graphrag/index/graph/utils/stable_lcc.py - Ensure an undirected graph with the same relationships will always be read the same way. - """ - fixed_graph = nx.DiGraph() if graph.is_directed() else nx.Graph() - - sorted_nodes = graph.nodes(data=True) - sorted_nodes = sorted(sorted_nodes, key=lambda x: x[0]) - - fixed_graph.add_nodes_from(sorted_nodes) - edges = list(graph.edges(data=True)) - - if not graph.is_directed(): - - def _sort_source_target(edge): - source, target, edge_data = edge - if source > target: - temp = source - source = target - target = temp - return source, target, edge_data - - edges = [_sort_source_target(edge) for edge in edges] - - def _get_edge_key(source: Any, target: Any) -> str: - return f"{source} -> {target}" - - edges = sorted(edges, key=lambda x: _get_edge_key(x[0], x[1])) - - fixed_graph.add_edges_from(edges) - return fixed_graph - - def __post_init__(self): - self._graphml_xml_file = os.path.join( - self.global_config["working_dir"], f"graph_{self.namespace}.graphml" - ) - preloaded_graph = NetworkXStorage.load_nx_graph(self._graphml_xml_file) - if preloaded_graph is not None: - logger.info( - f"Loaded graph from {self._graphml_xml_file} with {preloaded_graph.number_of_nodes()} nodes, {preloaded_graph.number_of_edges()} edges" - ) - self._graph = preloaded_graph or nx.Graph() - self._node_embed_algorithms = { - "node2vec": self._node2vec_embed, - } - - async def index_done_callback(self): - NetworkXStorage.write_nx_graph(self._graph, self._graphml_xml_file) - - async def has_node(self, node_id: str) -> bool: - return self._graph.has_node(node_id) - - async def has_edge(self, source_node_id: str, target_node_id: str) -> bool: - return self._graph.has_edge(source_node_id, target_node_id) - - async def get_node(self, node_id: str) -> Union[dict, None]: - return self._graph.nodes.get(node_id) - - async def node_degree(self, node_id: str) -> int: - return self._graph.degree(node_id) - - async def edge_degree(self, src_id: str, tgt_id: str) -> int: - return self._graph.degree(src_id) + self._graph.degree(tgt_id) - - async def get_edge( - self, source_node_id: str, target_node_id: str - ) -> Union[dict, None]: - return self._graph.edges.get((source_node_id, target_node_id)) - - async def get_node_edges(self, source_node_id: str): - if self._graph.has_node(source_node_id): - return list(self._graph.edges(source_node_id)) - return None - - async def upsert_node(self, node_id: str, node_data: dict[str, str]): - self._graph.add_node(node_id, **node_data) - - async def upsert_edge( - self, source_node_id: str, target_node_id: str, edge_data: dict[str, str] - ): - self._graph.add_edge(source_node_id, target_node_id, **edge_data) - - async def embed_nodes(self, algorithm: str) -> tuple[np.ndarray, list[str]]: - if algorithm not in self._node_embed_algorithms: - raise ValueError(f"Node embedding algorithm {algorithm} not supported") - return await self._node_embed_algorithms[algorithm]() - - async def _node2vec_embed(self): - from graspologic import embed - - embeddings, nodes = embed.node2vec_embed( - self._graph, - **self.global_config["node2vec_params"], - ) - - nodes_ids = [self._graph.nodes[node_id]["id"] for node_id in nodes] - return embeddings, nodes_ids diff --git a/lightrag/utils.py b/lightrag/utils.py deleted file mode 100644 index c75b4270..00000000 --- a/lightrag/utils.py +++ /dev/null @@ -1,165 +0,0 @@ -import asyncio -import html -import json -import logging -import os -import re -from dataclasses import dataclass -from functools import wraps -from hashlib import md5 -from typing import Any, Union - -import numpy as np -import tiktoken - -ENCODER = None - -logger = logging.getLogger("lightrag") - -def set_logger(log_file: str): - logger.setLevel(logging.DEBUG) - - file_handler = logging.FileHandler(log_file) - file_handler.setLevel(logging.DEBUG) - - formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') - file_handler.setFormatter(formatter) - - if not logger.handlers: - logger.addHandler(file_handler) - -@dataclass -class EmbeddingFunc: - embedding_dim: int - max_token_size: int - func: callable - - async def __call__(self, *args, **kwargs) -> np.ndarray: - return await self.func(*args, **kwargs) - -def locate_json_string_body_from_string(content: str) -> Union[str, None]: - """Locate the JSON string body from a string""" - maybe_json_str = re.search(r"{.*}", content, re.DOTALL) - if maybe_json_str is not None: - return maybe_json_str.group(0) - else: - return None - -def convert_response_to_json(response: str) -> dict: - json_str = locate_json_string_body_from_string(response) - assert json_str is not None, f"Unable to parse JSON from response: {response}" - try: - data = json.loads(json_str) - return data - except json.JSONDecodeError as e: - logger.error(f"Failed to parse JSON: {json_str}") - raise e from None - -def compute_args_hash(*args): - return md5(str(args).encode()).hexdigest() - -def compute_mdhash_id(content, prefix: str = ""): - return prefix + md5(content.encode()).hexdigest() - -def limit_async_func_call(max_size: int, waitting_time: float = 0.0001): - """Add restriction of maximum async calling times for a async func""" - - def final_decro(func): - """Not using async.Semaphore to aovid use nest-asyncio""" - __current_size = 0 - - @wraps(func) - async def wait_func(*args, **kwargs): - nonlocal __current_size - while __current_size >= max_size: - await asyncio.sleep(waitting_time) - __current_size += 1 - result = await func(*args, **kwargs) - __current_size -= 1 - return result - - return wait_func - - return final_decro - -def wrap_embedding_func_with_attrs(**kwargs): - """Wrap a function with attributes""" - - def final_decro(func) -> EmbeddingFunc: - new_func = EmbeddingFunc(**kwargs, func=func) - return new_func - - return final_decro - -def load_json(file_name): - if not os.path.exists(file_name): - return None - with open(file_name) as f: - return json.load(f) - -def write_json(json_obj, file_name): - with open(file_name, "w") as f: - json.dump(json_obj, f, indent=2, ensure_ascii=False) - -def encode_string_by_tiktoken(content: str, model_name: str = "gpt-4o"): - global ENCODER - if ENCODER is None: - ENCODER = tiktoken.encoding_for_model(model_name) - tokens = ENCODER.encode(content) - return tokens - - -def decode_tokens_by_tiktoken(tokens: list[int], model_name: str = "gpt-4o"): - global ENCODER - if ENCODER is None: - ENCODER = tiktoken.encoding_for_model(model_name) - content = ENCODER.decode(tokens) - return content - -def pack_user_ass_to_openai_messages(*args: str): - roles = ["user", "assistant"] - return [ - {"role": roles[i % 2], "content": content} for i, content in enumerate(args) - ] - -def split_string_by_multi_markers(content: str, markers: list[str]) -> list[str]: - """Split a string by multiple markers""" - if not markers: - return [content] - results = re.split("|".join(re.escape(marker) for marker in markers), content) - return [r.strip() for r in results if r.strip()] - -# Refer the utils functions of the official GraphRAG implementation: -# https://github.com/microsoft/graphrag -def clean_str(input: Any) -> str: - """Clean an input string by removing HTML escapes, control characters, and other unwanted characters.""" - # If we get non-string input, just give it back - if not isinstance(input, str): - return input - - result = html.unescape(input.strip()) - # https://stackoverflow.com/questions/4324790/removing-control-characters-from-a-string-in-python - return re.sub(r"[\x00-\x1f\x7f-\x9f]", "", result) - -def is_float_regex(value): - return bool(re.match(r"^[-+]?[0-9]*\.?[0-9]+$", value)) - -def truncate_list_by_token_size(list_data: list, key: callable, max_token_size: int): - """Truncate a list of data by token size""" - if max_token_size <= 0: - return [] - tokens = 0 - for i, data in enumerate(list_data): - tokens += len(encode_string_by_tiktoken(key(data))) - if tokens > max_token_size: - return list_data[:i] - return list_data - -def list_of_list_to_csv(data: list[list]): - return "\n".join( - [",\t".join([str(data_dd) for data_dd in data_d]) for data_d in data] - ) - -def save_data_to_file(data, file_name): - with open(file_name, 'w', encoding='utf-8') as f: - json.dump(data, f, ensure_ascii=False, indent=4) \ No newline at end of file