تستخدم Google تكنولوجيا الذكاء الاصطناعي لترجمة المحتوى إلى لغتك المفضّلة، وقد تتضمّن بعض الأخطاء.

مكافحة غسل الأموال ومنع الاحتيال باستخدام BigQuery GraphRAG

1. مقدمة

في هذا الدرس التطبيقي حول الترميز، ستنشئ حلاً للتوليد المعزّز بالاسترجاع باستخدام الرسوم البيانية (GraphRAG) من أجل رصد غسل الأموال والاحتيال المالي. ستستخدم Vertex AI وVector Search وإمكانات Graph الأصلية في 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
+------------------+

الإجراءات التي ستنفذّها

المرحلة 1: إعداد مجموعة البيانات والرسم البياني للعناصر: أنشئ جداول مالية ذات صلة وأنشئ PROPERTY GRAPH أصليًا في BigQuery.
المرحلة 2: إنشاء تضمينات متجهة دلالية: يمكنك إنشاء تضمينات نصية مباشرةً في SQL لسجلات التدقيق باستخدام AI.GENERATE_EMBEDDING (text-embedding-005).
المرحلة 3: أداة استرجاع مخصّصة من LangChain GraphRAG: أنشئ أداة استرجاع مخصّصة بلغة Python تجمع بين التشابه المتجهي (COSINE_DISTANCE) وعمليات اجتياز المسار ISO GQL.
المرحلة 4: عرض مسار التتبُّع واستخدام نماذج لغوية كبيرة لتحديد أسباب الاحتيال: تنفيذ سلسلة استنتاج من Gemini للكشف عن حلقات غسل الأموال غير المشروعة وعرض مسارات التتبُّع بشكل مرئي في BigQuery Studio

المتطلبات

متصفّح ويب، مثل Chrome
مشروع Google Cloud تم تفعيل الفوترة فيه

تم تصميم هذا الدرس التطبيقي حول الترميز للمطوّرين ومهندسي البيانات والمختصين في الذكاء الاصطناعي من جميع المستويات، بما في ذلك المبتدئين.

المدة المقدَّرة: 35 دقيقة
التكلفة المقدَّرة: أقل من 2.00 دولار أمريكي (يتم استخدام Vertex AI وBigQuery لمعالجة طلبات البحث بنظام الدفع حسب الاستخدام).

2. قبل البدء

إنشاء مشروع على Google Cloud

في Google Cloud Console، اختَر مشروعًا على Google Cloud أو أنشِئ مشروعًا.
تأكَّد من تفعيل الفوترة لمشروعك على السحابة الإلكترونية.

بدء Cloud Shell

انقر على تفعيل Cloud Shell في أعلى "وحدة تحكّم Google Cloud".
إثبات صحة المصادقة:

gcloud auth list

ضبط متغيرات البيئة في 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

تفعيل واجهات برمجة التطبيقات

نفِّذ الأمر التالي لتفعيل جميع واجهات برمجة التطبيقات المطلوبة:

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

3- الإعداد والتهيئة

في هذه الخطوة، سنُعدّ بيئة Python ونثبّت المكتبات المطلوبة ونبدأ عملاء BigQuery وVertex AI. يمكنك تنفيذ هذه الأوامر في Cloud Shell أو في بيئة دفتر ملاحظات Jupyter.

أنشئ بيئة Python افتراضية وفعِّلها:

python3 -m venv venv
source venv/bin/activate

ثبِّت حِزم Python المطلوبة:

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

أنشئ ملف Python graphrag_aml.py وأضِف رمز التهيئة. استبدِل برقم تعريف مشروعك على 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. إنشاء الجداول والمخططات

بعد ذلك، نحدّد المخطط البياني لـ "الرسم البياني المالي" من خلال إنشاء مجموعة بيانات وجداول عادية في BigQuery.

أنشئ مجموعة بيانات BigQuery:

bq mk --location=US --dataset fingraph_rag

أنشئ الجداول. يمكنك تنفيذ هذا الإجراء في واجهة مستخدم 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>);

5- إدراج مجموعة البيانات

سنُدرج الآن الكيانات وعلاقاتها لتشكيل مسار الأموال. تمثّل مجموعة البيانات هذه الأنشطة المشبوهة بين 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 Property Graph

بعد إعداد البيانات العلائقية، نحدّد FinGraph باستخدام لغة تعريف البيانات (DDL) الأصلية للرسومات البيانية في BigQuery. يؤدي ذلك إلى إنشاء طبقة دلالية فوق الجداول العلائقية الحالية بدون نسخ البيانات أو تكرارها.

مقدمة عن بنية ISO GQL

تستخدِم "رسومات بيانية للمواقع" في BigQuery أنماطًا عادية من لغة طلبات البحث في الرسومات البيانية (GQL) وفقًا للمعيار الدولي ISO:

تحدّد (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)
 );

لعرض الرسم البياني الكامل للحسابات والأشخاص والقروض، نفِّذ طلب بحث 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;

من المفترض أن تظهر لك نتيجة عرض بياني مشابهة لما يلي:

عرض الرسم البياني الكامل

7. إنشاء تضمينات لسجلّات التدقيق

لتفعيل جزء البحث المتّجه في مسار RAG، ننشئ تضمينات نصية لسجلات التدقيق غير المنظَّمة مباشرةً في BigQuery باستخدام AI.GENERATE_EMBEDDING Table-Valued Function (TVF).

إنشاء اتصال عن بُعد في BigQuery ومنح أذونات "إدارة الهوية وإمكانية الوصول"

تتطلّب BigQuery ML CLOUD_RESOURCE اتصالاً للتواصل بأمان مع نقاط نهاية التضمين في Vertex AI. نفِّذ أوامر bash التالية في Cloud Shell لإنشاء الاتصال والاطّلاع على حساب الخدمة الذي تم إنشاؤه تلقائيًا ومنح دور مستخدم 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

إنشاء نموذج تضمين عن بُعد

بعد ذلك، حدِّد نموذجًا عن بُعد في 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');

إنشاء تضمينات

الآن، أنشئ تضمينات لجدول AccountAudits من خلال استدعاء AI.GENERATE_EMBEDDING في عبارة FROM من عبارة 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;

التحقّق من أبعاد الرسومات المتجهة التي تم إنشاؤها

نفِّذ طلب البحث التالي للتأكّد من ملء عمليات التضمين المتجهة:

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 (يتم مصادقتها في Vertex AI من خلال اتصال CLOUD_RESOURCE) وطلب بحث إدخال يحتوي على عمود content. يتم نقل عملية تضمين الحساب مباشرةً إلى Vertex AI بدون أن تغادر البيانات BigQuery. تحسب COSINE_DISTANCE تشابه المتّجهات بين التضمينات المستهدَفة ومتّجهات طلب البحث لرصد المطابقات الدلالية التي لا ترصدها عمليات التحقّق المستندة إلى القواعد.

8. تحديد أداة استرجاع GraphRAG

سننشئ الآن أداة استرجاع مخصّصة من LangChain في بيئة Python. يجمع هذا المسترجِع بين "البحث الدلالي المتّجه" (للعثور على نقاط بداية ذات صلة) وطلبات بحث 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

بنية NL2SQL/GQL الديناميكية المتعددة الوكلاء:
في أنظمة الإنتاج المتقدّمة المتعددة الوكلاء، بدلاً من استخدام نماذج استعلامات SQL/GQL المحدّدة مسبقًا في أداة الاسترجاع، يمكنك نشر وكيل NL2SQL / NL2GQL مخصّص. يحتوي هذا الوكيل المتخصّص على تعريفات لمخطط الرسم البياني وأمثلة على طلبات البحث كمهارات، وينشئ ديناميكيًا طلبات بحث مخصّصة بلغة GQL استنادًا إلى أهداف التحقيق الأساسية للوكيل الرئيسي، ويجري عمليات تجريبية ويتحقّق من صحة طلبات البحث بلغة SQL في BigQuery، ويعرض نتائج الرسم البياني المستهدَفة. في حين أنّ إنشاء استعلامات SQL ديناميكية باستخدام وكيل يسلّط الضوء على المرونة الكاملة لسير العمل المتعدد الوكلاء، فإنّنا نستخدم استعلامات منظَّمة في هذا التمرين العملي من أجل الاختصار والوضوح.

9. تنفيذ "التحقيق في الاحتيال"

أخيرًا، ننفّذ مسار GraphRAG لإنشاء تقرير تفصيلي عن الاحتيال. سيستخدم النموذج اللغوي الكبير السياق الذي يسترجعه برنامج استرجاع الرسومات المخصّص للإجابة عن الطلب.

قيمة النطاق: حلّ مشكلة الإيجابية الخاطئة في عمليات مكافحة الاحتيال
تشير المراقبة التقليدية لمكافحة غسل الأموال المستندة إلى القواعد إلى ملايين الإيجابيات الخاطئة سنويًا لأنّ القواعد تقيّم المعاملات بشكل منفصل (على سبيل المثال، الإشارة إلى تحويل "سميث" مبلغ 150 دولارًا أمريكيًا إلى "مينفيل"). تجمع GraphRAG بين سياق سجلّ التدقيق الدلالي (تعذّر على Menville إكمال عملية التحقّق من الهوية) وبنية الرسم البياني المتعدّد الخطوات (Doe → Jacoby → Menville)، ما يتيح للنماذج اللغوية الكبيرة فلترة الضوضاء غير الضارة مع إظهار حلقات غسل الأموال عالية الخطورة.

أضِف الرمز التالي إلى النص البرمجي 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))

من المفترَض أن تظهر لك نتيجة مشابهة لنموذج تقرير تحليل النموذج اللغوي الكبير التالي:

تقرير تحليل مكافحة غسل الأموال لردّ النموذج اللغوي الكبير

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) وسداد القرض.