File size: 9,875 Bytes

8e874f5

"""
Context building for RAG query processing.

This module provides functions to build query context from the knowledge graph
based on different query modes (local, global, hybrid).
"""

import asyncio
from ..base import (
    BaseGraphStorage,
    BaseKVStorage,
    BaseVectorStorage,
    TextChunkSchema,
    QueryParam,
)
from ..utils import (
    logger,
    list_of_list_to_csv,
    csv_string_to_list,
)
from .clusters import find_flow_diffusion_clusters_and_summarize
from .text_units import find_most_related_text_unit_from_entities


async def build_query_context(
    query: list,
    knowledge_graph_inst: BaseGraphStorage,
    entities_vdb: BaseVectorStorage,
    relationships_vdb: BaseVectorStorage,
    text_chunks_db: BaseKVStorage[TextChunkSchema],
    query_param: QueryParam,
    global_config: dict,
):
    """
    Build query context based on extracted keywords and query mode.

    Parameters:
    -----------
    query : list
        List containing [ll_keywords, hl_keywords]
    knowledge_graph_inst : BaseGraphStorage
        Knowledge graph storage instance
    entities_vdb : BaseVectorStorage
        Entity vector database
    relationships_vdb : BaseVectorStorage
        Relationships vector database
    text_chunks_db : BaseKVStorage[TextChunkSchema]
        Text chunks database
    query_param : QueryParam
        Query parameters including mode (local/global/hybrid)
    global_config : dict
        Global configuration

    Returns:
    --------
    str
        Formatted context string for LLM response generation
    """
    ll_keywords, hl_keywords = query[0], query[1]

    # Initialize context variables
    entities_context, relations_context, text_units_context = "", "", ""

    if query_param.mode == "local":
        # Local mode: use ll_keywords only
        if ll_keywords == "":
            logger.warning("Low level keywords is empty for local mode")
            return "", "", ""

        (
            entities_context,
            relations_context,
            text_units_context,
        ) = await _get_node_data_with_flow_diffusion(
            ll_keywords,
            knowledge_graph_inst,
            entities_vdb,
            text_chunks_db,
            query_param,
            global_config,
        )

    elif query_param.mode == "global":
        # Global mode: use hl_keywords only
        if hl_keywords == "":
            logger.warning("High level keywords is empty for global mode")
            return "", "", ""

        (
            entities_context,
            relations_context,
            text_units_context,
        ) = await _get_node_data_with_flow_diffusion(
            hl_keywords,
            knowledge_graph_inst,
            entities_vdb,
            text_chunks_db,
            query_param,
            global_config,
        )

    elif query_param.mode == "hybrid":
        # Hybrid mode: use combined keywords (both ll_keywords and hl_keywords)
        if ll_keywords == "" and hl_keywords == "":
            logger.warning("Both Low Level and High Level keywords are empty for hybrid mode")
            return "", "", ""

        # Get local information using ll_keywords
        local_entities_context, local_relations_context, local_text_units_context = "", "", ""
        if ll_keywords:
            (
                local_entities_context,
                local_relations_context,
                local_text_units_context,
            ) = await _get_node_data_with_flow_diffusion(
                ll_keywords,
                knowledge_graph_inst,
                entities_vdb,
                text_chunks_db,
                query_param,
                global_config,
            )

        # Get global information using hl_keywords
        global_entities_context, global_relations_context, global_text_units_context = "", "", ""
        if hl_keywords:
            (
                global_entities_context,
                global_relations_context,
                global_text_units_context,
            ) = await _get_node_data_with_flow_diffusion(
                hl_keywords,
                knowledge_graph_inst,
                entities_vdb,
                text_chunks_db,
                query_param,
                global_config,
            )

    # Return context based on mode
    if query_param.mode == "local":
        if query_param.return_raw_entities:
            return entities_context
        elif query_param.return_raw_clusters:
            return relations_context

        return f"""
-----local-information-----
-----low-level entity information-----
```csv
{entities_context}
```
-----low-level relationship information-----
```csv
{relations_context}
```
-----Sources-----
```csv
{text_units_context}
```
"""
    elif query_param.mode == "global":
        if query_param.return_raw_entities:
            return entities_context
        elif query_param.return_raw_clusters:
            return relations_context

        return f"""
-----global-information-----
-----high-level entity information-----
```csv
{entities_context}
```
-----high-level relationship information-----
```csv
{relations_context}
```
-----Sources-----
```csv
{text_units_context}
```
"""
    elif query_param.mode == "hybrid":
        if query_param.return_raw_entities:
            # Merge local + global entities CSV into one CSV and reindex id
            merged_rows = []
            if local_entities_context:
                merged_rows += csv_string_to_list(local_entities_context)[1:]
            if global_entities_context:
                merged_rows += csv_string_to_list(global_entities_context)[1:]
            for idx, row in enumerate(merged_rows):
                if row:
                    row[0] = str(idx)
            merged_entities_csv = list_of_list_to_csv(
                [["id", "entity", "entity_type", "description", "rank"]] + merged_rows
            )
            return merged_entities_csv
        elif query_param.return_raw_clusters:
            return local_relations_context + global_relations_context

        return f"""
-----hybrid-information-----
-----local information (from low-level keywords)-----
-----local entity information-----
```csv
{local_entities_context}
```
-----local relationship information-----
```csv
{local_relations_context}
```
-----local sources-----
```csv
{local_text_units_context}
```
-----global information (from high-level keywords)-----
-----global entity information-----
```csv
{global_entities_context}
```
-----global relationship information-----
```csv
{global_relations_context}
```
-----global sources-----
```csv
{global_text_units_context}
```
"""
    else:
        return ""


async def _get_node_data_with_flow_diffusion(
    query,
    knowledge_graph_inst: BaseGraphStorage,
    entities_vdb: BaseVectorStorage,
    text_chunks_db: BaseKVStorage[TextChunkSchema],
    query_param: QueryParam,
    global_config: dict,
):
    """
    Get node data using flow diffusion for finding relationships.

    Parameters:
    -----------
    query : str
        Query string (can be either ll_keywords or hl_keywords)
    knowledge_graph_inst : BaseGraphStorage
        Knowledge graph storage instance
    entities_vdb : BaseVectorStorage
        Entity vector database
    text_chunks_db : BaseKVStorage[TextChunkSchema]
        Text chunks database
    query_param : QueryParam
        Query parameters
    global_config : dict
        Global configuration

    Returns:
    --------
    tuple
        (entities_context, relations_context, text_units_context)
    """
    results = await entities_vdb.query(query, top_k=query_param.max_source_nodes)
    if not len(results):
        return "", "", ""

    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 flow diffusion instead of the original relationship finding method
    use_relations = await find_flow_diffusion_clusters_and_summarize(
        node_datas, query, query_param, knowledge_graph_inst, global_config
    )

    logger.info(
        f"Flow diffusion query uses {len(node_datas)} entities, {len(use_relations)} cluster summaries, {len(use_text_units)} text units"
    )

    entites_section_list = [["id", "entity", "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 context: return JSON clusters when requested; otherwise CSV
    if query_param.return_raw_clusters:
        relations_context = use_relations
    else:
        relations_section_list = [["id", "cluster_summary"]]
        for i, summary in enumerate(use_relations):
            relations_section_list.append([i, summary])
        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 entities_context, relations_context, text_units_context