[draft] adk example

examples/java/adk/README.md

@@ -0,0 +1,109 @@
+# ADK AML Investigator Example
+
+This example demonstrates how to use the **Google Agent Development Kit (ADK)** with **Apache Beam** to build an AI-powered Anti-Money Laundering (AML) investigator.
+
+The pipeline monitors a **Cloud Spanner Change Stream** for new transactions and uses a **Gemini-powered Agent** to perform graph-based analysis (using Spanner Graph/GQL) to detect suspicious patterns.
+
+## Features
+
+- **Spanner Change Stream Integration**: Automatically triggers analysis on new `INSERT` events.
+- **ADK LlmAgent**: Uses Gemini 2.5 Flash to reason about transaction risks.
+- **Spanner Graph (GQL) Tools**: The agent is equipped with three GQL tools:
+    1. `detectCircularFlow`: Finds round-tripping loops (e.g., A -> B -> C -> A).
+    2. `detectFanInStructuring`: Finds smurfing patterns where multiple accounts funnel money to one collector.
+    3. `detectSharedIdentity`: Detects synthetic accounts sharing the same physical device.
+- **OpenTelemetry Tracing**: End-to-end observability of agent reasoning and tool execution.
+
+## Prerequisites
+
+- A Google Cloud Project.
+- `gcloud` CLI installed and authenticated.
+- Java 17 or higher.
+- Gradle.
+
+## Setup
+
+### 1. Spanner Database Setup
+
+Run the provided script to create a Spanner instance, database, schema, and seed data:
+
+```bash
+./examples/java/adk/setup-spanner.sh [INSTANCE_ID] [DATABASE_ID]
+```
+
+Defaults are `aml-instance` and `aml-db`.
+
+### 2. Run the Beam Pipeline
+
+#### Option A: Direct Runner (Local)
+Execute the pipeline locally:
+
+```bash
+JAVA_HOME=/Library/Java/JavaVirtualMachines/temurin-17.jdk/Contents/Home ./gradlew :examples:java:adk:execute --args=" \
+    --spannerInstance=aml-instance \
+    --spannerDatabase=aml-db \
+    --changeStreamName=TransactionsStream \
+    --outputTable=Transactions \
+    --project=radoslaws-playground-pso"
+```
+
+#### Option B: Dataflow Runner v1 (Legacy Worker)
+To run on Google Cloud Dataflow using the v1 runner (Legacy Worker), first build the worker JAR:
+
+```bash
+JAVA_HOME=/Library/Java/JavaVirtualMachines/temurin-17.jdk/Contents/Home ./gradlew :runners:google-cloud-dataflow-java:worker:shadowJar
+```
+
+Then, execute the pipeline:
+
+```bash
+JAVA_HOME=/Library/Java/JavaVirtualMachines/temurin-17.jdk/Contents/Home ./gradlew :examples:java:adk:execute --args=" \
+    --project=radoslaws-playground-pso \
+    --region=us-central1 \
+    --runner=DataflowRunner \
+    --spannerInstance=aml-instance \
+    --spannerDatabase=aml-db \
+    --changeStreamName=TransactionsStream \
+    --outputTable=Transactions \
+    --gcpTempLocation=gs://radoslaws-playground-pso/temp \
+    --tempLocation=gs://radoslaws-playground-pso/ \
+    --streaming=true \
+    --maxNumWorkers=1 \
+    --enableStreamingEngine \
+    --dataflowWorkerJar=runners/google-cloud-dataflow-java/worker/build/libs/beam-runners-google-cloud-dataflow-java-legacy-worker-2.76.0-SNAPSHOT.jar \
+    --experiments=disable_runner_v2,enable_otel_defaults \
+    --dataflowServiceOptions=enable_google_cloud_profiler,enable_google_cloud_heap_sampling"
+```
+
+## Demo Scenarios
+
+Once the pipeline is running, you can trigger AML scenarios by inserting "TRIGGER" transactions.
+
+### Scenario 1: Circular Flow (Round-Tripping)
+Trigger a loop where money goes Alice -> Bob -> Charlie -> Alice within 48 hours.
+
+```bash
+gcloud spanner databases execute-sql aml-db --instance=aml-instance --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_loop_03_TRIGGER', 'usr_charlie', 'usr_alice', 9500.00, 'PENDING', CURRENT_TIMESTAMP())"
+```
+
+### Scenario 2: Fan-In Structuring (Smurfing)
+Trigger a pattern where a third "mule" sends money to a collector who already received two sub-threshold transfers.
+
+```bash
+gcloud spanner databases execute-sql aml-db --instance=aml-instance --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_fan_03_TRIGGER', 'usr_bob', 'usr_collector', 9100.00, 'PENDING', CURRENT_TIMESTAMP())"
+```
+
+### Scenario 3: Shared Device (Co-location)
+Trigger a transfer between two accounts that are physically located on the same hardware device.
+
+```bash
+gcloud spanner databases execute-sql aml-db --instance=aml-instance --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_colocate_TRIGGER', 'usr_fraudA', 'usr_fraudB', 4500.00, 'PENDING', CURRENT_TIMESTAMP())"
+```
+
+## Observing Results
+
+Check the `Transactions` table in Spanner to see the `RiskReason` generated by the AI Agent:
+
+```bash
+gcloud spanner databases execute-sql aml-db --instance=aml-instance --sql="SELECT TransactionId, RiskReason FROM Transactions WHERE TransactionId LIKE '%_TRIGGER'"
+```

examples/java/adk/build.gradle

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+plugins {
+    id 'java'
+    id 'org.apache.beam.module'
+}
+
+applyJavaNature(
+        exportJavadoc: false,
+        automaticModuleName: 'org.apache.beam.examples.adk',
+)
+provideIntegrationTestingDependencies()
+enableJavaPerformanceTesting()
+
+description = "Apache Beam :: Examples :: Java :: ADK"
+ext.summary = "Apache Beam SDK examples for ADK."
+
+dependencies {
+    implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
+    implementation project(":sdks:java:extensions:google-cloud-platform-core")
+    implementation project(path: ":sdks:java:core", configuration: "shadow")
+    implementation project(path: ":sdks:java:io:google-cloud-platform")
+    implementation project(path: ":runners:google-cloud-dataflow-java")
+    testImplementation library.java.google_cloud_pubsub
+    //runtimeOnly project(path: ":runners:direct-java", configuration: "shadow")
+    runtimeOnly library.java.opentelemetry_exporter_otlp
+    runtimeOnly library.java.opentelemetry_extension_autoconfigure
+    runtimeOnly project(":sdks:java:extensions:opentelemetry-gcp-auth-extension")
+    // Google Cloud Spanner
+    implementation library.java.google_cloud_spanner
+
+    // OpenTelemetry
+    implementation library.java.opentelemetry_api
+    implementation library.java.opentelemetry_sdk
+
+    // ADK Dependencies
+    implementation "com.google.adk:google-adk:1.6.0"
+   // implementation "com.google.genai:google-genai:0.1.1"
+
+    implementation library.java.joda_time
+    implementation library.java.slf4j_api
+    implementation library.java.slf4j_simple
+}
+
+task execute (type:JavaExec) {
+    classpath = sourceSets.main.runtimeClasspath
+    mainClass = "org.apache.beam.examples.adk.AmlPipeline"
+}

examples/java/adk/revert-demo.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+# Configuration
+INSTANCE_ID=${1:-"aml-instance"}
+DATABASE_ID=${2:-"aml-db"}
+PROJECT_ID=$(gcloud config get-value project)
+
+echo "Using Project: $PROJECT_ID"
+echo "Using Instance: $INSTANCE_ID"
+echo "Using Database: $DATABASE_ID"
+
+echo "Reverting demo transactions..."
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="DELETE FROM Transactions WHERE TransactionId IN ('tx_loop_03_TRIGGER', 'tx_fan_03_TRIGGER', 'tx_colocate_TRIGGER');"
+
+echo "Demo reverted successfully. You can now re-trigger the scenarios."

examples/java/adk/setup-spanner.sh

@@ -0,0 +1,59 @@
+#!/bin/bash
+
+# Configuration
+INSTANCE_ID=${1:-"aml-instance"}
+DATABASE_ID=${2:-"aml-db"}
+PROJECT_ID=$(gcloud config get-value project)
+
+echo "Using Project: $PROJECT_ID"
+echo "Using Instance: $INSTANCE_ID"
+echo "Using Database: $DATABASE_ID"
+
+# 1. Create Instance (if not exists)
+if ! gcloud spanner instances describe "$INSTANCE_ID" > /dev/null 2>&1; then
+    echo "Creating Spanner instance $INSTANCE_ID..."
+    gcloud spanner instances create "$INSTANCE_ID" \
+        --config=regional-us-central1 \
+        --description="AML Demo Instance" \
+        --nodes=1 \
+        --edition=ENTERPRISE
+fi
+
+# 2. Create Database (if not exists)
+if ! gcloud spanner databases describe "$DATABASE_ID" --instance="$INSTANCE_ID" > /dev/null 2>&1; then
+    echo "Creating Spanner database $DATABASE_ID..."
+    gcloud spanner databases create "$DATABASE_ID" --instance="$INSTANCE_ID"
+fi
+
+# 3. Apply Schema
+echo "Applying DDL schema..."
+gcloud spanner databases ddl update "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --ddl-file="examples/java/adk/spanner-schema.sql"
+
+# 4. Seed Data
+echo "Seeding initial data..."
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO Account (AccountId, AccountHolder, Status, CreatedAt) VALUES ('usr_alice', 'Alice Smith', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_bob', 'Bob Jones', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_charlie', 'Charlie Brown', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_mule1', 'Mule One', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_mule2', 'Mule Two', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_collector', 'Collector Hub', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_fraudA', 'Fraud User A', 'CLEARED', CURRENT_TIMESTAMP()), ('usr_fraudB', 'Fraud User B', 'CLEARED', CURRENT_TIMESTAMP());"
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO SharedDevice (DeviceId, DeviceModel, FirstSeen) VALUES ('dev_hardware_xyz99', 'iPhone 15 Pro', CURRENT_TIMESTAMP());"
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_loop_01', 'usr_alice', 'usr_bob', 10000.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 10 HOUR)), ('tx_loop_02', 'usr_bob', 'usr_charlie', 9800.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 HOUR));"
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_fan_01', 'usr_mule1', 'usr_collector', 9200.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 DAY)), ('tx_fan_02', 'usr_mule2', 'usr_collector', 9400.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY));"
+
+gcloud spanner databases execute-sql "$DATABASE_ID" --instance="$INSTANCE_ID" \
+    --sql="INSERT INTO AccountDevice (AccountId, DeviceId, LinkedAt) VALUES ('usr_fraudA', 'dev_hardware_xyz99', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)), ('usr_fraudB', 'dev_hardware_xyz99', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY));"
+
+echo "Spanner setup complete."
+echo ""
+echo "To trigger Scenario 1 (Circular Flow), run:"
+echo "gcloud spanner databases execute-sql $DATABASE_ID --instance=$INSTANCE_ID --sql=\"INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_loop_03_TRIGGER', 'usr_charlie', 'usr_alice', 9500.00, 'PENDING', CURRENT_TIMESTAMP())\""
+echo ""
+echo "To trigger Scenario 2 (Fan-In), run:"
+echo "gcloud spanner databases execute-sql $DATABASE_ID --instance=$INSTANCE_ID --sql=\"INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_fan_03_TRIGGER', 'usr_bob', 'usr_collector', 9100.00, 'PENDING', CURRENT_TIMESTAMP())\""
+echo ""
+echo "To trigger Scenario 3 (Co-location), run:"
+echo "gcloud spanner databases execute-sql $DATABASE_ID --instance=$INSTANCE_ID --sql=\"INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES ('tx_colocate_TRIGGER', 'usr_fraudA', 'usr_fraudB', 4500.00, 'PENDING', CURRENT_TIMESTAMP())\""

examples/java/adk/spanner-schema.sql

@@ -0,0 +1,68 @@
+-- =============================================================================
+-- 1. BASE RELATIONAL TABLES
+-- =============================================================================
+
+-- Accounts Node Table
+CREATE TABLE Account (
+    AccountId STRING(64) NOT NULL,
+    AccountHolder STRING(256) NOT NULL,
+    Status STRING(32) NOT NULL, -- 'PENDING', 'CLEARED', 'REVIEW_REQUIRED'
+    CreatedAt TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp = true)
+) PRIMARY KEY (AccountId);
+
+-- Shared Attributes Node Tables (For Synthetic/Co-location Checks)
+CREATE TABLE SharedDevice (
+    DeviceId STRING(128) NOT NULL,
+    DeviceModel STRING(128),
+    FirstSeen TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp = true)
+) PRIMARY KEY (DeviceId);
+
+-- Edge Table: Linking Accounts to Devices
+CREATE TABLE AccountDevice (
+    AccountId STRING(64) NOT NULL,
+    DeviceId STRING(128) NOT NULL,
+    LinkedAt TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp = true),
+    FOREIGN KEY (AccountId) REFERENCES Account (AccountId),
+    FOREIGN KEY (DeviceId) REFERENCES SharedDevice (DeviceId)
+) PRIMARY KEY (AccountId, DeviceId);
+
+-- Edge Table: Financial Transactions (Edges between Accounts)
+CREATE TABLE Transactions (
+    TransactionId STRING(64) NOT NULL,
+    SenderId STRING(64) NOT NULL,
+    ReceiverId STRING(64) NOT NULL,
+    Amount NUMERIC NOT NULL,
+    Status STRING(32) NOT NULL, -- 'PENDING', 'CLEARED', 'REVIEW_REQUIRED'
+    RiskReason STRING(MAX),
+    Timestamp TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp = true),
+    ReviewedAt TIMESTAMP,
+    FOREIGN KEY (SenderId) REFERENCES Account (AccountId),
+    FOREIGN KEY (ReceiverId) REFERENCES Account (AccountId)
+) PRIMARY KEY (TransactionId);
+
+-- Change Stream for Transactions
+CREATE CHANGE STREAM TransactionsStream FOR Transactions
+  OPTIONS (
+    exclude_update = true,
+    exclude_delete = true
+  );
+
+-- =============================================================================
+-- 2. SPANNER PROPERTY GRAPH DEFINITION
+-- =============================================================================
+
+CREATE PROPERTY GRAPH FinancialGraph
+  NODE TABLES (
+    Account,
+    SharedDevice
+  )
+  EDGE TABLES (
+    Transactions
+      SOURCE KEY (SenderId) REFERENCES Account (AccountId)
+      DESTINATION KEY (ReceiverId) REFERENCES Account (AccountId)
+      LABEL TRANSFERRED_TO,
+    AccountDevice
+      SOURCE KEY (AccountId) REFERENCES Account (AccountId)
+      DESTINATION KEY (DeviceId) REFERENCES SharedDevice (DeviceId)
+      LABEL USED_DEVICE
+  );

examples/java/adk/spanner-seed.sql

@@ -0,0 +1,29 @@
+-- Seed Accounts
+INSERT INTO Account (AccountId, AccountHolder, Status, CreatedAt) VALUES
+  ('usr_alice', 'Alice Smith', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_bob', 'Bob Jones', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_charlie', 'Charlie Brown', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_mule1', 'Mule One', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_mule2', 'Mule Two', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_collector', 'Collector Hub', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_fraudA', 'Fraud User A', 'CLEARED', CURRENT_TIMESTAMP()),
+  ('usr_fraudB', 'Fraud User B', 'CLEARED', CURRENT_TIMESTAMP());
+
+-- Seed Shared Device
+INSERT INTO SharedDevice (DeviceId, DeviceModel, FirstSeen) VALUES
+  ('dev_hardware_xyz99', 'iPhone 15 Pro', CURRENT_TIMESTAMP());
+
+-- Scenario 1: Circular Flow Setup
+INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES
+  ('tx_loop_01', 'usr_alice', 'usr_bob', 10000.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 10 HOUR)),
+  ('tx_loop_02', 'usr_bob', 'usr_charlie', 9800.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 HOUR));
+
+-- Scenario 2: Fan-In Structuring Setup
+INSERT INTO Transactions (TransactionId, SenderId, ReceiverId, Amount, Status, Timestamp) VALUES
+  ('tx_fan_01', 'usr_mule1', 'usr_collector', 9200.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 2 DAY)),
+  ('tx_fan_02', 'usr_mule2', 'usr_collector', 9400.00, 'CLEARED', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY));
+
+-- Scenario 3: Shared Device Setup
+INSERT INTO AccountDevice (AccountId, DeviceId, LinkedAt) VALUES
+  ('usr_fraudA', 'dev_hardware_xyz99', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)),
+  ('usr_fraudB', 'dev_hardware_xyz99', TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY));