Hierarchical Indexing - Multi-Level Document Structure

from typing import List, Dict, Optional
import uuid

# ── Index Construction ────────────────────────────────────────────────────────

def build_hierarchical_index(
    documents: List[Dict],
    llm,
    vector_stores: Dict,   # {'l0': ..., 'l1': ..., 'l2': ...}
):
    """
    Build a three-level hierarchical index.
    Each document yields: 1 L0 summary, N L1 sections, M L2 sentences.
    """
    for doc in documents:
        doc_id = doc['id']

        # L0: Generate document-level summary (~200 tokens)
        summary = llm.invoke(
            f"Summarize this document in ~200 tokens, capturing key topics and structure:\n\n{doc['content']}"
        )
        l0_node = {'id': f"l0_{doc_id}", 'content': summary, 'doc_id': doc_id, 'level': 0}
        vector_stores['l0'].add(l0_node)

        # L1: Split into sections by heading or fixed ~500-token windows
        sections = split_into_sections(doc['content'], max_tokens=500)
        for i, section in enumerate(sections):
            l1_id = f"l1_{doc_id}_{i}"
            l1_node = {
                'id': l1_id,
                'content': section,
                'doc_id': doc_id,
                'l0_parent': f"l0_{doc_id}",
                'level': 1,
            }
            vector_stores['l1'].add(l1_node)

            # L2: Split each section into sentences (~100 tokens)
            sentences = split_into_sentences(section, max_tokens=100)
            for j, sent in enumerate(sentences):
                l2_node = {
                    'id': f"l2_{doc_id}_{i}_{j}",
                    'content': sent,
                    'doc_id': doc_id,
                    'l1_parent': l1_id,
                    'l0_parent': f"l0_{doc_id}",
                    'level': 2,
                }
                vector_stores['l2'].add(l2_node)


# ── Query Routing ─────────────────────────────────────────────────────────────

def route_query_level(query: str, llm) -> int:
    """
    Route the query to the appropriate index level.
    Returns 0 (overview), 1 (section), or 2 (specific fact).
    """
    prompt = f"""Classify this query by the granularity of answer it needs.

Query: {query}

Levels:
0 = Overview / summary (e.g., "What is X about?", "Compare X and Y", "Give me an overview")
1 = Section / explanation (e.g., "How does X work?", "Explain the process of Y")
2 = Specific fact (e.g., "What is the value of X?", "When was Y released?", "List the parameters")

Return only the level number (0, 1, or 2)."""

    response = llm.invoke(prompt).strip()
    try:
        return int(response[0])
    except (ValueError, IndexError):
        return 1  # default to section level


# ── Top-Down Drill-Down ───────────────────────────────────────────────────────

def hierarchical_retrieve_topdown(
    query: str,
    vector_stores: Dict,
    top_k: int = 5,
    drill_threshold: float = 0.75,
) -> List[Dict]:
    """
    Top-down retrieval: start at L0, drill to L1 if promising,
    optionally drill to L2 for high-specificity queries.
    """
    # Search L0 summaries
    l0_results = vector_stores['l0'].search(query, top_k=top_k)

    results = []
    for l0_hit in l0_results:
        if l0_hit['score'] >= drill_threshold:
            # This document looks relevant — drill to L1 sections
            l1_results = vector_stores['l1'].search(
                query,
                top_k=top_k,
                filter={'doc_id': l0_hit['doc_id']},
            )
            results.extend(l1_results)
        else:
            # Not specific enough to drill — return the L0 summary itself
            results.append(l0_hit)

    return results[:top_k]


# ── Full RAG Pipeline ─────────────────────────────────────────────────────────

def hierarchical_rag(
    query: str,
    vector_stores: Dict,
    llm,
    top_k: int = 5,
    use_routing: bool = True,
) -> Dict:
    """
    Hierarchical RAG: route query to the right level, retrieve, generate.
    """
    if use_routing:
        level = route_query_level(query, llm)
        index_name = ['l0', 'l1', 'l2'][level]
        results = vector_stores[index_name].search(query, top_k=top_k)
    else:
        # Default: top-down drill-down
        results = hierarchical_retrieve_topdown(query, vector_stores, top_k=top_k)

    context = "\n\n".join([r['content'] for r in results])
    answer = llm.invoke(f"Context:\n{context}\n\nQuestion: {query}\n\nAnswer:")

    return {
        'answer': answer,
        'level_used': level if use_routing else 'drill-down',
        'chunks_retrieved': len(results),
    }


# ── Helpers ───────────────────────────────────────────────────────────────────

def split_into_sections(text: str, max_tokens: int = 500) -> List[str]:
    """Split text into sections, preferring heading boundaries."""
    import re
    # Split on markdown headings or double newlines, then merge short sections
    raw = re.split(r'\n#{1,3} |\n\n', text)
    sections, current = [], ''
    for chunk in raw:
        if len((current + chunk).split()) > max_tokens and current:
            sections.append(current.strip())
            current = chunk
        else:
            current += '\n\n' + chunk
    if current.strip():
        sections.append(current.strip())
    return sections


def split_into_sentences(text: str, max_tokens: int = 100) -> List[str]:
    """Split text into sentence-level chunks."""
    import re
    sentences = re.split(r'(?<=[.!?])\s+', text)
    chunks, current = [], ''
    for sent in sentences:
        if len((current + sent).split()) > max_tokens and current:
            chunks.append(current.strip())
            current = sent
        else:
            current += ' ' + sent
    if current.strip():
        chunks.append(current.strip())
    return chunks