Przeciwdziałanie praniu pieniędzy i zapobieganie oszustwom za pomocą BigQuery GraphRAG

1. Wprowadzenie

W tym ćwiczeniu utworzysz rozwiązanie Graph Retrieval-Augmented Generation (GraphRAG) do wykrywania prania pieniędzy i oszustw finansowych. Będziesz używać Vertex AI, wyszukiwania wektorowego i natywnych funkcji grafu BigQuery, koordynowanych przez LangChain. Po ukończeniu tego modułu dowiesz się, jak duży model językowy (LLM) może identyfikować nielegalne przekazywanie środków, syntetyzując semantyczne dzienniki audytu i złożone sieci transakcyjne.

Przepływ architektury 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
+------------------+

Dlaczego warto używać GraphRAG? Standardowy RAG pobiera wyizolowane fragmenty tekstu na podstawie podobieństwa semantycznego, pomijając relacje wieloetapowe. GraphRAG używa wyszukiwania wektorowego do znajdowania jednostek początkowych, a potem BigQuery GQL do przechodzenia przez wieloetapowe ścieżki przepływu środków (np. Właściciel → Konto → Przelew → Konto → Pożyczka).

Jakie zadania wykonasz

• Faza 1. Konfiguracja zbioru danych i grafu właściwości: utwórz relacyjne tabele finansowe i skonstruuj natywny graf właściwości BigQuery PROPERTY GRAPH.
• Etap 2. Generowanie wektorów dystrybucyjnych semantycznych: generuj wektory dystrybucyjne tekstu bezpośrednio w SQL na potrzeby dzienników kontrolnych za pomocą funkcji AI.GENERATE_EMBEDDING (text-embedding-005).
• Faza 3. Niestandardowy moduł pobierania LangChain GraphRAG: utwórz niestandardowy moduł pobierania w Pythonie, który łączy podobieństwo wektorowe (COSINE_DISTANCE) i przechodzenie ścieżek ISO GQL.
• Faza 4. Rozumowanie LLM dotyczące oszustw i wizualizacja ścieżek: uruchom łańcuch rozumowania Gemini, aby ujawnić nielegalne pętle prania pieniędzy i wizualizować ścieżki w BigQuery Studio.

Czego potrzebujesz

• przeglądarka, np. Chrome;
• projekt Google Cloud z włączonymi płatnościami;

Te ćwiczenia są przeznaczone dla programistów, inżynierów danych i praktyków AI na wszystkich poziomach zaawansowania, w tym dla początkujących.

Szacowany czas trwania: 35 minut.
Szacowany koszt: poniżej 2 USD (korzysta z płatności według wykorzystania Vertex AI i przetwarzania zapytań BigQuery).

2. Zanim zaczniesz

Tworzenie projektu Google Cloud

1. W konsoli Google Cloud wybierz lub utwórz projekt w chmurze Google Cloud.
2. Sprawdź, czy w projekcie Cloud włączone są płatności.

Uruchamianie Cloud Shell

1. U góry konsoli Google Cloud kliknij Aktywuj Cloud Shell.
2. Potwierdź uwierzytelnianie:

gcloud auth list

3. Skonfiguruj zmienne środowiskowe w 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

Włącz interfejsy API

Aby włączyć wszystkie wymagane interfejsy API, uruchom to polecenie:

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

3. Konfiguracja i inicjowanie

W tym kroku skonfigurujemy środowisko Pythona, zainstalujemy wymagane biblioteki i zainicjujemy klientów BigQuery i Vertex AI. Te polecenia możesz uruchomić w Cloud Shell lub w środowisku notatnika Jupyter.

1. Utwórz i aktywuj środowisko wirtualne Pythona:

python3 -m venv venv
source venv/bin/activate

2. Zainstaluj wymagane pakiety Pythona:

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

3. Utwórz plik Pythona graphrag_aml.py i dodaj do niego kod inicjujący. Zastąp identyfikatorem projektu 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. Tworzenie tabel i schematu

Następnie definiujemy schemat wykresu finansowego, tworząc zbiór danych BigQuery i tabele standardowe.

1. Utwórz zbiór danych BigQuery:

bq mk --location=US --dataset fingraph_rag

2. Utwórz tabele. Możesz to zrobić w interfejsie BigQuery Studio lub w Cloud Shell. Oto kod 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. Wstawianie zbioru danych

Teraz wstawimy encje i ich relacje, aby utworzyć ślad pieniędzy. Ten zbiór danych przedstawia podejrzane działania między Doe (podejrzanym właścicielem spółki fasadowej), Jacoby (pośrednikiem), Menville (osobą, która nie przeszła weryfikacji tożsamości) i Smith (niewinnym obserwatorem).

Aby wypełnić tabele, uruchom ten kod 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.');

Weryfikowanie przetworzonych rekordów

Uruchom to zapytanie, aby sprawdzić liczbę rekordów w tabelach finansowych:

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

Powinny pojawić się dane wyjściowe zapytania potwierdzające wstawienie wiersza podobne do tych:

6. Tworzenie wykresu właściwości BigQuery

Po utworzeniu danych relacyjnych definiujemy FinGraph za pomocą natywnego języka DDL wykresów BigQuery. Tworzy to warstwę semantyczną na istniejących tabelach relacyjnych bez kopiowania ani duplikowania danych.

Podstawy składni GQL w ISO

Wykresy właściwości BigQuery korzystają ze standardowych wzorców języka zapytań GQL (Graph Query Language) zgodnych z normą ISO:

• (node:Label) definiuje węzły elementów (np. Account, Person, Loan).
• -[edge:LABEL]-> określa relacje kierunkowe (np. Transfers, Repays, Owns).

Aby utworzyć wykres właściwości, uruchom tę instrukcję 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)
 );

Paradygmat Zero-ETL: BigQuery CREATE PROPERTY GRAPH definiuje struktury grafów bezpośrednio w istniejących tabelach relacyjnych, eliminując niestabilne potoki ETL i duplikowanie danych.

Aby wizualizować cały graf kont, osób i pożyczek, uruchom w BigQuery Studio to zapytanie SQL:

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

Powinien pojawić się wynik wizualizacji wykresu podobny do tego:

7. Generowanie wektorów do logów kontrolnych

Aby włączyć część wyszukiwania wektorowego w naszym potoku RAG, generujemy osadzanie tekstu dla nieustrukturyzowanych dzienników kontrolnych bezpośrednio w BigQuery za pomocą AI.GENERATE_EMBEDDING funkcji zwracającej tabelę (TVF).

Tworzenie połączenia zdalnego z BigQuery i przyznawanie uprawnień IAM

BigQuery ML wymaga CLOUD_RESOURCE połączenia, aby bezpiecznie komunikować się z punktami końcowymi Vertex AI do tworzenia osadzania. Aby utworzyć połączenie, wykryć automatycznie wygenerowane konto usługi i przyznać rolę Użytkownik Vertex AI (roles/aiplatform.user), uruchom w Cloud Shell te polecenia w bashu:

# 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

Uwaga dotycząca promptów warunków uprawnień:
jeśli Twój projekt Google Cloud zawiera już powiązania warunkowe uprawnień, gcloud może interaktywnie poprosić Cię o wybranie opcji warunku (np. [1] ..., [2] ..., [3] None). Aby bezwarunkowo przyznać rolę roles/aiplatform.user w tym module, wybierz None (opcja 3). Dodanie --condition=None w powyższym poleceniu automatycznie pomija ten monit.

Tworzenie zdalnego modelu wektorów dystrybucyjnych

Następnie zdefiniuj model zdalny BigQuery ML, który łączy się z modelem text-embedding-005 Vertex AI za pomocą nowo autoryzowanego połączenia:

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

Opóźnienie w rozpowszechnianiu uprawnień IAM:
rozpowszechnienie nowo przyznanych ról IAM w usługach Google Cloud może potrwać od 1 do 3 minut. Jeśli utworzenie modelu lub wygenerowanie osadzania początkowo nie powiedzie się z powodu błędu uprawnień (np. The [e-mail] does not have the permission to access or use the endpoint...), poczekaj kilka minut na zsynchronizowanie powiązań IAM i ponów próbę wykonania zapytania.

Generowanie wektorów dystrybucyjnych

Teraz wygeneruj wektory dystrybucyjne dla tabeli AccountAudits, wywołując AI.GENERATE_EMBEDDING w klauzuli FROM instrukcji 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;

Sprawdzanie wygenerowanych wymiarów wektora

Aby sprawdzić, czy osadzanie wektorowe zostało wypełnione, uruchom to zapytanie:

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

Powinny pojawić się wyniki zapytania z 768-wymiarowymi wektorami osadzonymi podobnymi do tych:

Czym jest AI.GENERATE_EMBEDDING i wyszukiwanie wektorowe?
AI.GENERATE_EMBEDDING to funkcja zwracająca tabelę, którą należy wywołać w klauzuli FROM. Akceptuje zdalny model BigQuery ML MODEL (który uwierzytelnia się w Vertex AI za pomocą połączenia CLOUD_RESOURCE) i zapytanie wejściowe zawierające kolumnę content. Obliczenia związane z osadzaniem są przenoszone bezpośrednio do Vertex AI bez opuszczania BigQuery przez dane. COSINE_DISTANCE oblicza podobieństwo wektorowe między wektorami docelowymi a wektorami zapytania, aby wykrywać dopasowania semantyczne, które są pomijane przez sprawdzanie oparte na regułach.

8. Zdefiniuj moduł pobierania GraphRAG

Teraz utworzymy niestandardowy moduł pobierania LangChain w naszym środowisku Pythona. Ten moduł pobierania łączy semantyczne wyszukiwanie wektorowe (aby znaleźć odpowiednie punkty wyjścia) z natywnymi zapytaniami MATCH w grafie (aby przechodzić przez relacje).

Dodaj ten kod do skryptu w Pythonie 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

Zwróć uwagę, jak zapytanie o przechodzenie po grafie bez trudu łączy właścicieli kont, nadawców, przelewy i pożyczki za pomocą MATCH (sender_person:Person)-[:Owns]->... – nie wymaga to złożonych złączeń SQL.

Dynamiczna architektura wieloagentowa NL2SQL/GQL:
w zaawansowanych produkcyjnych systemach wieloagentowych zamiast korzystać z gotowych szablonów zapytań SQL/GQL w wyszukiwarce, możesz wdrożyć dedykowanego agenta NL2SQL / NL2GQL. Ten wyspecjalizowany agent, wyposażony w definicje schematu grafu i przykłady zapytań, dynamicznie generuje dostosowane zapytania GQL na podstawie głównych celów dochodzenia głównego agenta, przeprowadza testy i weryfikuje zapytania SQL w BigQuery oraz zwraca ukierunkowane wyniki grafu. Dynamiczne generowanie zapytań SQL przez agentów podkreśla pełną elastyczność przepływów pracy z wieloma agentami, ale w tym module używamy zapytań strukturalnych, aby zachować zwięzłość i jasność.

9. Przeprowadź analizę oszustwa

Na koniec uruchamiamy potok GraphRAG, aby wygenerować szczegółowy raport o oszustwach. Model LLM użyje kontekstu pobranego przez nasz niestandardowy moduł pobierania grafów, aby odpowiedzieć na prompta.

Wartość domeny: rozwiązanie problemu fałszywych alarmów w operacjach związanych z oszustwami
Tradycyjne monitorowanie AML oparte na regułach generuje rocznie miliony fałszywych alarmów, ponieważ reguły oceniają transakcje w izolacji (np. oznaczają przelew Smitha do Menville na kwotę 150 USD). GraphRAG łączy kontekst semantycznego logu kontrolnego (Menville nie przeszedł weryfikacji tożsamości) z topologią wykresu wielokrotnych połączeń (Doe → Jacoby → Menville), co umożliwia LLM odfiltrowanie nieistotnych szumów i wykrywanie pętli prania pieniędzy o wysokim ryzyku.

Dodaj ten kod do skryptu graphrag_aml.py i uruchom go za pomocą polecenia 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))

Ukończ skrypt graphrag_aml.py

Gotowy skrypt graphrag_aml.py powinien wyglądać tak:

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

Powinny wyświetlić się dane wyjściowe podobne do tego przykładowego raportu z analizy LLM:

10. Wizualizacja ścieżki prania pieniędzy

Aby wizualnie zrozumieć ślad prania pieniędzy, który właśnie odkryliśmy programowo, możesz uruchomić zapytanie wizualizujące wykres w konsoli BigQuery Studio.

Uruchom to zapytanie w BigQuery Studio. (Upewnij się, że funkcja wizualizacji grafu jest włączona, lub kliknij kartę Wykres, jeśli jest dostępna).

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;

To zapytanie GQL śledzi całą ścieżkę od podejrzanego właściciela spółki fasadowej (Doe) przez pośrednika (Jacoby) do ostatecznego celu (Menville) i spłaty pożyczki.

Powinien pojawić się wynik wizualizacji wykresu podobny do tego:

11. Czyszczenie danych

Aby uniknąć obciążenia konta Google Cloud bieżącymi opłatami, usuń zasoby utworzone podczas tego ćwiczenia.

Usuń zbiór danych BigQuery i połączenie z zasobem 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

Sprawdź, czy zasoby zostały usunięte:

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

12. Gratulacje

Gratulacje! Udało Ci się utworzyć aplikację RAG i przeanalizować jej działanie. Pokazujesz, jak używać natywnych funkcji BigQuery do wyszukiwania grafów i wyszukiwania wektorowego w celu przeprowadzania GraphRAG, wykrywając schemat prania pieniędzy bez ETL.

Czego się nauczysz

• Jak utworzyć wykres właściwości w BigQuery na podstawie standardowych tabel
• Generowanie i przechowywanie wektorów dystrybucyjnych za pomocą BigQuery ML
• Jak połączyć przechodzenie po grafach BigQuery i wyszukiwanie wektorowe w narzędziu do pobierania LangChain
• Jak duże modele językowe mogą syntetyzować semantyczne dzienniki kontrolne z topologią grafu, aby zmniejszyć liczbę wyników fałszywie pozytywnych

Co dalej

• Zapoznaj się z artykułem Omówienie wykresów właściwości BigQuery
• Wypróbuj więcej wzorców języka zapytań Graph Query Language (GQL) w BigQuery.

Dokumentacja

• Funkcja BigQuery ML AI.GENERATE_EMBEDDING
• Wyszukiwanie wektorowe i funkcje odległości w BigQuery
• Integracje LangChain z usługami Google