不能拍照的機密 SOP,怎麼手抄成 Neo4j 知識圖

這篇要解一道很具體的題:手邊有一本 700 頁的機密機械流程 SOP,是公司的資產,規定嚴到——書不能拆、進工作間不能帶手機不能拍照、書不能帶出來、只能帶紙筆進出、出來時紙還要被稽核。你想把它變成能被 AI 查詢的流程文件 RAG,該怎麼做?這是把 GraphRAG 落到一個「輸入通道被限成一支筆」的真實極端情境。

一本不能拍照的書:限制長什麼樣

先把限制攤開,因為每一條都砍掉一種常見做法:

限制	砍掉的做法
書不能拆	🚫 破壞式掃描(裁書脊進饋紙機)
不能帶手機、不能拍照	🚫 翻拍 App、OCR 數位化
書不能帶出來	🚫 帶回家慢慢處理
只能帶紙筆、出來要稽核	⚠️ 你能帶出的只有「手寫的東西」,而且會被檢查
申請就能進、時間不限	✅ 可以多趟、慢慢來

最後一條是唯一的好消息:時間不是瓶頸,你可以分很多趟。真正的瓶頸是那條被限成一支筆的輸入通道——所以整個策略,就是最大化每一頁紙的資訊密度。

第一個轉念:不是抄書,是萃取骨架

直覺會想「那我就一頁一頁手抄」。700 頁純手抄不值得做——那是好幾週的工、會抄錯,錯了 RAG 就跟著錯,而且逐字謄寫機密內容根本過不了稽核。

正確目標是:用人腦當感測器,把 700 頁壓縮成一疊能過稽核的手寫骨架,出來後重建成知識庫。你帶出來的每張紙,目的是在外面觸發你重建,不是取代那本書。這裡有一個關鍵技巧:

為什麼流程文件該進圖,不是向量 RAG

在動手前先判斷路線。一般向量 RAG 是「切塊 → 向量化 → 找最像的塊」,對「請假規定是什麼」這種知識問答很好。但流程文件的價值不在某段文字,而在關係——向量找不到「順序」、找不到「誰負責」、找不到「什麼條件跳哪站」。

一套 80 站的機械流 SOP,正是知識圖 / GraphRAG 的甜蜜點。用這張決策樹判斷就清楚:

SOP 幾乎必然落在最後一格。要注意的反面是:對「一本散文知識書」GraphRAG 常常 overkill(建圖貴、簡單查詢反而更差)——但這裡不是散文書,是高度結構化的站點流程,所以圖法穩贏。

帶一個固定模板:四格站點卡

SOP 的結構高度規則:80 站 × 每站同樣幾個欄位。這代表你不是在「抄書」,是在「填 80 次同一張表」。表格式的東西,手抄又快又不會漏。進去前,腦子裡帶好這個固定模板,每個站填這幾格:

格子	抄什麼	建圖後變成
站號 / 站名 / 上下站	流向:上站 → 本站 → 下站	節點 + ST_NEXT 邊
做什麼 Action	這站的動作	節點屬性
動什麼 Operate	部件 → 動作 → 目標值	節點屬性
⚠ 注意什麼 Caution	警告、門檻、連鎖條件	節點屬性(最重要)
分支 / 條件	若 ___ → 跳站 ___	ST_BRANCH 邊

「注意什麼」是全書的皇冠珠寶——「做什麼 / 動什麼」常常是標準動作,「注意什麼」才是老師傅用事故換來的 tribal knowledge(扭力多少、超過幾度材質劣化、爐門沒關禁止啟動)。這格抄最細、最原話。

分趟策略:先畫地圖,再逐站深挖

因為時間不限、可以多趟,就別想一趟啃完。分兩種趟:

1. 第一趟 = 只畫流程骨架。走完 80 站,每站只抄四樣:站號 / 站名 / 上站→下站 / 一句話這站幹嘛。目的是抓出整條流的形狀,尤其揪出分支、迴圈、返工、並行——這些是散文 SOP 最愛藏、卻是圖法唯一贏向量 RAG 的地方。出來當天打成 80 節點的流程圖,你就有全局地圖。
2. 第二趟起 = 逐站填滿卡片。拿著地圖,一站一站填完整四格。優先順序:①分支點 ②安全 / 品質關鍵站 ③你工作真的會碰的站。

為什麼地圖先行?80 站如果從第 1 站就深挖,可能還沒挖完就沒力氣、也看不到全貌。先拿到骨架,才能安排優先序、並且提早發現結構驚喜(第 60 站原來會跳回第 34 站,這會改變你讀前面 59 站的方式)。

照抄 vs 簡寫:事實與判斷

手寫速度有限,要分清楚哪些照抄、哪些簡寫。判準是「事後重建得出來嗎」:

類型	怎麼處理	例子
事實(記不回來)	照抄,漏一個永遠補不回	扭力、溫度、壓力、料號、門檻、簽核級數、注意事項原話
判斷(記得回來)	簡寫,關鍵詞 + 箭頭,回家補	標準動作的描述、流程大意

這正是正道腦的核心切法:事實是地基不會背叛,判斷會朽。概念你回家 Google 得到,但「這步驟門檻 30 天、簽核三級」這種公司特有數字,漏抄就永遠補不回。

一列 CSV 怎麼變成圖上一個站

出來之後回到熟悉的地盤。整條鏈四段:CSV(一列一站)→ Python 切成參數 → Cypher 用 MERGE 寫進 Neo4j → 圖成形。用一個有分支的站(檢驗分流)走一遍。

① 你準備的資料:CSV 的一列

no,name,next,action,operate,caution,branch_cond,branch_to,output,note
03,檢驗分流,04,量測外徑,量規 → 測 → Ø50.0±0.1mm,量規未歸零讀值失準,外徑超差,05,尺寸紀錄表,分支站

你要準備的就這樣:一列一站、逗號分欄;action / operate / caution 一格塞多項時用 | 隔開。人只做這一步——把手寫卡片打成 CSV。

② 透過程式:Python 讀成參數

import csv
SEP = "|"

def split_list(v):
    return [x.strip() for x in (v or "").split(SEP) if x.strip()]

# csv.DictReader 讀一列 → 多值欄位切成 list
action  = split_list("量測外徑")            # ["量測外徑"]
operate = split_list("量規 → 測 → Ø50.0±0.1mm")
caution = split_list("量規未歸零讀值失準")   # ["量規未歸零讀值失準"]

到這裡,一列文字變成了一包乾淨參數,程式還沒碰資料庫。

③ 怎麼寫入 Neo4j:參數 + Cypher

分兩趟——先建所有節點,再建邊,這樣邊的另一端保證存在。第一趟把參數代進 upsert:

MERGE (s:Station {no: "03"})              // 沒有就建、有就沿用(不重複)
SET s.name    = "檢驗分流",
    s.action  = ["量測外徑"],
    s.operate = ["量規 → 測 → Ø50.0±0.1mm"],
    s.caution = ["量規未歸零讀值失準"]       // list 直接掛節點上

第二趟用 next 和 branch_to 兩欄建邊:

// 順序邊:03 → 04
MATCH (a:Station {no:"03"})
MERGE (b:Station {no:"04"})               // 04 還沒建也沒關係,先補空殼
MERGE (a)-[:ST_NEXT]->(b)

// 分支邊:03 → 05,條件寫在邊上
MATCH (a:Station {no:"03"})
MERGE (b:Station {no:"05"})
MERGE (a)-[r:ST_BRANCH]->(b)
SET r.condition = "外徑超差"

完成的圖 + 自動驗證缺哪些卡

範例 6 站跑完,流程圖長這樣——藍色主鏈是順序,站 03 岔出分支到 05,站 05 又接回 03 形成返工迴圈:

這條返工迴圈,就是散文 SOP 最會藏、向量 RAG 抓不到、而圖一眼就看到的東西。跑完程式還會印出驗證,順便當作跨趟的 TODO 清單:

Station nodes : 6
ST_NEXT edges : 6
ST_BRANCH     : 1
dead-ends (no next out): ['06']     <- 只有末站,對
isolated: none                       <- 沒有孤兒站,對
!! CARDS STILL MISSING: ['47','48']  <- 被指到但還沒抄 = 下一趟 TODO

那個 MISSING 清單特別實用:被 next / branch 指到、但你還沒抄的站(名字是空的),就是下一趟該進去補哪幾張卡。跨趟迭代直接看它收斂到空為止。

圖建好之後,包成 MCP 服務,agent 問「站 47 要注意什麼」就是對這張圖一句 MATCH 秒回、還帶出處(那條查詢本身就是 provenance)。檢索邏輯寫一次,所有 agent 共用同一張圖。

稽核紅線:別夾帶,合法筆記自然過關

有一條紅線一定要釘死:別想把原文逐字帶出,也別用密碼 / 微縮字夾帶。稽核就是防這個,會被抓;而且逐字內容對建知識庫根本不是最好的原料。

你要帶出的是「你消化後的結構」——節點、箭頭圖、關鍵數字。這種東西天生又精簡、又明顯是讀書筆記不是影本,自然過得了稽核。方向對了,稽核就不是敵人:你在合法範圍內做的事(領域專家為工作學習、做筆記),剛好就是產出最好 RAG 原料的方式。用箭頭 / 節點的心智圖取代整段句子,密度高 10 倍,而且一看就是筆記。

什麼時候別這樣做

這套是為「結構化流程 + 你有合法存取權 + 值得長期反覆查」設計的。反過來,遇到這些情況先別建圖:

• 只查幾次:與其花好幾趟建圖,不如把地圖那層抄出來、問幾個具體問題就好。知識庫是「結果」不是「起點」——先用出效益,再收整成庫。
• 內容是散文知識、不是流程:那是向量 RAG 的活,GraphRAG 會 overkill。
• 沒有合法存取權:這整套的前提是「你能申請進去、可以合法做筆記」。沒有授權,不要碰。

但只要命中那三個條件,一套 80 站的 SOP 就是最理想的對象:進去把它壓成骨架,出來重建成一張會被 agent 反覆查詢的流程圖。萃取通道被限成一支筆,方法卻一點不含糊。

建圖程式:load_stations.py

把打好的 stations.csv 灌進 Neo4j。兩趟:先建全部節點、再建邊(MERGE 不用 MATCH,下一站可能後面才抄到);跑完印出還缺哪些卡。

#!/usr/bin/env python3
"""Load hand-captured SOP station cards (stations.csv) into Neo4j as a process graph.

One CSV row = one :Station node. Multi-value fields (action/operate/caution) are
split on " | " and stored as list properties (denormalized on purpose: keeps
"what to watch at station 47" a single MATCH, no hop -- small models write 1-2 hop
Cypher reliably, 3-hop they don't).

Edges built:
  (:Station)-[:ST_NEXT]->(:Station)                 flow order
  (:Station)-[:ST_BRANCH {condition}]->(:Station)   conditional jump / rework / loop

After loading it verifies flow integrity and prints which cards are still MISSING
(referenced by a next/branch but never captured) -- doubles as a cross-trip TODO list.

Env:  NEO4J_URI (default bolt://localhost:7687) | NEO4J_USER (default neo4j) | NEO4J_PASSWORD (required)
Run:  pip install neo4j ;  NEO4J_PASSWORD=... python load_stations.py [stations.csv]
"""
import csv
import os
import sys
import atexit
from neo4j import GraphDatabase

SEP = "|"                 # multi-value separator inside a single field
CSV_DEFAULT = "stations.csv"

_driver = None
def driver():
    """Module-level lazy singleton -- reuse one connection pool, close on exit."""
    global _driver
    if _driver is None:
        uri = os.environ.get("NEO4J_URI", "bolt://localhost:7687")
        user = os.environ.get("NEO4J_USER", "neo4j")
        pw = os.environ.get("NEO4J_PASSWORD")
        if not pw:
            sys.exit("NEO4J_PASSWORD not set (export it; do not hardcode in this file)")
        _driver = GraphDatabase.driver(uri, auth=(user, pw), connection_timeout=15)
        atexit.register(_driver.close)
    return _driver


def split_list(v):
    return [x.strip() for x in (v or "").split(SEP) if x.strip()]


def load_rows(path):
    with open(path, encoding="utf-8-sig", newline="") as f:      # utf-8-sig strips Excel BOM
        rows = []
        for r in csv.DictReader(f):
            no = (r.get("no") or "").strip()
            if not no or no.startswith("#"):                     # skip comment / blank lines
                continue
            rows.append(r)
    return rows


UPSERT = """
MERGE (s:Station {no: $no})
SET s.name = $name, s.action = $action, s.operate = $operate,
    s.caution = $caution, s.output = $output, s.note = $note
"""
# MERGE (not MATCH) the target: the next/branch station may be captured in a later
# row or a later trip. MATCH would silently no-op and leave the edge unbuilt.
NEXT = """
MATCH (a:Station {no: $a})
MERGE (b:Station {no: $b})
MERGE (a)-[:ST_NEXT]->(b)
"""
BRANCH = """
MATCH (a:Station {no: $a})
MERGE (b:Station {no: $b})
MERGE (a)-[r:ST_BRANCH]->(b)
SET r.condition = $cond
"""


def load(path):
    rows = load_rows(path)
    if not rows:
        sys.exit(f"No station rows in {path} (did you delete the # comment lines?)")
    with driver().session() as s:
        for r in rows:                       # pass 1: all nodes first
            no = r["no"].strip()
            s.run(UPSERT, no=no,
                  name=(r.get("name") or "").strip(),
                  action=split_list(r.get("action")),
                  operate=split_list(r.get("operate")),
                  caution=split_list(r.get("caution")),
                  output=(r.get("output") or "").strip(),
                  note=(r.get("note") or "").strip())
        for r in rows:                       # pass 2: edges (targets now guaranteed present)
            no = r["no"].strip()
            nxt = (r.get("next") or "").strip()
            if nxt and nxt != "-":
                s.run(NEXT, a=no, b=nxt)
            bt = (r.get("branch_to") or "").strip()
            if bt:
                s.run(BRANCH, a=no, b=bt, cond=(r.get("branch_cond") or "").strip())
    return len(rows)


def verify():
    q = {
        "stations":  "MATCH (s:Station) RETURN count(s) AS n",
        "next_edges":"MATCH ()-[r:ST_NEXT]->() RETURN count(r) AS n",
        "branches":  "MATCH ()-[r:ST_BRANCH]->() RETURN count(r) AS n",
    }
    with driver().session() as s:
        counts = {k: s.run(v).single()["n"] for k, v in q.items()}
        # referenced by an edge but never captured -> a card you still owe (name is null)
        missing = [r["no"] for r in s.run(
            "MATCH (s:Station) WHERE s.name IS NULL OR s.name = '' RETURN s.no AS no ORDER BY no")]
        # no outgoing ST_NEXT -> should be exactly the final station; more = flow gaps
        deadends = [r["no"] for r in s.run(
            "MATCH (s:Station) WHERE NOT (s)-[:ST_NEXT]->() AND (s.name IS NOT NULL AND s.name<>'') "
            "RETURN s.no AS no ORDER BY no")]
        # fully isolated (no next in or out) -> almost always a mistake
        isolated = [r["no"] for r in s.run(
            "MATCH (s:Station) WHERE NOT (s)-[:ST_NEXT]->() AND NOT (s)<-[:ST_NEXT]-() "
            "RETURN s.no AS no ORDER BY no")]

    print(f"\n  Station nodes : {counts['stations']}")
    print(f"  ST_NEXT edges : {counts['next_edges']}")
    print(f"  ST_BRANCH     : {counts['branches']}")
    print(f"  dead-ends (no next out): {deadends or 'none'}   <- expect exactly the last station")
    print(f"  isolated (no next at all): {isolated or 'none'}   <- should be empty")
    if missing:
        print(f"\n  !! CARDS STILL MISSING (referenced but not captured): {missing}")
        print(f"     -> these are your next-trip TODO list.")
    else:
        print("\n  OK: every referenced station has a captured card.")
    assert counts["stations"] > 0, "no stations loaded"
    assert counts["next_edges"] > 0, "no ST_NEXT edges built -- check the 'next' column"


if __name__ == "__main__":
    path = sys.argv[1] if len(sys.argv) > 1 else CSV_DEFAULT
    n = load(path)
    print(f"Loaded {n} station rows from {path}")
    verify()

向量 RAG 程式:rag_min.py(完整循環)

手冊那半的最小可跑版 —— 向量庫就是一個 list、cosine 自己算,把「原文→embedding→存」「問題→embedding→找最近→拉出」「拉出→餵 LLM→生成」整條看得一清二楚。正式版換成 pgvector 見 HR 站程式頁。

#!/usr/bin/env python3
"""rag_min.py — 向量 RAG 最小可跑範例,對照你講的那個循環。

  【建庫】  原文一段一段 → embedding → 存(向量 + 原文)
  【R 檢索】問題 → embedding → 找最接近的 top-k → 拉出原文
  【A+G】   拉出來 → 塞進 prompt → LLM 依據它生成答案 + 出處

零基礎設施:向量庫就是一個 list、cosine 自己算(看得見數學)。全本地。
前置: ollama pull bge-m3 ; ollama pull qwen2.5:7b ; pip install ollama
跑:   python rag_min.py "軸承過熱怎麼查?"
"""
import sys, math, ollama

EMBED = "bge-m3"        # 多語 embedding,1024 維(中英混雜)
GEN   = "qwen2.5:7b"    # 生成答案(可換)

# 原文:一段一段(手冊的每個站台條目)
DOCS = [
    ("車床 B12",     "主軸 2000rpm,夾持氣壓 0.5MPa。軸承型號 6205。主軸過熱 → 查潤滑油位與冷卻。每 500 小時換油。"),
    ("銑床 B07",     "進給 0.1mm/rev,刀具 T03。軸承型號 6205。加工振動過大 → 查刀具夾持與軸承磨耗。每 300 小時檢查軸承。"),
    ("熱處理爐 C03", "設定溫度 850±10°C,爐門電子連鎖。溫度不均 → 校正熱電偶。超過 870°C 材質劣化。每季校正溫控。"),
]

def embed(text):
    return ollama.embeddings(model=EMBED, prompt=text)["embedding"]     # 文字 → 向量

def cosine(a, b):                                                       # 兩向量有多接近(越接近 1 越像)
    dot = sum(x * y for x, y in zip(a, b))
    na  = math.sqrt(sum(x * x for x in a))
    nb  = math.sqrt(sum(y * y for y in b))
    return dot / (na * nb)

# ---------- 【建庫】原文 → embedding → 存(向量 + 原文 兩份一起存)----------
STORE = [{"name": n, "text": t, "vec": embed(t)} for n, t in DOCS]      # 向量庫 = 一個 list
print(f"建庫:{len(STORE)} 段,每段一個 {len(STORE[0]['vec'])} 維向量\n")

def rag(question, k=2):
    # ---------- 【R 檢索】問題也 embed → 算相似度 → 拉最接近 top-k ----------
    qv = embed(question)
    ranked = sorted(STORE, key=lambda d: cosine(qv, d["vec"]), reverse=True)
    print("相似度排名:")
    for d in ranked:
        print(f"  {cosine(qv, d['vec']):.3f}  {d['name']}")
    hits = ranked[:k]                                                   # 拉出來的原文(top-k)
    print(f"\n拉出前 {k} 段 → {', '.join(h['name'] for h in hits)}\n")

    # ---------- 【A 增強 + G 生成】拉出來塞進 prompt → LLM 只依據它作答 ----------
    context = "\n".join(f"【{h['name']}】{h['text']}" for h in hits)
    prompt  = ("只依據以下手冊內容回答,不要編造,並標明答案出自哪個站台。\n\n"
               f"{context}\n\n問題:{question}")
    resp = ollama.chat(model=GEN, messages=[{"role": "user", "content": prompt}])
    return resp["message"]["content"]

if __name__ == "__main__":
    q = " ".join(sys.argv[1:]) or "軸承過熱怎麼查?"
    print(f"問題:{q}\n")
    print("答案:\n" + rag(q))

實跑輸出 + 互動 demo

用真的 embedding(bge-m3 1024 維)+ 真的 LLM(qwen2.5:7b)跑出來:「過熱」在 B12 原文命中最高 0.718 = 語義而非關鍵字。第二題「哪些機台用 6205」這次對,是因為只有 3 段、k=2 剛好都撈到;真到 80 站有多台用 6205,top-k 會漏 → 這才是上 hybrid 關鍵字的時機。

問題:軸承過熱怎麼查?
相似度排名(找最接近的 vector):
  0.718  車床 B12      ← 最高
  0.602  銑床 B07
  0.573  熱處理爐 C03
拉出前 2 段 → 車床 B12, 銑床 B07
答案:主軸過熱應「查潤滑油位與冷卻」。出自【車床 B12】。

問題:哪些機台用 6205 軸承?
相似度排名:
  0.585  車床 B12
  0.523  銑床 B07
  0.314  熱處理爐 C03   ← 沒用 6205,明顯低
拉出前 2 段 → 車床 B12, 銑床 B07
答案:6205 軸承用於 車床 B12、銑床 B07。