利用 BigQuery GraphRAG 防范洗钱和欺诈行为

1. 简介

在此 Codelab 中，您将构建一个图检索增强生成 (GraphRAG) 解决方案，用于检测反洗钱 (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 使用 Vector Search 查找种子实体，然后使用 BigQuery GQL 遍历多跃点资金流向（例如，所有者 → 账号 → 转账 → 账号 → 贷款）。

您将执行的操作

• 第 1 阶段：数据集和属性图设置：创建关系型财务表并构建原生 BigQuery PROPERTY GRAPH。
• 第 2 阶段：生成语义向量嵌入：使用 AI.GENERATE_EMBEDDING (text-embedding-005) 直接在 SQL 中为审核日志生成文本嵌入。
• 第 3 阶段：自定义 LangChain GraphRAG 检索器：构建一个结合了向量相似度 (COSINE_DISTANCE) 和 ISO GQL 路径遍历的自定义 Python 检索器。
• 第 4 阶段：LLM 欺诈推理和轨迹可视化：执行 Gemini 推理链，揭露非法洗钱循环，并在 BigQuery Studio 中直观呈现路径轨迹。

所需条件

• 网络浏览器，例如 Chrome。
• 启用了结算功能的 Google Cloud 项目。

本 Codelab 专为各种水平的开发者、数据工程师和 AI 从业者（包括新手）而设计。

预计时长：35 分钟
预计费用：低于 2.00 美元（使用随用随付 Vertex AI 和 BigQuery 查询处理）。

2. 准备工作

创建 Google Cloud 项目

1. 在 Google Cloud 控制台中，选择或创建 Google Cloud 项目。
2. 确保您的 Cloud 项目已启用结算功能。

启动 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

3. 设置和初始化

在此步骤中，我们将设置 Python 环境、安装所需的库，并初始化 BigQuery 和 Vertex AI 客户端。您可以在 Cloud Shell 或 Jupyter 笔记本环境中运行这些命令。

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 项目 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. 创建表和架构

接下来，我们通过创建 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. 插入数据集

现在，我们将插入实体及其关系，以形成资金流向。此数据集表示 Doe（疑似空壳公司所有者）、Jacoby（中介）、Menville（未通过 KYC 的目标）和 Smith（无辜旁观者）之间的可疑活动。

运行以下 SQL 以填充表：

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

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

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

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

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

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

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

验证提取的记录

运行以下查询，验证财务表中的记录数：

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

您应该会看到确认行插入的查询输出，如下所示：

6. 创建 BigQuery 属性图

关系型数据准备就绪后，我们使用 BigQuery 的原生图 DDL 定义 FinGraph。这会在现有关系型表上创建一个语义层，而无需复制或重复数据。

ISO GQL 语法入门

BigQuery 属性图使用标准 ISO Graph Query Language (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)
 );

零 ETL 范式：BigQuery 的 CREATE PROPERTY GRAPH 直接在现有关系型表上定义图结构，从而消除了脆弱的 ETL 流水线和数据重复。

如需直观呈现由账号、人员和贷款构成的整个图，请在 BigQuery Studio 中运行以下 SQL 查询：

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

您应该会看到类似如下所示的图表可视化结果：

7. 为审核日志生成嵌入

为了启用 RAG 流水线的 Vector Search 部分，我们使用 AI.GENERATE_EMBEDDING 表值函数 (TVF) 直接在 BigQuery 中为非结构化审核日志生成文本嵌入。

创建 BigQuery 远程连接并授予 IAM 权限

BigQuery ML 需要 CLOUD_RESOURCE 连接才能与 Vertex AI 嵌入端点安全通信。在 Cloud Shell 中运行以下 bash 命令，以创建连接、发现其自动生成的服务账号，并授予 Vertex AI User (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 远程模型，通过您新获授权的连接关联到 Vertex AI 的 text-embedding-005 模型：

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

IAM 权限传播延迟：
新授予的 IAM 角色可能需要 1 到 3 分钟才能在 Google Cloud 服务中生效。如果模型创建或嵌入生成最初因权限错误（例如 The [e-mail] does not have the permission to access or use the endpoint...）而失败，请等待几分钟，让 IAM 绑定同步，然后重试查询。

生成嵌入

现在，通过在 UPDATE 语句的 FROM 子句中调用 AI.GENERATE_EMBEDDING，为 AccountAudits 表生成嵌入：

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 和 Vector Search？
AI.GENERATE_EMBEDDING 是一个表值函数 (TVF)，必须在 FROM 子句中调用。它接受 BigQuery ML 远程MODEL（通过 CLOUD_RESOURCE 连接向 Vertex AI 进行身份验证）和包含 content 列的输入查询。嵌入计算直接卸载到 Vertex AI，数据无需离开 BigQuery。COSINE_DISTANCE 会计算目标嵌入和查询向量之间的向量相似度，以捕获基于规则的检查所遗漏的语义匹配。

8. 定义 GraphRAG 检索器

现在，我们将在 Python 环境中创建一个自定义 LangChain 检索器。此检索器将语义 Vector Search（用于查找相关起点）与原生图 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 查询，针对 BigQuery 试运行并验证 SQL，然后返回有针对性的图表结果。虽然动态 Agentic Workflows SQL 生成功能充分展示了多智能体工作流的灵活性，但为了简洁明了，我们在本实验中使用结构化查询。

9. 运行欺诈调查

最后，我们运行 GraphRAG 流水线，生成详细的欺诈报告。LLM 将使用自定义图检索器检索到的上下文来回答提示。

领域价值：解决欺诈运营中的假正例问题
传统的基于规则的反洗钱监控每年会标记数百万个假正例，因为规则会孤立地评估交易（例如，标记 Smith 向 Menville 转账 150 美元的交易）。GraphRAG 将语义审核日志上下文（Menville 未通过 KYC）与多跃点图拓扑（Doe → Jacoby → Menville）相结合，使 LLM 能够滤除良性噪声，同时发现高风险洗钱循环。

将以下代码添加到脚本 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 账号持续收取费用，请删除在此 Codelab 期间创建的资源。

删除 BigQuery 数据集和 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

验证资源是否已删除：

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

12. 恭喜

恭喜！您已成功构建 RAG 应用并分析了其行为。您演示了如何使用 BigQuery 的原生图和 Vector Search 功能执行 GraphRAG，在无需 ETL 的情况下检测反洗钱计划。

您学到的内容

• 如何在 BigQuery 中基于标准表构建属性图
• 如何使用 BigQuery ML 生成和存储向量嵌入
• 如何将 BigQuery 图遍历和 Vector Search 组合到 LangChain 检索器中
• LLM 如何将语义审核日志与图表拓扑相结合，以减少假正例

后续步骤

• 探索 BigQuery 属性图概览
• 尝试更多 BigQuery Graph Query Language (GQL) 模式。

参考文档

• BigQuery ML AI.GENERATE_EMBEDDING 函数
• BigQuery Vector Search 和距离函数
• LangChain Google 集成