Geldwäschebekämpfung und Betrugsprävention mit BigQuery GraphRAG

1. Einführung

In diesem Codelab erstellen Sie eine Graph Retrieval-Augmented Generation-Lösung (GraphRAG), um Geldwäsche und Finanzbetrug zu erkennen. Sie verwenden Vertex AI, Vector Search und die nativen Graph-Funktionen von BigQuery, die über LangChain koordiniert werden. Am Ende dieses Labs werden Sie sehen, wie ein Large Language Model (LLM) illegale Geldflüsse erkennen kann, indem es semantische Audit-Logs und komplexe Transaktionsnetzwerke zusammenführt.

GraphRAG-Architekturfluss

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

Warum GraphRAG? Bei Standard-RAG werden isolierte Textblöcke mithilfe der semantischen Ähnlichkeit abgerufen, wobei mehrstufige Beziehungen nicht berücksichtigt werden. GraphRAG verwendet die Vektorsuche, um Seed-Entitäten zu finden, und dann BigQuery GQL, um Multi-Hop-Geldflüsse zu durchlaufen (z.B. Owner → Account → Transfer → Account → Loan).

Aufgaben

• Phase 1: Dataset und Property Graph einrichten: Erstellen Sie relationale Finanztabellen und einen nativen BigQuery-PROPERTY GRAPH.
• Phase 2: Generierung semantischer Vektoreinbettungen: Generieren Sie Texteinbettungen direkt in SQL für Audit-Logs mit AI.GENERATE_EMBEDDING (text-embedding-005).
• Phase 3: Benutzerdefinierter LangChain GraphRAG Retriever: Erstellen Sie einen benutzerdefinierten Python-Retriever, der Vektorähnlichkeit (COSINE_DISTANCE) und ISO-GQL-Pfad-Traversierungen kombiniert.
• Phase 4: LLM-Betrugsbegründung und Visualisierung von Pfaden: Führen Sie eine Gemini-Begründungskette aus, um illegale Geldwäschekreisläufe aufzudecken und Pfade in BigQuery Studio zu visualisieren.

Voraussetzungen

• Ein Webbrowser wie Chrome.
• Google Cloud-Projekt mit aktivierter Abrechnungsfunktion.

Dieses Codelab richtet sich an Entwickler, Data Engineers und KI-Experten aller Erfahrungsstufen, auch an Anfänger.

Geschätzte Dauer:35 Minuten
Geschätzte Kosten:Weniger als 2,00 $ (nutzt die nutzungsbasierte Abrechnung für die Verarbeitung von Vertex AI- und BigQuery-Abfragen).

2. Hinweis

Google Cloud-Projekt erstellen

1. Wählen Sie in der Google Cloud Console ein Google Cloud-Projekt aus oder erstellen Sie eines.
2. Die Abrechnung für das Cloud-Projekt muss aktiviert sein.

Cloud Shell starten

1. Klicken Sie oben in der Google Cloud Console auf Cloud Shell aktivieren.
2. Authentifizierung überprüfen:

gcloud auth list

3. Konfigurieren Sie Umgebungsvariablen in 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

APIs aktivieren

Führen Sie diesen Befehl aus, um alle erforderlichen APIs zu aktivieren:

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

3. Einrichtung und Initialisierung

In diesem Schritt richten wir eine Python-Umgebung ein, installieren die erforderlichen Bibliotheken und initialisieren die BigQuery- und Vertex AI-Clients. Sie können diese Befehle in Cloud Shell oder in einer Jupyter-Notebook-Umgebung ausführen.

1. Erstellen und aktivieren Sie eine virtuelle Python-Umgebung:

python3 -m venv venv
source venv/bin/activate

2. Installieren Sie die erforderlichen Python-Pakete:

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

3. Erstellen Sie eine Python-Datei graphrag_aml.py und fügen Sie den Initialisierungscode hinzu. Ersetzen Sie durch Ihre Google Cloud-Projekt-ID.

import vertexai
from google.cloud import bigquery

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

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

4. Tabellen und Schema erstellen

Als Nächstes definieren wir das Schema für unser Finanzdiagramm, indem wir ein BigQuery-Dataset und Standardtabellen erstellen.

1. BigQuery-Dataset erstellen:

bq mk --location=US --dataset fingraph_rag

2. Erstellen Sie die Tabellen. Sie können diesen Befehl in der BigQuery Studio-Benutzeroberfläche oder über Cloud Shell ausführen. Hier ist der SQL-Code:

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. Dataset einfügen

Wir fügen nun die Entitäten und ihre Beziehungen ein, um den Geldfluss zu erstellen. Dieses Dataset stellt die verdächtigen Aktivitäten zwischen Doe (mutmaßlicher Inhaber einer Briefkastenfirma), Jacoby (Vermittler), Menville (Ziel, das die KYC-Prüfung nicht besteht) und Smith (unschuldiger Dritter) dar.

Führen Sie den folgenden SQL-Befehl aus, um die Tabellen zu befüllen:

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

Aufgenommene Datensätze prüfen

Führen Sie diese Abfrage aus, um die Anzahl der Datensätze in Ihren Finanztabellen zu prüfen:

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

Die Ausgabe der Abfrage sollte in etwa so aussehen und die Einfügung der Zeile bestätigen:

6. BigQuery-Eigenschaftsgraphen erstellen

Nachdem wir unsere relationalen Daten eingerichtet haben, definieren wir die FinGraph mit der nativen Graph-DDL von BigQuery. So wird eine semantische Ebene über vorhandenen relationalen Tabellen erstellt, ohne dass Daten kopiert oder dupliziert werden.

ISO GQL-Syntax – Grundlagen

Für BigQuery-Eigenschaftsgraphen werden Standardmuster der ISO Graph Query Language (GQL) verwendet:

• (node:Label) definiert Entitätsknoten (z.B. Account, Person, Loan).
• -[edge:LABEL]-> definiert gerichtete Beziehungen (z.B. Transfers, Repays, Owns).

Führen Sie die folgende SQL-Anweisung aus, um den Eigenschaftsgraphen zu erstellen:

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

Das Zero-ETL-Paradigma: Mit CREATE PROPERTY GRAPH von BigQuery werden Diagrammstrukturen direkt in vorhandenen relationalen Tabellen definiert. So werden anfällige ETL-Pipelines und Datenduplizierung vermieden.

Wenn Sie das gesamte Diagramm mit Konten, Personen und Darlehen visualisieren möchten, führen Sie die folgende SQL-Abfrage in BigQuery Studio aus:

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

Das Ergebnis sollte etwa so aussehen:

7. Eingebettete Darstellungen für Audit-Logs generieren

Um den Vektorsuchteil unserer RAG-Pipeline zu aktivieren, generieren wir Text-Embeddings für die unstrukturierten Audit-Logs direkt in BigQuery mit der Tabellenwertfunktion (Table-Valued Function, TVF) AI.GENERATE_EMBEDDING.

BigQuery-Remote-Verbindung erstellen und IAM-Berechtigungen erteilen

Für BigQuery ML ist eine CLOUD_RESOURCE-Verbindung erforderlich, um sicher mit Vertex AI-Einbettungsendpunkten zu kommunizieren. Führen Sie die folgenden Bash-Befehle in Cloud Shell aus, um die Verbindung zu erstellen, das automatisch generierte Dienstkonto zu ermitteln und die Rolle Vertex AI-Nutzer (roles/aiplatform.user) zuzuweisen:

# 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

Hinweis zu Aufforderungen für IAM-Bedingungen
:Wenn Ihr Google Cloud-Projekt bereits bedingte IAM-Bindungen enthält, werden Sie von gcloud möglicherweise interaktiv aufgefordert, eine Bedingungsoption auszuwählen (z.B. [1] ..., [2] ..., [3] None). Wählen Sie None (Option 3) aus, um die Rolle roles/aiplatform.user für dieses Lab bedingungslos zu gewähren. Wenn Sie --condition=None im obigen Befehl hinzufügen, wird diese Aufforderung automatisch umgangen.

Remote-Einbettungsmodell erstellen

Definieren Sie als Nächstes ein BigQuery ML-Remote-Modell, das über die neu autorisierte Verbindung mit dem text-embedding-005-Modell von Vertex AI verknüpft wird:

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

Verzögerung bei der Weitergabe von IAM-Berechtigungen:Es kann 1 bis 3 Minuten dauern, bis neu zugewiesene IAM-Rollen in allen Google Cloud-Diensten wirksam werden.
Wenn die Modellerstellung oder die Generierung von Einbettungen anfangs mit einem Berechtigungsfehler (z.B. The [e-mail] does not have the permission to access or use the endpoint...) fehlschlägt, warten Sie einige Minuten, bis die IAM-Bindungen synchronisiert sind, und wiederholen Sie die Anfrage.

Einbettungen generieren

Generieren Sie nun Einbettungen für die Tabelle AccountAudits, indem Sie AI.GENERATE_EMBEDDING in der FROM-Klausel einer UPDATE-Anweisung aufrufen:

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;

Generierte Vektordimensionen prüfen

Führen Sie die folgende Abfrage aus, um zu prüfen, ob Vektoreinbettungen eingefügt wurden:

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

Die Abfrageausgabe sollte 768-dimensionale Vektoreinbettungen wie diese enthalten:

Was ist AI.GENERATE_EMBEDDING und die Vektorsuche?
AI.GENERATE_EMBEDDING ist eine Tabellenwertfunktion (Table-Valued Function, TVF), die in einer FROM-Klausel aufgerufen werden muss. Sie akzeptiert ein BigQuery ML-Remote-MODEL (das sich über eine CLOUD_RESOURCE-Verbindung bei Vertex AI authentifiziert) und eine Eingabeabfrage mit einer content-Spalte. Die Berechnung von Einbettungen wird direkt an Vertex AI ausgelagert, ohne dass Daten BigQuery verlassen. COSINE_DISTANCE berechnet die Vektorähnlichkeit zwischen Zieleinbettungen und Anfragevektoren, um semantische Übereinstimmungen zu erkennen, die bei regelbasierten Prüfungen nicht berücksichtigt werden.

8. GraphRAG-Retriever definieren

Wir erstellen jetzt einen benutzerdefinierten LangChain-Retriever in unserer Python-Umgebung. Dieser Retriever kombiniert die semantische Vektorsuche (um relevante Ausgangspunkte zu finden) mit nativen Graph-MATCH-Abfragen (um Beziehungen zu durchlaufen).

Fügen Sie Ihrem Python-Skript graphrag_aml.py den folgenden Code hinzu:

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

Beachten Sie, wie die Graph-Traversal-Abfrage die Kontoinhaber, Absender, Überweisungen und Darlehen mühelos über MATCH (sender_person:Person)-[:Owns]->... zusammenführt – ohne komplexe SQL-JOINs.

Dynamische Multi-Agent-NL2SQL-/GQL-Architektur
:In komplexen Multi-Agent-Systemen für die Produktion können Sie anstelle von vordefinierten SQL-/GQL-Abfragevorlagen im Retriever einen dedizierten NL2SQL-/NL2GQL-Agent bereitstellen. Dieser spezialisierte Agent ist mit Definitionen für das Graphenschema und Abfragebeispielen ausgestattet. Er generiert dynamisch maßgeschneiderte GQL-Abfragen basierend auf den primären Untersuchungszielen des Haupt-Agents, führt Testläufe durch und validiert die SQL-Abfragen in BigQuery. Anschließend gibt er gezielte Grafikergebnisse zurück. Die dynamische agentenbasierte SQL-Generierung verdeutlicht die volle Flexibilität von Multi-Agent-Workflows. In diesem Lab verwenden wir jedoch strukturierte Abfragen, um die Übersichtlichkeit zu wahren.

9. Betrugsprüfung ausführen

Schließlich führen wir die GraphRAG-Pipeline aus, um einen detaillierten Betrugsbericht zu erstellen. Das LLM verwendet den Kontext, der von unserem benutzerdefinierten Graph-Retriever abgerufen wurde, um den Prompt zu beantworten.

Domain Value: Solving the False Positive Problem in Fraud Operations
(Domain Value: Das Problem der falsch positiven Ergebnisse bei Betrugsvorgängen lösen) Bei der herkömmlichen regelbasierten AML-Überwachung werden jährlich Millionen von falsch positiven Ergebnissen gemeldet, da Transaktionen isoliert bewertet werden (z.B. wird die Überweisung von 150 $ von Smith an Menville gemeldet). GraphRAG kombiniert semantischen Audit-Log-Kontext (Menville hat die KYC-Prüfung nicht bestanden) mit Multi-Hop-Graphtopologie (Doe → Jacoby → Menville). So können LLMs harmloses Rauschen herausfiltern und gleichzeitig Geldwäsche-Schleifen mit hohem Risiko erkennen.

Fügen Sie Ihrem Skript graphrag_aml.py den folgenden Code hinzu und führen Sie es mit python graphrag_aml.py aus:

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-Script fertigstellen

Ihr vollständiges graphrag_aml.py-Skript sollte so aussehen:

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

Die Ausgabe sollte in etwa wie in diesem Beispiel für einen LLM-Analysebericht aussehen:

10. Geldwäschepfad visualisieren

Um den Geldwäschepfad, den wir gerade programmatisch ermittelt haben, visuell nachzuvollziehen, können Sie eine Abfrage zur Visualisierung von Grafiken in der BigQuery Studio-Konsole ausführen.

Führen Sie diese Abfrage in BigQuery Studio aus. Achten Sie darauf, dass die Funktion zur Visualisierung von Grafiken aktiviert ist, oder klicken Sie auf den Tab „Graph“, falls verfügbar.

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;

Diese GQL-Abfrage verfolgt den gesamten Pfad vom verdächtigen Inhaber der Briefkastenfirma (Doe) über den Vermittler (Jacoby) bis zum endgültigen Ziel (Menville) und der Rückzahlung des Kredits.

Das Ergebnis sollte etwa so aussehen:

11. Bereinigen

Löschen Sie die in diesem Codelab erstellten Ressourcen, um laufende Gebühren für Ihr Google Cloud-Konto zu vermeiden.

Löschen Sie das BigQuery-Dataset und die Cloud-Ressourcenverbindung:

# 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

Prüfen Sie, ob die Ressourcen gelöscht wurden:

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

12. Glückwunsch

Glückwunsch! Sie haben eine RAG-Anwendung erstellt und ihr Verhalten analysiert. Sie haben gezeigt, wie Sie die nativen Graph- und Vektorsuchfunktionen von BigQuery verwenden können, um GraphRAG auszuführen und ein Geldwäscheschema ohne ETL zu erkennen.

Das haben Sie gelernt

• Property Graph in BigQuery auf Grundlage von Standardtabellen erstellen
• Vektoreinbettungen mit BigQuery ML generieren und speichern
• BigQuery Graph-Traversierungen und Vektorsuche in einem LangChain-Retriever kombinieren
• Wie LLMs semantische Audit-Logs mit der Graphtopologie synthetisieren können, um falsch positive Ergebnisse zu reduzieren

Nächste Schritte

• BigQuery-Eigenschaftsgraphen
• Weitere GQL-Muster für BigQuery Graph ausprobieren

Referenzdokumente

• BigQuery ML AI.GENERATE_EMBEDDING-Funktion
• BigQuery-Vektorsuche und ‑Distanzfunktionen
• LangChain Google-Integrationen