从零搭建本地 AI 知识库：双索引 RAG 与自动化流水线（附可直接交给 AI 的完整代码）

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

┌─────────────────────────────────────────┐
│  Layer 1 · 记忆层（Agent 多级记忆）       │
│  身份 / 用户档案 / 长期记忆 / 每日日志     │
│  + 聊天备份→每周蒸馏→RAG 三级记忆闭环     │
├─────────────────────────────────────────┤
│  Layer 2 · 知识库层（PARA + 领域 wiki）    │
│  10-PROJECTS / 20-AREAS / 30-RESOURCES   │
│  + 业务真实数据（软链穿透）               │
├─────────────────────────────────────────┤
│  Layer 3 · 索引层（单生产索引）           │
│  full_workspace：全库 + 表格聚合         │
│  （双索引已于 2026-08 简化为单索引）      │
├─────────────────────────────────────────┤
│  Layer 4 · 自动化层（13 个 automation）   │
│  增量同步+verify+体检 / 备份 / 蒸馏      │
└─────────────────────────────────────────┘

1

scan → embed → merge → verify → sync_obsidian

 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

你是本地 AI 知识库搭建助手。请严格按以下要求执行：

1. 读取本页面附录 A 中的三个代码块：setup.sh、build_index.py、ask.py，
   将它们完整写入用户指定目录（如 ~/kb-demo/）
2. 执行 bash setup.sh <目标目录>，完成目录创建、Ollama 检查与模型拉取
3. 运行 python3 build_index.py 构建向量索引
4. 依次运行以下 3 个检索验证：
   python3 ask.py "采购决策应该关注什么"
   python3 ask.py "供应商评估看什么"
   python3 ask.py "什么是复利思维"
5. 汇报执行结果：文件数、索引块数、每次检索的相似度分数与命中文件

验收标准：
- build_index.py 输出 [done] 且无报错
- 至少 2 个检索问题的 Top-1 相似度 > 0.5
- 所有路径使用绝对路径，不遗留临时文件

防坑要求（重要）：
- 排除清单优先：40-ARCHIVE/ 和 .kb_index/ 本身绝不入索引（防自引用）；
  任何含密钥、密码、token 的文件名/目录名发现即跳过并提醒用户
- fail-closed：任何脚本写入路径必须先校验目标不是系统目录、
  不在排除清单内；校验失败直接退出并报错，不要"尝试继续"
- 产物验证：索引建好后必须实际跑一次检索并展示命中内容，
  不允许只说"构建成功"就结束——没有检索验证的索引等于没建
- 静默失败检查：后续若设置自动同步任务，每次运行后必须检查
  索引文件 mtime 确实更新、chunk 数确实变化，两者都没变=任务空转，需报警

若 Ollama 未安装，先输出安装指引并停止，不自行安装。

1
2
3
4
5
6

目标目录/
├── 10-PROJECTS/     # 项目（有明确目标与截止时间）
├── 20-AREAS/        # 领域（长期责任，如采购、财务、健康）
├── 30-RESOURCES/    # 资源（参考资料、书籍、模型）
├── 40-ARCHIVE/      # 归档（不再活跃的内容）
└── .kb_index/       # 向量索引（脚本自动生成）

 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

#!/bin/bash
# 一键搭建最小本地 AI 知识库
# 用法: bash setup.sh /path/to/kb-root
set -e

TARGET="${1:-./kb}"
echo "▶ 目标目录: $TARGET"

# 1. 建目录骨架（PARA）
mkdir -p "$TARGET/10-PROJECTS" "$TARGET/20-AREAS" "$TARGET/30-RESOURCES" "$TARGET/40-ARCHIVE"
echo "✓ 目录骨架已创建"

# 2. 检查 Ollama
if ! curl -s http://localhost:11434/api/tags > /dev/null 2>&1; then
  echo "✗ 未检测到 Ollama，请先安装: https://ollama.com"
  exit 1
fi
echo "✓ Ollama 运行中"

# 3. 拉取 embedding 模型（首次约 400MB）
if ! curl -s http://localhost:11434/api/tags | grep -q "qwen3-embedding"; then
  echo "▶ 拉取 qwen3-embedding:0.6b ..."
  ollama pull qwen3-embedding:0.6b
fi
echo "✓ embedding 模型就绪"

# 4. 创建示例笔记
cat > "$TARGET/20-AREAS/示例-采购方法论.md" << 'EOF'
---
title: 示例-采购方法论
---
采购决策的四个关键：供应商评估、价格谈判、风险分散、长期关系。
供应商评估要看质量、交付、成本、服务四维。
EOF
cat > "$TARGET/30-RESOURCES/示例-复利思维.md" << 'EOF'
---
title: 示例-复利思维
---
复利思维：长期稳定的小幅增长，最终超过短期爆发。
关键是持续投入与足够的时间窗口。
EOF
echo "✓ 示例笔记已创建"

# 5. 提示放入自己的笔记
echo "✓ 现在把你的 .md 笔记放进 10-PROJECTS / 20-AREAS / 30-RESOURCES"
echo "  然后运行: python3 build_index.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

#!/usr/bin/env python3
"""最小本地 RAG 索引构建脚本
用法: python3 build_index.py
依赖: pip install numpy requests
模型: Ollama 本地 embedding（qwen3-embedding:0.6b）
"""
import hashlib
import json
import os
import re
import sys

import numpy as np
import requests

KB_ROOT = os.path.dirname(os.path.abspath(__file__))
INDEX_DIR = os.path.join(KB_ROOT, ".kb_index")
OLLAMA_URL = "http://localhost:11434/api/embed"
EMBED_MODEL = "qwen3-embedding:0.6b"   # 可换任何 Ollama embedding 模型
CHUNK_CHARS = 300                       # 分块长度
CHUNK_OVERLAP = 50                      # 分块重叠
IGNORE_DIRS = {".kb_index", ".git", "node_modules", "40-ARCHIVE"}


def sha256(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def collect_md_files(root):
    files = []
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in IGNORE_DIRS]
        for fn in filenames:
            if fn.endswith(".md"):
                files.append(os.path.join(dirpath, fn))
    return sorted(files)


def chunk_text(text):
    text = re.sub(r"\n{3,}", "\n\n", text.strip())
    chunks = []
    i = 0
    while i < len(text):
        chunks.append(text[i:i + CHUNK_CHARS])
        i += CHUNK_CHARS - CHUNK_OVERLAP
    return chunks or [""]


def embed(texts):
    if not texts:
        return []
    r = requests.post(OLLAMA_URL, json={"model": EMBED_MODEL, "input": texts}, timeout=120)
    r.raise_for_status()
    return r.json()["embeddings"]


def main():
    os.makedirs(INDEX_DIR, exist_ok=True)
    files = collect_md_files(KB_ROOT)
    print(f"[scan] 发现 {len(files)} 个 md 文件")

    # 增量：指纹对比，只处理变更文件
    fp_path = os.path.join(INDEX_DIR, "file_fingerprint.json")
    old_fp = {}
    if os.path.exists(fp_path):
        with open(fp_path, "r") as f:
            old_fp = json.load(f)

    chunks, metas, dirty = [], [], 0
    for path in files:
        rel = os.path.relpath(path, KB_ROOT)
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            text = f.read()
        fp = sha256(text)
        if old_fp.get(rel) == fp:      # 未变更，跳过
            continue
        dirty += 1
        for c in chunk_text(text):
            chunks.append(c)
            metas.append({"file": rel, "fingerprint": fp})
    print(f"[scan] 变更文件 {dirty} 个，待向量化块 {len(chunks)} 个")

    if chunks:
        vecs = np.array(embed(chunks), dtype=np.float32)
        # 最小版本：变更时全量重建（大规模场景可改细粒度 merge）
        np.save(os.path.join(INDEX_DIR, "vectors.npy"), vecs)
        with open(os.path.join(INDEX_DIR, "metas.json"), "w") as f:
            json.dump(metas, f, ensure_ascii=False)
        with open(fp_path, "w") as f:
            json.dump({os.path.relpath(p, KB_ROOT):
                       sha256(open(p, encoding="utf-8", errors="replace").read())
                       for p in files}, f, ensure_ascii=False, indent=1)
    print(f"[done] 索引 {len(chunks)} 块 → {INDEX_DIR}/vectors.npy")
    print(f"快照: 文件 {len(files)} · 块 {len(chunks)} · 模型 {EMBED_MODEL}")


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

#!/usr/bin/env python3
"""最小本地 RAG 检索脚本
用法: python3 ask.py "你的问题"
"""
import json
import os
import sys

import numpy as np
import requests

INDEX_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".kb_index")
OLLAMA_URL = "http://localhost:11434/api/embed"
EMBED_MODEL = "qwen3-embedding:0.6b"
TOP_K = 3


def embed(text):
    r = requests.post(OLLAMA_URL, json={"model": EMBED_MODEL, "input": [text]}, timeout=60)
    r.raise_for_status()
    return np.array(r.json()["embeddings"][0], dtype=np.float32)


def cosine(a, b):
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))


def main():
    if len(sys.argv) < 2:
        print("用法: python3 ask.py '你的问题'")
        sys.exit(1)
    query = sys.argv[1]

    vecs = np.load(os.path.join(INDEX_DIR, "vectors.npy"))
    with open(os.path.join(INDEX_DIR, "metas.json"), "r") as f:
        metas = json.load(f)

    qv = embed(query)
    scores = [cosine(qv, v) for v in vecs]
    top = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:TOP_K]

    print(f"问题: {query}\n")
    for rank, idx in enumerate(top, 1):
        print(f"[{rank}] 相似度 {scores[idx]:.3f} | {metas[idx]['file']}")
        src = os.path.join(os.path.dirname(os.path.abspath(__file__)), metas[idx]["file"])
        with open(src, encoding="utf-8", errors="replace") as f:
            print("    " + f.read()[:120].replace("\n", " "))


if __name__ == "__main__":
    main()

1
2
3
4
5

# crontab -e 添加（macOS/Linux）：
# 每日 12:15 增量同步
15 12 * * * cd /path/to/kb && python3 build_index.py >> .kb_index/sync.log 2>&1
# 每周日 02:00 全量重建兜底
0 2 * * 0 cd /path/to/kb && rm -rf .kb_index/vectors.npy && python3 build_index.py >> .kb_index/rebuild.log 2>&1

1
2
3

用户问：哪些供应商横跨 3 个品类且累计金额超 500 万？
向量 RAG：召回 5 段采购规则片段 → LLM 编造答案（幻觉）
多跳查询：pandas 直查 75,189 行明细 → group by 供应商 → filter 品类数 ≥ 3 → filter 金额 ≥ 500 万

  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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230

#!/usr/bin/env python3
"""
multi_hop_query.py - 跨表多跳查询器（v1.0）

解决的问题：向量 RAG 只能召回 500 字碎片，跨表关系题（供应商×品类×金额×年度）
的答案不存在于任何单文档 → 残缺/幻觉。本工具绕过向量检索，pandas 直查黄金源
Excel，任意交叉维度一次算清，零幻觉，附证据行号。

数据源（黄金源，唯一权威）：
  data/sales_master.xlsx  ←  你替换成自己的数据源
  · 完整合并明细 sheet：N 行 × M 列
  · 必备列：日期 / 供应商 / 分类 / 物料名称 / 采购金额 / 实收数量

支持 7 类查询（自然语言 → 意图识别 → 精确计算）：
  1. supplier_rank          供应商排名（全库/按年/按品类）
  2. cross_category         横跨 N 个品类且金额超 X 的供应商
  3. supplier_detail        某供应商的品类构成/年度趋势/明细
  4. material_supplier      某物料的供应商构成
  5. year_compare           年度对比（总金额/供应商数/品类数）
  6. supplier_pair          ★两家供应商的物料交集（COL_MATERIAL only）
  7. supplier_by_category   ★跨 ≥2 品类的供应商清单（group by supplier）

用法：
  python multi_hop_query.py "哪些供应商横跨3个品类且累计金额超500万"
  python multi_hop_query.py "帝伟的采购构成"
  python multi_hop_query.py "帝伟和汇富豪达的共同物料"
  python multi_hop_query.py "哪些供应商同时供应门和板材"
  python multi_hop_query.py "2025年供应商排名前10"

依赖：openpyxl / pandas
"""
import re
import sys
from pathlib import Path

import pandas as pd

# ============ 改成你的数据源 ============
ROOT = Path(__file__).resolve().parent
XLSX = ROOT / "data" / "sales_master.xlsx"
SHEET = "完整合并明细"
# ======================================

COL_SUPPLIER = "供应商"
COL_CAT = "分类"
COL_AMOUNT = "采购金额/加工费"
COL_DATE = "日期"
COL_MATERIAL = "物料名称"


def load_df():
    df = pd.read_excel(XLSX, sheet_name=SHEET)
    df["_year"] = pd.to_datetime(df[COL_DATE], errors="coerce").dt.year
    df["_amount"] = pd.to_numeric(df[COL_AMOUNT], errors="coerce")
    return df


def fmt_money(v):
    """金额格式化：亿元/万元/元"""
    if abs(v) >= 1e8:
        return f"¥{v/1e8:.2f}亿"
    if abs(v) >= 1e4:
        return f"¥{v/1e4:.1f}万"
    return f"¥{v:,.2f}"


def extract_years(q):
    """从查询里抽年份列表"""
    return [int(y) for y in re.findall(r"20\d{2}", q)]


def extract_threshold(q):
    """'超500万' / '超过2000万' / '大于100万' → 元。默认 None。"""
    m = re.search(r"(?:超|超过|大于|高于)?\s*(\d+(?:\.\d+)?)\s*(亿|万)", q)
    if not m:
        return None
    val = float(m.group(1))
    if m.group(2) == "亿":
        return val * 1e8
    return val * 1e4


def extract_supplier_pair(q):
    """'A 和 B 的共同物料' → [A, B]"""
    m = re.search(r"([\u4e00-\u9fa5\w]+(?:[\u4e00-\u9fa5\w]+)*?)\s*(?:和|与|跟|及)\s*([\u4e00-\u9fa5\w]+(?:[\u4e00-\u9fa5\w]+)*)", q)
    if m and "公司" in q:
        return [m.group(1), m.group(2)]
    return None


def query_supplier_rank(df, q):
    """Top N 供应商排名"""
    n_m = re.search(r"前\s*(\d+)", q)
    n = int(n_m.group(1)) if n_m else 10
    years = extract_years(q)
    sub = df[df["_year"].isin(years)] if years else df
    grouped = sub.groupby(COL_SUPPLIER)["_amount"].sum().sort_values(ascending=False).head(n)
    out = [f"## Top {n} 供应商排名（{'/'.join(map(str, years)) or '全库'}）\n"]
    out.append("| 排名 | 供应商 | 金额 |")
    out.append("|---:|---|---:|")
    for i, (name, amt) in enumerate(grouped.items(), 1):
        out.append(f"| {i} | {name} | {fmt_money(amt)} |")
    return "\n".join(out)


def query_supplier_pair(df, q):
    """★v1.0：两家供应商的物料交集（仅物料，不混品类）"""
    pair = extract_supplier_pair(q)
    if not pair:
        return None
    a, b = pair
    sub = df[df[COL_SUPPLIER].isin([a, b])]
    if sub.empty:
        return f"未找到供应商：{a} / {b}"
    mats_a = set(sub[sub[COL_SUPPLIER] == a][COL_MATERIAL].dropna().unique())
    mats_b = set(sub[sub[COL_SUPPLIER] == b][COL_MATERIAL].dropna().unique())
    overlap = mats_a & mats_b
    only_a = mats_a - mats_b
    only_b = mats_b - mats_a
    if not overlap:
        return f"## {a} × {b} 无共同物料\n仅 {a}: {len(only_a)} 个，仅 {b}: {len(only_b)} 个"
    # 按两家合计金额排序
    amt_overlap = sub[sub[COL_MATERIAL].isin(overlap)].groupby(COL_MATERIAL)["_amount"].sum().sort_values(ascending=False)
    out = [f"## {a} × {b} 物料关系\n"]
    out.append(f"**共同物料数**：{len(overlap)} · **仅 {a}**：{len(only_a)} · **仅 {b}**：{len(only_b)}\n")
    out.append(f"\n**前 20 共同物料（按两家合计金额降序）**：\n")
    out.append("| 物料 | 共同采购合计 |")
    out.append("|---|---:|")
    for mat, amt in amt_overlap.head(20).items():
        out.append(f"| {mat} | {fmt_money(amt)} |")
    out.append(f"\n---\n📎 计算口径：黄金源 {XLSX.name} · 两家供应商 COL_MATERIAL 集合交集（仅物料，不混品类），共 {len(overlap)} 个共同物料 · 零向量检索")
    return "\n".join(out)


def query_supplier_by_category(df, q):
    """★v1.0：跨 ≥2 品类的供应商清单（group by supplier，never invent）"""
    cats_m = re.findall(r"([\u4e00-\u9fa5]+(?:[\u4e00-\u9fa5]+)*?)(?:和|与|跟|及|同时供应|同时供)", q)
    cats = [c for c in cats_m if len(c) >= 2 and c not in ["哪些", "什么", "这个", "那个", "供应商", "品类", "供应", "横跨"]][:5]
    if len(cats) < 2:
        # 尝试从 query 中识别品类词
        for c in ["门", "板材", "五金", "压缩机", "电机", "管路", "钣金", "焊接", "塑料", "橡胶"]:
            if c in q:
                cats.append(c)
        cats = list(set(cats))
    if len(cats) < 2:
        return None
    sub = df[df[COL_CAT].isin(cats)]
    if sub.empty:
        return f"未找到品类 {cats} 的数据"
    grp = sub.groupby(COL_SUPPLIER).agg(
        cats=("COL_CAT", lambda x: set(x)),
        total=("COL_CAT", "count"),
    )
    qualified = grp[grp["cats"].apply(lambda s: cats[0] in s and cats[1] in s)]
    out = [f"## 同时供应「{'」+「'.join(cats)}」的供应商（共 {len(qualified)} 家）\n"]
    if qualified.empty:
        out.append("（无供应商同时供应这些品类）")
    else:
        out.append("| 供应商 | 两品类合计金额 |")
        out.append("|---|---:|")
        for sup in qualified.index:
            amt = sub[sub[COL_SUPPLIER] == sup]["_amount"].sum()
            out.append(f"| {sup} | {fmt_money(amt)} |")
    out.append(f"\n---\n📎 计算口径：黄金源 {XLSX.name} · 按供应商 group by COL_CAT，品类集合 ⊇ {{{', '.join(cats)}}} · 永不臆造供应商对 · 零向量检索")
    return "\n".join(out)


def query_year_compare(df, q):
    """年度对比"""
    years = extract_years(q)
    if len(years) < 2:
        return None
    out = [f"## {' vs '.join(map(str, years))} 年度对比\n"]
    out.append("| 指标 | " + " | ".join(map(str, years)) + " |")
    out.append("|---|" + "---:|" * len(years))
    metrics = [("_amount", "总金额"), (COL_SUPPLIER, "供应商数"), (COL_CAT, "品类数"), (COL_MATERIAL, "物料数")]
    for col, label in metrics:
        cells = []
        for y in years:
            sub = df[df["_year"] == y]
            if col == "_amount":
                cells.append(fmt_money(sub[col].sum()))
            else:
                cells.append(str(sub[col].nunique()))
        out.append(f"| {label} | " + " | ".join(cells) + " |")
    return "\n".join(out)


def route_query(df, q):
    """意图路由：自然语言 → 决策表 → 执行函数"""
    q_lower = q.lower()

    # 优先级 1：supplier_pair（两家供应商对，含"和/与/跟/及"）
    if extract_supplier_pair(q) and ("共同" in q or "相同" in q or "交集" in q or "都" in q):
        r = query_supplier_pair(df, q)
        if r: return r

    # 优先级 2：supplier_by_category（"哪些供应商同时供应 X 和 Y"）
    if "供应" in q and ("同时" in q or "和" in q or "与" in q):
        r = query_supplier_by_category(df, q)
        if r: return r

    # 优先级 3：supplier_rank（"前 N 大供应商"）
    if "前" in q and ("供应商" in q or "排名" in q or "TOP" in q.upper()):
        return query_supplier_rank(df, q)

    # 优先级 4：year_compare（"X 年 vs Y 年" / "X-Y 年度对比"）
    years = extract_years(q)
    if len(years) >= 2 and ("对比" in q or "比较" in q or "vs" in q_lower or "VS" in q):
        return query_year_compare(df, q)

    return None  # 非多跳 → 回退向量检索


def main():
    if len(sys.argv) < 2:
        print("用法: python multi_hop_query.py \"<自然语言查询>\"")
        print('示例: python multi_hop_query.py "帝伟和汇富豪达的共同物料"')
        sys.exit(1)
    query = sys.argv[1]
    df = load_df()
    result = route_query(df, query)
    if result:
        print(result)
    else:
        print(f"未识别为多跳查询，建议回退向量检索。原始 query: {query}")


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
 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

#!/usr/bin/env python3
"""
kb_star_map.py - KB 全景星图生成器（v1.0）

扫描 KB 全库 Markdown，解析 [[wikilink]] 关系，构建节点+边的图结构，
输出 JSON（节点 + 连接）供前端可视化，或直接生成静态 HTML 看板。

用法：
  python kb_star_map.py /path/to/kb /path/to/output
"""
import json
import re
import sys
from pathlib import Path
from collections import Counter


def extract_wikilinks(text):
    """提取 [[link]] 和 [[link|alias]] 格式"""
    return re.findall(r"\[\[([^\]\|#]+)(?:\|[^\]]+)?(?:#[^\]]+)?\]\]", text)


def scan_vault(root):
    """扫描所有 .md，构建节点和边"""
    nodes = {}  # path → {title, tags, size}
    edges = []  # [(src, dst), ...]

    for md in root.rglob("*.md"):
        rel = str(md.relative_to(root))
        nodes[rel] = {
            "path": rel,
            "title": md.stem,
            "size_kb": round(md.stat().st_size / 1024, 1),
        }
        try:
            text = md.read_text(encoding="utf-8", errors="replace")
        except Exception:
            continue
        for link in extract_wikilinks(text):
            dst = link.strip() + ".md"
            # 模糊匹配（处理子目录）
            for node_path in nodes:
                if node_path.endswith(dst) or dst.endswith(node_path):
                    edges.append((rel, node_path))
                    break

    return nodes, edges


def generate_dashboard(nodes, edges, output_dir):
    """输出 JSON + 简易 HTML 看板"""
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)

    # 节点入度（被多少节点引用）
    in_deg = Counter(dst for _, dst in edges)
    for path, node in nodes.items():
        node["in_degree"] = in_deg.get(path, 0)
        node["out_degree"] = sum(1 for s, _ in edges if s == path)

    data = {
        "generated_at": str(Path(__file__).stat().st_mtime),
        "node_count": len(nodes),
        "edge_count": len(edges),
        "nodes": list(nodes.values()),
        "edges": [{"src": s, "dst": d} for s, d in edges],
        "orphan_count": sum(1 for n in nodes.values() if n["in_degree"] == 0 and n["out_degree"] == 0),
    }
    (out / "kb_star_map.json").write_text(json.dumps(data, ensure_ascii=False, indent=2))

    # 简易 HTML（force-graph 或 d3-force）
    html = f"""<!DOCTYPE html>
<html><head><meta charset="utf-8"><title>KB 全景看板</title>
<script src="https://unpkg.com/force-graph"></script>
<style>body{{font-family:sans-serif;margin:0}}#info{{position:fixed;top:10px;left:10px;background:#fff;padding:10px;border-radius:8px;box-shadow:0 0 10px rgba(0,0,0,.1)}}</style>
</head><body>
<div id="info"><b>KB 全景看板</b><br/>节点 {len(nodes)} · 连接 {len(edges)} · 孤儿 {data['orphan_count']}</div>
<div id="graph"></div>
<script>
fetch('kb_star_map.json').then(r=>r.json()).then(data=>{{
  const nodes = data.nodes.map(n=>({{id:n.path,name:n.title,size:n.size_kb,in_degree:n.in_degree}}));
  const links = data.edges;
  ForceGraph()(document.getElementById('graph')).graphData({{nodes, links}});
}});
</script>
</body></html>"""
    (out / "index.html").write_text(html)

    print(f"✅ 节点 {len(nodes)} · 连接 {len(edges)} · 孤儿 {data['orphan_count']}")
    print(f"   JSON: {out / 'kb_star_map.json'}")
    print(f"   HTML: {out / 'index.html'}")


def main():
    if len(sys.argv) < 3:
        print("用法: python kb_star_map.py <kb_root> <output_dir>")
        sys.exit(1)
    root = Path(sys.argv[1])
    nodes, edges = scan_vault(root)
    generate_dashboard(nodes, edges, sys.argv[2])


if __name__ == "__main__":
    main()

1
2

节点 3,263 · 连接 4,609 · 孤儿 487
JSON: 3.2 MB · HTML 实时渲染

 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

#!/usr/bin/env python3
"""
mcp_kb_server.py - KB Auto-Retrieval MCP server (stdio)

把 kb_search.sh 包装为 MCP 工具 kb_auto_search，使任何 MCP 客户端
（WorkBuddy / Codex / ZCode / Kimi Work）都能通过标准 MCP 协议调用。

注册方式（~/.workbuddy/.mcp.json）：
{
  "mcpServers": {
    "kb_search": {
      "command": "python3",
      "args": ["/path/to/mcp_kb_server.py"]
    }
  }
}

工具暴露：kb_auto_search
  参数：message (str) — 用户原始消息
  返回：JSON（hits[] 或 answer）
"""
import json
import os
import subprocess
import sys
from pathlib import Path

KB_AUTO_HOOK = Path(__file__).resolve().parent / "kb_auto_hook.py"

SERVER_INFO = {"name": "kb_search", "version": "0.1.0"}
SERVER_CAPABILITIES = {"tools": {}}


def call_kb_auto_search(message):
    """调用 kb_auto_hook.py auto "<message>"，返回 stdout JSON"""
    result = subprocess.run(
        [sys.executable, str(KB_AUTO_HOOK), "auto", message],
        capture_output=True, text=True, timeout=30,
    )
    if result.returncode != 0:
        return {"status": "error", "stderr": result.stderr[:500]}
    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return {"status": "error", "raw": result.stdout[:500]}


def handle_request(req):
    method = req.get("method")
    req_id = req.get("id")

    if method == "initialize":
        return {"jsonrpc": "2.0", "id": req_id, "result": {
            "protocolVersion": "2024-11-05",
            "serverInfo": SERVER_INFO,
            "capabilities": SERVER_CAPABILITIES,
        }}
    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": req_id, "result": {"tools": [{
            "name": "kb_auto_search",
            "description": "知识库自动检索。先用此工具查询知识库，再回答非简单问候/操作指令类消息。",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "message": {"type": "string", "description": "用户原始消息"},
                },
                "required": ["message"],
            },
        }]}}
    if method == "tools/call":
        params = req.get("params", {})
        if params.get("name") == "kb_auto_search":
            message = params.get("arguments", {}).get("message", "")
            result = call_kb_auto_search(message)
            return {"jsonrpc": "2.0", "id": req_id, "result": {
                "content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}],
                "isError": False,
            }}
    return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": "method not found"}}


def main():
    """MCP stdio 主循环：逐行读 JSON-RPC 请求，逐行写响应"""
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
        except json.JSONDecodeError:
            continue
        resp = handle_request(req)
        sys.stdout.write(json.dumps(resp) + "\n")
        sys.stdout.flush()


if __name__ == "__main__":
    main()

1
2
3
4
5
6
7
8

{
  "mcpServers": {
    "kb_search": {
      "command": "python3",
      "args": ["/Users/you/Documents/KnowledgeBase/.workbuddy/scripts/mcp_kb_server.py"]
    }
  }
}

1
2
3
4
5
6

除以下两类外的所有用户消息，必须先调用 kb_auto_search 工具再回答。
- ❌ 简单问候：「你好」「hi」「hello」等 → 不检索
- ❌ 纯操作指令：「帮我跑个命令」「写个脚本」等 → 不检索
工具返回 JSON：status ∈ {executed, executed_but_failed, skipped, uncertain} + hits[] 或 answer
引用规则：自然融入回答，不机械贴段落；引用具体数字时保留计算口径。
失败静默：检索失败/无命中/分类为 uncertain 时，正常回答，不向用户说"检索失败"。

1
2
3
4
5
6
7
8
9

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