BigQuery GraphRAG를 사용한 자금 세탁 방지 및 사기 방지

1. 소개

이 Codelab에서는 자금세탁 방지 (AML) 및 금융 사기를 감지하는 그래프 검색 증강 생성 (GraphRAG) 솔루션을 빌드합니다. LangChain을 통해 조정된 Vertex AI, 벡터 검색, BigQuery의 기본 그래프 기능을 사용합니다. 이 실습을 마치면 대규모 언어 모델 (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단계: 데이터 세트 및 속성 그래프 설정: 관계형 금융 테이블을 만들고 네이티브 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달러 미만 (사용한 만큼만 지불하는 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 UI 또는 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 파이프라인의 벡터 검색 부분을 사용 설정하기 위해 AI.GENERATE_EMBEDDING 테이블 값 함수 (TVF)를 사용하여 BigQuery 내에서 직접 구조화되지 않은 감사 로그의 텍스트 임베딩을 생성합니다.

BigQuery 원격 연결 만들기 및 IAM 권한 부여

BigQuery ML은 Vertex AI 임베딩 엔드포인트와 안전하게 통신하기 위해 CLOUD_RESOURCE 연결이 필요합니다. Cloud Shell에서 다음 bash 명령어를 실행하여 연결을 만들고, 자동 생성된 서비스 계정을 검색하고, 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)을 선택하라는 메시지를 대화형으로 표시할 수 있습니다. 이 실습에서 roles/aiplatform.user 역할을 무조건 부여하려면 None (옵션 3)을 선택하세요. 위의 명령어에 --condition=None를 추가하면 이 프롬프트가 자동으로 우회됩니다.

원격 임베딩 모델 만들기

다음으로, 새로 승인된 연결을 통해 Vertex AI의 text-embedding-005 모델에 연결되는 BigQuery ML 원격 모델을 정의합니다.

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 바인딩이 동기화될 때까지 몇 분 정도 기다린 후 쿼리를 다시 시도하세요.

임베딩 생성

이제 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 및 벡터 검색이란 무엇인가요?
AI.GENERATE_EMBEDDING은 FROM 절에서 호출해야 하는 테이블 값 함수 (TVF)입니다. MODEL (CLOUD_RESOURCE 연결을 통해 Vertex AI에 인증됨) 및 content 열이 포함된 입력 쿼리를 허용합니다. 컴퓨팅 삽입은 BigQuery를 벗어나지 않고 Vertex AI로 직접 오프로드됩니다. COSINE_DISTANCE는 타겟 임베딩과 쿼리 벡터 간의 벡터 유사성을 계산하여 규칙 기반 검사에서 놓친 의미적 일치를 포착합니다.

8. GraphRAG 검색기 정의

이제 Python 환경에서 맞춤 LangChain 리트리버를 만듭니다. 이 리트리버는 시맨틱 벡터 검색 (관련 시작점 찾기)과 네이티브 그래프 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

그래프 순회 쿼리가 복잡한 SQL JOIN 없이 MATCH (sender_person:Person)-[:Owns]->...를 통해 계정 소유자, 발신자, 이체, 대출을 손쉽게 가져오는 방법을 확인하세요.

동적 멀티 에이전트 NL2SQL/GQL 아키텍처:
고급 프로덕션 멀티 에이전트 시스템에서는 리트리버에서 사전 정의된 SQL/GQL 쿼리 템플릿을 사용하는 대신 전용 NL2SQL / NL2GQL 에이전트를 배포할 수 있습니다. 그래프 스키마 정의와 쿼리 예시를 기술로 로드한 이 전문 에이전트는 기본 에이전트의 기본 조사 목표에 따라 맞춤 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의 기본 그래프 및 벡터 검색 기능을 사용하여 ETL 없이 자금 세탁 방지 계획을 감지하는 GraphRAG를 수행하는 방법을 보여주었습니다.

학습한 내용

• 표준 테이블을 기반으로 BigQuery에서 속성 그래프를 빌드하는 방법
• BigQuery ML을 사용하여 벡터 임베딩을 생성하고 저장하는 방법
• BigQuery 그래프 순회와 벡터 검색을 LangChain 리트리버로 결합하는 방법
• LLM이 그래프 토폴로지로 시맨틱 감사 로그를 합성하여 거짓양성을 줄이는 방법

다음 단계

• BigQuery 속성 그래프 개요 살펴보기
• BigQuery Graph Query Language (GQL) 패턴을 더 많이 사용해 보세요.

참조 문서

• BigQuery ML AI.GENERATE_EMBEDDING 함수
• BigQuery 벡터 검색 및 거리 함수
• LangChain Google 통합