مبارزه با پولشویی و پیشگیری از کلاهبرداری با BigQuery GraphRAG

۱. مقدمه

در این آزمایشگاه کد، شما یک راهکار بازیابی گراف-تقویت‌شده نسل (GraphRAG) برای تشخیص مبارزه با پولشویی (AML) و کلاهبرداری مالی خواهید ساخت. شما از Vertex AI، Vector Search و قابلیت‌های بومی گراف 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 برای پیمایش مسیرهای مالی چند مرحله‌ای استفاده می‌کند (مثلاً مالک → حساب → انتقال → حساب → وام ).

کاری که انجام خواهید داد

• مرحله ۱: تنظیم مجموعه داده‌ها و نمودار ویژگی‌ها : ایجاد جداول مالی رابطه‌ای و ساخت یک PROPERTY GRAPH بومی BigQuery.
• مرحله ۲: تولید جاسازی بردار معنایی : با استفاده از AI.GENERATE_EMBEDDING ( text-embedding-005 )، جاسازی‌های متنی را مستقیماً در SQL برای گزارش‌های حسابرسی ایجاد کنید.
• مرحله ۳: بازیابی‌کننده سفارشی LangChain GraphRAG : ساخت یک بازیابی‌کننده پایتون سفارشی با ترکیب شباهت برداری ( COSINE_DISTANCE ) و پیمایش‌های مسیر ISO GQL.
• مرحله ۴: استدلال و تجسم دنباله کلاهبرداری LLM : اجرای یک زنجیره استدلال Gemini برای افشای حلقه‌های پولشویی غیرقانونی و تجسم مسیرهای مسیر در BigQuery Studio.

آنچه نیاز دارید

• یک مرورگر وب مانند کروم .
• یک پروژه گوگل کلود با قابلیت پرداخت.

این آزمایشگاه کد برای توسعه‌دهندگان، مهندسان داده و متخصصان هوش مصنوعی در تمام سطوح، از جمله مبتدیان، طراحی شده است.

مدت زمان تخمینی: ۳۵ دقیقه
هزینه تخمینی: کمتر از ۲ دلار آمریکا (از هوش مصنوعی Vertex و پردازش پرس‌وجوی BigQuery به صورت پرداخت در محل استفاده می‌کند).

۲. قبل از شروع

ایجاد یک پروژه ابری گوگل

1. در کنسول گوگل کلود ، یک پروژه گوگل کلود انتخاب یا ایجاد کنید .
2. مطمئن شوید که پرداخت برای پروژه ابری شما فعال است.

شروع پوسته ابری

1. روی فعال کردن Cloud Shell در بالای کنسول Google Cloud کلیک کنید.
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 \
 bigqueryreservation.googleapis.com

ایجاد رزرو و تخصیص BigQuery

برای اجرای کوئری‌های GQL، باید رزروی داشته باشید که از نسخه Enterprise یا Enterprise Plus استفاده کند، یک رزرو Enterprise Edition با قابلیت مقیاس‌بندی خودکار در Cloud Shell ایجاد کنید:

# 1. Create a BigQuery Enterprise reservation with 0 baseline slots and 100 max autoscaling slots
bq mk --reservation \
  --project_id=${GCP_PROJECT} \
  --location=US \
  --edition=ENTERPRISE \
  --slots=0 \
  --autoscale_max_slots=100 \
  --ignore_idle_slots=true \
  graphrag-bigquery-reservation

# 2. Assign your Cloud project to the newly created reservation for query execution
bq mk --reservation_assignment \
  --project_id=${GCP_PROJECT} \
  --location=US \
  --reservation_id=graphrag-bigquery-reservation \
  --job_type=QUERY \
  --assignee_type=PROJECT \
  --assignee_id=${GCP_PROJECT}

محاسبه مقیاس‌بندی خودکار مقرون‌به‌صرفه :
با پیکربندی --slots=0 (0 اسلات پایه) و --autoscale_max_slots=100 ، رزرو BigQuery شما در حالت غیرفعال، هزینه پایه صفر را متحمل می‌شود. BigQuery به طور خودکار تا 100 اسلات اختصاصی افزایش می‌یابد، در حالی که جستجوها یا جستجوی برداری شما به طور فعال اجرا می‌شوند، سپس پس از اتمام، به 0 کاهش می‌یابد.

۳. راه‌اندازی و مقداردهی اولیه

در این مرحله، یک محیط پایتون راه‌اندازی می‌کنیم، کتابخانه‌های مورد نیاز را نصب می‌کنیم و کلاینت‌های BigQuery و Vertex AI را راه‌اندازی اولیه می‌کنیم. می‌توانید این دستورات را در Cloud Shell یا محیط Jupyter notebook اجرا کنید.

1. ایجاد و فعال‌سازی یک محیط مجازی پایتون:

python3 -m venv venv
source venv/bin/activate

2. بسته‌های پایتون مورد نیاز را نصب کنید:

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

3. یک فایل پایتون به graphrag_aml.py ایجاد کنید و کد مقداردهی اولیه را به آن اضافه کنید. جایگزین کنید با شناسه پروژه گوگل کلود شما.

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)

۴. ایجاد جداول و طرحواره

در مرحله بعد، با ایجاد یک مجموعه داده BigQuery و جداول استاندارد، طرحواره (schema) را برای نمودار مالی خود تعریف می‌کنیم.

1. مجموعه داده BigQuery را ایجاد کنید:

bq mk --location=US --dataset fingraph_rag

2. جداول را ایجاد کنید. می‌توانید این را در رابط کاربری BigQuery Studio یا از طریق 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>);

۵. مجموعه داده را وارد کنید

اکنون نهادها و روابط آنها را برای تشکیل مسیر مالی خود وارد می‌کنیم. این مجموعه داده‌ها نشان‌دهنده فعالیت‌های مشکوک بین Doe (مالک مشکوک شرکت صوری)، Jacoby (واسطه)، Menville (هدف ناموفق در احراز هویت مشتری) و 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

با قرار دادن داده‌های رابطه‌ای در جای خود، FinGraph با استفاده از Graph DDL بومی BigQuery تعریف می‌کنیم. این کار بدون کپی کردن یا تکثیر داده‌ها، یک لایه معنایی روی جداول رابطه‌ای موجود ایجاد می‌کند.

آغازگر نحو ISO GQL

نمودارهای ویژگی BigQuery از الگوهای استاندارد زبان پرس‌وجوی نمودار ISO (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)
 );

الگوی Zero-ETL: قابلیت CREATE PROPERTY GRAPH در BigQuery، ساختارهای گراف را مستقیماً روی جداول رابطه‌ای موجود تعریف می‌کند و خطوط لوله ETL شکننده و تکثیر داده‌ها را از بین می‌برد.

برای نمایش نمودار کامل حساب‌ها، اشخاص و وام‌ها، کوئری SQL زیر را در BigQuery Studio اجرا کنید:

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

شما باید نتیجه تجسم نموداری مشابه این را ببینید:

۷. ایجاد جاسازی‌ها برای گزارش‌های حسابرسی

برای فعال کردن بخش جستجوی برداری در خط لوله RAG، ما جاسازی‌های متنی را برای گزارش‌های حسابرسی بدون ساختار مستقیماً در BigQuery با استفاده از تابع AI.GENERATE_EMBEDDING Table-Valued (TVF) ایجاد می‌کنیم.

ایجاد اتصال از راه دور BigQuery و اعطای مجوزهای IAM

BigQuery ML برای برقراری ارتباط ایمن با نقاط پایانی تعبیه‌شده‌ی Vertex AI به یک اتصال CLOUD_RESOURCE نیاز دارد. دستورات bash زیر را در Cloud Shell اجرا کنید تا اتصال ایجاد شود، حساب سرویس خودکار ایجاد شده‌ی آن را پیدا کنید و نقش Vertex AI User ( 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 ). برای اعطای بدون قید و شرط نقش roles/aiplatform.user برای این آزمایش، None (گزینه 3 ) را انتخاب کنید. اضافه کردن --condition=None در دستور بالا، این اعلان را به طور خودکار دور می‌زند.

ایجاد مدل جاسازی از راه دور

در مرحله بعد، یک مدل از راه دور BigQuery ML تعریف کنید که از طریق اتصال تازه مجاز شما به مدل text-embedding-005 مربوط به Vertex AI لینک شود:

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

تأخیر انتشار مجوز IAM:
نقش‌های IAM تازه اعطا شده می‌توانند ۱ تا ۳ دقیقه طول بکشند تا در سرویس‌های Google Cloud منتشر شوند. اگر ایجاد مدل یا تولید جاسازی در ابتدا با خطای مجوز مواجه شد (مثلاً، The [e-mail] does not have the permission to access or use the endpoint... )، چند دقیقه صبر کنید تا اتصالات IAM همگام‌سازی شوند و دوباره پرس‌وجو را امتحان کنید.

ایجاد جاسازی‌ها

اکنون، با فراخوانی AI.GENERATE_EMBEDDING در بند FROM از دستور UPDATE ، جاسازی‌هایی برای جدول 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`;

شما باید خروجی پرس و جو را مشاهده کنید که جاسازی‌های برداری ۷۶۸ بعدی مشابه این را نشان می‌دهد:

هوش مصنوعی، تولید جاسازی (AI.GENERATE_EMBEDDING) و جستجوی برداری (Vector Search) چیست؟
AI.GENERATE_EMBEDDING یک تابع Table-Valued (TVF) است که باید در یک عبارت FROM فراخوانی شود. این تابع یک MODEL از راه دور BigQuery ML (که از طریق اتصال CLOUD_RESOURCE به Vertex AI احراز هویت می‌شود) و یک پرس‌وجوی ورودی حاوی یک ستون content را می‌پذیرد. محاسبه جاسازی مستقیماً به Vertex AI منتقل می‌شود بدون اینکه داده‌ها از BigQuery خارج شوند. COSINE_DISTANCE شباهت برداری بین جاسازی‌های هدف و بردارهای پرس‌وجو را محاسبه می‌کند تا تطابق‌های معنایی را که بررسی‌های مبتنی بر قانون از دست می‌دهند، پیدا کند.

۸. تعریف GraphRAG Retriever

اکنون یک بازیابی‌کننده‌ی سفارشی LangChain در محیط پایتون خود ایجاد خواهیم کرد. این بازیابی‌کننده، جستجوی برداری معنایی (برای یافتن نقاط شروع مرتبط) را با پرس‌وجوهای بومی Graph MATCH (برای پیمایش روابط) ترکیب می‌کند.

کد زیر را به اسکریپت پایتون 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]->... جمع‌آوری می‌کند - هیچ JOIN پیچیده‌ای در SQL لازم نیست!

معماری پویای چندعاملی NL2SQL/GQL:
در سیستم‌های چندعاملی پیشرفته تولیدی، به جای استفاده از الگوهای پرس‌وجوی از پیش تعریف‌شده SQL/GQL در بازیابی‌کننده، می‌توانید یک عامل اختصاصی NL2SQL / NL2GQL را مستقر کنید. این عامل تخصصی که مملو از تعاریف طرحواره گراف و نمونه‌های پرس‌وجو به عنوان مهارت است، به صورت پویا پرس‌وجوهای GQL متناسب با اهداف اصلی تحقیق عامل اصلی را تولید می‌کند، SQL را در BigQuery اجرا و اعتبارسنجی می‌کند و نتایج گراف هدفمند را برمی‌گرداند. در حالی که تولید SQL عامل‌دار پویا، انعطاف‌پذیری کامل گردش‌های کاری چندعاملی را برجسته می‌کند، ما در این آزمایشگاه از پرس‌وجوهای ساختاریافته برای اختصار و وضوح استفاده می‌کنیم.

۹. تحقیقات مربوط به کلاهبرداری را اجرا کنید

در نهایت، ما خط لوله GraphRAG را برای تولید یک گزارش کلاهبرداری دقیق اجرا می‌کنیم. LLM از زمینه بازیابی شده توسط بازیابی کننده گراف سفارشی ما برای پاسخ به سوال استفاده خواهد کرد.

ارزش دامنه: حل مشکل مثبت کاذب در عملیات کلاهبرداری
نظارت سنتی AML مبتنی بر قانون، سالانه میلیون‌ها مورد مثبت کاذب را علامت‌گذاری می‌کند، زیرا قوانین، تراکنش‌ها را به صورت جداگانه ارزیابی می‌کنند (مثلاً علامت‌گذاری انتقال ۱۵۰ دلاری اسمیت به منویل). GraphRAG زمینه لاگ حسابرسی معنایی (KYC ناموفق منویل) را با توپولوژی گراف چند گامی (Doe → Jacoby → منویل) ترکیب می‌کند و 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 نمونه را ببینید:

۱۰. مسیر پولشویی را تجسم کنید

برای درک بصری رد پولشویی که به تازگی از طریق برنامه‌نویسی کشف کردیم، می‌توانید یک کوئری تجسم نموداری را در کنسول BigQuery Studio اجرا کنید.

این کوئری را در BigQuery Studio اجرا کنید. (مطمئن شوید که ویژگی تجسم نمودار (Graph visualization) را فعال کرده‌اید یا در صورت وجود، روی تب نمودار (Graph) کلیک کنید.)

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 ) و بازپرداخت وام ردیابی می‌کند.

شما باید نتیجه تجسم نموداری مشابه این را ببینید:

۱۱. تمیز کردن

برای جلوگیری از هزینه‌های مداوم برای حساب Google Cloud خود، منابع ایجاد شده در طول این codelab را حذف کنید.

مجموعه داده BigQuery و اتصال منابع ابری را حذف کنید:

# 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

۱۲. تبریک

تبریک! شما با موفقیت یک برنامه RAG ساختید و رفتار آن را تجزیه و تحلیل کردید. شما نشان دادید که چگونه از قابلیت‌های بومی جستجوی گراف و وکتور BigQuery برای انجام GraphRAG استفاده کنید و یک طرح مبارزه با پولشویی را بدون ETL شناسایی کنید.

آنچه آموخته‌اید

• نحوه ساخت نمودار ویژگی در BigQuery بر روی جداول استاندارد
• نحوه تولید و ذخیره Vector Embeddings با استفاده از BigQuery ML
• چگونه پیمایش‌های گراف BigQuery و جستجوی برداری را در یک LangChain Retriever ترکیب کنیم؟
• چگونه LLM ها می‌توانند لاگ‌های حسابرسی معنایی را با توپولوژی گراف ترکیب کنند تا موارد مثبت کاذب را کاهش دهند

مراحل بعدی

• بررسی اجمالی نمودارهای ویژگی BigQuery
• الگوهای بیشتری از زبان پرس‌وجوی گراف (GQL) در BigQuery را امتحان کنید.

اسناد مرجع

• تابع GENERATE_EMBEDDING در BigQuery ML AI
• جستجوی برداری BigQuery و توابع فاصله
• ادغام‌های گوگل با LangChain