LightRAG — Graph RAG nhẹ, nhanh và cập nhật tăng dần

def chunking_by_token_size(
    tokenizer,
    content,
    split_by_character=None,
    split_by_character_only=False,
    chunk_overlap_token_size=100,
    chunk_token_size=1200,
):
    tokens = tokenizer.encode(content)
    results = []
    for index, start in enumerate(
        range(0, len(tokens), chunk_token_size - chunk_overlap_token_size)
    ):
        chunk_content = tokenizer.decode(tokens[start : start + chunk_token_size])
        results.append({
            "tokens": min(chunk_token_size, len(tokens) - start),
            "content": chunk_content.strip(),
            "chunk_order_index": index,
        })
    return results

@dataclass
class QueryParam:
    mode: Literal["local", "global", "hybrid", "naive", "mix", "bypass"] = "mix"
    only_need_context: bool = False
    only_need_prompt: bool = False
    response_type: str = "Multiple Paragraphs"
    stream: bool = False
    top_k: int = int(os.getenv("TOP_K", str(DEFAULT_TOP_K)))
    chunk_top_k: int = int(os.getenv("CHUNK_TOP_K", str(DEFAULT_CHUNK_TOP_K)))
    enable_rerank: bool = os.getenv("RERANK_BY_DEFAULT", "true").lower() == "true"
    include_references: bool = False

if query_param.mode == "mix" and chunks_vdb:
    vector_chunks = await _get_vector_context(
        query,
        chunks_vdb,
        query_param,
        query_embedding,
    )
    for i, chunk in enumerate(vector_chunks):
        chunk_id = chunk.get("chunk_id") or chunk.get("id")
        if chunk_id:
            chunk_tracking[chunk_id] = {
                "source": "C",
                "frequency": 1,
                "order": i + 1,
            }

self.text_chunks = self.key_string_value_json_storage_cls(
    namespace=NameSpace.KV_STORE_TEXT_CHUNKS,
    workspace=self.workspace,
    embedding_func=self.embedding_func,
)
self.chunk_entity_relation_graph = self.graph_storage_cls(
    namespace=NameSpace.GRAPH_STORE_CHUNK_ENTITY_RELATION,
    workspace=self.workspace,
    embedding_func=self.embedding_func,
)
self.entities_vdb = self.vector_db_storage_cls(
    namespace=NameSpace.VECTOR_STORE_ENTITIES,
    workspace=self.workspace,
    embedding_func=self.embedding_func,
    meta_fields={"entity_name", "source_id", "content", "file_path"},
)

def aggregate_chunk_scores(chunk_results, doc_indices, num_original_docs, aggregation="max"):
    doc_scores = {i: [] for i in range(num_original_docs)}
    for result in chunk_results:
        chunk_idx = result["index"]
        score = result["relevance_score"]
        if 0 <= chunk_idx < len(doc_indices):
            doc_scores[doc_indices[chunk_idx]].append(score)

    aggregated_results = []
    for doc_idx, scores in doc_scores.items():
        if not scores:
            continue
        final_score = max(scores) if aggregation == "max" else sum(scores) / len(scores)
        aggregated_results.append({"index": doc_idx, "relevance_score": final_score})
    return sorted(aggregated_results, key=lambda x: x["relevance_score"], reverse=True)