Chunking Strategiyalari

TEST_QUESTIONS = [
    "Ucell uchun qanday loyiha qilingan?",
    "Kompaniya qachon tashkil topgan?",
    "Bank sohasida qanday tajriba bor?",
    "Loyihalar necha oyda yakunlanadi?",
    "Qaysi texnologiyalar ishlatiladi?",
]

CANDIDATE_SIZES = [400, 600, 800, 1200, 1600]

JUDGE = (
    "Quyidagi javob berilgan kontekstga asoslanganmi? "
    "Faqat bitta raqam qaytar: 1 (ha, to'liq asoslangan) yoki 0 (yo'q). "
    "Boshqa hech narsa yozma.\n\nKontekst:\n{ctx}\n\nJavob:\n{ans}"
)

def judge_faithfulness(context: str, answer: str) -> int:
    out = _ollama_chat(
        [{"role": "user", "content": JUDGE.format(ctx=context, ans=answer)}],
        temperature=0.0, max_tokens=5,
    )
    return 1 if "1" in (out or "") else 0

import time

def benchmark_chunk_size(session, body: str, size: int) -> dict:
    session.execute(text("TRUNCATE chunks"))
    chunks = chunk_text(body, size=size, overlap=int(size * 0.12))
    for i, ch in enumerate(chunks):
        emb, err = generate_query_embedding(ch)
        if not err and emb:
            session.execute(text(INSERT_SQL),
                            {"t": "test", "c": ch, "i": i, "e": str(emb)})
    session.commit()

    scores, latencies = [], []
    for q in TEST_QUESTIONS:
        t0 = time.time()
        emb, _ = generate_query_embedding(q)
        found = retrieve_chunks(session, emb, 5)
        context = "\n\n".join(c["content"] for c in found)
        answer = ask(session, q)
        latencies.append(time.time() - t0)
        scores.append(judge_faithfulness(context, answer))

    return {
        "size": size,
        "chunks": len(chunks),
        "faithfulness": sum(scores) / len(scores),
        "latency": sum(latencies) / len(latencies),
    }


body = open("nihol_corpus.txt", encoding="utf-8").read()
results = [benchmark_chunk_size(session, body, s) for s in CANDIDATE_SIZES]

print(f"{'size':<8}{'chunks':<10}{'faithful':<12}{'latency':<10}")
for r in results:
    print(f"{r['size']:<8}{r['chunks']:<10}{r['faithfulness']:<12.2f}{r['latency']:<10.2f}")

import re
import numpy as np

def split_sentences(body: str) -> list[str]:
    body = " ".join(body.split())
    parts = re.split(r"(?<=[.!?])\s+", body)
    return [p.strip() for p in parts if len(p.strip()) > 15]

def cosine(a: np.ndarray, b: np.ndarray) -> float:
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))


def semantic_chunk(body: str, percentile: int = 80, max_chars: int = 1500) -> list[str]:
    sentences = split_sentences(body)
    if len(sentences) < 3:
        return sentences

    vectors = []
    for s in sentences:
        emb, err = generate_query_embedding(s)
        vectors.append(np.array(emb) if not err and emb else None)

    distances = []
    for i in range(len(sentences) - 1):
        a, b = vectors[i], vectors[i + 1]
        distances.append(1 - cosine(a, b) if a is not None and b is not None else 0.0)

    threshold = np.percentile(distances, percentile)

    chunks, current = [], [sentences[0]]
    for i, d in enumerate(distances):
        too_long = sum(len(x) for x in current) > max_chars
        if d > threshold or too_long:
            chunks.append(" ".join(current))
            current = []
        current.append(sentences[i + 1])
    if current:
        chunks.append(" ".join(current))
    return chunks

body = open("nihol_corpus.txt", encoding="utf-8").read()
chunks = semantic_chunk(body, percentile=80)

print(f"Gaplar soni      : {len(split_sentences(body))}")
print(f"Chunk soni       : {len(chunks)}")
print(f"O'rtacha uzunlik : {sum(len(c) for c in chunks) // len(chunks)} belgi\n")

for c in chunks[:3]:
    print("-" * 70)
    print(c[:180], "...")

PARENT_SQL = (
    "INSERT INTO parent_chunks (title, content) "
    "VALUES (:t, :c) RETURNING id"
)
CHILD_SQL = (
    "INSERT INTO child_chunks (parent_id, content, embedding) "
    "VALUES (:p, :c, :e)"
)

def index_parent_child(session, title: str, body: str) -> tuple[int, int]:
    parents = chunk_text(body, size=2000, overlap=200)
    n_parent = n_child = 0

    for parent_text in parents:
        parent_id = session.execute(
            text(PARENT_SQL), {"t": title, "c": parent_text}
        ).scalar_one()
        n_parent += 1

        for child in chunk_text(parent_text, size=400, overlap=50):
            emb, err = generate_query_embedding(child)
            if err or emb is None:
                continue
            session.execute(text(CHILD_SQL),
                            {"p": parent_id, "c": child, "e": str(emb)})
            n_child += 1

    session.commit()
    return n_parent, n_child

LOOKUP_SQL = (
    "SELECT c.parent_id, p.title, p.content, "
    "       1 - (c.embedding <=> CAST(:e AS vector)) AS score "
    "FROM child_chunks c "
    "JOIN parent_chunks p ON p.id = c.parent_id "
    "ORDER BY c.embedding <=> CAST(:e AS vector) "
    "LIMIT :k"
)

def parent_retrieve(session, question: str, top_k: int = 4) -> list[dict]:
    emb, _ = generate_query_embedding(question)
    rows = session.execute(
        text(LOOKUP_SQL), {"e": str(emb), "k": top_k * 3}
    ).mappings().all()

    seen, out = set(), []
    for r in rows:
        if r["parent_id"] in seen:
            continue
        seen.add(r["parent_id"])
        out.append(dict(r))
        if len(out) == top_k:
            break
    return out


for c in parent_retrieve(session, "Ucell billing tizimi"):
    print(f"{c['score']:.3f}  {c['title']}  ({len(c['content'])} belgi)")

PROPOSITION_PROMPT = (
    "Quyidagi matnni mustaqil faktlarga ajrat. Qoidalar:\n"
    "1. Har bir fakt alohida qatorda\n"
    "2. Har bir fakt o'zi tushunarli bo'lsin — olmoshlarni (u, bu, o'sha) "
    "aniq nom bilan almashtir\n"
    "3. Matnda yo'q ma'lumot qo'shma\n"
    "4. Raqamlama, izoh yozma\n\n"
    "Matn: {chunk}"
)

def generate_propositions(chunk: str) -> list[str]:
    out = _ollama_chat(
        [{"role": "user", "content": PROPOSITION_PROMPT.format(chunk=chunk)}],
        temperature=0.0, max_tokens=600,
    )
    items = [p.strip(" -•0123456789.") for p in (out or "").split("\n")]
    return [p for p in items if len(p) > 25]


sample = (
    "Nihol 2015-yilda tashkil topgan. U dastlab veb-sayt ishlab chiqish bilan "
    "shug'ullangan, keyin korporativ tizimlarga o'tgan. Hozirda kompaniyada "
    "40 dan ortiq mutaxassis ishlaydi."
)
for p in generate_propositions(sample):
    print("-", p)

QUALITY_PROMPT = (
    "Quyidagi faktni 4 mezon bo'yicha bahola. Har biriga 1 (ha) yoki 0 (yo'q):\n"
    "aniqlik — asl matnga mos\n"
    "mustaqillik — kontekstsiz tushunarli, olmosh yo'q\n"
    "toliqlik — tugallangan fikr\n"
    "qisqalik — faqat bitta fakt\n\n"
    "Faqat 4 ta raqamni vergul bilan yoz, masalan: 1,1,1,0\n\n"
    "Asl matn: {source}\n\nFakt: {prop}"
)

def check_quality(proposition: str, source: str) -> dict:
    out = _ollama_chat(
        [{"role": "user", "content": QUALITY_PROMPT.format(source=source[:800],
                                                            prop=proposition)}],
        temperature=0.0, max_tokens=20,
    )
    digits = [c for c in (out or "") if c in "01"][:4]
    if len(digits) < 4:
        return {"passed": False, "scores": [0, 0, 0, 0]}
    scores = [int(d) for d in digits]
    return {"passed": all(s == 1 for s in scores), "scores": scores}


for p in generate_propositions(sample):
    r = check_quality(p, sample)
    mark = "OK  " if r["passed"] else "RAD "
    print(f"{mark} {r['scores']}  {p[:60]}")

PROP_SQL = (
    "INSERT INTO propositions (parent_id, content, embedding) "
    "VALUES (:p, :c, :e)"
)

def index_propositions(session, title: str, body: str) -> dict:
    stats = {"chunks": 0, "generated": 0, "passed": 0}

    for chunk in chunk_text(body, size=800):
        parent_id = session.execute(
            text(PARENT_SQL), {"t": title, "c": chunk}
        ).scalar_one()
        stats["chunks"] += 1

        for prop in generate_propositions(chunk):
            stats["generated"] += 1
            if not check_quality(prop, chunk)["passed"]:
                continue
            emb, err = generate_query_embedding(prop)
            if err or emb is None:
                continue
            session.execute(text(PROP_SQL),
                            {"p": parent_id, "c": prop, "e": str(emb)})
            stats["passed"] += 1

    session.commit()
    return stats


body = open("nihol_corpus.txt", encoding="utf-8").read()
stats = index_propositions(session, "Nihol korpusi", body)
print(stats)
print(f"O'tish darajasi: {stats['passed'] / stats['generated']:.0%}")

PROP_SEARCH = (
    "SELECT p.content AS proposition, pc.content AS parent, pc.title, "
    "       1 - (p.embedding <=> CAST(:e AS vector)) AS score "
    "FROM propositions p "
    "JOIN parent_chunks pc ON pc.id = p.parent_id "
    "ORDER BY p.embedding <=> CAST(:e AS vector) "
    "LIMIT :k"
)

def proposition_search(session, question: str, top_k: int = 3) -> list[dict]:
    emb, _ = generate_query_embedding(question)
    rows = session.execute(text(PROP_SEARCH),
                           {"e": str(emb), "k": top_k}).mappings().all()
    return [dict(r) for r in rows]


for r in proposition_search(session, "Nihol qachon tashkil topgan?"):
    print(f"{r['score']:.3f}  {r['proposition'][:65]}")

from bs4 import BeautifulSoup

def chunks_with_headers_from_html(html: str, doc_title: str,
                                  size: int = 800) -> list[dict]:
    soup = BeautifulSoup(html, "html.parser")
    out = []
    current_section = ""

    for el in soup.find_all(["h1", "h2", "h3", "p", "li"]):
        if el.name in ("h1", "h2", "h3"):
            current_section = el.get_text(strip=True)
            continue

        body = el.get_text(" ", strip=True)
        if len(body) < 60:
            continue

        for piece in chunk_text(body, size=size):
            header = f"Hujjat: {doc_title}"
            if current_section:
                header += f"\nBo'lim: {current_section}"
            out.append({
                "header": header,
                "body": piece,
                "embed_text": f"{header}\n---\n{piece}",
            })
    return out

CCH_SQL = (
    "INSERT INTO chunks (title, header, content, embedding) "
    "VALUES (:t, :h, :c, :e)"
)

def index_with_headers(session, html: str, doc_title: str) -> int:
    saved = 0
    for item in chunks_with_headers_from_html(html, doc_title):
        emb, err = generate_query_embedding(item["embed_text"])
        if err or emb is None:
            continue
        session.execute(text(CCH_SQL), {
            "t": doc_title,
            "h": item["header"],
            "c": item["body"],
            "e": str(emb),
        })
        saved += 1
    session.commit()
    return saved

body = "Ikkinchi bosqichda to'lov shlyuzi ulandi va hisobot moduli qo'shildi."
header = "Hujjat: Ucell billing tizimi modernizatsiyasi\nBo'lim: Loyiha bosqichlari"

question = "Ucell to'lov shlyuzi qachon ulangan?"
q_emb, _ = generate_query_embedding(question)

plain_emb, _ = generate_query_embedding(body)
cch_emb, _ = generate_query_embedding(f"{header}\n---\n{body}")

print(f"CCH siz  : {cosine_similarity(q_emb, plain_emb):.4f}")
print(f"CCH bilan: {cosine_similarity(q_emb, cch_emb):.4f}")

TITLE_PROMPT = (
    "Quyidagi hujjat boshidan uning mavzusini bitta qisqa jumlada yoz. "
    "Maksimal 10 so'z. Faqat sarlavhani yoz.\n\n{head}"
)

_title_cache: dict[str, str] = {}

def infer_title(doc_id: str, body: str) -> str:
    if doc_id in _title_cache:
        return _title_cache[doc_id]
    out = _ollama_chat(
        [{"role": "user", "content": TITLE_PROMPT.format(head=body[:1500])}],
        temperature=0.2, max_tokens=30,
    )
    title = (out or "").strip().strip('"')
    _title_cache[doc_id] = title
    return title

QUESTIONS_PROMPT = (
    "Quyidagi matn qaysi savollarga javob beradi? "
    "3 ta savol yoz. Har birini yangi qatorda, raqamlamasdan. "
    "Savollar bir-biridan farq qilsin.\n\nMatn: {chunk}"
)

def generate_questions(chunk: str, limit: int = 3) -> list[str]:
    out = _ollama_chat(
        [{"role": "user", "content": QUESTIONS_PROMPT.format(chunk=chunk)}],
        temperature=0.5, max_tokens=250,
    )
    items = [q.strip(" -•0123456789.") for q in (out or "").split("\n")]
    items = [q for q in items if q.endswith("?") and len(q) > 15]
    return items[:limit]


chunk = ("Billing moduli 2023-yil iyunda Click va Payme bilan integratsiya "
         "qilindi. Ulanish REST API orqali amalga oshirilgan.")
for q in generate_questions(chunk):
    print("-", q)

GENERIC = ("bu matn", "ushbu matn", "matnda nima", "nima haqida", "qanday ma'lumot")

def filter_questions(questions: list[str], max_similarity: float = 0.9) -> list[str]:
    kept, vectors = [], []
    for q in questions:
        low = q.lower()
        if any(g in low for g in GENERIC):
            continue
        emb, err = generate_query_embedding(q)
        if err or emb is None:
            continue
        if any(cosine_similarity(emb, v) > max_similarity for v in vectors):
            continue
        kept.append(q)
        vectors.append(emb)
    return kept

QUESTION_SQL = (
    "INSERT INTO chunk_questions (chunk_id, question, embedding) "
    "VALUES (:c, :q, :e)"
)

def augment_chunk(session, chunk_id: int, chunk: str) -> int:
    questions = filter_questions(generate_questions(chunk))
    saved = 0
    for q in questions:
        emb, err = generate_query_embedding(q)
        if err or emb is None:
            continue
        session.execute(text(QUESTION_SQL),
                        {"c": chunk_id, "q": q, "e": str(emb)})
        saved += 1
    session.commit()
    return saved

Q_SEARCH = (
    "SELECT c.id, c.title, c.content, q.question, "
    "       1 - (q.embedding <=> CAST(:e AS vector)) AS score "
    "FROM chunk_questions q JOIN chunks c ON c.id = q.chunk_id "
    "ORDER BY q.embedding <=> CAST(:e AS vector) "
    "LIMIT :k"
)

def augmented_search(session, question: str, top_k: int = 5) -> list[dict]:
    emb, _ = generate_query_embedding(question)

    by_text = retrieve_chunks(session, emb, top_k=10)
    rows = session.execute(text(Q_SEARCH),
                           {"e": str(emb), "k": 10}).mappings().all()
    by_question = [dict(r) for r in rows]

    fused = rrf_fuse(by_text, by_question, top_k=top_k)

    matched = {r["id"]: r["question"] for r in by_question}
    for row in fused:
        row["matched_question"] = matched.get(row["id"], "")
    return fused


for c in augmented_search(session, "Click qachon ulangan?"):
    hint = c["matched_question"] or "(matn bo'yicha)"
    print(f"{c['rrf_score']:.5f}  {hint[:55]}")