Lutter contre le blanchiment d'argent et la fraude avec BigQuery GraphRAG

1. Introduction

Dans cet atelier de programmation, vous allez créer une solution de génération augmentée par récupération de graphiques (GraphRAG) pour détecter le blanchiment d'argent et la fraude financière. Vous utiliserez Vertex AI, Vector Search et les fonctionnalités graphiques natives de BigQuery, coordonnées via LangChain. À la fin de cet atelier, vous verrez comment un grand modèle de langage (LLM) peut identifier l'acheminement illicite de fonds en synthétisant des journaux d'audit sémantiques et des réseaux transactionnels complexes.

Flux de l'architecture 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
+------------------+

Pourquoi GraphRAG ? La RAG standard récupère des blocs de texte isolés à l'aide de la similarité sémantique, mais ne tient pas compte des relations en plusieurs étapes. GraphRAG utilise la recherche vectorielle pour trouver des entités de départ, puis utilise BigQuery GQL pour parcourir les pistes de transactions à plusieurs sauts (par exemple, Propriétaire → Compte → Transfert → Compte → Prêt).

Objectifs de l'atelier

• Étape 1 : Configuration de l'ensemble de données et du graphique de propriété : créez des tables financières relationnelles et construisez un PROPERTY GRAPH BigQuery natif.
• Phase 2 : Génération d'embeddings vectoriels sémantiques : générez des embeddings de texte directement dans SQL pour les journaux d'audit à l'aide de AI.GENERATE_EMBEDDING (text-embedding-005).
• Phase 3 : Récupérateur GraphRAG LangChain personnalisé : créez un récupérateur Python personnalisé combinant la similarité vectorielle (COSINE_DISTANCE) et les traversées de chemin GQL ISO.
• Phase 4 : Raisonnement LLM sur la fraude et visualisation des chemins : exécutez une chaîne de raisonnement Gemini pour exposer les boucles de blanchiment d'argent illicites et visualiser les chemins dans BigQuery Studio.

Prérequis

• Un navigateur Web tel que Chrome.
• Un projet Google Cloud avec facturation activée.

Cet atelier de programmation s'adresse aux développeurs, aux ingénieurs de données et aux spécialistes de l'IA de tous niveaux, y compris aux débutants.

Durée estimée : 35 minutes
Coût estimé : moins de 2 $ (utilise le traitement des requêtes Vertex AI et BigQuery au paiement à l'utilisation).

2. Avant de commencer

Créer un projet Google Cloud

1. Dans la console Google Cloud, sélectionnez ou créez un projet Google Cloud.
2. Assurez-vous que la facturation est activée pour votre projet Cloud.

Démarrer Cloud Shell

1. Cliquez sur Activer Cloud Shell en haut de la console Google Cloud.
2. Vérifiez l'authentification :

gcloud auth list

3. Configurez les variables d'environnement dans 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

Activer les API

Exécutez la commande suivante pour activer toutes les API requises :

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

3. Configuration et initialisation

Dans cette étape, nous allons configurer un environnement Python, installer les bibliothèques requises et initialiser les clients BigQuery et Vertex AI. Vous pouvez exécuter ces commandes dans Cloud Shell ou dans un environnement de notebook Jupyter.

1. Créez et activez un environnement virtuel Python :

python3 -m venv venv
source venv/bin/activate

2. Installez les packages Python requis :

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

3. Créez un fichier Python graphrag_aml.py et ajoutez le code d'initialisation. Remplacez par l'ID de votre projet 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. Créer des tables et un schéma

Nous allons ensuite définir le schéma de notre graphique financier en créant un ensemble de données et des tables standards BigQuery.

1. Créez l'ensemble de données BigQuery :

bq mk --location=US --dataset fingraph_rag

2. Créez les tables. Vous pouvez exécuter cette requête dans l'UI BigQuery Studio ou via Cloud Shell. Voici le code 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. Insérer l'ensemble de données

Nous allons maintenant insérer les entités et leurs relations pour former notre chaîne de transactions. Cet ensemble de données représente les activités suspectes entre Doe (propriétaire présumé de la société écran), Jacoby (intermédiaire), Menville (cible ne respectant pas la procédure KYC) et Smith (simple passant).

Exécutez le code SQL suivant pour remplir les tables :

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

Vérifier les enregistrements ingérés

Exécutez cette requête pour vérifier le nombre d'enregistrements dans vos tables financières :

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

Le résultat de la requête doit confirmer l'insertion de la ligne. Il doit ressembler à ceci :

6. Créer un graphique de propriété BigQuery

Une fois nos données relationnelles en place, nous définissons FinGraph à l'aide du langage DDL Graph natif de BigQuery. Cela crée une couche sémantique sur les tables relationnelles existantes sans copier ni dupliquer les données.

Présentation de la syntaxe ISO GQL

Les graphiques de propriétés BigQuery utilisent des modèles standards du langage GQL (Graph Query Language) ISO :

• (node:Label) définit les nœuds d'entité (par exemple, Account, Person, Loan).
• -[edge:LABEL]-> définit les relations orientées (par exemple, Transfers, Repays, Owns).

Exécutez l'instruction SQL suivante pour créer le graphique de propriétés :

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

Le paradigme Zero-ETL : CREATE PROPERTY GRAPH de BigQuery définit des structures de graphiques directement sur les tables relationnelles existantes, ce qui élimine les pipelines ETL fragiles et la duplication des données.

Pour visualiser l'intégralité du graphique des comptes, des personnes et des prêts, exécutez la requête SQL suivante dans BigQuery Studio :

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

Vous devriez obtenir un résultat de visualisation de graphique semblable à celui-ci :

7. Générer des embeddings pour les journaux d'audit

Pour activer la partie recherche vectorielle de notre pipeline RAG, nous générons des embeddings de texte pour les journaux d'audit non structurés directement dans BigQuery à l'aide de la fonction de valeur de table (TVF) AI.GENERATE_EMBEDDING.

Créer une connexion BigQuery à distance et accorder des autorisations IAM

BigQuery ML nécessite une connexion CLOUD_RESOURCE pour communiquer de manière sécurisée avec les points de terminaison d'embedding Vertex AI. Exécutez les commandes Bash suivantes dans Cloud Shell pour créer la connexion, découvrir son compte de service généré automatiquement et attribuer le rôle Utilisateur 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

Remarque sur les invites de condition IAM
 : Si votre projet Google Cloud contient déjà des liaisons IAM conditionnelles, gcloud peut vous inviter de manière interactive à sélectionner une option de condition (par exemple, [1] ..., [2] ..., [3] None). Sélectionnez None (option 3) pour accorder le rôle roles/aiplatform.user de manière inconditionnelle pour cet atelier. L'ajout de --condition=None dans la commande ci-dessus permet de contourner automatiquement cette invite.

Créer un modèle d'embedding distant

Ensuite, définissez un modèle distant BigQuery ML qui établit un lien vers le modèle text-embedding-005 de Vertex AI via votre nouvelle connexion autorisée :

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

Délai de propagation des autorisations IAM : il peut s'écouler entre une et trois minutes avant que les rôles IAM nouvellement accordés ne se propagent aux services Google Cloud.
Si la création de modèle ou la génération d'embeddings échoue initialement en raison d'une erreur d'autorisation (par exemple, The [e-mail] does not have the permission to access or use the endpoint...), attendez quelques minutes que les liaisons IAM se synchronisent, puis réessayez d'exécuter la requête.

Générer des embeddings

Générez maintenant des embeddings pour la table AccountAudits en appelant AI.GENERATE_EMBEDDING dans la clause FROM d'une instruction 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;

Vérifier les dimensions vectorielles générées

Exécutez la requête suivante pour vérifier que les embeddings vectoriels ont été renseignés :

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

Le résultat de la requête doit afficher des embeddings vectoriels de dimension 768 semblables à ceux-ci :

Qu'est-ce que AI.GENERATE_EMBEDDING et Vector Search ?
AI.GENERATE_EMBEDDING est une fonction de valeur de table (TVF) qui doit être appelée dans une clause FROM. Il accepte un MODEL BigQuery ML distant (qui s'authentifie auprès de Vertex AI via une connexion CLOUD_RESOURCE) et une requête d'entrée contenant une colonne content. Le calcul des embeddings est directement déchargé sur Vertex AI sans que les données ne quittent BigQuery. COSINE_DISTANCE calcule la similarité vectorielle entre les embeddings cibles et les vecteurs de requête pour détecter les correspondances sémantiques que les vérifications basées sur des règles ne détectent pas.

8. Définir le Retriever GraphRAG

Nous allons maintenant créer un récupérateur LangChain personnalisé dans notre environnement Python. Ce récupérateur combine la recherche vectorielle sémantique (pour trouver des points de départ pertinents) avec des requêtes MATCH de graphe natif (pour parcourir les relations).

Ajoutez le code suivant à votre script 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

Remarquez comment la requête de parcours de graphique rassemble sans effort les propriétaires de comptes, les expéditeurs, les transferts et les prêts via MATCH (sender_person:Person)-[:Owns]->..., sans avoir besoin de jointures SQL complexes.

Architecture NL2SQL/GQL multi-agents dynamique
 : dans les systèmes multi-agents de production avancés, au lieu d'utiliser des modèles de requêtes SQL/GQL prédéfinis dans le récupérateur, vous pouvez déployer un agent NL2SQL / NL2GQL dédié. Chargé de définitions de schéma de graphe et d'exemples de requêtes en tant que compétences, cet agent spécialisé génère dynamiquement des requêtes GQL personnalisées en fonction des principaux objectifs d'investigation de l'agent principal, effectue des simulations et valide le code SQL par rapport à BigQuery, puis renvoie des résultats de graphe ciblés. Bien que la génération dynamique de requêtes SQL par les agents mette en évidence toute la flexibilité des workflows multi-agents, nous utilisons des requêtes structurées dans cet atelier pour plus de concision et de clarté.

9. Exécuter l'investigation sur la fraude

Enfin, nous exécutons le pipeline GraphRAG pour générer un rapport détaillé sur la fraude. Le LLM utilisera le contexte récupéré par notre récupérateur de graphiques personnalisé pour répondre à la requête.

Valeur du domaine : résoudre le problème des faux positifs dans les opérations de lutte contre la fraude
 La surveillance traditionnelle de la lutte contre le blanchiment d'argent basée sur des règles signale des millions de faux positifs chaque année, car les règles évaluent les transactions de manière isolée (par exemple, en signalant le transfert de 150 $ de Smith à Menville). GraphRAG combine le contexte sémantique des journaux d'audit (Menville a échoué à la procédure KYC) avec la topologie de graphe multihop (Doe → Jacoby → Menville). Les LLM peuvent ainsi filtrer le bruit bénin tout en faisant ressortir les boucles de blanchiment à haut risque.

Ajoutez le code suivant à votre script graphrag_aml.py et exécutez-le à l'aide de 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))

Script complet graphrag_aml.py

Pour référence, votre script graphrag_aml.py complet devrait ressembler à ceci :

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

Vous devriez obtenir un résultat semblable à cet exemple de rapport d'analyse LLM :

10. Visualiser la piste de blanchiment d'argent

Pour comprendre visuellement la piste de blanchiment d'argent que nous venons de découvrir de manière programmatique, vous pouvez exécuter une requête de visualisation de graphique dans la console BigQuery Studio.

Exécutez cette requête dans BigQuery Studio. (Assurez-vous d'activer la fonctionnalité de visualisation des graphiques ou de cliquer sur l'onglet "Graphique" si disponible.)

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;

Cette requête GQL retrace l'intégralité du chemin parcouru par le propriétaire de la société écran suspecte (Doe) en passant par l'intermédiaire (Jacoby) jusqu'à la cible finale (Menville) et le remboursement du prêt.

Vous devriez obtenir un résultat de visualisation de graphique semblable à celui-ci :

11. Effectuer un nettoyage

Pour éviter que les ressources créées lors de cet atelier de programmation ne soient facturées en permanence sur votre compte Google Cloud, supprimez-les.

Supprimez l'ensemble de données BigQuery et la connexion aux ressources 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

Vérifiez que les ressources ont été supprimées :

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

12. Félicitations

Félicitations ! Vous avez créé une application RAG et analysé son comportement. Vous avez montré comment utiliser les fonctionnalités natives de recherche vectorielle et de graphiques de BigQuery pour effectuer GraphRAG, en détectant un système de blanchiment d'argent sans ETL.

Connaissances acquises

• Créer un graphique de propriétés dans BigQuery à partir de tables standards
• Générer et stocker des embeddings vectoriels à l'aide de BigQuery ML
• Combiner les traversées de graphiques BigQuery et la recherche vectorielle dans un Retriever LangChain
• Comment les LLM peuvent synthétiser des journaux d'audit sémantiques avec une topologie de graphe pour réduire les faux positifs

Étapes suivantes

• Découvrez la présentation des graphiques de propriétés BigQuery.
• Essayez d'autres modèles BigQuery Graph Query Language (GQL).

Documents de référence

• Fonction BigQuery ML AI.GENERATE_EMBEDDING
• Recherche vectorielle et fonctions de distance BigQuery
• Intégrations Google LangChain