Противодействие отмыванию денег и предотвращение мошенничества с помощью BigQuery GraphRAG

1. Введение

В этом практическом занятии вы создадите решение на основе алгоритма GraphRAG (Graph Retrieval-Augmented Generation) для обнаружения противодействия отмыванию денег (AML) и финансовому мошенничеству. Вы будете использовать Vertex AI, Vector Search и встроенные возможности BigQuery для работы с графами, скоординированные с помощью LangChain. К концу занятия вы увидите, как модель обработки больших языковых данных (LLM) может выявлять незаконные операции с денежными средствами путем синтеза семантических журналов аудита и сложных транзакционных сетей.

Архитектурный поток GraphRAG

+------------------+     1. Vector Search      +---------------------+
| User Prompt /    | ------------------------> | BigQuery ML         |
| Investigation    |                           | (AccountAudits)     |
+------------------+                           +---------------------+
         |                                                |
         |                                                | 2. Seed Entity ID
         v                                                v
+------------------+     3. GQL Traversal      +---------------------+
| LangChain        | <------------------------ | BigQuery Property   |
| Graph Retriever  |                           | Graph (FinGraph)    |
+------------------+                           +---------------------+
         |
         | 4. Synthesized Context
         v
+------------------+
| Gemini 2.5 Flash | ---> Detailed Fraud Report
+------------------+

Почему GraphRAG? Стандартный RAG извлекает отдельные фрагменты текста, используя семантическое сходство, но не учитывает многоступенчатые связи. GraphRAG использует векторный поиск для поиска начальных сущностей, а затем использует BigQuery GQL для обхода многошаговых денежных потоков (например, Владелец → Счет → Перевод → Счет → Заем ).

Что вы будете делать

• Этап 1: Настройка набора данных и графа свойств : Создание реляционных финансовых таблиц и построение собственного PROPERTY GRAPH BigQuery.
• Этап 2: Генерация семантических векторных представлений : Генерация текстовых представлений непосредственно в SQL для журналов аудита с использованием AI.GENERATE_EMBEDDING ( text-embedding-005 ).
• Этап 3: Создание пользовательского средства поиска GraphRAG на основе LangChain : разработка пользовательского средства поиска на Python, объединяющего векторное сходство ( COSINE_DISTANCE ) и обход путей в формате ISO GQL.
• Этап 4: Анализ мошеннических схем и визуализация путей их распространения в рамках программы LLM : Выполнить цепочку логических рассуждений Gemini для выявления незаконных путей отмывания денег и визуализации этих путей в BigQuery Studio.

Что вам понадобится

• Веб-браузер, например Chrome .
• Проект Google Cloud с включенной функцией выставления счетов.

Этот практический семинар предназначен для разработчиков, инженеров данных и специалистов по искусственному интеллекту всех уровней, включая начинающих.

Примерное время: 35 минут
Ориентировочная стоимость: менее 2,00 долларов США (используется система оплаты по мере использования Vertex AI и обработка запросов BigQuery).

2. Прежде чем начать

Создайте проект в Google Cloud.

1. В консоли Google Cloud выберите или создайте проект Google Cloud .
2. Убедитесь, что для вашего облачного проекта включена функция выставления счетов.

Запустить Cloud Shell

1. В верхней части консоли Google Cloud нажмите кнопку «Активировать Cloud Shell» .
2. Проверка подлинности:

gcloud auth list

3. Настройка переменных среды в Cloud Shell:

export GCP_PROJECT=$(gcloud config get-value project)
export REGION="us-central1"
export BQ_DATASET="fingraph_rag"
gcloud config set project $GCP_PROJECT

Включить API

Выполните эту команду, чтобы включить все необходимые API:

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

Создание резервирования и назначения в BigQuery

Для выполнения GQL-запросов необходимо иметь резервирование, использующее версию Enterprise или Enterprise Plus. Создайте резервирование версии Enterprise с автомасштабированием в Cloud Shell:

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

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

Экономически эффективное автоматическое масштабирование вычислительных ресурсов :
При настройке --slots=0 (0 базовых слотов) и --autoscale_max_slots=100 , резервирование слотов в BigQuery не влечет за собой базовых затрат в режиме ожидания. BigQuery автоматически увеличивает количество выделенных слотов до 100 только во время выполнения запросов или векторного поиска, а затем уменьшает их до 0 после завершения.

3. Настройка и инициализация

На этом этапе мы настроим среду Python, установим необходимые библиотеки и инициализируем клиенты BigQuery и Vertex AI. Вы можете выполнить эти команды в Cloud Shell или в среде Jupyter Notebook.

1. Создайте и активируйте виртуальную среду Python:

python3 -m venv venv
source venv/bin/activate

2. Установите необходимые пакеты Python:

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

3. Создайте файл 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 и стандартные таблицы.

1. Создайте набор данных BigQuery:

bq mk --location=US --dataset fingraph_rag

2. Создайте таблицы. Вы можете выполнить это в пользовательском интерфейсе BigQuery Studio или через Cloud Shell. Вот SQL-запрос:

CREATE TABLE IF NOT EXISTS `fingraph_rag.Account` (id INT64, create_time TIMESTAMP, is_blocked BOOL, type STRING);
CREATE TABLE IF NOT EXISTS `fingraph_rag.Loan` (id INT64, loan_amount FLOAT64, balance FLOAT64, create_time TIMESTAMP, interest_rate FLOAT64);
CREATE TABLE IF NOT EXISTS `fingraph_rag.Person` (id INT64, name STRING);
CREATE TABLE IF NOT EXISTS `fingraph_rag.AccountRepayLoan` (id INT64, loan_id INT64, amount FLOAT64, create_time TIMESTAMP);
CREATE TABLE IF NOT EXISTS `fingraph_rag.AccountTransferAccount` (id INT64, to_id INT64, amount FLOAT64, create_time TIMESTAMP);
CREATE TABLE IF NOT EXISTS `fingraph_rag.PersonOwnAccount` (id INT64, account_id INT64, create_time TIMESTAMP);
CREATE TABLE IF NOT EXISTS `fingraph_rag.AccountAudits` (id INT64, audit_timestamp TIMESTAMP, audit_details STRING, embedding ARRAY<FLOAT64>);

5. Вставьте набор данных.

Теперь мы добавим сущности и их взаимосвязи, чтобы сформировать наш денежный след. Этот набор данных отражает подозрительную деятельность между Доу (предполагаемый владелец подставной компании), Джейкоби (посредник), Менвиллем (объект, не прошедший процедуру KYC) и Смитом (невинный свидетель).

Выполните следующий SQL-запрос для заполнения таблиц:

INSERT INTO `fingraph_rag.Account` VALUES 
  (10,'2020-01-10 06:22:20.222',false,'brokerage account'), 
  (20,'2020-01-27 17:55:09.206',false,'checking account'), 
  (30,'2020-02-15 09:12:33.111',false,'savings account'), 
  (40,'2019-11-05 14:33:10.000',false,'business account');

INSERT INTO `fingraph_rag.Loan` VALUES 
  (100,2022278.5,123359.0,'2020-03-18 16:42:57.719',0.064), 
  (200,50000.0,45000.0,'2020-03-23 19:03:05.567',0.097), 
  (300, 15000.0, 10000.0, '2020-05-10 10:00:00.000', 0.05);

INSERT INTO `fingraph_rag.Person` VALUES 
  (1,'Jacoby'), (2,'Menville'), (3,'Smith'), (4,'Doe');

INSERT INTO `fingraph_rag.AccountTransferAccount` VALUES 
  (40,10,25000.0,'2020-08-01 10:00:00.000'), 
  (10,20,24000.0,'2020-08-29 15:28:58.647'), 
  (30,20,150.0,'2020-09-01 12:00:00.000');

INSERT INTO `fingraph_rag.AccountRepayLoan` VALUES 
  (10,100,56809.8,'2020-12-12 07:25:02.597'), 
  (20,200,20000.0,'2021-01-18 01:40:25.317');

INSERT INTO `fingraph_rag.PersonOwnAccount` VALUES 
  (1,10,'2020-01-10 06:22:20.222'), (2,20,'2020-01-27 17:55:09.206'), 
  (3,30,'2020-02-15 09:12:33.111'), (4,40,'2019-11-05 14:33:10.000');

INSERT INTO `fingraph_rag.AccountAudits` (id, audit_timestamp, audit_details) VALUES 
  (10, '2020-05-14 06:57:02', 'Account 10 (Jacoby) flagged by AML system for suspicious high-volume transfers from offshore business accounts.'), 
  (20, '2021-03-09 02:51:45', 'Account 20 (Menville) failed KYC verification. Linked source of funds is unverified and customer is unresponsive.'), 
  (40, '2020-07-20 09:00:00', 'Account 40 (Doe) under investigation as a suspected shell company involved in illicit activities.');

Проверьте введенные записи.

Выполните этот запрос, чтобы проверить количество записей в ваших финансовых таблицах:

SELECT 'Account' AS entity_table, COUNT(*) AS row_count FROM `fingraph_rag.Account`
UNION ALL SELECT 'Loan', COUNT(*) FROM `fingraph_rag.Loan`
UNION ALL SELECT 'Person', COUNT(*) FROM `fingraph_rag.Person`
UNION ALL SELECT 'AccountAudits', COUNT(*) FROM `fingraph_rag.AccountAudits`;

В результате выполнения запроса, подтверждающего вставку строки, вы должны увидеть примерно следующее:

6. Создайте граф свойств BigQuery.

Имея на руках реляционные данные, мы определяем FinGraph , используя встроенный в BigQuery Graph DDL. Это создает семантический слой поверх существующих реляционных таблиц без копирования или дублирования данных.

Основы синтаксиса ISO GQL

Графы свойств BigQuery используют стандартные шаблоны языка запросов к графам ISO (GQL):

• (node:Label) определяет узлы сущностей (например, Account , Person , Loan ).
• -[edge:LABEL]-> определяет направленные отношения (например, Transfers , Repays , Owns ).

Выполните следующий SQL-запрос для создания графа свойств:

CREATE OR REPLACE PROPERTY GRAPH `fingraph_rag.FinGraph`
 NODE TABLES (
   `fingraph_rag.Account` KEY (id) LABEL Account PROPERTIES (id, type, is_blocked),
   `fingraph_rag.Loan` KEY (id) LABEL Loan PROPERTIES (id, loan_amount, balance),
   `fingraph_rag.Person` KEY (id) LABEL Person PROPERTIES (id, name)
 )
 EDGE TABLES(
   `fingraph_rag.AccountRepayLoan`
     KEY (id, loan_id, create_time)
     SOURCE KEY (id) REFERENCES `fingraph_rag.Account` (id)
     DESTINATION KEY (loan_id) REFERENCES `fingraph_rag.Loan` (id)
     LABEL Repays PROPERTIES (amount, create_time),
   `fingraph_rag.AccountTransferAccount`
     KEY (id, to_id, create_time)
     SOURCE KEY (id) REFERENCES `fingraph_rag.Account` (id)
     DESTINATION KEY (to_id) REFERENCES `fingraph_rag.Account` (id)
     LABEL Transfers PROPERTIES (amount, create_time),
   `fingraph_rag.PersonOwnAccount`
     KEY (id, account_id)
     SOURCE KEY (id) REFERENCES `fingraph_rag.Person` (id)
     DESTINATION KEY (account_id) REFERENCES `fingraph_rag.Account` (id)
     LABEL Owns PROPERTIES (create_time)
 );

Парадигма Zero-ETL: функция CREATE PROPERTY GRAPH в BigQuery определяет структуры графов непосредственно в существующих реляционных таблицах, устраняя ненадежные конвейеры ETL и дублирование данных.

Для визуализации всей графической структуры счетов, лиц и займов выполните следующий SQL-запрос в BigQuery Studio:

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

В результате визуализации графика вы должны увидеть изображение, похожее на это:

7. Создание встраиваний для журналов аудита.

Для активации части нашего конвейера RAG, отвечающей за векторный поиск, мы генерируем текстовые встраивания для неструктурированных журналов аудита непосредственно в BigQuery, используя табличную функцию AI.GENERATE_EMBEDDING (TVF).

Создайте удаленное подключение к BigQuery и предоставьте разрешения IAM.

Для безопасной связи с конечными точками встраивания Vertex AI ML требуется соединение CLOUD_RESOURCE . Выполните следующие команды 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

Примечание к подсказкам условий IAM:
Если ваш проект Google Cloud уже содержит условные привязки IAM, gcloud может в интерактивном режиме предложить вам выбрать вариант условия (например, [1] ... , [2] ... , [3] None ). Выберите None (вариант 3 ), чтобы предоставить роль 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');

Задержка распространения разрешений IAM:
Внедрение новых ролей IAM в сервисы Google Cloud может занять от 1 до 3 минут. Если создание модели или генерация встраивания изначально завершается ошибкой доступа (например, The [e-mail] does not have the permission to access or use the endpoint... "), подождите несколько минут, пока синхронизируются привязки IAM, и повторите запрос.

Сгенерировать векторные представления

Теперь сгенерируйте векторные представления для таблицы 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 . Она принимает удаленную MODEL BigQuery ML (которая аутентифицируется в Vertex AI через соединение CLOUD_RESOURCE ) и входной запрос, содержащий столбец content . Вычисления для встраивания передаются непосредственно в Vertex AI без выхода данных из BigQuery. COSINE_DISTANCE вычисляет сходство векторов между целевыми встраиваниями и векторами запроса для выявления семантических совпадений, которые пропускают проверки на основе правил.

8. Определите средство извлечения данных GraphRAG.

Теперь мы создадим собственный ретривер LangChain в нашей среде Python. Этот ретривер сочетает семантический векторный поиск (для нахождения релевантных начальных точек) с нативными запросами Graph 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

Обратите внимание, как запрос обхода графа без труда объединяет владельцев учетных записей, отправителей, переводы и займы с помощью MATCH (sender_person:Person)-[:Owns]->... — без необходимости сложных SQL-соединений!

Динамическая многоагентная архитектура NL2SQL/GQL:
В современных многоагентных системах, используемых в производственной среде, вместо предопределенных шаблонов SQL/GQL-запросов в рекрутере можно развернуть выделенный агент NL2SQL/NL2GQL . Этот специализированный агент, обладающий необходимыми навыками, генерирует индивидуальные GQL-запросы на основе основных целей исследования главного агента, выполняет пробные запуски и проверку SQL-запросов в BigQuery, а также возвращает целевые результаты для графа. Хотя динамическая генерация SQL-запросов агентом демонстрирует всю гибкость многоагентных рабочих процессов, в этой лабораторной работе для краткости и ясности мы используем структурированные запросы.

9. Проведите расследование мошенничества.

Наконец, мы запускаем конвейер GraphRAG для генерации подробного отчета о мошенничестве. LLM будет использовать контекст, полученный с помощью нашего пользовательского средства получения графов, для ответа на запрос.

Ценность домена: решение проблемы ложных срабатываний в мошеннических операциях.
Традиционный мониторинг противодействия отмыванию денег на основе правил ежегодно выявляет миллионы ложных срабатываний, поскольку правила оценивают транзакции изолированно (например, отмечают перевод Смитом 150 долларов в Менвилль). GraphRAG объединяет семантический контекст журнала аудита (Менвилль не прошел процедуру KYC) с многошаговой топологией графа (Доу → Джейкоби → Менвилль), позволяя системам мониторинга отмывания денег отфильтровывать безобидный шум и выявлять высокорискованные схемы отмывания денег.

Добавьте следующий код в свой скрипт graphrag_aml.py и запустите его с помощью python graphrag_aml.py :

from langchain_google_vertexai import ChatVertexAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Initialize the LLM and the Retriever
llm = ChatVertexAI(model_name=MODEL_NAME)
retriever = FinGraphRetriever(project=GCP_PROJECT_ID, dataset=BQ_DATASET_ID)

# Define the Prompt
prompt = ChatPromptTemplate.from_template("""
You are a Lead Fraud Analyst. Use the following audit logs and graph transaction history to answer the question.
Your goal is to connect the dots between the entities and explain the flow of funds.
If you see transfers from flagged users or shell companies, highlight the money laundering risk.

Context: {context}

Question: {question}

Detailed Fraud Report:
""")

# Create the LangChain
chain = (
    {"context": retriever , "question": lambda x: x}
    | prompt
    | llm
    | StrOutputParser()
)

# Execute the chain
question = "Why is Menville's loan repayment at risk? Flag any suspicious activity if you notice."
print(chain.invoke(question))

Полный скрипт graphrag_aml.py

Для справки, ваш полный скрипт graphrag_aml.py должен выглядеть следующим образом:

import vertexai
from google.cloud import bigquery
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from typing import List
from langchain_google_vertexai import ChatVertexAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

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

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

class FinGraphRetriever(BaseRetriever):
    project: str
    dataset: str

    def _get_relevant_documents(self, query: str) -> List[Document]:
        # 1. Vector Search using Cosine Distance
        vector_query = f"""
            SELECT id, audit_details
            FROM `{self.dataset}.AccountAudits`
            ORDER BY COSINE_DISTANCE(
                embedding,
                (
                    SELECT embedding
                    FROM AI.GENERATE_EMBEDDING(
                        MODEL `{self.dataset}.embedding_model`,
                        (SELECT @query AS content)
                    )
                )
            )
            LIMIT 1
        """
        res = bq_client.query(vector_query, job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ScalarQueryParameter("query", "STRING", query)]
        )).result()

        start_id = None
        audit_text = ""
        for row in res:
            start_id = row.id
            audit_text = row.audit_details

        if not start_id: return []

        # 2. Native Graph Traversal (GQL MATCH)
        graph_query = f"""
            GRAPH `{self.dataset}.FinGraph`
            MATCH
              (sender_person:Person)-[:Owns]->(sender_acc:Account)
              -[tx:Transfers]->
              (a:Account)
              -[repays:Repays]->(l:Loan),
              (owner:Person)-[:Owns]->(a)
            WHERE a.id = @id
            RETURN
              owner.name as owner_name,
              a.type as account_type,
              sender_person.name as sender_name,
              tx.amount as transfer_amount,
              repays.amount as repayment_amount,
              l.id as loan_id
        """
        graph_res = bq_client.query(graph_query, job_config=bigquery.QueryJobConfig(
            query_parameters=[bigquery.ScalarQueryParameter("id", "INT64", start_id)]
        )).result()

        context_docs = [Document(page_content=f"Primary Audit Log (Target Account): {audit_text}")]
        sender_names = []
        for row in graph_res:
            sender_names.append(row['sender_name'])
            doc_str = (f"Account Owner: {row['owner_name']} (Account Type: {row['account_type']}). "
                       f"Received transfer of ${row['transfer_amount']} from {row['sender_name']}. "
                       f"Made loan repayment of ${row['repayment_amount']} to Loan {row['loan_id']}.")
            context_docs.append(Document(page_content=doc_str))

        if sender_names:
            names_list = "','".join(sender_names)
            sender_audit_query = f"""
                SELECT p.name, au.audit_details
                FROM `{self.dataset}.AccountAudits` au
                JOIN `{self.dataset}.Account` a ON au.id = a.id
                JOIN `{self.dataset}.PersonOwnAccount` poa ON a.id = poa.account_id
                JOIN `{self.dataset}.Person` p ON poa.id = p.id
                WHERE p.name IN ('{names_list}')
            """
            sender_audits = bq_client.query(sender_audit_query).result()
            for row in sender_audits:
                context_docs.append(Document(page_content=f"Audit Log for Sender {row['name']}: {row['audit_details']}"))

        return context_docs

# Initialize LLM & Retriever
llm = ChatVertexAI(model_name=MODEL_NAME)
retriever = FinGraphRetriever(project=GCP_PROJECT_ID, dataset=BQ_DATASET_ID)

prompt = ChatPromptTemplate.from_template("""
You are a Lead Fraud Analyst. Use the following audit logs and graph transaction history to answer the question.
Your goal is to connect the dots between the entities and explain the flow of funds.
If you see transfers from flagged users or shell companies, highlight the money laundering risk.

Context: {context}

Question: {question}

Detailed Fraud Report:
""")

chain = (
    {"context": retriever, "question": lambda x: x}
    | prompt
    | llm
    | StrOutputParser()
)

question = "Why is Menville's loan repayment at risk? Flag any suspicious activity if you notice."
print(chain.invoke(question))

В результате вы должны увидеть данные, аналогичные приведенному в этом примере отчета об анализе LLM:

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 ) и погашения кредита .

В результате визуализации графика вы должны увидеть изображение, похожее на это:

11. Уборка

Чтобы избежать дальнейших списаний средств с вашего аккаунта Google Cloud, удалите ресурсы, созданные в ходе этого практического занятия.

Удалите набор данных BigQuery и подключение к облачным ресурсам:

# Delete the BigQuery dataset
bq rm -r -f $PROJECT_ID:fingraph_rag

# Delete the BigQuery Cloud Resource Connection
bq rm --connection --location=us vertex_ai_conn

Убедитесь, что ресурсы были удалены:

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

12. Поздравляем!

Поздравляем! Вы успешно создали RAG-приложение и проанализировали его поведение. Вы продемонстрировали, как использовать встроенные возможности BigQuery по поиску по графам и векторам для выполнения GraphRAG, обнаружив схему противодействия отмыванию денег с нулевым объемом ETL-процессов.

Что вы узнали

• Как построить граф свойств в BigQuery на основе стандартных таблиц
• Как создавать и хранить векторные представления с помощью BigQuery ML
• Как объединить обход графов BigQuery и векторный поиск в средство поиска на основе цепочек языков (LangChain Retriever)
• Как LLM-ы могут синтезировать семантические журналы аудита с учетом топологии графа для уменьшения количества ложных срабатываний

Следующие шаги

• Изучите обзор графов свойств BigQuery.
• Попробуйте другие шаблоны языка запросов BigQuery Graph Query Language (GQL) .

Справочная документация

• Функция BigQuery ML AI.GENERATE_EMBEDDING
• Функции векторного поиска и расстояния в BigQuery
• Интеграции LangChain с Google