Repository: HKUDS/LightRAG
File: lightrag/utils.py (Python)

Code before:
                else 0.0,
                "description": edge.find("./data[@key='d4']", namespace).text
                if edge.find("./data[@key='d4']", namespace) is not None
                else "",
                "keywords": edge.find("./data[@key='d5']", namespace).text
                if edge.find("./data[@key='d5']", namespace) is not None
                else "",
                "source_id": edge.find("./data[@key='d6']", namespace).text
                if edge.find("./data[@key='d6']", namespace) is not None
                else "",
            }
            data["edges"].append(edge_data)

        # Print the number of nodes and edges found
        print(f"Found {len(data['nodes'])} nodes and {len(data['edges'])} edges")

        return data
    except ET.ParseError as e:
        print(f"Error parsing XML file: {e}")
        return None
    except Exception as e:
        print(f"An error occurred: {e}")
        return None


def process_combine_contexts(hl_context: dict, ll_context: dict):
    seen_content = {}
    combined_data = []

    for item in hl_context + ll_context:
        content_key = {k: v for k, v in item.items() if k != 'id'}
        content_key_str = str(content_key)
        if content_key_str not in seen_content:
            seen_content[content_key_str] = item
            combined_data.append(item)

    for i, item in enumerate(combined_data):
        item['id'] = i

    return combined_data


async def get_best_cached_response(
    hashing_kv,
    current_embedding,
    similarity_threshold=0.95,
    mode="default",
    use_llm_check=False,
    llm_func=None,
    original_prompt=None,
    cache_type=None,

Review comment (suggestion):
types should be: list[dict[str, str]] like you stated in `list_of_list_to_json`

Code after:
                "description": edge.find("./data[@key='d4']", namespace).text
                if edge.find("./data[@key='d4']", namespace) is not None
                else "",
                "keywords": edge.find("./data[@key='d5']", namespace).text
                if edge.find("./data[@key='d5']", namespace) is not None
                else "",
                "source_id": edge.find("./data[@key='d6']", namespace).text
                if edge.find("./data[@key='d6']", namespace) is not None
                else "",
            }
            data["edges"].append(edge_data)

        # Print the number of nodes and edges found
        print(f"Found {len(data['nodes'])} nodes and {len(data['edges'])} edges")

        return data
    except ET.ParseError as e:
        print(f"Error parsing XML file: {e}")
        return None
    except Exception as e:
        print(f"An error occurred: {e}")
        return None


def process_combine_contexts(
    hl_context: list[dict[str, str]], ll_context: list[dict[str, str]]
):
    seen_content = {}
    combined_data = []

    for item in hl_context + ll_context:
        content_dict = {k: v for k, v in item.items() if k != "id"}
        content_key = tuple(sorted(content_dict.items()))
        if content_key not in seen_content:
            seen_content[content_key] = item
            combined_data.append(item)

    for i, item in enumerate(combined_data):
        item["id"] = str(i)

    return combined_data


async def get_best_cached_response(
    hashing_kv,
    current_embedding,
    similarity_threshold=0.95,
    mode="default",
    use_llm_check=False,
    llm_func=None,
    original_prompt=None,