Classify and Route Events with Vertex AI

Put a Gemini model from Vertex AI inline in your event pipeline as an automated guardrail. Every Pub/Sub message starts one Kestra execution, a Gemini model on Vertex AI classifies the payload as high or low priority, and an If branch routes the high-priority ones to a BigQuery audit table and a Discord channel. The classification decision is code, not a human pause, so routine event routing happens in milliseconds with a full execution record per message.

This solves a common gap in event-driven LLM pipelines: you want the language model in the decision loop, but you also want deterministic routing, retries on transient failures, and a queryable history of every classification. Wiring that together with a Pub/Sub push subscriber, a Cloud Function, and ad-hoc BigQuery inserts is brittle. Kestra gives you one declarative YAML flow with the trigger, the model call, the conditional routing, and the error handler in scope.

How it works

1. The io.kestra.plugin.gcp.pubsub.RealtimeTrigger (on_event) subscribes to the events topic via the kestra-events-sub subscription and starts one execution per message, deserializing the payload as JSON.
2. io.kestra.plugin.gcp.vertexai.ChatCompletion (classify) sends the event body to gemini-1.5-flash in us-central1 with a tight prompt that forces a single-word label, high or low.
3. io.kestra.plugin.core.flow.If (guardrail) evaluates outputs.classify.predictions[0] | lower == 'high' and only enters the then-branch when the model returns high priority.
4. io.kestra.plugin.gcp.bigquery.Query (record) inserts the message id, priority, and timestamp into the analytics.priority_events audit table.
5. io.kestra.plugin.discord.DiscordIncomingWebhook (escalate) posts a structured alert to the configured Discord channel with the message id and the execution id.
6. A flow-level errors block pages the same Discord webhook if anything in the pipeline fails, so silent drops are not possible.

What you get

• Sub-second LLM classification on every Pub/Sub event with no human in the loop.
• A queryable BigQuery audit trail of every high-priority event.
• Real-time Discord alerts for escalations and for flow failures.
• One YAML file you can version, review, and roll back.

Who it's for

• Platform and data engineers wiring Vertex AI into event-driven workflows.
• SRE and on-call teams that want LLM-assisted triage of alert streams.
• Trust and safety, abuse, and fraud teams routing user events by model verdict.
• Product analytics teams logging classified events to BigQuery for downstream analysis.

Why orchestrate this with Kestra

Vertex AI is a model API, not a workflow engine. Pub/Sub push delivery can invoke a Cloud Function, but you still need to glue together the classification, the conditional logic, the BigQuery write, the Discord alert, and the failure path, and you still have no central place to see what ran. Kestra wraps all of it in a declarative YAML flow with built-in retries via pluginDefaults, an errors block for alerting, lineage and outputs preserved for every execution, and a UI to inspect each event run. The RealtimeTrigger keeps the event-per-execution model that Pub/Sub natively supports, so you do not lose the streaming feel.

Prerequisites

• A GCP project with Pub/Sub, Vertex AI, and BigQuery APIs enabled.
• A Pub/Sub topic named events and a subscription named kestra-events-sub.
• A BigQuery dataset analytics with a priority_events table (message_id STRING, priority STRING, logged_at TIMESTAMP).
• A Discord channel with an incoming webhook configured.
• Gemini (gemini-1.5-flash) enabled in the us-central1 region.

Secrets

• GCP_PROJECT_ID: the GCP project hosting Pub/Sub, Vertex AI, and BigQuery.
• GCP_SERVICE_ACCOUNT: service account key JSON with Pub/Sub subscriber, Vertex AI user, and BigQuery data editor roles.
• DISCORD_WEBHOOK_URL: Discord incoming webhook URL for escalations and failure alerts.

Quick start

1. Add the three secrets to your Kestra namespace.
2. Point the RealtimeTrigger topic and subscription at your Pub/Sub resources.
3. Adjust the prompt and the If condition to match your label vocabulary.
4. Remove disabled: true from the trigger and publish a test message to the topic.
5. Confirm the execution in the Kestra UI, the row in BigQuery, and the Discord post.

How to extend

• Add more labels (urgent, spam, sensitive) and expand the If into a Switch task for per-label routing.
• Swap gemini-1.5-flash for gemini-1.5-pro when accuracy matters more than latency.
• Replace the BigQuery insert with a streaming write to a different table, or fan out to Kafka or Cloud Storage.
• Add a second classification pass for low-priority events that look suspicious, using a stricter prompt.
• Wire the escalation step to PagerDuty, Slack, or ServiceNow instead of (or alongside) Discord.

Links

• Kestra GCP plugin
• Vertex AI orchestration with Kestra