知识库 2.0 升级全解：11 个升级点的逻辑与好处（附 AI 可执行搭建包）

1
2
3
4
5
6
7
8
9

┌─────────────────────────────────────┐
│  L4 输出层  聚合报告（TOP10/预警/对标）  │
├─────────────────────────────────────┤
│  L3 分析层  粗筛→族归组→下钻（prompt链） │
├─────────────────────────────────────┤
│  L2 索引层  SPU 卡片 × 23,743         │
├─────────────────────────────────────┤
│  L1 数据层  行级明细 78,647 行（黄金源） │
└─────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

你是「知识库 2.0 卡片索引层」搭建助手。请严格按以下要求执行：

1. 读取本文档附录中的代码块（generate_spu_cards.py / spu_loader.py /
   spu_screen.py / spu_cards_autoupdate.py / verify_cards_conservation.py），
   完整写入目标目录（如 ~/kb2-demo/），不遗漏、不改动逻辑
2. 准备或生成一份行级明细 Excel（列：日期/供应商/物料编码/物料名称/分类/
   数量/单价/金额），记录其【行数】与【总额】作为 P0 期望值
3. 依次执行并记录输出：
   a. python3 generate_spu_cards.py --source <明细> --out spu_reference \
        --expected-rows <行数> --expected-total <总额>
   b. python3 verify_cards_conservation.py --cards spu_reference \
        --expected-total <总额>
   c. python3 spu_screen.py --cards spu_reference --top 20
   d. 对 SPU-00001 及任意 4 张卡运行
      python3 spu_loader.py --cards spu_reference --source <明细> --spu <SPU>
      验证行号回溯与卡片金额一致
4. 配置幂等自动刷新：crontab 或自动化平台每日触发
   python3 spu_cards_autoupdate.py（参数同上）
5. 反孤岛自检：新产出的每份分析报告正文末尾必须带「关联导航」段（出链>0）；
   跑一次死链扫描，确认 0 新增
6. 逐项完成下方验收清单，汇报每项 PASS/FAIL 与关键输出

验收标准（全部满足才算完成）：
- P0 双锚点通过（行数/总额），无「带病生成」
- 全量对数验证差额 0.00
- 粗筛候选全部为真实品类（无「杂码桶」类结构性噪声）
- loader 下钻行号指回原始明细，行数/金额与卡片一致
- 第二次运行 autoupdate 且源文件未变 → 输出 [skip]（幂等生效）
- 卡片 JSON 未混入 RAG 索引（确认排除清单含 spu_reference/）

防坑要求：
- 金额一律 Decimal，禁止 float 求和（精度污染）
- 供应商真名只写进 _supplier_mask.json，卡片里必须是 SUP-xxx
- 任何脚本写入路径先校验不在排除清单内，校验失败直接退出报错
- 不运行任何真实业务数据——用演示数据完成全流程即可

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

~/kb2-demo/
├── data/                        # 行级明细（黄金源，唯一权威）
│   └── 采购明细.xlsx
├── spu_reference/               # 卡片索引层（刻意不进 RAG）
│   ├── cards/SPU-00001.json …   # 每物料编码一张
│   ├── _manifest.json           # 生成元信息（卡数/时间/源SHA）
│   ├── _source_sha.json         # 幂等刷新锚点（源文件哈希）
│   └── _supplier_mask.json      # 真名→代码映射（独立存放）
├── outputs/                     # 粗筛候选/分析报告（自带关联导航）
├── generate_spu_cards.py
├── spu_loader.py
├── spu_screen.py
├── spu_cards_autoupdate.py
└── verify_cards_conservation.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

#!/usr/bin/env python3
"""
generate_spu_cards.py — SPU 卡片索引生成器（v1.0）

输入：行级采购明细（Excel/CSV）
输出：SPU 卡片（JSON，每物料编码一张）+ _manifest.json + _source_sha.json

核心设计：
1. P0 双锚点预检（行数 + 总额，Decimal 精确）——对不上就停，不带病生成
2. 按物料编码聚合：累计金额 / 年度金额 / 年度均价（加权=金额÷数量）/
   供应商集(脱敏) / 单价区间 / 价格离散系数 / 数据行号指针
3. 卡片 JSON 刻意不进 RAG 索引——语义检索不擅长精确算数，各守边界

用法：
  python3 generate_spu_cards.py \
      --source data/采购明细.xlsx --out spu_reference \
      --expected-rows 78647 --expected-total 214956793.43
依赖：pip install pandas openpyxl
"""
import argparse
import hashlib
import json
import sys
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path

import pandas as pd

# ============ 字段映射（换成你的表头） ============
COL_DATE, COL_SUPPLIER = "日期", "供应商"
COL_CODE, COL_NAME, COL_CAT = "物料编码", "物料名称", "分类"
COL_QTY, COL_PRICE, COL_AMOUNT = "数量", "单价", "金额"
# ==================================================

DISCRETE_WARN = 2.0   # 离散系数（max/min）≥ 该值 → 波动预警


def money(v: Decimal) -> str:
    """Decimal 精确两位，字符串存储避免 float 精度污染"""
    return str(v.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))


def mask_supplier(name: str, mapping: dict) -> str:
    """供应商脱敏：真名 → SUP-xxx。真名只存于 _supplier_mask.json。"""
    if name not in mapping:
        mapping[name] = f"SUP-{len(mapping) + 1:03d}"
    return mapping[name]


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def p0_precheck(df: pd.DataFrame, expected_rows: int, expected_total: str) -> None:
    """P0 门禁：行数 + 总额双锚点。任一不符 → 中止，不许带病生成。"""
    rows = len(df)
    total = sum((Decimal(str(v)) for v in df[COL_AMOUNT].fillna(0)), Decimal("0"))
    print(f"[P0] 行数 {rows}（期望 {expected_rows}）· 总额 {money(total)}（期望 {expected_total}）")
    if rows != expected_rows:
        sys.exit(f"✗ P0 行数不符：{rows} != {expected_rows}，停止生成")
    if total != Decimal(str(expected_total)):
        sys.exit(f"✗ P0 总额不符：{total} != {expected_total}，停止生成")
    print("[P0] ✓ 双锚点通过，可以生成卡片")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--source", required=True, help="行级明细 Excel/CSV")
    ap.add_argument("--out", required=True, help="卡片输出目录")
    ap.add_argument("--expected-rows", type=int, required=True)
    ap.add_argument("--expected-total", type=str, required=True)
    args = ap.parse_args()

    src, out = Path(args.source), Path(args.out)
    cards_dir = out / "cards"
    cards_dir.mkdir(parents=True, exist_ok=True)
    src_sha = sha256_file(src)

    df = pd.read_excel(src) if src.suffix in (".xlsx", ".xls") else pd.read_csv(src)
    df[COL_AMOUNT] = pd.to_numeric(df[COL_AMOUNT], errors="coerce").fillna(0)
    df[COL_QTY] = pd.to_numeric(df[COL_QTY], errors="coerce").fillna(0)
    df["_year"] = pd.to_datetime(df[COL_DATE], errors="coerce").dt.year
    df["_row"] = range(2, len(df) + 2)          # Excel 真实行号（1=表头）

    p0_precheck(df, args.expected_rows, args.expected_total)

    supplier_map: dict = {}
    seq = 0
    for code, g in df.groupby(COL_CODE, sort=False):
        seq += 1
        total = sum((Decimal(str(v)) for v in g[COL_AMOUNT]), Decimal("0"))
        prices = g.loc[g[COL_PRICE] > 0, COL_PRICE]
        p_min, p_max = float(prices.min()), float(prices.max())
        discrete = round(p_max / p_min, 2) if p_min > 0 else None
        top_sup = g[COL_SUPPLIER].value_counts().index[0]
        year_amt = {
            str(y): money(sum((Decimal(str(v)) for v in s[COL_AMOUNT]), Decimal("0")))
            for y, s in g.groupby("_year")
        }
        year_avg = {
            str(y): money((Decimal(str(s[COL_AMOUNT].sum())) / Decimal(str(s[COL_QTY].sum())))
                          if s[COL_QTY].sum() else Decimal("0"))
            for y, s in g.groupby("_year")
        }
        card = {
            "spu_id": f"SPU-{seq:05d}",
            "物料编码": code,
            "物料名称": g[COL_NAME].iloc[0],
            "分类": g[COL_CAT].iloc[0],
            "累计金额": money(total),
            "数据行号": g["_row"].tolist(),        # ← 指回 Excel 原始行
            "年度金额": year_amt,
            "年度均价": year_avg,
            "主供应商_脱敏": mask_supplier(top_sup, supplier_map),
            "供应商数": int(g[COL_SUPPLIER].nunique()),
            "价格离散系数": discrete,
            "波动预警": bool(discrete is not None and discrete >= DISCRETE_WARN),
            "数据源SHA256": src_sha,
            "p0_行数校验": True,
            "p0_总额校验": True,
        }
        (cards_dir / f"{card['spu_id']}.json").write_text(
            json.dumps(card, ensure_ascii=False, indent=2), encoding="utf-8")

    (out / "_supplier_mask.json").write_text(
        json.dumps(supplier_map, ensure_ascii=False, indent=2), encoding="utf-8")
    (out / "_source_sha.json").write_text(
        json.dumps({"path": str(src), "sha256": src_sha}, indent=2), encoding="utf-8")
    (out / "_manifest.json").write_text(json.dumps({
        "generated_at": pd.Timestamp.now().isoformat(timespec="seconds"),
        "card_count": seq,
        "source": str(src),
        "source_sha256": src_sha,
    }, ensure_ascii=False, indent=2), encoding="utf-8")

    print(f"[done] 卡片 {seq} 张 → {cards_dir}")
    print(f"[done] 供应商脱敏映射 {len(supplier_map)} 条（真名仅存此处）")


if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

#!/usr/bin/env python3
"""
spu_loader.py — 卡片 → 原始行的一跳桥（v1.0）

输入 spu_id，按卡片里的「数据行号」指针从源文件抽对应行。
任何卡片数字都能一键回溯原始明细——审计、复盘、跟供应商对账。

用法：
  python3 spu_loader.py --cards spu_reference --source data/采购明细.xlsx --spu SPU-005
"""
import argparse
import json
from pathlib import Path

import pandas as pd


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--cards", required=True, help="卡片目录")
    ap.add_argument("--source", required=True, help="行级明细源文件")
    ap.add_argument("--spu", required=True, help="SPU id，如 SPU-005")
    args = ap.parse_args()

    card = json.loads((Path(args.cards) / "cards" / f"{args.spu}.json").read_text(encoding="utf-8"))
    rows = card["数据行号"]
    print(f"▶ {args.spu} · {card['物料名称']} · 累计金额 ¥{card['累计金额']} · 命中 {len(rows)} 行")

    src = Path(args.source)
    df = pd.read_excel(src) if src.suffix in (".xlsx", ".xls") else pd.read_csv(src)
    sub = df.iloc[[r - 2 for r in rows]]           # Excel 行号 → 0 基索引
    print(sub.to_string(max_rows=30))
    total = float(sub["金额"].astype(float).sum())
    print(f"\n📎 合计 {len(sub)} 行 · 金额合计 ¥{total:,.2f}（应等于卡片累计金额）")


if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

#!/usr/bin/env python3
"""
spu_screen.py — 粗筛分析链（v1.0）

三信号排序，输出候选清单（每条带卡片指针）：
1. 价格离散：分层阈值——大金额档从严（离散≥2.0 预警），小金额档从宽（≥5.0）
2. 单价同比异动：年度均价同比 ≥10% 预警线
3. 供应商集中度：只在大额档看「单一供应商占比」

LLM 只做最后的解读，不做任何数据计算——数字全部来自卡片。

用法：
  python3 spu_screen.py --cards spu_reference --top 20
"""
import argparse
import json
from decimal import Decimal
from pathlib import Path

AMOUNT_BIG = Decimal("1000000")   # 大金额档阈值（元）


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--cards", required=True)
    ap.add_argument("--top", type=int, default=20)
    args = ap.parse_args()

    hits = []
    for p in sorted((Path(args.cards) / "cards").glob("SPU-*.json")):
        c = json.loads(p.read_text(encoding="utf-8"))
        total = Decimal(c["累计金额"])
        warn_threshold = 2.0 if total >= AMOUNT_BIG else 5.0   # 分层阈值

        sig = []
        d = c["价格离散系数"]
        if d is not None and d >= warn_threshold:
            sig.append(f"离散{d}×")
        avgs = [Decimal(v) for v in c["年度均价"].values()]
        if len(avgs) >= 2 and avgs[-2] and (avgs[-1] - avgs[-2]) / avgs[-2] >= Decimal("0.10"):
            sig.append(f"同比涨{(avgs[-1] / avgs[-2] - 1) * 100:.1f}%")
        if total >= AMOUNT_BIG and c["供应商数"] == 1:
            sig.append("单一来源")

        if sig:
            hits.append((total, c["spu_id"], c["物料名称"], " / ".join(sig), c["数据行号"][:3]))

    hits.sort(key=lambda x: x[0], reverse=True)
    print(f"候选 {len(hits)} 个（取 Top {args.top}）\n")
    print("| SPU | 物料 | 累计金额 | 信号 | 行号指针 |")
    print("|---|---|---|---|---|")
    for total, spu, name, sig, rows in hits[: args.top]:
        print(f"| {spu} | {name} | ¥{total:,.2f} | {sig} | {rows} |")
    print("\n→ 交给 LLM 解读：为什么涨、是否单一来源、该不该谈——数据已锚定卡片。")


if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

#!/usr/bin/env python3
"""
spu_cards_autoupdate.py — 幂等自动刷新（v1.0）

记录源文件 SHA-256：变了才重生成，没变就跳过。配合 crontab / 自动化平台
每天跑都行，零副作用（idempotent）。

用法（crontab 每日）：
  30 2 * * * cd ~/kb2-demo && python3 spu_cards_autoupdate.py \
      --source data/采购明细.xlsx --out spu_reference \
      --expected-rows 78647 --expected-total 214956793.43
"""
import argparse
import json
import subprocess
import sys
from pathlib import Path

from generate_spu_cards import sha256_file


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--source", required=True)
    ap.add_argument("--out", required=True)
    ap.add_argument("--expected-rows", type=int, required=True)
    ap.add_argument("--expected-total", type=str, required=True)
    args = ap.parse_args()

    src, out = Path(args.source), Path(args.out)
    sha_file = out / "_source_sha.json"
    current_sha = sha256_file(src)

    if sha_file.exists():
        old = json.loads(sha_file.read_text(encoding="utf-8"))
        if old.get("sha256") == current_sha:
            print(f"[skip] 源文件未变更（{current_sha[:12]}…），卡片保持现状")
            return

    print(f"[rebuild] 源文件已变更（{current_sha[:12]}…），重新生成卡片…")
    subprocess.run(
        [sys.executable, "generate_spu_cards.py",
         "--source", str(src), "--out", str(out),
         "--expected-rows", str(args.expected_rows),
         "--expected-total", args.expected_total],
        check=True,
    )
    print("[done] 幂等刷新完成")


if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

#!/usr/bin/env python3
"""
verify_cards_conservation.py — 全量对数验证（v1.0）

卡片累计金额合计 vs 权威源总额 → 差额必须 0.00。
「先验证再生产」的最后一道闸：任何一张卡算错，这里都会现形。

用法：
  python3 verify_cards_conservation.py --cards spu_reference --expected-total 214956793.43
"""
import argparse
import json
import sys
from decimal import Decimal
from pathlib import Path


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--cards", required=True)
    ap.add_argument("--expected-total", type=str, required=True)
    args = ap.parse_args()

    total = sum(
        (Decimal(json.loads(p.read_text(encoding="utf-8"))["累计金额"])
         for p in (Path(args.cards) / "cards").glob("SPU-*.json")),
        Decimal("0"),
    )
    expect = Decimal(args.expected_total)
    diff = total - expect
    status = "✓ 差额 0.00" if diff == 0 else f"✗ 差额 {diff}"
    print(f"卡片合计 ¥{total:,.2f} vs 权威总额 ¥{expect:,.2f} → {status}")
    if diff != 0:
        sys.exit(1)


if __name__ == "__main__":
    main()