Chống rửa tiền và ngăn chặn gian lận bằng BigQuery GraphRAG

1. Giới thiệu

Trong lớp học lập trình này, bạn sẽ xây dựng một giải pháp Tạo sinh tăng cường truy xuất đồ thị (GraphRAG) để phát hiện hoạt động Chống rửa tiền (AML) và gian lận tài chính. Bạn sẽ sử dụng Vertex AI, Vector Search và các chức năng Graph gốc của BigQuery, được điều phối thông qua LangChain. Khi kết thúc bài thực hành này, bạn sẽ thấy cách Mô hình ngôn ngữ lớn (LLM) có thể xác định hoạt động chuyển tiền bất hợp pháp bằng cách tổng hợp nhật ký kiểm tra ngữ nghĩa và mạng lưới giao dịch phức tạp.

Luồng kiến trúc 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
+------------------+

Tại sao nên dùng GraphRAG? RAG tiêu chuẩn truy xuất các đoạn văn bản riêng lẻ bằng cách sử dụng sự tương đồng về ngữ nghĩa, thiếu các mối quan hệ nhiều bước. GraphRAG sử dụng Vector Search để tìm các thực thể ban đầu, sau đó sử dụng BigQuery GQL để duyệt qua các đường đi tiền nhiều bước (ví dụ: Chủ sở hữu → Tài khoản → Chuyển tiền → Tài khoản → Khoản vay).

Bạn sẽ thực hiện

• Giai đoạn 1: Thiết lập tập dữ liệu và biểu đồ thuộc tính: Tạo các bảng tài chính quan hệ và tạo PROPERTY GRAPH BigQuery gốc.
• Giai đoạn 2: Tạo vectơ nhúng ngữ nghĩa: Tạo vectơ nhúng văn bản trực tiếp trong SQL cho nhật ký kiểm tra bằng cách sử dụng AI.GENERATE_EMBEDDING (text-embedding-005).
• Giai đoạn 3: Trình truy xuất GraphRAG LangChain tuỳ chỉnh: Tạo một trình truy xuất Python tuỳ chỉnh kết hợp khả năng so sánh độ tương tự của vectơ (COSINE_DISTANCE) và các lượt truyền tải đường dẫn ISO GQL.
• Giai đoạn 4: LLM Fraud Reasoning & Trail Visualization (Lý luận về gian lận bằng LLM và trực quan hoá dấu vết): Thực thi chuỗi lý luận của Gemini để phát hiện các vòng rửa tiền bất hợp pháp và trực quan hoá dấu vết đường dẫn trong BigQuery Studio.

Bạn cần có

• Một trình duyệt web như Chrome.
• Một dự án trên Google Cloud đã bật tính năng thanh toán.

Lớp học lập trình này dành cho nhà phát triển, kỹ sư dữ liệu và chuyên gia AI ở mọi cấp độ, kể cả người mới bắt đầu.

Thời lượng ước tính: 35 phút
Chi phí ước tính: Dưới 2 USD (sử dụng Vertex AI và thao tác xử lý truy vấn BigQuery theo mô hình trả tiền khi sử dụng).

2. Trước khi bắt đầu

Tạo một dự án trên Google Cloud

1. Trong Google Cloud Console, hãy chọn hoặc tạo một dự án trên đám mây của Google Cloud.
2. Đảm bảo bạn đã bật tính năng thanh toán cho dự án trên Cloud.

Khởi động Cloud Shell

1. Nhấp vào Kích hoạt Cloud Shell ở đầu bảng điều khiển Google Cloud.
2. Xác minh quy trình xác thực:

gcloud auth list

3. Định cấu hình các biến môi trường trong 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

Bật API

Chạy lệnh này để bật tất cả các API bắt buộc:

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

3. Thiết lập và khởi động

Trong bước này, chúng ta sẽ thiết lập một môi trường Python, cài đặt các thư viện cần thiết và khởi chạy các ứng dụng BigQuery và Vertex AI. Bạn có thể chạy các lệnh này trong Cloud Shell hoặc môi trường sổ tay Jupyter.

1. Tạo và kích hoạt môi trường ảo Python:

python3 -m venv venv
source venv/bin/activate

2. Cài đặt các gói Python bắt buộc:

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

3. Tạo một tệp Python graphrag_aml.py rồi thêm mã khởi chạy. Thay thế bằng mã dự án trên Google Cloud.

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. Tạo bảng và giản đồ

Tiếp theo, chúng ta sẽ xác định giản đồ cho Biểu đồ tài chính bằng cách tạo một tập dữ liệu và các bảng chuẩn BigQuery.

1. Tạo tập dữ liệu BigQuery:

bq mk --location=US --dataset fingraph_rag

2. Tạo bảng. Bạn có thể chạy lệnh này trong giao diện người dùng BigQuery Studio hoặc thông qua Cloud Shell. Sau đây là 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. Chèn tập dữ liệu

Bây giờ, chúng ta sẽ chèn các thực thể và mối quan hệ của chúng để tạo thành chuỗi tiền. Tập dữ liệu này thể hiện các hoạt động đáng ngờ giữa Doe (người bị tình nghi là chủ sở hữu công ty ma), Jacoby (bên trung gian), Menville (mục tiêu không vượt qua quy trình KYC) và Smith (người qua đường vô tội).

Chạy SQL sau để điền dữ liệu vào các bảng:

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.');

Xác minh bản ghi đã nhập

Chạy truy vấn này để xác minh số lượng bản ghi trên các bảng tài chính của bạn:

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

Bạn sẽ thấy kết quả truy vấn xác nhận việc chèn hàng tương tự như sau:

6. Tạo biểu đồ thuộc tính BigQuery

Khi đã có dữ liệu quan hệ, chúng ta sẽ xác định FinGraph bằng Graph DDL gốc của BigQuery. Thao tác này sẽ tạo một lớp ngữ nghĩa trên các bảng quan hệ hiện có mà không cần sao chép hoặc sao chép dữ liệu.

Tài liệu giới thiệu về cú pháp GQL theo tiêu chuẩn ISO

Đồ thị thuộc tính BigQuery sử dụng các mẫu Ngôn ngữ truy vấn đồ thị (GQL) theo tiêu chuẩn ISO:

• (node:Label) xác định các nút thực thể (ví dụ: Account, Person, Loan).
• -[edge:LABEL]-> xác định các mối quan hệ có hướng (ví dụ: Transfers, Repays, Owns).

Chạy câu lệnh SQL sau để tạo biểu đồ thuộc tính:

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

Mô hình Zero-ETL: CREATE PROPERTY GRAPH của BigQuery xác định cấu trúc đồ thị ngay trên các bảng quan hệ hiện có, loại bỏ các quy trình ETL dễ bị lỗi và tình trạng trùng lặp dữ liệu.

Để trực quan hoá toàn bộ biểu đồ tài khoản, người và khoản vay, hãy chạy truy vấn SQL sau trong BigQuery Studio:

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

Bạn sẽ thấy kết quả trực quan hoá biểu đồ tương tự như sau:

7. Tạo các vectơ nhúng cho nhật ký kiểm tra

Để bật phần tìm kiếm vectơ của quy trình RAG, chúng tôi tạo các mục nhúng văn bản cho nhật ký kiểm tra không có cấu trúc ngay trong BigQuery bằng cách sử dụng Hàm có giá trị bảng (TVF) AI.GENERATE_EMBEDDING.

Tạo mối kết nối từ xa với BigQuery và cấp quyền IAM

BigQuery ML yêu cầu có một kết nối CLOUD_RESOURCE để giao tiếp an toàn với các điểm cuối nhúng Vertex AI. Chạy các lệnh bash sau trong Cloud Shell để tạo kết nối, khám phá tài khoản dịch vụ được tạo tự động và cấp vai trò Người dùng 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

Lưu ý về lời nhắc điều kiện IAM:
Nếu dự án Google Cloud của bạn đã chứa các liên kết IAM có điều kiện, gcloud có thể tương tác với bạn để nhắc bạn chọn một lựa chọn điều kiện (ví dụ: [1] ..., [2] ..., [3] None). Chọn None (lựa chọn 3) để cấp vai trò roles/aiplatform.user vô điều kiện cho bài thực hành này. Việc thêm --condition=None vào lệnh trên sẽ tự động bỏ qua lời nhắc này.

Tạo mô hình nhúng từ xa

Tiếp theo, hãy xác định một mô hình từ xa BigQuery ML liên kết với mô hình text-embedding-005 của Vertex AI thông qua mối kết nối mới được uỷ quyền:

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

Độ trễ khi truyền quyền IAM:
Các vai trò IAM mới được cấp có thể mất từ 1 đến 3 phút để truyền trên các dịch vụ của Google Cloud. Nếu quá trình tạo mô hình hoặc tạo mục nhúng ban đầu không thành công do lỗi về quyền (ví dụ: The [e-mail] does not have the permission to access or use the endpoint...), hãy đợi vài phút để các liên kết IAM đồng bộ hoá rồi thử lại truy vấn.

Tạo vectơ nhúng

Bây giờ, hãy tạo các mục nhúng cho bảng AccountAudits bằng cách gọi AI.GENERATE_EMBEDDING trong mệnh đề FROM của câu lệnh UPDATE:

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;

Xác minh các phương diện vectơ được tạo

Chạy truy vấn sau để xác minh rằng các mục nhúng vectơ đã được điền sẵn:

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

Bạn sẽ thấy kết quả truy vấn cho thấy các mục nhúng vectơ 768 chiều tương tự như sau:

AI.GENERATE_EMBEDDING và Tìm kiếm vectơ là gì?
AI.GENERATE_EMBEDDING là một Hàm trả về bảng (TVF) phải được gọi trong mệnh đề FROM. Hàm này chấp nhận một MODEL từ xa BigQuery ML (xác thực với Vertex AI thông qua một kết nối CLOUD_RESOURCE) và một truy vấn đầu vào chứa cột content. Việc nhúng dữ liệu tính toán được chuyển trực tiếp sang Vertex AI mà không cần dữ liệu rời khỏi BigQuery. COSINE_DISTANCE tính toán độ tương đồng vectơ giữa các mục nhúng mục tiêu và vectơ truy vấn để nắm bắt các kết quả khớp ngữ nghĩa mà các lượt kiểm tra dựa trên quy tắc bỏ lỡ.

8. Xác định GraphRAG Retriever

Giờ đây, chúng ta sẽ tạo một trình truy xuất LangChain tuỳ chỉnh trong môi trường Python. Trình truy xuất này kết hợp tính năng Tìm kiếm vectơ ngữ nghĩa (để tìm các điểm bắt đầu có liên quan) với các truy vấn MATCH đồ thị gốc (để duyệt qua các mối quan hệ).

Thêm mã sau vào tập lệnh 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

Lưu ý cách truy vấn duyệt qua biểu đồ dễ dàng kết hợp chủ sở hữu tài khoản, người gửi, giao dịch chuyển tiền và khoản vay thông qua MATCH (sender_person:Person)-[:Owns]->... mà không cần các lệnh JOIN phức tạp trong SQL!

Cấu trúc NL2SQL/GQL đa tác nhân linh hoạt:
Trong các hệ thống đa tác nhân sản xuất nâng cao, thay vì sử dụng các mẫu truy vấn SQL/GQL được xác định trước trong trình truy xuất, bạn có thể triển khai một tác nhân NL2SQL / NL2GQL chuyên dụng. Được trang bị các định nghĩa về giản đồ đồ thị và ví dụ về truy vấn dưới dạng kỹ năng, tác nhân chuyên biệt này sẽ tạo động các truy vấn GQL phù hợp dựa trên các mục tiêu điều tra chính của tác nhân chính, chạy thử và xác thực SQL dựa trên BigQuery, đồng thời trả về kết quả đồ thị nhắm đến mục tiêu. Mặc dù tính năng tạo SQL dựa trên tác nhân động làm nổi bật tính linh hoạt hoàn toàn của quy trình làm việc có nhiều tác nhân, nhưng chúng tôi sử dụng các truy vấn có cấu trúc trong phòng thí nghiệm này để đảm bảo tính ngắn gọn và rõ ràng.

9. Tiến hành điều tra hành vi gian lận

Cuối cùng, chúng tôi chạy quy trình GraphRAG để tạo một báo cáo chi tiết về hành vi gian lận. LLM sẽ sử dụng ngữ cảnh do trình truy xuất biểu đồ tuỳ chỉnh của chúng tôi truy xuất để trả lời câu lệnh.

Giá trị của miền: Giải quyết vấn đề về kết quả dương tính giả trong hoạt động chống gian lận
Hoạt động giám sát AML dựa trên quy tắc truyền thống đánh dấu hàng triệu kết quả dương tính giả mỗi năm vì các quy tắc đánh giá giao dịch riêng lẻ (ví dụ: đánh dấu giao dịch chuyển 150 USD của Smith cho Menville). GraphRAG kết hợp bối cảnh nhật ký kiểm tra ngữ nghĩa (Menville không xác minh được danh tính) với cấu trúc liên kết đồ thị nhiều bước (Doe → Jacoby → Menville), cho phép LLM lọc bỏ nhiễu vô hại trong khi làm nổi bật các vòng rửa tiền có rủi ro cao.

Thêm mã sau vào tập lệnh graphrag_aml.py rồi chạy tập lệnh đó bằng 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))

Tập lệnh graphrag_aml.py hoàn chỉnh

Để tham khảo, tập lệnh graphrag_aml.py hoàn chỉnh của bạn sẽ có dạng như sau:

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

Bạn sẽ thấy kết quả tương tự như báo cáo phân tích LLM mẫu này:

10. Trực quan hoá dấu vết rửa tiền

Để hiểu rõ bằng hình ảnh về dấu vết rửa tiền mà chúng ta vừa phát hiện được theo phương thức lập trình, bạn có thể chạy một truy vấn trực quan hoá biểu đồ trong bảng điều khiển BigQuery Studio.

Chạy truy vấn này trong BigQuery Studio. (Đảm bảo bạn bật tính năng trực quan hoá Biểu đồ hoặc nhấp vào thẻ Biểu đồ nếu có).

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;

Truy vấn GQL này theo dõi toàn bộ đường dẫn từ chủ sở hữu công ty vỏ bọc đáng ngờ (Doe) thông qua bên trung gian (Jacoby) đến mục tiêu cuối cùng (Menville) và việc trả Khoản vay.

Bạn sẽ thấy kết quả trực quan hoá biểu đồ tương tự như sau:

11. Dọn dẹp

Để tránh các khoản phí phát sinh cho tài khoản Google Cloud của bạn, hãy xoá các tài nguyên đã tạo trong lớp học lập trình này.

Xoá tập dữ liệu BigQuery và Cloud Resource Connection:

# 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

Xác minh rằng các tài nguyên đã bị xoá:

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

12. Xin chúc mừng

Xin chúc mừng! Bạn đã tạo thành công một ứng dụng RAG và phân tích hành vi của ứng dụng đó. Bạn đã minh hoạ cách sử dụng các chức năng Tìm kiếm vectơ và Đồ thị gốc của BigQuery để thực hiện GraphRAG, phát hiện một kế hoạch rửa tiền mà không cần ETL.

Kiến thức bạn học được

• Cách tạo Biểu đồ thuộc tính trong BigQuery dựa trên các bảng chuẩn
• Cách tạo và lưu trữ vectơ nhúng bằng BigQuery ML
• Cách kết hợp các thao tác duyệt qua biểu đồ BigQuery và tính năng Tìm kiếm vectơ thành một Trình truy xuất LangChain
• Cách LLM có thể tổng hợp nhật ký kiểm tra ngữ nghĩa bằng cấu trúc liên kết đồ thị để giảm kết quả dương tính giả

Các bước tiếp theo

• Khám phá Tổng quan về biểu đồ thuộc tính BigQuery
• Hãy thử thêm các mẫu Ngôn ngữ truy vấn đồ thị (GQL) của BigQuery.

Tài liệu tham khảo

• Hàm AI.GENERATE_EMBEDDING của BigQuery ML
• Hàm khoảng cách và tính năng Tìm kiếm vectơ của BigQuery
• Các dịch vụ tích hợp của Google với LangChain