"""第 32 章：Markdown 片段、BM25、实测稠密向量与 RRF；默认不下载模型。"""

from collections import Counter
import math
import re
from prompt_contracts import read_json, fingerprint

EMBEDDING_MODEL = "BAAI/bge-small-zh-v1.5"
EMBEDDING_REVISION = "7999e1d3359715c523056ef9478215996d62a620"
QUERY_PREFIX = "为这个句子生成表示以用于检索相关文章："


def chunk_documents(documents, max_chars=180):
    """只处理本书使用的二级标题与纯文本段落，保留正文字符偏移。"""
    if max_chars < 1:
        raise ValueError("块长必须为正")
    chunks = []
    for doc in documents:
        # 本例只有公开读者这一种身份；生产系统需由服务端身份决定 ACL。
        if not doc["current"] or doc["visibility"] != "public":
            continue
        body = doc["body"]
        sections = list(re.finditer(r"(?m)^## ([^\n]+)\n([\s\S]*?)(?=^## |\Z)", body))
        if not sections or body[:sections[0].start()].strip():
            raise ValueError("本例要求正文从二级标题开始")
        part = 0
        for section in sections:
            for paragraph in re.finditer(r"[^\n]+", section.group(2)):
                start = section.start(2)+paragraph.start()
                remaining = paragraph.group()
                while remaining:
                    end = min(max_chars, len(remaining))
                    if end < len(remaining):
                        boundary = remaining.rfind("。", 0, end)
                        if boundary >= 0:
                            end = boundary+1
                    content = remaining[:end]
                    part += 1
                    chunks.append(dict(
                        id=f"{doc['id']}:{part}", document=doc["id"], version=doc["version"],
                        start=start, end=start+end, content=content,
                        text=f"{doc['title']} / {section.group(1)}\n{content}",
                        document_sha256=fingerprint(body),
                    ))
                    start += end
                    remaining = remaining[end:]
    if len({chunk["id"] for chunk in chunks}) != len(chunks):
        raise ValueError("片段 ID 重复")
    return chunks


def terms(text):
    result = []
    for run in re.findall(r"[\u4e00-\u9fff]+|[a-z0-9]+", text.lower()):
        if re.fullmatch(r"[a-z0-9]+", run):
            result.append(run)
        else:
            result.extend(run)
            result.extend(run[i:i+2] for i in range(len(run)-1))
    return result


def bm25(query, chunks, k1=1.2, b=.75):
    counts = [Counter(terms(chunk["text"])) for chunk in chunks]
    lengths = [sum(count.values()) for count in counts]
    average = sum(lengths)/len(lengths) if lengths else 1
    scores = [0.0]*len(chunks)
    for term in set(terms(query)):
        df = sum(term in count for count in counts)
        idf = math.log(1+(len(chunks)-df+.5)/(df+.5))
        for i, (count, length) in enumerate(zip(counts, lengths)):
            tf = count[term]
            denominator = tf+k1*(1-b+b*length/average)
            scores[i] += idf*tf*(k1+1)/denominator if denominator else 0
    return sorted(zip(chunks, scores), key=lambda row: (-row[1], row[0]["id"]))


def vector_inputs(chunks, questions):
    return ([dict(key=chunk["id"], text=chunk["text"]) for chunk in chunks]
            + [dict(key=case["id"], text=QUERY_PREFIX+case["question"]) for case in questions])


def load_vectors(chunks, questions, name="workshop-vectors.json"):
    saved = read_json(name)
    expected = vector_inputs(chunks, questions)
    if (saved["inputs_sha256"] != fingerprint(expected)
            or saved["model"] != EMBEDDING_MODEL or saved["revision"] != EMBEDDING_REVISION):
        raise ValueError("向量对应的文本或模型已变化，请重新生成向量")
    vectors = saved["vectors"]
    if set(vectors) != {item["key"] for item in expected}:
        raise ValueError("向量条目不完整")
    for vector in vectors.values():
        if len(vector) != 512 or any(not math.isfinite(x) for x in vector):
            raise ValueError("向量形状或数值错误")
        if abs(sum(x*x for x in vector)-1) > 1e-5:
            raise ValueError("向量未归一化")
    return vectors


def dense(case, chunks, vectors):
    q = vectors[case["id"]]
    return sorted(((chunk, sum(a*b for a, b in zip(q, vectors[chunk["id"]])))
                   for chunk in chunks), key=lambda row: (-row[1], row[0]["id"]))


def reciprocal_rank_fusion(rankings, c=60, candidate_k=5):
    scores, by_id = Counter(), {}
    for ranking in rankings:
        for position, (chunk, _) in enumerate(ranking[:candidate_k], 1):
            scores[chunk["id"]] += 1/(c+position)
            by_id[chunk["id"]] = chunk
    return [(by_id[key], score) for key, score in
            sorted(scores.items(), key=lambda row: (-row[1], row[0]))]


def retrieve(case, chunks, vectors, method="rrf", k=2):
    sparse = bm25(case["question"], chunks)
    neural = dense(case, chunks, vectors)
    ranked = {"bm25": sparse, "dense": neural,
              "rrf": reciprocal_rank_fusion([sparse, neural])}
    if method not in ranked:
        raise ValueError("未知检索方法")
    return [chunk for chunk, _ in ranked[method][:k]]


def retrieval_metrics(cases, rankings):
    answerable = [case for case in cases if case["gold"]]
    if not answerable:
        raise ValueError("没有有答案的问题，召回率无定义")
    recall = complete = hit1 = 0
    for case in answerable:
        expected = set(case["gold"])
        found = [chunk["id"] for chunk in rankings[case["id"]]]
        recall += len(expected.intersection(found))/len(expected)
        complete += expected.issubset(found)
        hit1 += bool(found and found[0] in expected)
    count = len(answerable)
    return dict(count=count, recall=recall/count, complete=complete, hit1=hit1)


def run():
    docs = read_json("workshop-documents.json")
    chunks = chunk_documents(docs)
    questions = read_json("workshop-questions.json")
    vectors = load_vectors(chunks, questions)
    cases = [case for case in questions if case["split"] == "test"]
    report = {}
    for method in ["bm25", "dense", "rrf"]:
        rankings = {case["id"]: retrieve(case, chunks, vectors, method) for case in cases}
        report[method] = dict(metrics=retrieval_metrics(cases, rankings),
                              rankings={key: [row["id"] for row in value]
                                        for key, value in rankings.items()})
    return dict(documents=len(docs), chunks=len(chunks), methods=report)


def main():
    report = run()
    print(f"原始文档={report['documents']}；公开且生效的片段={report['chunks']}；每次取 2 块")
    for method, row in report["methods"].items():
        m = row["metrics"]
        print(f"{method}：首位命中={m['hit1']}/{m['count']}；"
              f"平均 Recall@2={m['recall']:.3f}；证据齐全={m['complete']}/{m['count']}")
    print("多来源问题："+", ".join(report["methods"]["rrf"]["rankings"]["test-two"]))
    print("无答案问题仍会返回："+", ".join(report["methods"]["rrf"]["rankings"]["test-unknown"]))
    print("排名靠前不等于资料中有答案；引用存在不等于引用支持结论。")


if __name__ == "__main__":
    main()

"""核对并汇总第 31、32 章保存的真实模型记录；不会重新调用模型。"""

from prompt_contracts import read_json, fingerprint, messages, grade, strict_json
from retrieval_qa import chunk_documents, load_vectors, retrieve
from local_application_models import MODEL, REVISION


def read_record(mode):
    report = read_json(f"workshop-{mode}.json")
    docs = read_json("workshop-documents.json")
    questions = read_json("workshop-questions.json")
    chunks = chunk_documents(docs)
    by_id = {chunk["id"]: chunk for chunk in chunks}
    vectors = load_vectors(chunks, questions) if mode == "rag" else None
    assert report["complete"] and report["mode"] == mode, "实验未完成或类型错误"
    assert (report["model"], report["revision"]) == (MODEL, REVISION), "模型版本错误"
    assert report["documents_sha256"] == fingerprint(docs), "资料已变化"
    assert report["questions_sha256"] == fingerprint(questions), "问题已变化"
    assert report["max_new_tokens"] == 96 and report["do_sample"] is False
    assert report["environment"]["transformers"] == "4.57.3"
    settings = (["v0", "v1", "v2"] if mode == "prompts" else
                ["v1-schema"] if mode == "schema" else ["none", "bm25", "rrf", "oracle"])
    cases = (questions if mode == "prompts" else
             [q for q in questions if q["split"] == ("dev" if mode == "schema" else "test")])
    expected_keys = {(setting, case["id"]) for setting in settings for case in cases}
    actual_keys = [(row["setting"], row["case_id"]) for row in report["rows"]]
    assert len(actual_keys) == len(set(actual_keys)) and set(actual_keys) == expected_keys
    by_case = {case["id"]: case for case in cases}
    for row in report["rows"]:
        setting, case = row["setting"], by_case[row["case_id"]]
        version = setting if mode == "prompts" else "v1" if mode == "schema" else "v2"
        if mode in {"prompts", "schema"} or setting == "oracle":
            selected = [by_id[key] for key in case["gold"]]
        elif setting == "none":
            selected = []
        else:
            selected = retrieve(case, chunks, vectors, setting)
        conversation = messages(case["question"], selected, version)
        assert row["messages"] == conversation
        assert row["prompt_sha256"] == fingerprint(conversation)
        assert row["chunks"] == [chunk["id"] for chunk in selected]
        assert row["split"] == case["split"]
        assert 0 < row["output_tokens"] <= 96 and row["input_tokens"]+96 <= 2048
        for key, value in grade(row["text"], case, selected).items():
            assert row[key] == value, (mode, setting, case["id"], key)
    return report


def schema_shape(text):
    """单独统计字段集合与类型；引用存在性和拒答约定由 validate 继续检查。"""
    try:
        value = strict_json(text)
        return (isinstance(value, dict) and set(value) == {"answer", "sources", "unknown"}
                and isinstance(value["answer"], str) and type(value["unknown"]) is bool
                and isinstance(value["sources"], list)
                and all(isinstance(item, str) for item in value["sources"]))
    except ValueError:
        return False


def main():
    for mode in ["prompts", "schema", "rag"]:
        report = read_record(mode)
        groups = sorted({(row["setting"], row["split"]) for row in report["rows"]})
        print(f"{mode}：读取并复核 {len(report['rows'])} 条真实生成记录")
        for setting, split in groups:
            rows = [row for row in report["rows"] if (row["setting"], row["split"]) == (setting, split)]
            total = len(rows)
            print(f"  {setting}/{split}：字段形状={sum(schema_shape(r['text']) for r in rows)}/{total}；"
                  f"契约={sum(r['valid'] for r in rows)}/{total}；"
                  f"答案与引用={sum(r['grounded'] for r in rows)}/{total}；"
                  f"正常结束={sum(r['stopped'] for r in rows)}/{total}")


if __name__ == "__main__":
    main()

1. 旧向量对应旧文本，片段位置和引用也可能失效。要更新受影响的表示，并记录版本。
2. 模型可能只看到操作而没看到限制，或只看到时间却不知道限制什么。块边界应尽量保留完整条件。
3. $\log(1+9.5/1.5)\approx1.992$。
4. 两个长度都是 1，余弦公式的分母变为 1。
5. $1/61+1/63\approx0.03227$。不应与另一种检索器的原始分数直接比较。
6. 首位命中可以为真；Recall@1 为 0.5；证据不齐全。命中一条不能代表完成整个问题。
7. 不可以。排序和相似度不是答案存在或事实正确的概率。
8. 它构造了明确的证据缺失。再与人工提供正确证据的结果比较，可以区分检索缺失与模型误读；不要把模型猜对当成检索成功。
9. 同时记录有答案题的证据覆盖和无答案题的实际生成行为。若据此反复改配置，这些题就成为开发集，另留新题检验效果。