利用 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 检索器：构建一个自定义 Python 检索器，将向量相似度 (COSINE_DISTANCE) 和 ISO GQL 路径遍历相结合。
• 第 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 的原生 Graph 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 流水线的向量搜索部分，我们使用 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 JOIN！

动态多代理 NL2SQL/GQL 架构：
在高级生产多代理系统中，您可以部署专用 NL2SQL/NL2GQL 代理，而不是在检索器中使用预定义的 SQL / GQL 查询模板。该专业代理加载了图表架构定义和查询示例作为技能，可根据主代理的主要调查目标动态生成量身定制的 GQL 查询，针对 BigQuery 进行 SQL 试运行和验证，并返回目标图表结果。虽然动态代理 SQL 生成功能充分展现了多代理工作流的灵活性，但为了简洁明了，我们在本实验中使用了结构化查询。

9. 运行欺诈调查

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

领域价值：解决欺诈运营中的假正例问题
传统的基于规则的 AML 监控每年会标记数百万个假正例，因为规则会单独评估交易（例如，标记 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) 的整个路径，以及 Loan 还款。

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

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 图遍历和向量搜索功能组合到 LangChain Retriever 中
• LLM 如何通过图拓扑结构合成语义审核日志来减少假正例

后续步骤

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

参考文档

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