RAG 的链路与失效模式

from dataclasses import dataclass, field
from typing import Callable
import re
@dataclass
class Chunk:
    id: str
    text: str
    parent_id: str | None = None     # 父子块：指向所属的大块
    path: str = ""                   # 层级路径：文档 > 章节 > 小节
    meta: dict = field(default_factory=dict)   # source, updated_at, version, lang
    def embed_text(self) -> str:
        """参与嵌入的文本：带上层级路径，提升语义匹配质量"""
        return f"{self.path}\n{self.text}" if self.path else self.text
@dataclass
class RagConfig:
    top_k: int = 20                  # 召回条数（先宽）
    top_n: int = 6                   # 精排后保留条数
    min_score: float = 0.0           # 分数下限，低于此值视为无关
    require_meta: dict | None = None # 必须满足的元数据过滤，如 {"lang": "zh"}
    max_chars_per_chunk: int = 3000  # 单块入上下文的字符上限
def retrieve(query: str, cfg: RagConfig, embed, search) -> list[Chunk]:
    """
    召回阶段。注意三点：
      1. 元数据过滤必须在召回时做，而不是后面过滤 —— 否则 top-k 会被无关内容占满
      2. 先宽召回，再精排（top_k > top_n）
      3. 分数下限很重要：宁可返回空，也不要用低相关内容凑数
    """
    vec = embed(query)
    raw = search(vec, k=cfg.top_k, filters=cfg.require_meta)
    picked = [c for c, score in raw if score >= cfg.min_score]
    return picked
def expand_parents(chunks: list[Chunk], get_parent: Callable[[str], Chunk]) -> list[Chunk]:
    """命中子块后膨胀为父块，但去重 —— 命中同一父块的多个子块只保留一个"""
    seen, out = set(), []
    for c in chunks:
        if c.parent_id and c.parent_id not in seen:
            seen.add(c.parent_id)
            out.append(get_parent(c.parent_id))
        elif not c.parent_id:
            out.append(c)
    return out
def build_rag_context(query: str, cfg: RagConfig, embed, search,
                      rerank, get_parent) -> tuple[str, dict]:
    """
    完整链路 + 诊断信息。
    诊断信息是必须的 —— 没有它，出了问题只能靠猜。
    """
    diag = {}
    # ④ 召回
    cand = retrieve(query, cfg, embed, search)
    diag["召回条数"] = len(cand)
    diag["召回来源分布"] = {}
    for c in cand:
        s = c.meta.get("source", "unknown")
        diag["召回来源分布"][s] = diag["召回来源分布"].get(s, 0) + 1
    if not cand:
        return ("（未检索到相关资料。请明确说明信息不足，不要凭推测作答。）", diag)
    # ⑤ 重排
    ranked = rerank(query, cand)[: cfg.top_n]
    diag["重排后条数"] = len(ranked)
    diag["最高分来源"] = ranked[0].meta.get("source") if ranked else None
    # 父子块膨胀
    expanded = expand_parents(ranked, get_parent)
    diag["膨胀后条数"] = len(expanded)
    # ⑥ 组装：必须带来源、时间、层级；超限要明确标注截断
    lines, used, dropped = [], 0, 0
    for c in expanded:
        body = c.text
        if len(body) > cfg.max_chars_per_chunk:
            body = body[: cfg.max_chars_per_chunk] + "……[内容过长已截断]"
        block = (f"【来源 {c.meta.get('source','未知')}｜"
                 f"更新 {c.meta.get('updated_at','未知')}｜"
                 f"层级 {c.path or '—'}】\n{body}")
        lines.append(block)
        used += len(block)
    if dropped:
        lines.append(f"（另有 {dropped} 条因长度限制未纳入）")
    head = ("以下为检索到的材料。请严格基于这些材料回答，"
            "并在结论处标注依据的来源。若材料不足以回答，请直接说明不足，不要推测。\n\n")
    diag["上下文总字符"] = used
    return head + "\n\n".join(lines), diag
def split_by_structure(doc: str, doc_title: str, target: int = 600,
                       overlap: int = 100) -> list[Chunk]:
    """
    结构化切分（默认推荐）：按标题切，超长再按段落切，段落仍超长才按长度切。
    每块都带上层级路径。
    """
    chunks: list[Chunk] = []
    # 简易实现：以 Markdown 标题为界
    sections = re.split(r"\n(?=#{1,3} )", doc)
    idx = 0
    for sec in sections:
        title = ""
        m = re.match(r"(#{1,3}) (.+)", sec)
        if m:
            title = m.group(2).strip()
        path = f"{doc_title} > {title}" if title else doc_title
        # 超长段落再按长度切，保留重叠
        if len(sec) <= target:
            chunks.append(Chunk(id=f"c{idx}", text=sec.strip(), path=path)); idx += 1
        else:
            start = 0
            while start < len(sec):
                piece = sec[start: start + target]
                chunks.append(Chunk(id=f"c{idx}", text=piece.strip(), path=path))
                idx += 1
                start += target - overlap
    return chunks
def table_to_text(rows: list[list[str]]) -> str:
    """
    表格处理：不要按字符切表格。
    转成「每行一句」的自然语言，作为一个整体块。
    """
    if not rows:
        return ""
    header = rows[0]
    lines = []
    for r in rows[1:]:
        pairs = ", ".join(f"{h}为{v}" for h, v in zip(header, r))
        lines.append(f"该表一行记录：{pairs}")
    return f"表头：{', '.join(header)}\n" + "\n".join(lines)