Google Vertex AI
Overview
Prerequisites
pip install google-cloud-aiplatformIntegration Code
from google.cloud import aiplatform
import requests
from datetime import datetime, timezone
ATHENA_API_KEY = "sk_live_xxxxx"
ATHENA_API_URL = "https://api.athenatrust.ai/v1"
def send_vertex_attributions_to_athena(
endpoint_id: str,
project: str,
location: str,
model_id: str,
protected_attribute: str
):
"""
Fetch Vertex AI attributions and send to ATHENA.
"""
aiplatform.init(project=project, location=location)
endpoint = aiplatform.Endpoint(endpoint_id)
# Get model monitoring results
monitoring_stats = endpoint.list_model_monitoring_jobs()
results = []
for job in monitoring_stats:
if job.state.name == "JOB_STATE_SUCCEEDED":
# Extract attribution data
stats = job.stats
# Check for protected attribute drift
if protected_attribute in stats.feature_stats:
attr_stats = stats.feature_stats[protected_attribute]
payload = {
"externalToolId": "google_vertex",
"modelId": model_id,
"metricName": "feature_attribution",
"metricValue": normalize_attribution(attr_stats.attribution_score),
"protectedAttribute": protected_attribute,
"rawPayload": {
"attribution_score": attr_stats.attribution_score,
"drift_score": attr_stats.drift_score,
"job_id": job.name
},
"signalTimestamp": datetime.now(timezone.utc).isoformat()
}
response = requests.post(
f"{ATHENA_API_URL}/model-fairness-signals",
headers={
"Authorization": f"Bearer {ATHENA_API_KEY}",
"Content-Type": "application/json"
},
json=payload
)
results.append({
"status": "success" if response.status_code == 201 else "failed",
"signalId": response.json().get("signalId")
})
return results
def normalize_attribution(score: float) -> float:
"""Normalize attribution score to 0 to 1."""
return min(1.0, max(0.0, abs(score)))When to Use
Custom Metrics
Next Steps
Last updated