iT邦幫忙

2026 iThome 鐵人賽

DAY 19
0
AI Engineering

AI Agent 系統開發 30 天系列 第 19 篇

用 GraphRAG 串接跨文件關聯

  • 分享至 

  • xImage
  •  

向量 RAG 依賴問題與文字片段之間的語意相似度。當推導答案需要跨越多份文件,且中繼段落與問題用詞沒有交集時,向量檢索往往會遺漏關鍵事實,導致模型無法得出結論。

例如在電商商城客服中,顧客詢問:

「音波極淨電動牙刷-Pro8 拆封後可以退貨嗎?」

答案需要拼起兩個事實:「這款牙刷屬於衛生用品」與「衛生用品拆封不能退」。但這兩個事實分別記錄在不同文件、用詞完全不重疊的兩個 Chunk 裡:

  • Chunk A(商品規格目錄):講震動頻率、防水等級、刷頭配件,說明該型號屬於「個人衛生與貼身用品」,整篇沒出現「退貨」或「拆封」。
  • Chunk B(退貨政策):講「拆封」「退貨」「鑑賞期」的規則本身,規定個人衛生用品拆封後排除鑑賞期,完全沒有提到特定商品型號。

範例使用三份 Markdown 文件,完整程式放在 graphrag-return-policy

  1. 01-商城退貨與售後服務政策.md:鑑賞期原則、特殊商品例外(第 2.1 節個人衛生用品拆封不退、第 2.2 節生鮮短效期食品排除退貨、第 2.3 節客製化給付排除退貨)與瑕疵換貨規範。
  2. 02-商品規格與品類目錄.md:各型號商品介紹與所屬品類(如「音波極淨電動牙刷-Pro8」屬於「個人衛生與貼身用品」)。
  3. 03-一般退貨常見問題.md:退貨運費與退款天數等語意相近的干擾文件。
data/docs/
├── 01-商城退貨與售後服務政策.md
├── 02-商品規格與品類目錄.md
└── 03-一般退貨常見問題.md

先用一般的向量 RAG 檢索這三份文件:

$ uv run gr-baseline "音波極淨電動牙刷-Pro8 拆封後可以退貨嗎?"
檢索結果(top_k=2):
  [1] 距離 0.281 | 01-商城退貨政策 § 2.1 個人衛生與貼身用品 (✅ 找到 Chunk B)
  [2] 距離 0.435 | 01-商城退貨政策 § 1.1 七日鑑賞期規範   (❌ 無關的一般條款)
  (⚠️ Chunk A 商品規格目錄相似度過低,完全沒被撈到!)

問題和 Chunk B 用詞高度重疊,相似度自然高;Chunk A 記錄的是硬體規格,與問題中的「退貨」沒有用詞交集,因相似度過低而被過濾掉。

最後送進模型的只有 Chunk B。模型知道「衛生用品不能退」,但手上完全沒有「這款牙刷到底算不算衛生用品」的事實依據,只能含糊回答或憑空猜測。

這反映了向量檢索的局限:它只能拿使用者的問題,個別比對單一片段的文字相似度。但真實問題往往需要串接多個事實——先從「商品」對應到「品類」,再從「品類」對應到「規定」。只要中間銜接的事實(Chunk A:牙刷屬於個人衛生用品)沒有包含問題裡的關鍵字(退貨、拆封),向量檢索就會漏掉它,導致推導過程缺少關鍵的中繼證據。

GraphRAG 怎麼把分散的事實串起來

核心想法:把「商品種類」這種原本只是商品身上一個屬性欄位的東西,升格成一個獨立節點,讓不同來源的文件可以透過同一個節點接軌。

如果品類只是商品 JSON 裡的一個字串屬性:

{ "商品名稱": "音波極淨電動牙刷-Pro8", "所屬品類": "個人衛生與貼身用品" }

那「商品」和「退貨政策」在圖上依然是兩座孤島,什麼演算法都跨不過去。所以改成:

  • 節點:個人衛生與貼身用品(獨立於商品之外的實體)
  • 關係:(音波極淨電動牙刷-Pro8) ──[BELONGS_TO_CATEGORY]──▶ (個人衛生與貼身用品)

這樣一來,未來不管加幾款商品(牙刷、刮鬍刀、貼身衣物),都只要新增一條指向這個節點的 BELONGS_TO_CATEGORY 關係,就能透過它連到退貨政策。

逐篇抽取實體與關係,儲存為中繼檔(gr-extract)

圖資料庫無法直接解析長篇自然語言文章。要建立圖譜,必須先讓模型將每篇 Markdown 文章拆解為結構化的「實體(節點)」與「關係(連線)」。

為了避免模型自行發明隨意名稱(例如把分類關係寫成 IS_A 或 KIND_OF),程式在 schema.py 中預先規範允許的實體與關係型別:

  • 實體型別(Entity Types):Product(商品)、Category(品類)、Policy(退換貨規定)。
  • 關係型別(Relation Types):BELONGS_TO_CATEGORY(商品屬於品類)、SUBJECT_TO_POLICY(品類適用規定)。

執行抽取指令:

$ uv run gr-extract
01-商城退貨與售後服務政策:實體 2 個、關係 1 條
    個人衛生與貼身用品 -[SUBJECT_TO_POLICY]-> 排除七日鑑賞期
02-商品規格與品類目錄:實體 3 個、關係 2 條
    音波極淨電動牙刷-Pro8 -[BELONGS_TO_CATEGORY]-> 個人衛生與貼身用品
    雙刀頭水洗刮鬍刀-S3 -[BELONGS_TO_CATEGORY]-> 個人衛生與貼身用品
...
合計:實體 5 個、關係 3 條 → data/extracted.json

抽取程式(extract.py)會逐一讀取各篇文件,每次獨立呼叫模型抽取該篇的事實,並將所有抽取結果儲存為中繼檔 data/extracted.json。

將抽取結果存為檔案有兩個目的:

  1. 避免重複消耗 Token 與等待時間:逐篇抽取需要呼叫模型。將抽取結果快取為本機 JSON 檔後,後續調整資料庫寫入邏輯或反覆測試查詢時,可以直接讀取該檔案,不必每次重新呼叫 API。
  2. 在寫入前檢查事實正確性:在匯入資料庫前可以直接檢視 JSON,確認模型抽出的主詞、述詞、受詞與原文佐證(evidence)是否完全忠於原文。

產出的 data/extracted.json 包含圖譜的兩大核心要素:節點清單(entities) 與 連線清單(relations):

  • entities(點):列出這張圖上有哪些名詞節點及其型別(商品 Product、品類 Category、規定 Policy)。
  • relations(線):描述節點之間的箭頭。每條關係的 subject(起點)與 object(終點),就是直接對應到 entities 裡的實體 name。
{
  "entities": [
    // 來自【01-商城退貨與售後服務政策.md】
    { "name": "個人衛生與貼身用品", "type": "Category", "doc_id": "01" },
    { "name": "排除七日鑑賞期", "type": "Policy", "doc_id": "01" },

    // 來自【02-商品規格與品類目錄.md】
    { "name": "音波極淨電動牙刷-Pro8", "type": "Product", "doc_id": "02" },
    { "name": "個人衛生與貼身用品", "type": "Category", "doc_id": "02" }
  ],
  "relations": [
    // 來自【01-商城退貨與售後服務政策.md】
    {
      "subject": "個人衛生與貼身用品",
      "predicate": "SUBJECT_TO_POLICY",
      "object": "排除七日鑑賞期",
      "evidence": "個人衛生與貼身用品一旦售出並拆封...排除七日鑑賞期",
      "doc_id": "01"
    },
    // 來自【02-商品規格與品類目錄.md】
    {
      "subject": "音波極淨電動牙刷-Pro8",
      "predicate": "BELONGS_TO_CATEGORY",
      "object": "個人衛生與貼身用品",
      "evidence": "所屬品類:個人衛生與貼身用品",
      "doc_id": "02"
    }
  ]
}

這兩組清單的對應關係如下:

  1. relations 裡的第二條紀錄,以 音波極淨電動牙刷-Pro8 為起點(subject)、個人衛生與貼身用品 為終點(object),在兩顆節點之間拉出 BELONGS_TO_CATEGORY 箭頭。
  2. 第一條紀錄則以 個人衛生與貼身用品 為起點、排除七日鑑賞期 為終點,拉出 SUBJECT_TO_POLICY 箭頭,並將來源條文存為 evidence。

注意此時的資料狀態:每份文件是各自獨立抽取的。在 data/extracted.json 裡,個人衛生與貼身用品 分別出現在兩份文件的抽取清單中,兩條關係也只是文字檔裡的兩筆記錄,彼此還沒有接軌,也還不是連通的圖。

用 Cypher MERGE 寫入 Neo4j 並接軌成圖

在上一步,我們已經把各篇文件中的實體與關係抽取出來,存成中繼檔 data/extracted.json。接下來要將這些零散的事實匯入圖資料庫,讓不同文件透過相同的節點接成一張連通的圖。

Neo4j 是目前最主流的圖資料庫(Graph Database) 之一。與傳統關聯式資料庫(以資料表、欄位與列儲存)不同,Neo4j 是直接以「節點(Node)」與「關係(Relationship)」來儲存與索引資料。

之所以需要圖資料庫,而不是直接在 JSON 檔裡搜尋或用傳統 SQL,主要有兩個原因:

  1. 原生支援多跳路徑遍歷:在關聯式資料庫中,要查詢「商品 ➔ 品類 ➔ 規定」這種跨層關係,每多跨一層就需要做一次成本高昂的 JOIN,而且難以處理不固定長度的路徑;Neo4j 在內部直接以指標儲存實體間的關係,系統可以毫秒級地順著箭頭一路走訪兩段、三段甚至更長的路徑。
  2. 同名實體自動接軌:當多份文件提到同一個品類(如「個人衛生與貼身用品」)時,圖資料庫能將它們合併為同一個實體節點,讓分散在不同文件的事實自然接通。

專案根目錄已加上 docker-compose.yml,使用 Docker 啟動本機的 Neo4j 容器:

docker compose up -d

容器啟動後,Neo4j 提供了內建的 Web 視覺化管理介面(Neo4j Browser):

  • 網址(URL):http://localhost:7474
  • 預設帳號:neo4j
  • 預設密碼:graphrag-demo

在瀏覽器打開這個 URL 並登入後,就能看見 Neo4j Browser。後續寫入資料後,可以在這個介面中輸入查詢語法,直接檢視圖譜上的圓圈節點與連線箭頭。

圖資料庫使用的操作語言稱為 Cypher(相當於圖資料庫的 SQL)。在 Cypher 中:

  • CREATE 代表「強制新建」:如果已經有同名節點,會重複建出兩顆獨立的節點。
  • MERGE 則代表**「存在就沿用,不存在才新建」**(類似關聯式資料庫的 UPSERT 去重寫入)。這正是跨文件資料能夠自動接軌的核心機制。

執行寫入指令:

uv run gr-load

終端機輸出:

$ uv run gr-load
抽取結果:實體 5 個、關係 3 條
寫入 Neo4j:4 個節點、3 條關係

再回到neo4j,可以看到這樣的結果。

https://ithelp.ithome.com.tw/upload/images/20260930/20111896GtqpDRQTH2.png

注意數字的變化:原本從兩份文件總共抽出了 5 個實體物件,但寫入 Neo4j 後只有 4 個節點。這是因為寫入程式(graph.py)分成了兩步執行:

1. 先寫入節點(Node MERGE)

程式逐一走訪 entities 清單,用 MERGE 寫入節點:

session.run(
    f"MERGE (n:{label} {{name: $name}}) SET n.aliases = $aliases",
    name=entity.name,
    aliases=entity.aliases,
)

當程式處理到兩份文件的節點時:

  1. 讀取《01-退貨政策》的實體 個人衛生與貼身用品 ➔ 資料庫裡尚未存在,建立新節點。
  2. 讀取《02-商品目錄》的實體 個人衛生與貼身用品 ➔ 資料庫裡已經有了同名節點,直接沿用既有節點,不重複建立。

兩份文件獨立抽出的同名實體,在資料庫中自動合而為一。

2. 再寫入連線(Edge MERGE)

節點建立完成後,程式走訪 relations 清單,找到兩端的節點拉出關係箭頭:

session.run(
    f"MATCH (s {{name: $subject}}), (o {{name: $object}}) "
    f"MERGE (s)-[r:{predicate} {{doc_id: $doc_id}}]->(o) "
    f"SET r.evidence = $evidence",
    subject=relation.subject,
    object=relation.object,
    doc_id=relation.doc_id,
    evidence=relation.evidence,
)
  1. 連上《02-商品目錄》的關係:
    (音波極淨電動牙刷-Pro8) ──[BELONGS_TO_CATEGORY]──▶ (個人衛生與貼身用品)
  2. 連上《01-退貨政策》的關係:
    (個人衛生與貼身用品) ──[SUBJECT_TO_POLICY]──▶ (排除七日鑑賞期)

3. 跨文件接軌完成

兩份原本毫無關聯的文件,在圖資料庫裡因為指到了同一顆共享節點(個人衛生與貼身用品),自動接通成一條跨文件的完整因果鏈:

https://ithelp.ithome.com.tw/upload/images/20260930/20111896wDmnmiLSkk.png

從起點實體沿關係展開因果鏈

有了圖,查詢時系統要做兩件事:先找起點,再沿線走。

https://ithelp.ithome.com.tw/upload/images/20260930/201118968wSAY5DfGf.png

尋找起點實體

要把「音波極淨電動牙刷-Pro8」對應到圖裡的同名節點,直覺的做法是把圖上所有節點名稱都抓到 Python 裡,逐一檢查有沒有整段或拆字後的片段出現在問題裡。這個做法在商品只有幾筆的示範資料裡看不出問題,但節點一多就會卡:Product、Category、Policy 各查一次資料庫、把該類型底下所有節點名稱整批搬回來,再用迴圈對問題文字逐一比對——資料庫查詢次數跟節點類型數成正比,比對次數則跟節點總數成正比,節點上萬筆時,每問一次都要重新掃過所有名稱。

改成交給 Neo4j 的全文索引(Full-text Index)處理。在 graph.py 的 reset() 裡,對 Product、Category、Policy 三種節點的 name 建一個全文索引,並指定 cjk analyzer:

labels = "|".join(ENTITY_TYPES)
session.run(
    f"CREATE FULLTEXT INDEX {FULLTEXT_INDEX_NAME} IF NOT EXISTS "
    f"FOR (n:{labels}) ON EACH [n.name] "
    "OPTIONS { indexConfig: { `fulltext.analyzer`: 'cjk' } }"
)

cjk analyzer 建索引時,會把每個中文名稱拆成重疊的兩字詞元(bigram),例如「音波極淨電動牙刷」會拆成「音波」「波極」「極淨」「淨電」……每個詞元都指向這個節點。查詢時只要問題裡出現任兩個連續字命中其中一個詞元,索引就能直接定位到節點,不必逐一比對每個節點的完整名稱:

def search_entities(driver: Driver, text: str, *, limit: int = 5) -> list[tuple[str, str]]:
    with driver.session() as session:
        records = session.run(
            f"CALL db.index.fulltext.queryNodes('{FULLTEXT_INDEX_NAME}', $text) "
            "YIELD node, score "
            "RETURN node.name AS name, labels(node)[0] AS label, score "
            "ORDER BY score DESC LIMIT $limit",
            text=_escape_lucene(text),
            limit=limit,
        ).data()
    return [(record["name"], record["label"]) for record in records]

answer.py 的 find_start_entities 改成呼叫這個查詢,取回的候選裡優先挑 Product 型別,並濾掉被其他候選整段包住的較短名稱:

def find_start_entities(driver: Driver, question: str, *, limit: int = 5) -> list[str]:
    hits = graph.search_entities(driver, question, limit=limit)
    priority_hits = [name for name, label in hits if label == "Product"]
    names = priority_hits or [name for name, _label in hits]
    names.sort(key=len, reverse=True)
    return [
        name
        for name in names
        if not any(other != name and name in other for other in names)
    ]

沿關係展開,串起中間的規定節點

從起點 音波極淨電動牙刷-Pro8 出發,先沿 BELONGS_TO_CATEGORY 走到 個人衛生與貼身用品,再沿 SUBJECT_TO_POLICY 走到 排除七日鑑賞期——這道問題的答案要靠這兩條關係接起來才完整。如果查詢只抓起點的直接鄰居,會停在 個人衛生與貼身用品 就停下來,抓不到最後的 排除七日鑑賞期;所以查詢要沿關係箭頭連續往外走兩段,才能把兩份文件裡的事實串在一起。

在 graph.py 中,使用 Cypher 查詢從起點實體往外展開,最多經過兩段關係:

TRAVERSE_QUERY = """
MATCH path = (start:Entity {name: $start_name})-[*1..2]-(connected)
UNWIND relationships(path) AS r
WITH DISTINCT r
RETURN startNode(r).name AS source, type(r) AS rel_type,
       r.description AS description, endNode(r).name AS target,
       r.evidence AS evidence, r.doc_id AS doc_id
"""

執行多跳檢索問答:

uv run gr-ask "音波極淨電動牙刷-Pro8 拆封後可以退貨嗎?"

終端機輸出:

$ uv run gr-ask "音波極淨電動牙刷-Pro8 拆封後可以退貨嗎?"
起點實體:音波極淨電動牙刷-Pro8
取回關聯路徑:2 條
[1] 音波極淨電動牙刷-Pro8 -[BELONGS_TO_CATEGORY]-> 個人衛生與貼身用品
[2] 個人衛生與貼身用品 -[SUBJECT_TO_POLICY]-> 排除七日鑑賞期

客服回覆:
依據商品規格,「音波極淨電動牙刷-Pro8」屬於「個人衛生與貼身用品」[1]。
根據退貨政策第 2.1 節,個人衛生與貼身用品一旦拆封,除原廠瑕疵外,
依法排除七日鑑賞期,恕無法受理退貨[2]。

系統把展開得到的關係鏈連同原文證據組進 Prompt,模型拿到的不再是零散關鍵字,而是一條結構完整的跨文件因果鏈。

依問題目標分流:不是每個問題都要走 GraphRAG

在前面的範例中,GraphRAG 成功補齊了跨文件的因果關係,解決了向量檢索遺漏中繼資訊的問題。但這並不代表所有查詢都該送進圖資料庫。

建立與維護圖譜需要抽取實體、維護結構約束,每次多跳走訪也有額外的延遲與查詢成本。如果問題本身很單純,或者需要精確的數值計算,硬走 GraphRAG 反而浪費資源。

因此在實際的 Agent 架構中,通常會由 Router 依據問題性質進行分流:

https://ithelp.ithome.com.tw/upload/images/20260930/201118962aIZ257xSg.png

分流時的核心原則是:用成本最低、且能精確命中事實的機制解決問題。常見的三種情境如下:

  • 單點條款查詢(走向量 RAG):當問題可以直接在單一章節找到答案(例如「標準鑑賞期有幾天?」),文字相似度檢索就能精準命中,不需要動用圖資料庫,延遲最低且成本極低。
  • 精確統計與數值計算(走結構化查詢):當問題涉及「上個月退款超過千元的訂單總額」等數值計算或欄位篩選時,交給語言模型自由推導極易產生幻覺,應改交由程式端組出的參數化查詢直接計算。這類查詢要如何限制模型只能在合法欄位與數值範圍內操作,下一篇會用 QuerySpec 架構說明。
  • 跨文件實體關聯推導(走 GraphRAG):當問題圍繞特定商品或實體(例如「這款電動牙刷拆封後能退嗎?」),且答案散落在不同文件、中間缺少文字交集時,才透過圖節點與箭頭展開多跳因果鏈。

上一篇
Agentic RAG:由 Agent 決定是否檢索、拆解問題與驗證證據
下一篇
用預先定義的欄位約束 Agent 的資料查詢
系列文
AI Agent 系統開發 30 天 共 21 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言