使用 BigQuery GraphRAG 防範洗錢和詐欺行為

1. 簡介

在本程式碼研究室中，您將建構圖形檢索增強生成 (GraphRAG) 解決方案，偵測洗錢防制 (AML) 和金融詐欺。您將使用 Vertex AI、向量搜尋和 BigQuery 的原生圖形功能，並透過 LangChain 協調作業。在本實驗室結束時，您會瞭解大型語言模型 (LLM) 如何綜合語意稽核記錄和複雜的交易網路，找出非法資金路徑。

GraphRAG 架構流程

+------------------+     1. Vector Search      +---------------------+
| User Prompt /    | ------------------------> | BigQuery ML         |
| Investigation    |                           | (AccountAudits)     |
+------------------+                           +---------------------+
         |                                                |
         |                                                | 2. Seed Entity ID
         v                                                v
+------------------+     3. GQL Traversal      +---------------------+
| LangChain        | <------------------------ | BigQuery Property   |
| Graph Retriever  |                           | Graph (FinGraph)    |
+------------------+                           +---------------------+
         |
         | 4. Synthesized Context
         v
+------------------+
| Gemini 2.5 Flash | ---> Detailed Fraud Report
+------------------+

為什麼要使用 GraphRAG？標準 RAG 會使用語意相似度擷取獨立的文字分塊，但會遺漏多步驟關係。GraphRAG 會使用向量搜尋尋找種子實體，然後使用 BigQuery GQL 遍歷多跳貨幣路徑 (例如「擁有者 → 帳戶 → 轉移 → 帳戶 → 貸款」)。

學習內容

• 第 1 階段：資料集和屬性圖設定：建立關聯式財務資料表，並建構原生 BigQuery PROPERTY GRAPH。
• 階段 2：生成語意向量嵌入：使用 AI.GENERATE_EMBEDDING (text-embedding-005) 直接在 SQL 中生成稽核記錄的文字嵌入。
• 第 3 階段：自訂 LangChain GraphRAG 檢索器：建構自訂 Python 檢索器，結合向量相似度 (COSINE_DISTANCE) 和 ISO GQL 路徑遍歷。
• 第 4 階段：LLM 詐欺推理和追蹤路徑視覺化：執行 Gemini 推理鏈，揭露非法洗錢迴圈，並在 BigQuery Studio 中將路徑追蹤視覺化。

軟硬體需求

• 網路瀏覽器，例如 Chrome。
• 已啟用計費功能的 Google Cloud 專案。

本程式碼研究室適合各種程度的開發人員、資料工程師和 AI 從業人員 (包括初學者) 參加。

預估時間：35 分鐘
預估費用：不到 $2.00 美元 (使用即付即用的 Vertex AI 和 BigQuery 查詢處理)。

2. 事前準備

建立 Google Cloud 專案

1. 在 Google Cloud 控制台中，選取或建立 Google Cloud 專案。
2. 確認 Cloud 專案已啟用計費功能。

啟動 Cloud Shell

1. 按一下 Google Cloud 控制台頂端的「啟用 Cloud Shell」。
2. 驗證：

gcloud auth list

3. 在 Cloud Shell 中設定環境變數：

export GCP_PROJECT=$(gcloud config get-value project)
export REGION="us-central1"
export BQ_DATASET="fingraph_rag"
gcloud config set project $GCP_PROJECT

啟用 API

執行下列指令，啟用所有必要的 API：

gcloud services enable \
 bigquery.googleapis.com \
 aiplatform.googleapis.com

3. 設定和初始化

在這個步驟中，我們將設定 Python 環境、安裝必要程式庫，並初始化 BigQuery 和 Vertex AI 用戶端。您可以在 Cloud Shell 或 Jupyter 筆記本環境中執行這些指令。

1. 建立並啟用 Python 虛擬環境：

python3 -m venv venv
source venv/bin/activate

2. 安裝必要的 Python 套件：

pip install langchain-google-vertexai langchain-core google-cloud-bigquery vertexai

3. 建立 Python 檔案 graphrag_aml.py，並加入初始化程式碼。將 替換為您的 Google Cloud 專案 ID。

import vertexai
from google.cloud import bigquery

# Configuration
GCP_PROJECT_ID = "<YOUR_PROJECT_ID>"
REGION = "us-central1"
BQ_DATASET_ID = "fingraph_rag"
MODEL_NAME = "gemini-2.5-flash"

# Initialize clients
bq_client = bigquery.Client(project=GCP_PROJECT_ID)
vertexai.init(project=GCP_PROJECT_ID, location=REGION)

4. 建立資料表和結構定義

接著，我們要建立 BigQuery 資料集和標準資料表，定義財務圖表的結構定義。

1. 建立 BigQuery 資料集：

bq mk --location=US --dataset fingraph_rag

2. 建立資料表。您可以在 BigQuery Studio UI 或透過 Cloud Shell 執行這項作業。SQL 查詢如下：

CREATE TABLE IF NOT EXISTS `fingraph_rag.Account` (id INT64, create_time TIMESTAMP, is_blocked BOOL, type STRING);
CREATE TABLE IF NOT EXISTS `fingraph_rag.Loan` (id INT64, loan_amount FLOAT64, balance FLOAT64, create_time TIMESTAMP, interest_rate FLOAT64);
CREATE TABLE IF NOT EXISTS `fingraph_rag.Person` (id INT64, name STRING);
CREATE TABLE IF NOT EXISTS `fingraph_rag.AccountRepayLoan` (id INT64, loan_id INT64, amount FLOAT64, create_time TIMESTAMP);
CREATE TABLE IF NOT EXISTS `fingraph_rag.AccountTransferAccount` (id INT64, to_id INT64, amount FLOAT64, create_time TIMESTAMP);
CREATE TABLE IF NOT EXISTS `fingraph_rag.PersonOwnAccount` (id INT64, account_id INT64, create_time TIMESTAMP);
CREATE TABLE IF NOT EXISTS `fingraph_rag.AccountAudits` (id INT64, audit_timestamp TIMESTAMP, audit_details STRING, embedding ARRAY<FLOAT64>);

5. 插入資料集

現在我們要插入實體及其關係，形成資金流向。這個資料集代表 Doe (疑似空殼公司擁有者)、Jacoby (中介機構)、Menville (未通過 KYC 的目標) 和 Smith (無辜的旁觀者) 之間的疑似可疑活動。

執行下列 SQL 陳述式，填入資料表：

INSERT INTO `fingraph_rag.Account` VALUES 
  (10,'2020-01-10 06:22:20.222',false,'brokerage account'), 
  (20,'2020-01-27 17:55:09.206',false,'checking account'), 
  (30,'2020-02-15 09:12:33.111',false,'savings account'), 
  (40,'2019-11-05 14:33:10.000',false,'business account');

INSERT INTO `fingraph_rag.Loan` VALUES 
  (100,2022278.5,123359.0,'2020-03-18 16:42:57.719',0.064), 
  (200,50000.0,45000.0,'2020-03-23 19:03:05.567',0.097), 
  (300, 15000.0, 10000.0, '2020-05-10 10:00:00.000', 0.05);

INSERT INTO `fingraph_rag.Person` VALUES 
  (1,'Jacoby'), (2,'Menville'), (3,'Smith'), (4,'Doe');

INSERT INTO `fingraph_rag.AccountTransferAccount` VALUES 
  (40,10,25000.0,'2020-08-01 10:00:00.000'), 
  (10,20,24000.0,'2020-08-29 15:28:58.647'), 
  (30,20,150.0,'2020-09-01 12:00:00.000');

INSERT INTO `fingraph_rag.AccountRepayLoan` VALUES 
  (10,100,56809.8,'2020-12-12 07:25:02.597'), 
  (20,200,20000.0,'2021-01-18 01:40:25.317');

INSERT INTO `fingraph_rag.PersonOwnAccount` VALUES 
  (1,10,'2020-01-10 06:22:20.222'), (2,20,'2020-01-27 17:55:09.206'), 
  (3,30,'2020-02-15 09:12:33.111'), (4,40,'2019-11-05 14:33:10.000');

INSERT INTO `fingraph_rag.AccountAudits` (id, audit_timestamp, audit_details) VALUES 
  (10, '2020-05-14 06:57:02', 'Account 10 (Jacoby) flagged by AML system for suspicious high-volume transfers from offshore business accounts.'), 
  (20, '2021-03-09 02:51:45', 'Account 20 (Menville) failed KYC verification. Linked source of funds is unverified and customer is unresponsive.'), 
  (40, '2020-07-20 09:00:00', 'Account 40 (Doe) under investigation as a suspected shell company involved in illicit activities.');

驗證擷取的記錄

執行這項查詢，驗證財務資料表中的記錄數量：

SELECT 'Account' AS entity_table, COUNT(*) AS row_count FROM `fingraph_rag.Account`
UNION ALL SELECT 'Loan', COUNT(*) FROM `fingraph_rag.Loan`
UNION ALL SELECT 'Person', COUNT(*) FROM `fingraph_rag.Person`
UNION ALL SELECT 'AccountAudits', COUNT(*) FROM `fingraph_rag.AccountAudits`;

您應該會看到類似下方的查詢輸出內容，確認資料列已插入：

6. 建立 BigQuery 屬性圖形

有了關聯式資料後，我們使用 BigQuery 的原生圖形 DDL 定義 FinGraph。這樣一來，您就能在現有的關聯式資料表上建立語意層，不必複製或重複資料。

ISO GQL 語法入門

BigQuery 屬性圖形使用標準 ISO Graph Query Language (GQL) 模式：

• (node:Label) 定義實體節點 (例如 Account、Person、Loan)。
• -[edge:LABEL]-> 定義有向關係 (例如 Transfers、Repays、Owns)。

執行下列 SQL 陳述式來建立屬性圖：

CREATE OR REPLACE PROPERTY GRAPH `fingraph_rag.FinGraph`
 NODE TABLES (
   `fingraph_rag.Account` KEY (id) LABEL Account PROPERTIES (id, type, is_blocked),
   `fingraph_rag.Loan` KEY (id) LABEL Loan PROPERTIES (id, loan_amount, balance),
   `fingraph_rag.Person` KEY (id) LABEL Person PROPERTIES (id, name)
 )
 EDGE TABLES(
   `fingraph_rag.AccountRepayLoan`
     KEY (id, loan_id, create_time)
     SOURCE KEY (id) REFERENCES `fingraph_rag.Account` (id)
     DESTINATION KEY (loan_id) REFERENCES `fingraph_rag.Loan` (id)
     LABEL Repays PROPERTIES (amount, create_time),
   `fingraph_rag.AccountTransferAccount`
     KEY (id, to_id, create_time)
     SOURCE KEY (id) REFERENCES `fingraph_rag.Account` (id)
     DESTINATION KEY (to_id) REFERENCES `fingraph_rag.Account` (id)
     LABEL Transfers PROPERTIES (amount, create_time),
   `fingraph_rag.PersonOwnAccount`
     KEY (id, account_id)
     SOURCE KEY (id) REFERENCES `fingraph_rag.Person` (id)
     DESTINATION KEY (account_id) REFERENCES `fingraph_rag.Account` (id)
     LABEL Owns PROPERTIES (create_time)
 );

零 ETL 範例：BigQuery 的 CREATE PROPERTY GRAPH 直接在現有的關聯資料表上定義圖表結構，免除脆弱的 ETL 管道和資料重複問題。

如要以圖表呈現帳戶、人員和貸款的完整關係，請在 BigQuery Studio 中執行下列 SQL 查詢：

GRAPH `fingraph_rag.FinGraph`
MATCH (src)-[e]->(dst)
RETURN TO_JSON([
  TO_JSON(src),
  TO_JSON(e),
  TO_JSON(dst)
  ]) AS result;

您應該會看到類似下方的圖形視覺化結果：

7. 生成稽核記錄的嵌入內容

如要啟用 RAG 管道的向量搜尋部分，我們直接在 BigQuery 中使用 AI.GENERATE_EMBEDDING 表格值函式 (TVF)，為非結構化稽核記錄產生文字嵌入。

建立 BigQuery 遠端連線並授予 IAM 權限

BigQuery ML 必須CLOUD_RESOURCE連線，才能與 Vertex AI 嵌入端點安全地通訊。在 Cloud Shell 中執行下列 Bash 指令，建立連線、探索自動產生的服務帳戶，並授予 Vertex AI 使用者 (roles/aiplatform.user) 角色：

# 1. Set environment variables
export PROJECT_ID=$(gcloud config get-value project)
export LOCATION="us"
export CONNECTION_ID="vertex_ai_conn"

# 2. Create the BigQuery Cloud Resource Connection
bq mk --connection \
    --location=${LOCATION} \
    --project_id=${PROJECT_ID} \
    --connection_type=CLOUD_RESOURCE \
    ${CONNECTION_ID}

# 3. Retrieve the auto-generated Service Account ID associated with the connection
SA_ID=$(bq show --format=json --location=${LOCATION} --connection ${CONNECTION_ID} | jq -r '.cloudResource.serviceAccountId')
echo "Connection Service Account: ${SA_ID}"

# 4. Grant Vertex AI User (roles/aiplatform.user) permission to the Service Account
gcloud projects add-iam-policy-binding ${PROJECT_ID} \
    --member="serviceAccount:${SA_ID}" \
    --role="roles/aiplatform.user" \
    --condition=None

IAM 條件提示注意事項：
如果 Google Cloud 專案已包含條件式 IAM 繫結，gcloud 可能會以互動方式提示您選取條件選項 (例如 [1] ...、[2] ...、[3] None)。請選取 None (選項 3)，無條件授予本實驗室的 roles/aiplatform.user 角色。在上述指令中加入 --condition=None，即可自動略過這個提示。

建立遠端嵌入模型

接著，定義連結至 Vertex AI text-embedding-005 模型的 BigQuery ML 遠端模型，方法是透過您新授權的連線：

CREATE OR REPLACE MODEL `fingraph_rag.embedding_model`
  REMOTE WITH CONNECTION `us.vertex_ai_conn`
  OPTIONS(ENDPOINT = 'text-embedding-005');

IAM 權限傳播延遲：
新授予的 IAM 角色可能需要 1 到 3 分鐘，才能在 Google Cloud 服務中傳播。如果模型建立或嵌入生成作業一開始因權限錯誤 (例如 The [e-mail] does not have the permission to access or use the endpoint...) 而失敗，請稍候幾分鐘，等待 IAM 繫結同步處理完成，然後重試查詢。

生成嵌入

現在，請在 UPDATE 陳述式的 FROM 子句中呼叫 AI.GENERATE_EMBEDDING，為 AccountAudits 資料表生成嵌入：

UPDATE `fingraph_rag.AccountAudits` target
SET embedding = source.embedding
FROM AI.GENERATE_EMBEDDING(
  MODEL `fingraph_rag.embedding_model`,
  (SELECT id, audit_details AS content FROM `fingraph_rag.AccountAudits` WHERE ARRAY_LENGTH(embedding) = 0)
) source
WHERE target.id = source.id;

驗證產生的向量維度

執行下列查詢，確認向量嵌入已填入：

SELECT id, audit_details, ARRAY_LENGTH(embedding) AS embedding_dim 
FROM `fingraph_rag.AccountAudits`;

您應該會看到查詢輸出內容，顯示類似以下的 768 維向量嵌入：

什麼是 AI.GENERATE_EMBEDDING 和向量搜尋？
AI.GENERATE_EMBEDDING 是資料表值函式 (TVF)，必須在 FROM 子句中呼叫。這項函式會接受 BigQuery ML 遠端 MODEL (透過 CLOUD_RESOURCE 連線向 Vertex AI 驗證) 和包含 content 資料欄的輸入查詢。嵌入運算作業會直接卸載至 Vertex AI，資料不會離開 BigQuery。COSINE_DISTANCE 會計算目標嵌入與查詢向量之間的向量相似度，以找出規則式檢查遺漏的語意相符項目。

8. 定義 GraphRAG 檢索器

我們現在要在 Python 環境中建立自訂 LangChain 檢索器。這個檢索器結合了語意向量搜尋 (尋找相關起點) 和原生圖表 MATCH 查詢 (遍歷關係)。

在 Python 指令碼 graphrag_aml.py 中加入下列程式碼：

from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from typing import List

class FinGraphRetriever(BaseRetriever):
    project: str
    dataset: str

    def _get_relevant_documents(self, query: str) -> List[Document]:
        # 1. Vector Search
        vector_query = f"""
            SELECT id, audit_details
            FROM `{self.dataset}.AccountAudits`
            ORDER BY COSINE_DISTANCE(
                embedding,
                (
                    SELECT embedding
                    FROM AI.GENERATE_EMBEDDING(
                        MODEL `{self.dataset}.embedding_model`,
                        (SELECT @query AS content)
                    )
                )
            )
            LIMIT 1
        """
        res = bq_client.query(vector_query, job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ScalarQueryParameter("query", "STRING", query)]
        )).result()

        start_id = None
        audit_text = ""
        for row in res:
            start_id = row.id
            audit_text = row.audit_details

        if not start_id: return []

        # 2. Native Graph Traversal
        graph_query = f"""
            GRAPH `{self.dataset}.FinGraph`
            MATCH
              (sender_person:Person)-[:Owns]->(sender_acc:Account)
              -[tx:Transfers]->
              (a:Account)
              -[repays:Repays]->(l:Loan),
              (owner:Person)-[:Owns]->(a)
            WHERE a.id = @id
            RETURN
              owner.name as owner_name,
              a.type as account_type,
              sender_person.name as sender_name,
              tx.amount as transfer_amount,
              repays.amount as repayment_amount,
              l.id as loan_id
        """
        graph_res = bq_client.query(graph_query, job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ScalarQueryParameter("id", "INT64", start_id)]
        )).result()

        context_docs = [Document(page_content=f"Primary Audit Log (Target Account): {audit_text}")]
        sender_names = []
        for row in graph_res:
            sender_names.append(row['sender_name'])
            doc_str = (f"Account Owner: {row['owner_name']} (Account Type: {row['account_type']}). "
                       f"Received transfer of ${row['transfer_amount']} from {row['sender_name']}. "
                       f"Made loan repayment of ${row['repayment_amount']} to Loan {row['loan_id']}.")
            context_docs.append(Document(page_content=doc_str))

        if sender_names:
            names_list = "','".join(sender_names)
            sender_audit_query = f"""
                SELECT p.name, au.audit_details
                FROM `{self.dataset}.AccountAudits` au
                JOIN `{self.dataset}.Account` a ON au.id = a.id
                JOIN `{self.dataset}.PersonOwnAccount` poa ON a.id = poa.account_id
                JOIN `{self.dataset}.Person` p ON poa.id = p.id
                WHERE p.name IN ('{names_list}')
            """
            sender_audits = bq_client.query(sender_audit_query).result()
            for row in sender_audits:
                context_docs.append(Document(page_content=f"Audit Log for Sender {row['name']}: {row['audit_details']}"))

        return context_docs

請注意，圖形遍歷查詢如何透過 MATCH (sender_person:Person)-[:Owns]->... 輕鬆整合帳戶擁有者、寄件者、轉移和貸款，完全不需要複雜的 SQL JOIN！

動態多代理程式 NL2SQL/GQL 架構：
在進階的正式版多代理程式系統中，您可以部署專屬的 NL2SQL/NL2GQL 代理程式，而非在擷取器中使用預先定義的 SQL / GQL 查詢範本。這個專用代理程式會載入圖表結構定義和查詢範例做為技能，根據主要代理程式的主要調查目標動態產生量身打造的 GQL 查詢，針對 BigQuery 執行 SQL 查詢並驗證，然後傳回目標圖表結果。動態代理 SQL 生成功能充分展現多代理工作流程的彈性，但為了簡潔明瞭，本實驗室使用結構化查詢。

9. 執行詐欺調查

最後，我們執行 GraphRAG 管道，產生詳細的詐欺報告。LLM 會使用自訂圖表檢索器檢索到的脈絡資訊，回答提示。

網域價值：解決詐欺作業中的偽陽性問題
傳統的 AML 監控以規則為準，每年會標記數百萬個偽陽性結果，因為規則會單獨評估交易 (例如標記 Smith 轉給 Menville 的 $150 美元)。GraphRAG 會結合語意稽核記錄內容 (Menville 未通過 KYC) 和多跳圖形拓撲 (Doe → Jacoby → Menville)，讓 LLM 篩除良性雜訊，同時找出高風險的洗錢迴圈。

將下列程式碼新增至指令碼 graphrag_aml.py，然後使用 python graphrag_aml.py 執行：

from langchain_google_vertexai import ChatVertexAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Initialize the LLM and the Retriever
llm = ChatVertexAI(model_name=MODEL_NAME)
retriever = FinGraphRetriever(project=GCP_PROJECT_ID, dataset=BQ_DATASET_ID)

# Define the Prompt
prompt = ChatPromptTemplate.from_template("""
You are a Lead Fraud Analyst. Use the following audit logs and graph transaction history to answer the question.
Your goal is to connect the dots between the entities and explain the flow of funds.
If you see transfers from flagged users or shell companies, highlight the money laundering risk.

Context: {context}

Question: {question}

Detailed Fraud Report:
""")

# Create the LangChain
chain = (
    {"context": retriever , "question": lambda x: x}
    | prompt
    | llm
    | StrOutputParser()
)

# Execute the chain
question = "Why is Menville's loan repayment at risk? Flag any suspicious activity if you notice."
print(chain.invoke(question))

完成 graphrag_aml.py 指令碼

如需參考，完整的 graphrag_aml.py 指令碼應如下所示：

import vertexai
from google.cloud import bigquery
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from typing import List
from langchain_google_vertexai import ChatVertexAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Configuration
GCP_PROJECT_ID = "<YOUR_PROJECT_ID>"
REGION = "us-central1"
BQ_DATASET_ID = "fingraph_rag"
MODEL_NAME = "gemini-2.5-flash"

# Initialize clients
bq_client = bigquery.Client(project=GCP_PROJECT_ID)
vertexai.init(project=GCP_PROJECT_ID, location=REGION)

class FinGraphRetriever(BaseRetriever):
    project: str
    dataset: str

    def _get_relevant_documents(self, query: str) -> List[Document]:
        # 1. Vector Search using Cosine Distance
        vector_query = f"""
            SELECT id, audit_details
            FROM `{self.dataset}.AccountAudits`
            ORDER BY COSINE_DISTANCE(
                embedding,
                (
                    SELECT embedding
                    FROM AI.GENERATE_EMBEDDING(
                        MODEL `{self.dataset}.embedding_model`,
                        (SELECT @query AS content)
                    )
                )
            )
            LIMIT 1
        """
        res = bq_client.query(vector_query, job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ScalarQueryParameter("query", "STRING", query)]
        )).result()

        start_id = None
        audit_text = ""
        for row in res:
            start_id = row.id
            audit_text = row.audit_details

        if not start_id: return []

        # 2. Native Graph Traversal (GQL MATCH)
        graph_query = f"""
            GRAPH `{self.dataset}.FinGraph`
            MATCH
              (sender_person:Person)-[:Owns]->(sender_acc:Account)
              -[tx:Transfers]->
              (a:Account)
              -[repays:Repays]->(l:Loan),
              (owner:Person)-[:Owns]->(a)
            WHERE a.id = @id
            RETURN
              owner.name as owner_name,
              a.type as account_type,
              sender_person.name as sender_name,
              tx.amount as transfer_amount,
              repays.amount as repayment_amount,
              l.id as loan_id
        """
        graph_res = bq_client.query(graph_query, job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ScalarQueryParameter("id", "INT64", start_id)]
        )).result()

        context_docs = [Document(page_content=f"Primary Audit Log (Target Account): {audit_text}")]
        sender_names = []
        for row in graph_res:
            sender_names.append(row['sender_name'])
            doc_str = (f"Account Owner: {row['owner_name']} (Account Type: {row['account_type']}). "
                       f"Received transfer of ${row['transfer_amount']} from {row['sender_name']}. "
                       f"Made loan repayment of ${row['repayment_amount']} to Loan {row['loan_id']}.")
            context_docs.append(Document(page_content=doc_str))

        if sender_names:
            names_list = "','".join(sender_names)
            sender_audit_query = f"""
                SELECT p.name, au.audit_details
                FROM `{self.dataset}.AccountAudits` au
                JOIN `{self.dataset}.Account` a ON au.id = a.id
                JOIN `{self.dataset}.PersonOwnAccount` poa ON a.id = poa.account_id
                JOIN `{self.dataset}.Person` p ON poa.id = p.id
                WHERE p.name IN ('{names_list}')
            """
            sender_audits = bq_client.query(sender_audit_query).result()
            for row in sender_audits:
                context_docs.append(Document(page_content=f"Audit Log for Sender {row['name']}: {row['audit_details']}"))

        return context_docs

# Initialize LLM & Retriever
llm = ChatVertexAI(model_name=MODEL_NAME)
retriever = FinGraphRetriever(project=GCP_PROJECT_ID, dataset=BQ_DATASET_ID)

prompt = ChatPromptTemplate.from_template("""
You are a Lead Fraud Analyst. Use the following audit logs and graph transaction history to answer the question.
Your goal is to connect the dots between the entities and explain the flow of funds.
If you see transfers from flagged users or shell companies, highlight the money laundering risk.

Context: {context}

Question: {question}

Detailed Fraud Report:
""")

chain = (
    {"context": retriever, "question": lambda x: x}
    | prompt
    | llm
    | StrOutputParser()
)

question = "Why is Menville's loan repayment at risk? Flag any suspicious activity if you notice."
print(chain.invoke(question))

您應該會看到類似以下 LLM 分析報告範例的輸出內容：

10. 以視覺化方式呈現洗錢路徑

如要以視覺化方式瞭解我們剛透過程式發現的洗錢路徑，可以在 BigQuery Studio 控制台中執行圖表視覺化查詢。

在 BigQuery Studio 中執行這項查詢。(請務必啟用「圖表」視覺化功能，或點選「圖表」分頁標籤 (如有)。)

GRAPH `fingraph_rag.FinGraph`
 MATCH
   (p_shell:Person)-[o1:Owns]->(acc_shell:Account)-[t1:Transfers]->(acc_fraud:Account)-[t2:Transfers]->(acc_target:Account)-[r:Repays]->(l:Loan),
   (p_fraud:Person)-[o2:Owns]->(acc_fraud),
   (p_target:Person)-[o3:Owns]->(acc_target)
 WHERE p_target.name = 'Menville' AND p_fraud.name = 'Jacoby' AND p_shell.name = 'Doe'
 RETURN TO_JSON([
  TO_JSON(p_shell), TO_JSON(o1), TO_JSON(acc_shell),
  TO_JSON(t1), TO_JSON(acc_fraud), TO_JSON(p_fraud), TO_JSON(o2),
  TO_JSON(t2), TO_JSON(acc_target), TO_JSON(p_target), TO_JSON(o3),
  TO_JSON(r), TO_JSON(l)
]) AS result;

這項 GQL 查詢會追蹤從可疑空殼公司業主 (Doe) 到中介機構 (Jacoby) 的整個路徑，最後抵達最終目標 (Menville) 和貸款還款。

您應該會看到類似下方的圖形視覺化結果：

11. 清理

如要避免系統持續向您的 Google Cloud 帳戶收取費用，請刪除本程式碼研究室建立的資源。

刪除 BigQuery 資料集和 Cloud 資源連結：

# Delete the BigQuery dataset
bq rm -r -f $PROJECT_ID:fingraph_rag

# Delete the BigQuery Cloud Resource Connection
bq rm --connection --location=us vertex_ai_conn

確認資源已刪除：

bq ls --project_id $PROJECT_ID
bq ls --connection --location=us

12. 恭喜

恭喜！您已成功建構 RAG 應用程式並分析其行為。您示範了如何使用 BigQuery 的原生圖表和向量搜尋功能執行 GraphRAG，並在零 ETL 的情況下偵測洗錢計畫。

目前所學內容

• 如何在標準資料表上方的 BigQuery 中建構屬性圖
• 如何使用 BigQuery ML 生成及儲存向量嵌入
• 如何將 BigQuery 圖形遍歷和向量搜尋合併至 LangChain Retriever
• 大型語言模型如何透過圖形拓撲合成語意稽核記錄，減少偽陽性情形

後續步驟

• 探索 BigQuery 屬性圖總覽
• 試試更多 BigQuery Graph Query Language (GQL) 模式。

參考文件

• BigQuery ML AI.GENERATE_EMBEDDING 函式
• BigQuery 向量搜尋和距離函式
• LangChain Google 整合