from langchain_text_splitters import CharacterTextSplitter


splitter = CharacterTextSplitter(
    separator="",        # 純按長(zhǎng)度切
    chunk_size=600,      # 依據(jù)實(shí)驗(yàn)與模型上限調(diào)整
    chunk_overlap=90,    # 15% 重疊
)
chunks = splitter.split_text(text)

import re


def split_sentences_zh(text: str):
    # 在句末標(biāo)點(diǎn)（。！？；）后面帶可選引號(hào)的場(chǎng)景斷句
    pattern = re.compile(r'([^。?。?；]*[。?。?；]+|[^。?。?；]+$)')
    sentences = [m.group(0).strip() for m in pattern.finditer(text) if m.group(0).strip()]
    return sentences


def sentence_chunk(text: str, chunk_size=600, overlap=80):
    sents = split_sentences_zh(text)
    chunks, buf = [], ""
    for s in sents:
        if len(buf) + len(s) <= chunk_size:
            buf += s
        else:
            if buf:
                chunks.append(buf)
            # 簡(jiǎn)單重疊：從當(dāng)前塊尾部截取 overlap 字符與下一句拼接
            buf = (buf[-overlap:] if overlap > 0 and len(buf) > overlap else "") + s
    if buf:
        chunks.append(buf)
    return chunks


chunks = sentence_chunk(text, chunk_size=600, overlap=90)

from hanlp_common.constant import ROOT
import hanlp


tokenizer = hanlp.load('PKU_NAME_MERGED_SIX_MONTHS_CONVSEG')  # 或句法/句子級(jí)管線
# HanLP 高層 API 通常通過句法/語料管線獲得句子邊界，具體以所用版本 API 為準(zhǔn)
# 將句子列表再做聚合為 chunk_size

import re
from langchain_text_splitters import RecursiveCharacterTextSplitter


separators = [
    r"\n#{1,6}\s",                 # 標(biāo)題
    r"\n\d+(?:\.\d+)*\s",          # 數(shù)字編號(hào)標(biāo)題 1. / 2.3. 等
    "\n\n",                        # 段落
    "\n",                          # 行
    " ",                           # 空格
    "",                            # 兜底字符級(jí)
]
splitter = RecursiveCharacterTextSplitter(
    separators=separators,
    chunk_size=700,
    chunk_overlap=100,
    is_separator_regex=True,       # 告訴分割器上面包含正則
)
chunks = splitter.split_text(text)

import re
from typing import List, Dict


heading_pat = re.compile(r'^(#{1,6})\s+(.*)$')  # 標(biāo)題
fence_pat = re.compile(r'^```')                 # fenced code fence


def split_markdown_structure(text: str, chunk_size=900, min_chunk=250, overlap_ratio=0.1) -> List[Dict]:
    lines = text.splitlines()
    sections = []
    in_code = False
    current = {"level": 0, "title": "", "content": [], "path": []}
    
    path_stack = []  # [(level, title)]
    
    for ln in lines:
        if fence_pat.match(ln):
            in_code = not in_code
        m = heading_pat.match(ln) if not in_code else None
        if m:
            if current["content"]:
                sections.append(current)
            level = len(m.group(1))
            title = m.group(2).strip()


            while path_stack and path_stack[-1][0] >= level:
                path_stack.pop()
            path_stack.append((level, title))
            breadcrumbs = [t for _, t in path_stack]
            current = {"level": level, "title": title, "content": [], "path": breadcrumbs}
        else:
            current["content"].append(ln)
    
    if current["content"]:
        sections.append(current)
    
    # 通過二次拆分/合并將部分平鋪成塊
    chunks = []
    def emit_chunk(text_block: str, path: List[str], level: int):
        chunks.append({
            "text": text_block.strip(),
            "meta": {
                "section_title": path[-1] if path else "",
                "breadcrumbs": path,
                "section_level": level,
            }
        })
    
    for sec in sections:
        raw = "\n".join(sec["content"]).strip()
        if not raw:
            continue
        if len(raw) <= chunk_size:
            emit_chunk(raw, sec["path"], sec["level"])
        else:
            paras = [p.strip() for p in raw.split("\n\n") if p.strip()]
            buf = ""
            for p in paras:
                if len(buf) + len(p) + 2 <= chunk_size:
                    buf += (("\n\n" + p) if buf else p)
                else:
                    if buf:
                        emit_chunk(buf, sec["path"], sec["level"])
                    buf = p
            if buf:
                emit_chunk(buf, sec["path"], sec["level"])
    
    merged = []
    for ch in chunks:
        if not merged:
            merged.append(ch)
            continue
        if len(ch["text"]) < min_chunk and merged[-1]["meta"]["breadcrumbs"] == ch["meta"]["breadcrumbs"]:
            merged[-1]["text"] += "\n\n" + ch["text"]
        else:
            merged.append(ch)
    
    overlap = int(chunk_size * overlap_ratio)
    for ch in merged:
        bc = " > ".join(ch["meta"]["breadcrumbs"][-3:])
        prefix = f"[{bc}]\n" if bc else ""
        if prefix and not ch["text"].startswith(prefix):
            ch["text"] = prefix + ch["text"]
        # optional character overlap can在檢索階段用鄰接聚合替代，這里略
    
    return merged

from typing import List, Dict


def chunk_dialogue(turns: List[Dict], max_turns=10, max_chars=900, overlap_turns=2):
    """
    turns: [{"speaker":"User","text":"..." , "ts_start":123, "ts_end":130}, ...]
    """
    chunks = []
    i = 0
    while i < len(turns):
        j = i
        char_count = 0
        speakers = set()
        while j < len(turns):
            t = turns[j]
            uttr_len = len(t["text"])
            # 若單條超長(zhǎng)，允許在句級(jí)二次切分（此處略），但不跨 speaker
            if (j - i + 1) > max_turns or (char_count + uttr_len) > max_chars:
                break
            char_count += uttr_len
            speakers.add(t["speaker"])
            j += 1
        
        if j > i:
            window = turns[i:j]
        elif i < len(turns):
            window = [turns[i]]
        else:
            break
        text = "\n".join([f'{t["speaker"]}: {t["text"]}' for t in window])
        meta = {
            "speakers": list(speakers),
            "turns_range": (i, j - 1),
            "ts_start": window[0].get("ts_start"),
            "ts_end": window[-1].get("ts_end"),
        }
        chunks.append({"text": text, "meta": meta})
        
        # 按輪次重疊回退
        if j >= len(turns):
            break
        next_start = i + len(window) - overlap_turns
        i = max(next_start, i + 1)  # 確保至少前進(jìn)1步
    return chunks

from typing import List, Dict, Tuple
import numpy as np
from sentence_transformers import SentenceTransformer
import re


def split_sentences_zh(text: str) -> List[str]:
    # 簡(jiǎn)易中文分句，可替換為 HanLP/Stanza 更穩(wěn)健的實(shí)現(xiàn)
    pattern = re.compile(r'([^。?。?；]*[。！？；]+|[^。??？；]+$)')
    return [m.group(0).strip() for m in pattern.finditer(text) if m.group(0).strip()]


def rolling_mean(vecs: np.ndarray, i: int, w: int) -> np.ndarray:
    s = max(0, i - w)
    e = min(len(vecs), i + w + 1)
    return vecs[s:e].mean(axis=0)


def semantic_chunk(
    text: str,
    model_name: str = "BAAI/bge-m3",
    window_size: int = 2,
    min_chars: int = 350,
    max_chars: int = 1100,
    lambda_std: float = 0.8,
    overlap_chars: int = 80,
) -> List[Dict]:
    sents = split_sentences_zh(text)
    if not sents:
        return []
    
    model = SentenceTransformer(model_name)
    emb = model.encode(sents, normalize_embeddings=True, batch_size=64, show_progress_bar=False)
    emb = np.asarray(emb)
    
    # 基于窗口均值的“新穎度”分?jǐn)?shù)
    novelties = []
    for i in range(len(sents)):
        ref = rolling_mean(emb, i-1, window_size) if i > 0 else emb[0]
        ref = ref / (np.linalg.norm(ref) + 1e-8)
        novelty = 1.0 - float(np.dot(emb[i], ref))
        novelties.append(novelty)
    novelties = np.array(novelties)
    
    # 相對(duì)閾值：μ + λσ
    mu, sigma = float(novelties.mean()), float(novelties.std() + 1e-8)
    threshold = mu + lambda_std * sigma
    
    chunks, buf, start_idx = [], "", 0
    def flush(end_idx: int):
        nonlocal buf, start_idx
        if buf.strip():
            chunks.append({
                "text": buf.strip(),
                "meta": {"start_sent": start_idx, "end_sent": end_idx-1}
            })
        buf, start_idx = "", end_idx
    
    for i, s in enumerate(sents):
        # 若超長(zhǎng)則先沖洗
        if len(buf) + len(s) > max_chars and len(buf) >= min_chars:
            flush(i)
            # 結(jié)構(gòu)化重疊：附加上一個(gè)塊的尾部
            if overlap_chars > 0 and len(s) < overlap_chars:
                buf = s
                continue
        
        buf += s
        
        # 達(dá)到最小長(zhǎng)度后遇到突變則切分
        if len(buf) >= min_chars and novelties[i] > threshold:
            flush(i + 1)
    
    if buf:
        flush(len(sents))
    
    return chunks

from typing import List, Dict
import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.cluster import KMeans
import re


def split_sentences_zh(text: str) -> List[str]:
    pattern = re.compile(r'([^。??？；]*[。??？；]+|[^。??？；]+$)')
    return [m.group(0).strip() for m in pattern.finditer(text) if m.group(0).strip()]


def topic_chunk(
    text: str,
    k_topics: int = 5,
    min_chars: int = 500,
    max_chars: int = 1400,
    smooth_window: int = 2,
    model_name: str = "BAAI/bge-m3"
) -> List[Dict]:
    sents = split_sentences_zh(text)
    if not sents:
        return []
    
    model = SentenceTransformer(model_name)
    emb = model.encode(sents, normalize_embeddings=True, batch_size=64, show_progress_bar=False)
    emb = np.asarray(emb)
    
    km = KMeans(n_clusters=k_topics, n_init="auto", random_state=42)
    labels = km.fit_predict(emb)
    
    # 簡(jiǎn)單序列平滑：滑窗多數(shù)投票
    smoothed = labels.copy()
    for i in range(len(labels)):
        s = max(0, i - smooth_window)
        e = min(len(labels), i + smooth_window + 1)
        window = labels[s:e]
        vals, counts = np.unique(window, return_counts=True)
        smoothed[i] = int(vals[np.argmax(counts)])
    
    chunks, buf, start_idx, cur_label = [], "", 0, smoothed[0]
    def flush(end_idx: int):
        nonlocal buf, start_idx
        if buf.strip():
            chunks.append({
                "text": buf.strip(),
                "meta": {"start_sent": start_idx, "end_sent": end_idx-1, "topic": int(cur_label)}
            })
        buf, start_idx = "", end_idx
    
    for i, s in enumerate(sents):
        switched = smoothed[i] != cur_label
        over_max = len(buf) + len(s) > max_chars
        under_min = len(buf) < min_chars
        
        # 嘗試延后切分，保證最小塊長(zhǎng)
        if switched and not under_min:
            flush(i)
            cur_label = smoothed[i]
        
        if over_max and not under_min:
            flush(i)
        
        buf += s
    
    if buf:
        flush(len(sents))
    
    return chunks

# 離線：構(gòu)建小塊索引，并保存 parent_id -> 大塊文本 的映射
# 在線檢索：
small_hits = small_index.search(embed(query), top_k=30)
groups = group_by_parent(small_hits)
scored_parents = score_groups(groups, agg="max")
candidates = top_m(scored_parents, m=3)


# 交叉編碼重排
rerank_inputs = [(query, parent_text(pid)) for pid in candidates]
reranked = cross_encoder_rerank(rerank_inputs)


# 組裝上下文：對(duì)每個(gè)父塊，僅保留命中句及其鄰近窗口，并加上標(biāo)題路徑
contexts = []
for pid, _ in reranked:
    hits = groups[pid]
    context = build_local_window(parent_text(pid), hits, window_sents=1)
    contexts.append(prefix_with_breadcrumbs(pid) + context)


final_context = pack_under_budget(contexts, token_budget=3000)    # 留出回答空間

from typing import List, Dict, Tuple
import numpy as np
from sentence_transformers import SentenceTransformer




embedder = SentenceTransformer("BAAI/bge-m3")


def search_parent_child(query: str, top_k_child=40, top_m_parent=3, window_chars=180):
    q = embedder.encode([query], normalize_embeddings=True)[0]
    hits = small_index.search(q, top_k=top_k_child)  # 返回 [(child_id, score), ...]
    # 分組
    groups: Dict[str, List[Tuple[str, float]]] = {}
    for cid, score in hits:
        p = child_parent_id[cid]
        groups.setdefault(p, []).append((cid, float(score)))
    
    # 聚合打分（max + coverage）
    scored = []
    for pid, items in groups.items():
        scores = np.array([s for _, s in items])
        agg = 0.7 * scores.max() + 0.3 * (len(items) / (len(parents[pid]["sent_spans"]) + 1e-6))
        scored.append((pid, float(agg)))
    scored.sort(key=lambda x: x[1], reverse=True)
    candidates = [pid for pid, _ in scored[:top_m_parent]]
    
    # 為每個(gè)父塊構(gòu)造“命中窗口”
    contexts = []
    for pid in candidates:
        ptext = parents[pid]["text"]
        # 找到子塊命中區(qū)間并合并窗口
        spans = sorted([(children[cid]["start"], children[cid]["end"]) for cid, _ in groups[pid]])
        merged = []
        for s, e in spans:
            s = max(0, s - window_chars)
            e = min(len(ptext), e + window_chars)
            if not merged or s > merged[-1][1] + 50:
                merged.append([s, e])
            else:
                merged[-1][1] = max(merged[-1][1], e)
        windows = [ptext[s:e] for s, e in merged]
        prefix = " > ".join(parents[pid]["meta"].get("breadcrumbs", [])[-3:])
        contexts.append((pid, f"[{prefix}]\n" + "\n...\n".join(windows)))
    
    # 交叉編碼重排（此處用占位函數(shù)）
    reranked = cross_encoder_rerank(query, [c[1] for c in contexts])  # 返回 indices 順序
    ordered = [contexts[i] for i in reranked]
    return ordered  # [(parent_id, context_text), ...]

系統(tǒng)：你是分塊器。目標(biāo)：為RAG檢索創(chuàng)建高內(nèi)聚、可追溯的塊。規(guī)則：
1) 不得在代碼/表格/公式中間切分；
2) 每塊400-1000字；
3) 保持標(biāo)題路徑完整；
4) 盡量讓“定義+解釋”在同一塊；
5) 輸出JSON，含 start_offset/end_offset/title_path。


用戶：<文檔片段文本>
助手（示例輸出）：
{
  "segments": [
    {"start": 0, "end": 812, "title_path": ["指南","安裝"], "reason": "完整步驟+注意事項(xiàng)"},
    {"start": 813, "end": 1620, "title_path": ["指南","配置"], "reason": "參數(shù)表與示例緊密相關(guān)"}
  ]
}

from typing import List, Dict


def hybrid_chunk(
    doc_text: str,
    parse_structure,          # 函數(shù)：返回 [{'type': 'text|code|table|dialogue', 'text': str, 'breadcrumbs': [...], 'anchor': str}]
    recursive_splitter,       # 函數(shù)：text -> [{'text': str}]
    sentence_splitter,        # 函數(shù)：text -> [{'text': str}]
    semantic_splitter,        # 函數(shù)：text -> [{'text': str}]
    dialogue_splitter,        # 函數(shù)：turns(list) -> [{'text': str}]，若無對(duì)話則忽略
    max_coarse_len: int = 1100,
    min_chunk_len: int = 320,
    target_len: int = 750,
    overlap_ratio: float = 0.1,
) -> List[Dict]:
    """
    返回格式: [{'text': str, 'meta': {...}}]
    """
    blocks = parse_structure(doc_text)  # 先拿到結(jié)構(gòu)塊
    chunks: List[Dict] = []
    
    def emit(t: str, meta_base: Dict):
        t = t.strip()
        if not t:
            return
        # 結(jié)構(gòu)重疊前綴（標(biāo)題路徑）
        bc = " > ".join(meta_base.get("breadcrumbs", [])[-3:])
        prefix = f"[{bc}]\n" if bc else ""
        chunks.append({
            "text": (prefix + t) if not t.startswith(prefix) else t,
            "meta": meta_base
        })
    
    for b in blocks:
        t = b["text"]
        btype = b.get("type", "text")
        
        # 原子塊：代碼/表格
        if btype in {"code", "table", "formula"}:
            emit(t, {**b, "splitter": "atomic"})
            continue
        
        # 對(duì)話塊
        if btype == "dialogue":
            for ck in dialogue_splitter(b.get("turns", [])):
                emit(ck["text"], {**b, "splitter": "dialogue"})
            continue
        
        # 普通文本：依據(jù)長(zhǎng)度與“可讀性”啟用不同細(xì)分器
        if len(t) <= max_coarse_len:
            # 中短文本：遞歸 or 句子
            sub = recursive_splitter(t)
            # 合并過短子塊
            buf = ""
            for s in sub:
                txt = s["text"]
                if len(buf) + len(txt) < min_chunk_len:
                    buf += txt
                else:
                    emit(buf or txt, {**b, "splitter": "recursive"})
                    buf = "" if buf else ""
            if buf:
                emit(buf, {**b, "splitter": "recursive"})
        else:
            # 超長(zhǎng)文本：語義分塊優(yōu)先
            for ck in semantic_splitter(t):
                emit(ck["text"], {**b, "splitter": "semantic"})
    
    # 輕量字符重疊（可選）
    if overlap_ratio > 0:
        overlapped = []
        for i, ch in enumerate(chunks):
            overlapped.append(ch)
            if i + 1 < len(chunks) and ch["meta"].get("breadcrumbs") == chunks[i+1]["meta"].get("breadcrumbs"):
                # 在相鄰?fù)鹿?jié)塊間引入小比例重疊
                ov = int(len(ch["text"]) * overlap_ratio)
                if ov > 0:
                    head = ch["text"][-ov:]
                    chunks[i+1]["text"] = head + chunks[i+1]["text"]
        chunks = overlapped
    
    return chunks