from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

def chunk_document(text: str, doc_type: str) -> list[dict]:
    """
    Use structural chunking for documents with clear headers (markdown, PDFs with sections),
    fall back to semantic chunking for unstructured prose.
    """
    if doc_type in ("markdown", "structured_pdf"):
        # Header-based: preserves document hierarchy as metadata
        header_splitter = MarkdownHeaderTextSplitter(
            headers_to_split_on=[
                ("#", "h1"), ("##", "h2"), ("###", "h3")
            ]
        )
        header_chunks = header_splitter.split_text(text)

        # Second pass: limit chunk size within each header section
        char_splitter = RecursiveCharacterTextSplitter(
            chunk_size=800,
            chunk_overlap=100,
            separators=["\n\n", "\n", ". ", " "]
        )
        chunks = []
        for doc in header_chunks:
            sub_chunks = char_splitter.split_documents([doc])
            chunks.extend(sub_chunks)
        return chunks

    else:
        # Semantic chunking for unstructured text
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=600,
            chunk_overlap=80,
            separators=["\n\n", "\n", ". ", "! ", "? ", " "]
        )
        return splitter.create_documents([text])

-- Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Document chunks table with rich metadata for filtering
CREATE TABLE document_chunks (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
    content     TEXT NOT NULL,
    embedding   VECTOR(1536),           -- OpenAI text-embedding-3-small
    chunk_index INTEGER NOT NULL,
    metadata    JSONB DEFAULT '{}',     -- {source, section, doc_type, language, created_at}
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- HNSW index: better recall at query time vs. IVFFlat
-- ef_construction=128 and m=16 is a solid default for most workloads
-- Increase ef_construction for higher recall at the cost of index build time
CREATE INDEX ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 128);

-- Index metadata for fast pre-filtering
CREATE INDEX ON document_chunks USING GIN (metadata);
CREATE INDEX ON document_chunks (document_id);

import asyncio
from openai import AsyncOpenAI
from pgvector.asyncpg import register_vector
import asyncpg

client = AsyncOpenAI()

async def embed_batch(texts: list[str], batch_size: int = 100) -> list[list[float]]:
    """Batch embedding with rate limit handling."""
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = await client.embeddings.create(
            input=batch,
            model="text-embedding-3-small"
        )
        all_embeddings.extend([r.embedding for r in response.data])
    return all_embeddings

async def ingest_document(document_id: str, chunks: list[dict], pool: asyncpg.Pool):
    """
    Async ingest: embed all chunks in parallel batches, then bulk-insert.
    Never ingest one chunk at a time — it's 100x slower.
    """
    texts = [c["content"] for c in chunks]
    embeddings = await embed_batch(texts)

    async with pool.acquire() as conn:
        await register_vector(conn)
        # Bulk insert with executemany for efficiency
        await conn.executemany(
            """
            INSERT INTO document_chunks
                (document_id, content, embedding, chunk_index, metadata)
            VALUES ($1, $2, $3, $4, $5)
            """,
            [
                (document_id, c["content"], emb, idx, c.get("metadata", {}))
                for idx, (c, emb) in enumerate(zip(chunks, embeddings))
            ]
        )

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text

async def hybrid_search(
    query: str,
    query_embedding: list[float],
    db: AsyncSession,
    metadata_filter: dict | None = None,
    top_k: int = 10,
    alpha: float = 0.7,  # weight for semantic vs. keyword; tune per domain
) -> list[dict]:
    """
    Reciprocal Rank Fusion of semantic and full-text search.
    alpha=0.7 favors semantic; lower it for keyword-heavy domains.
    """
    filter_clause = ""
    params = {"embedding": query_embedding, "query": query, "top_k": top_k * 2}

    if metadata_filter:
        filter_clause = "AND metadata @> :filter"
        params["filter"] = metadata_filter

    result = await db.execute(text(f"""
        WITH semantic AS (
            SELECT id, content, metadata,
                   1 - (embedding <=> :embedding::vector) AS score,
                   ROW_NUMBER() OVER (ORDER BY embedding <=> :embedding::vector) AS rank
            FROM document_chunks
            WHERE 1=1 {filter_clause}
            ORDER BY embedding <=> :embedding::vector
            LIMIT :top_k
        ),
        keyword AS (
            SELECT id, content, metadata,
                   ts_rank(to_tsvector('english', content),
                           plainto_tsquery('english', :query)) AS score,
                   ROW_NUMBER() OVER (
                       ORDER BY ts_rank(to_tsvector('english', content),
                                        plainto_tsquery('english', :query)) DESC
                   ) AS rank
            FROM document_chunks
            WHERE to_tsvector('english', content) @@ plainto_tsquery('english', :query)
            {filter_clause}
            LIMIT :top_k
        ),
        fused AS (
            SELECT
                COALESCE(s.id, k.id) AS id,
                COALESCE(s.content, k.content) AS content,
                COALESCE(s.metadata, k.metadata) AS metadata,
                (
                    {alpha} * COALESCE(1.0 / (60 + s.rank), 0) +
                    (1 - {alpha}) * COALESCE(1.0 / (60 + k.rank), 0)
                ) AS rrf_score
            FROM semantic s
            FULL OUTER JOIN keyword k ON s.id = k.id
        )
        SELECT * FROM fused ORDER BY rrf_score DESC LIMIT :top_k
    """), params)

    return [dict(row) for row in result.fetchall()]

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, candidates: list[dict], top_n: int = 5) -> list[dict]:
    """
    Re-rank retrieved candidates with a cross-encoder.
    Only use when retrieval precision is the bottleneck — adds ~50-150ms latency.
    """
    pairs = [(query, c["content"]) for c in candidates]
    scores = reranker.predict(pairs)

    ranked = sorted(
        zip(candidates, scores),
        key=lambda x: x[1],
        reverse=True
    )
    return [doc for doc, score in ranked[:top_n] if score > -5.0]  # threshold, not top-k blind

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class RAGState(TypedDict):
    query: str
    reformulated_query: str | None
    retrieved_chunks: list[dict]
    context: str
    answer: str
    retrieval_attempts: int

def should_retry_retrieval(state: RAGState) -> str:
    """
    Decide whether to retry with query reformulation.
    Retry if: insufficient chunks returned and we haven't tried twice.
    """
    if len(state["retrieved_chunks"]) < 3 and state["retrieval_attempts"] < 2:
        return "reformulate"
    return "generate"

def build_rag_graph():
    graph = StateGraph(RAGState)

    graph.add_node("retrieve", retrieve_node)
    graph.add_node("reformulate", reformulate_query_node)
    graph.add_node("rerank", rerank_node)
    graph.add_node("generate", generate_node)

    graph.set_entry_point("retrieve")
    graph.add_conditional_edges("retrieve", should_retry_retrieval, {
        "reformulate": "reformulate",
        "generate": "rerank"
    })
    graph.add_edge("reformulate", "retrieve")
    graph.add_edge("rerank", "generate")
    graph.add_edge("generate", END)

    return graph.compile()

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

def run_rag_eval(test_cases: list[dict]) -> dict:
    """
    Evaluate pipeline on a golden dataset.
    Run this on every chunking/index/retrieval change — not just before release.

    test_cases: [{"question": ..., "ground_truth": ..., "answer": ..., "contexts": [...]}]
    """
    dataset = Dataset.from_list(test_cases)

    results = evaluate(
        dataset=dataset,
        metrics=[
            faithfulness,         # Does the answer stay grounded in retrieved context?
            answer_relevancy,     # Does the answer actually address the question?
            context_precision,    # Are the retrieved chunks relevant to the question?
            context_recall,       # Did retrieval surface all necessary information?
        ]
    )

    return results